fix: provenance tracking improvements with hash generation

Author: satra

nipype/pipeline/engine.py

@@ -689,10 +689,10 @@ def run(self, plugin=None, plugin_args=None, updatehash=False):
         runner.run(execgraph, updatehash=updatehash, config=self.config)
         datestr = datetime.utcnow().strftime('%Y%m%dT%H%M%S')
         if str2bool(self.config['execution']['write_provenance']):
-            write_workflow_prov(execgraph,
-                                os.path.join(self.base_dir,
-                                             'workflow_provenance_%s' % datestr),
-                                format='all')
+            prov_base = os.path.join(self.base_dir,
+                                     'workflow_provenance_%s' % datestr)
+            logger.info('Provenance file prefix: %s' % prov_base)
+            write_workflow_prov(execgraph, prov_base, format='all')
         return execgraph
 
     # PRIVATE API AND FUNCTIONS

nipype/pipeline/utils.py

@@ -1070,6 +1070,11 @@ def merge_dict(d1, d2, merge=lambda x, y: y):
     return result
 
 
+def merge_bundles(g1, g2):
+    for rec in g2.get_records():
+        g1._add_record(rec)
+    return g1
+
 def write_workflow_prov(graph, filename=None, format='turtle'):
     """Write W3C PROV Model JSON file
     """
@@ -1102,12 +1107,12 @@ def write_workflow_prov(graph, filename=None, format='turtle'):
                         if isdefined(values):
                             subresult.outputs[key] = values[idx]
                 sub_bundle = ProvStore().add_results(subresult)
-                ps.g.add_bundle(sub_bundle)
+                ps.g = merge_bundles(ps.g, sub_bundle)
                 ps.g.wasGeneratedBy(sub_bundle, process)
         else:
             process.add_extra_attributes({pm.PROV["type"]: nipype_ns["Node"]})
             result_bundle = ProvStore().add_results(result)
-            ps.g.add_bundle(result_bundle)
+            ps.g = merge_bundles(ps.g, result_bundle)
             ps.g.wasGeneratedBy(result_bundle, process)
         processes.append(process)
 

nipype/utils/provenance.py

@@ -31,7 +31,7 @@
 get_id = lambda: niiri[uuid1().hex]
 
 def get_attr_id(attr, skip=None):
-    dictwithhash, hashval = get_hashval(attr, skip=None)
+    dictwithhash, hashval = get_hashval(attr, skip=skip)
     return niiri[hashval]
 
 max_text_len = 1024000
@@ -59,10 +59,19 @@ def get_hashval(inputdict, skip=None):
 
     dict_withhash = {}
     dict_nofilename = OrderedDict()
-    for name, val in sorted(inputdict.items()):
-        if skip is not None and name in skip:
+    keys = {}
+    for key in inputdict:
+        if skip is not None and key in skip:
             continue
-        outname = name.get_uri()
+        keys[key.get_uri()] = key
+    for key in sorted(keys):
+        val = inputdict[keys[key]]
+        outname = key
+        try:
+            if isinstance(val, pm.URIRef):
+                val = val.decode()
+        except AttributeError:
+            pass
         if isinstance(val, pm.QName):
             val = val.get_uri()
         if isinstance(val, pm.Literal):
@@ -180,14 +189,19 @@ def prov_encode(graph, value, create_container=True):
                 entities = []
                 for item in value:
                     item_entity = prov_encode(graph, item)
-                    if 'file://' not in item_entity.get_value():
-                        raise ValueError('No file found')
                     entities.append(item_entity)
+                    if isinstance(item, list):
+                        continue
+                    if not isinstance(item_entity.get_value()[0], basestring):
+                        raise ValueError('Not a string literal')
+                    if 'file://' not in item_entity.get_value()[0]:
+                        raise ValueError('No file found')
                 id = get_id()
                 entity = graph.collection(identifier=id)
                 for item_entity in entities:
-                    graph.hadMember(id, item_entity.get_identifier())
-            except ValueError:
+                    graph.hadMember(id, item_entity)
+            except ValueError, e:
+                iflogger.debug(e)
                 entity = prov_encode(graph, value, create_container=False)
         else:
             entity = prov_encode(graph, value[0])