Fixed some errors

Author: pintohutch

nipype/pipeline/plugins/callback_log.py

@@ -3,54 +3,63 @@
 """Callback logger for recording workflow and node run stats
 """
 
-# Import packages
-import datetime
-import logging
 
 # Log node stats function
 def log_nodes_cb(node, status):
     """Function to record node run statistics to a log file as json
     dictionaries
+
+    Parameters
+    ----------
+    node : nipype.pipeline.engine.Node
+        the node being logged
+    status : string
+        acceptable values are 'start', 'end'; otherwise it is
+        considered and error
+
+    Returns
+    -------
+    None
+        this function does not return any values, it logs the node
+        status info to the callback logger
     """
 
-    # Init variables
-    logger = logging.getLogger('callback')
+    # Import packages
+    import datetime
+    import logging
+    import json
 
     # Check runtime profile stats
     if node.result is not None:
         try:
             runtime = node.result.runtime
             runtime_memory_gb = runtime.runtime_memory_gb
             runtime_threads = runtime.runtime_threads
-        except:
-            runtime_memory_gb = runtime_threads = 'Unkown'
+        except AttributeError:
+            runtime_memory_gb = runtime_threads = 'Unknown'
     else:
         runtime_memory_gb = runtime_threads = 'N/A'
 
+    # Init variables
+    logger = logging.getLogger('callback')
+    status_dict = {'name' : node.name,
+                   'id' : node._id,
+                   'estimated_memory_gb' : node._interface.estimated_memory_gb,
+                   'num_threads' : node._interface.num_threads}
+
     # Check status and write to log
     # Start
     if status == 'start':
-        message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' +\
-        node._id + '"' + ',"start":' + '"' +str(datetime.datetime.now()) +\
-        '"' + ',"estimated_memory_gb":' + str(node._interface.estimated_memory_gb) + \
-        ',"num_threads":' + str(node._interface.num_threads) + '}'
-
-        logger.debug(message)
+        status_dict['start'] = str(datetime.datetime.now())
     # End
     elif status == 'end':
-        message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' + \
-        node._id + '"' + ',"finish":' + '"' + str(datetime.datetime.now()) +  \
-        '"' + ',"estimated_memory_gb":' + '"'+ str(node._interface.estimated_memory_gb) + \
-        '"'+ ',"num_threads":' + '"'+ str(node._interface.num_threads) + '"'+ \
-        ',"runtime_threads":' + '"'+ str(runtime_threads) + '"'+ \
-        ',"runtime_memory_gb":' + '"'+ str(runtime_memory_gb) + '"' + '}'
-
-        logger.debug(message)
+        status_dict['finish'] = str(datetime.datetime.now())
+        status_dict['runtime_threads'] = runtime_threads
+        status_dict['runtime_memory_gb'] = runtime_memory_gb
     # Other
     else:
-        message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' + \
-        node._id + '"' + ',"finish":' + '"' + str(datetime.datetime.now()) +\
-        '"' + ',"estimated_memory_gb":' + str(node._interface.estimated_memory_gb) + \
-        ',"num_threads":' + str(node._interface.num_threads) + ',"error":"True"}'
+        status_dict['finish'] = str(datetime.datetime.now())
+        status_dict['error'] = True
 
-        logger.debug(message)
+    # Dump string to log
+    logger.debug(json.dumps(status_dict))

nipype/pipeline/plugins/multiproc.py

@@ -124,8 +124,8 @@ class MultiProcPlugin(DistributedPluginBase):
     Currently supported options are:
 
     - non_daemon : boolean flag to execute as non-daemon processes
-    - num_threads: maximum number of threads to be executed in parallel
-    - estimated_memory_gb: maximum memory (in GB) that can be used at once.
+    - n_procs: maximum number of threads to be executed in parallel
+    - memory_gb: maximum memory (in GB) that can be used at once.
 
     """
 
@@ -180,11 +180,8 @@ def _clear_task(self, taskid):
 
     def _submit_job(self, node, updatehash=False):
         self._taskid += 1
-        try:
-            if node.inputs.terminal_output == 'stream':
-                node.inputs.terminal_output = 'allatonce'
-        except:
-            pass
+        if node.inputs.terminal_output == 'stream':
+            node.inputs.terminal_output = 'allatonce'
 
         self._taskresult[self._taskid] = \
             self.pool.apply_async(run_node,
@@ -282,8 +279,10 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
                                  % self.procs[jobid])
                     try:
                         self.procs[jobid].run()
-                    except Exception:
-                        self._clean_queue(jobid, graph)
+                    except:
+                        etype, eval, etr = sys.exc_info()
+                        formatted_exc = format_exception(etype, eval, etr)
+                        logger.debug('Traceback:\n%s' % '\n'.join(formatted_exc))
                     self._task_finished_cb(jobid)
                     self._remove_node_dirs()
 

nipype/utils/draw_gantt_chart.py

@@ -43,22 +43,10 @@ def create_event_dict(start_time, nodes_list):
     events = {}
     for node in nodes_list:
         # Format node fields
-        try:
-            estimated_threads = float(node['num_threads'])
-        except:
-            estimated_threads = 1
-        try:
-            estimated_memory_gb = float(node['estimated_memory_gb'])
-        except:
-            estimated_memory_gb = 1.0
-        try:
-            runtime_threads = float(node['runtime_threads'])
-        except:
-            runtime_threads = 0
-        try:
-            runtime_memory_gb = float(node['runtime_memory_gb'])
-        except:
-            runtime_memory_gb = 0.0
+        estimated_threads = int(node.get('num_threads'), 1)
+        estimated_memory_gb = float(node.get('estimated_memory_gb', 1.0))
+        runtime_threads = int(node.get('runtime_threads'), 0)
+        runtime_memory_gb = float(node.get('runtime_memory_gb', 0.0))
 
         # Init and format event-based nodes 
         node['estimated_threads'] =  estimated_threads
@@ -119,7 +107,7 @@ def log_to_dict(logfile):
             node = None
             try:
                 node = json.loads(l)
-            except Exception:
+            except ValueError:
                 pass
 
             if not node: