Added runtime_profile to run by default unless the necessary packages arent available

Author: pintohutch

nipype/interfaces/base.py

@@ -1032,6 +1032,7 @@ def run(self, **inputs):
         self._check_mandatory_inputs()
         self._check_version_requirements(self.inputs)
         interface = self.__class__
+
         # initialize provenance tracking
         env = deepcopy(dict(os.environ))
         runtime = Bunch(cwd=os.getcwd(),
@@ -1211,7 +1212,6 @@ def _get_num_threads(proc):
 
     # Import packages
     import psutil
-    import logging as lg
 
     # Init variables
     num_threads = proc.num_threads()
@@ -1255,13 +1255,19 @@ def run_command(runtime, output=None, timeout=0.01, redirect_x=False):
     The returned runtime contains a merged stdout+stderr log with timestamps
     """
 
-    # Import packages
+    # Init logger
+    logger = logging.getLogger('workflow')
+
+    # Default to profiling the runtime
     try:
         import memory_profiler
         import psutil
-        mem_prof = True
-    except:
-        mem_prof = False
+        runtime_profile = True
+    except ImportError as exc:
+        logger.info('Unable to import packages needed for runtime '\
+                    'profiling. Turning off runtime profiler.\n'\
+                    'Error: %s' % exc)
+        runtime_profile = False
 
     # Init variables
     PIPE = subprocess.PIPE
@@ -1317,7 +1323,7 @@ def _process(drain=0):
                 for stream in res[0]:
                     stream.read(drain)
         while proc.returncode is None:
-            if mem_prof:
+            if runtime_profile:
                 mem_mb, num_threads = \
                     _get_max_resources_used(proc, mem_mb, num_threads)
             proc.poll()
@@ -1335,7 +1341,7 @@ def _process(drain=0):
         result['merged'] = [r[1] for r in temp]
 
     if output == 'allatonce':
-        if mem_prof:
+        if runtime_profile:
             while proc.returncode is None:
                 mem_mb, num_threads = \
                     _get_max_resources_used(proc, mem_mb, num_threads, poll=True)
@@ -1355,7 +1361,7 @@ def _process(drain=0):
         result['stderr'] = str(stderr).split('\n')
         result['merged'] = ''
     if output == 'file':
-        if mem_prof:
+        if runtime_profile:
             while proc.returncode is None:
                 mem_mb, num_threads = \
                     _get_max_resources_used(proc, mem_mb, num_threads, poll=True)
@@ -1366,7 +1372,7 @@ def _process(drain=0):
         result['stderr'] = [line.strip() for line in open(errfile).readlines()]
         result['merged'] = ''
     if output == 'none':
-        if mem_prof:
+        if runtime_profile:
             while proc.returncode is None:
                 mem_mb, num_threads = \
                     _get_max_resources_used(proc, mem_mb, num_threads, poll=True)

nipype/interfaces/tests/test_runtime_profiler.py

@@ -137,7 +137,6 @@ def _run_workflow(self):
         # Run workflow
         plugin_args = {'n_procs' : num_procs,
                        'memory' : num_gb,
-                       'runtime_profile' : True,
                        'status_callback' : log_nodes_cb}
         wf.run(plugin='ResourceMultiProc', plugin_args=plugin_args)
 

nipype/pipeline/plugins/multiproc.py

@@ -23,7 +23,7 @@
 logger = logging.getLogger('workflow')
 
 # Run node
-def run_node(node, updatehash, runtime_profile=False):
+def run_node(node, updatehash):
     """docstring
     """
 
@@ -33,29 +33,22 @@ def run_node(node, updatehash, runtime_profile=False):
     # Init variables
     result = dict(result=None, traceback=None)
 
-    # If we're profiling the run
-    if runtime_profile:
-        try:
-            start = datetime.datetime.now()
-            retval = node.run(updatehash=updatehash)
-            run_secs = (datetime.datetime.now() - start).total_seconds()
-            result['result'] = retval
-            result['runtime_seconds'] = run_secs
-            if hasattr(retval.runtime, 'get'):
-                result['runtime_memory'] = retval.runtime.get('runtime_memory')
-                result['runtime_threads'] = retval.runtime.get('runtime_threads')
-        except:
-            etype, eval, etr = sys.exc_info()
-            result['traceback'] = format_exception(etype,eval,etr)
-            result['result'] = node.result
-    # Otherwise, execute node.run as normal
-    else:
-        try:
-            result['result'] = node.run(updatehash=updatehash)
-        except:
-            etype, eval, etr = sys.exc_info()
-            result['traceback'] = format_exception(etype,eval,etr)
-            result['result'] = node.result
+    # 
+    try:
+        start = datetime.datetime.now()
+        retval = node.run(updatehash=updatehash)
+        run_secs = (datetime.datetime.now() - start).total_seconds()
+        result['result'] = retval
+        result['runtime_seconds'] = run_secs
+        if hasattr(retval.runtime, 'get'):
+            result['runtime_memory'] = retval.runtime.get('runtime_memory')
+            result['runtime_threads'] = retval.runtime.get('runtime_threads')
+    except:
+        etype, eval, etr = sys.exc_info()
+        result['traceback'] = format_exception(etype,eval,etr)
+        result['result'] = node.result
+
+    # Return the result dictionary
     return result
 
 
@@ -160,13 +153,10 @@ def _submit_job(self, node, updatehash=False):
                 node.inputs.terminal_output = 'allatonce'
         except:
             pass
-        try:
-            runtime_profile = self.plugin_args['runtime_profile']
-        except:
-            runtime_profile = False
+
         self._taskresult[self._taskid] = \
             self.pool.apply_async(run_node,
-                                  (node, updatehash, runtime_profile),
+                                  (node, updatehash),
                                   callback=release_lock)
         return self._taskid