collect resource_monitor info after run

Author: oesteban

nipype/interfaces/base.py

@@ -1135,10 +1135,17 @@ def run(self, **inputs):
                 # Read .prof file in and set runtime values
                 vals = np.loadtxt(mon_fname, delimiter=',')
                 if vals.tolist():
-                    _, mem_peak_gb, nthreads = np.atleast_2d(vals).max(0).astype(float).tolist()
-                    runtime.mem_peak_gb = mem_peak_gb / 1024
+                    vals = np.atleast_2d(vals)
+                    _, mem_peak_mb, nthreads = vals.max(0).astype(float).tolist()
+                    runtime.mem_peak_gb = mem_peak_mb / 1024
                     runtime.nthreads_max = int(nthreads)
 
+                    runtime.prof_dict = {
+                        'time': vals[:, 0].tolist(),
+                        'mem_gb': (vals[:, 1] / 1024).tolist(),
+                        'cpus': vals[:, 2].astype(int).tolist(),
+                    }
+
         return results
 
     def _list_outputs(self):

nipype/pipeline/engine/utils.py

@@ -1296,6 +1296,51 @@ def write_workflow_prov(graph, filename=None, format='all'):
     return ps.g
 
 
+def write_workflow_resources(graph, filename=None):
+    import simplejson as json
+    if not filename:
+        filename = os.path.join(os.getcwd(), 'resource_monitor.json')
+
+    big_dict = {
+        'time': [],
+        'name': [],
+        'interface': [],
+        'mem_gb': [],
+        'cpus': [],
+        'mapnode': [],
+        'params': [],
+    }
+
+    for idx, node in enumerate(graph.nodes()):
+        nodename = node.fullname
+        classname = node._interface.__class__.__name__
+
+        params = ''
+        if node.parameterization:
+            params = '_'.join(['{}'.format(p)
+                              for p in node.parameterization])
+
+        rt_list = node.result.runtime
+        if not isinstance(rt_list, list):
+            rt_list = [rt_list]
+
+        for subidx, runtime in enumerate(rt_list):
+            nsamples = len(runtime.prof_dict['time'])
+
+            for key in ['time', 'mem_gb', 'cpus']:
+                big_dict[key] += runtime.prof_dict[key]
+
+            big_dict['interface'] += [classname] * nsamples
+            big_dict['name'] += [nodename] * nsamples
+            big_dict['mapnode'] += [subidx] * nsamples
+            big_dict['params'] += [params] * nsamples
+
+    with open(filename, 'w') as rsf:
+        json.dump(big_dict, rsf)
+
+    return filename
+
+
 def topological_sort(graph, depth_first=False):
     """Returns a depth first sorted order if depth_first is True
     """

nipype/pipeline/engine/workflows.py

@@ -51,6 +51,7 @@
                                 write_rst_list, to_str)
 from .utils import (generate_expanded_graph, modify_paths,
                     export_graph, make_output_dir, write_workflow_prov,
+                    write_workflow_resources,
                     clean_working_directory, format_dot, topological_sort,
                     get_print_name, merge_dict, evaluate_connect_function,
                     _write_inputs, format_node)
@@ -593,6 +594,9 @@ def run(self, plugin=None, plugin_args=None, updatehash=False):
                                 'workflow_provenance_%s' % datestr)
             logger.info('Provenance file prefix: %s' % prov_base)
             write_workflow_prov(execgraph, prov_base, format='all')
+
+        if str2bool(self.config['execution'].get('resource_monitor', 'false')):
+            write_workflow_resources(execgraph)
         return execgraph
 
     # PRIVATE API AND FUNCTIONS