fix: updated provenance support to use pypi prov package

Author: satra

nipype/pipeline/tests/test_utils.py

@@ -335,10 +335,11 @@ def test_provenance():
     out_dir = mkdtemp()
     metawf = pe.Workflow(name='meta')
     metawf.base_dir = out_dir
-    metawf.add_nodes([create_wf('wf%d' % i) for i in range(30)])
+    metawf.add_nodes([create_wf('wf%d' % i) for i in range(1)])
     eg = metawf.run(plugin='Linear')
     prov_base = os.path.join(out_dir,
                              'workflow_provenance_test')
     psg = write_workflow_prov(eg, prov_base, format='all')
-    yield assert_equal, psg, 1
+    yield assert_equal, len(psg.bundles), 2
+    yield assert_equal, len(psg.get_records()), 7
     rmtree(out_dir)

nipype/pipeline/utils.py

@@ -1168,14 +1168,27 @@ def write_workflow_prov(graph, filename=None, format='all'):
                         values = getattr(result.outputs, key)
                         if isdefined(values) and idx < len(values):
                             subresult.outputs[key] = values[idx]
-                sub_bundle = ProvStore().add_results(subresult)
-                ps.g = merge_bundles(ps.g, sub_bundle)
-                ps.g.wasGeneratedBy(sub_bundle, process)
+                sub_doc = ProvStore().add_results(subresult)
+                sub_bundle = pm.ProvBundle(sub_doc.get_records(),
+                                           identifier=get_id())
+                ps.g.add_bundle(sub_bundle)
+                bundle_entity = ps.g.entity(sub_bundle.identifier,
+                                            other_attributes={'prov:type':
+                                                               pm.PROV_BUNDLE})
+                ps.g.wasGeneratedBy(bundle_entity, process)
         else:
             process.add_attributes({pm.PROV["type"]: nipype_ns["Node"]})
-            result_bundle = ProvStore().add_results(result).bundle()
-            ps.g = merge_bundles(ps.g, result_bundle)
-            ps.g.wasGeneratedBy(result_bundle, process)
+            if result.provenance:
+                prov_doc = result.provenance
+            else:
+                prov_doc = ProvStore().add_results(result)
+            result_bundle = pm.ProvBundle(prov_doc.get_records(),
+                                          identifier=get_id())
+            ps.g.add_bundle(result_bundle)
+            bundle_entity = ps.g.entity(result_bundle.identifier,
+                                        other_attributes={'prov:type':
+                                                              pm.PROV_BUNDLE})
+            ps.g.wasGeneratedBy(bundle_entity, process)
         processes.append(process)
 
     # add dependencies (edges)

nipype/utils/provenance.py

@@ -2,6 +2,7 @@
 standard_library.install_aliases()
 from builtins import object
 
+from copy import deepcopy
 from pickle import dumps
 import simplejson
 import os
@@ -244,18 +245,15 @@ def write_provenance(results, filename='provenance', format='turtle'):
 class ProvStore(object):
 
     def __init__(self):
-        self.g = pm.ProvDocument() #ProvBundle(identifier=get_id())
+        self.g = pm.ProvDocument()
         self.g.add_namespace(foaf)
         self.g.add_namespace(dcterms)
         self.g.add_namespace(nipype_ns)
         self.g.add_namespace(niiri)
 
-    def add_results(self, results):
-        if results.provenance:
-            try:
-                self.g.add_bundle(results.provenance)
-            except pm.ProvException:
-                self.g.add_bundle(results.provenance, get_id())
+    def add_results(self, results, keep_provenance=False):
+        if keep_provenance and results.provenance:
+            self.g = deepcopy(results.provenance)
             return self.g
         runtime = results.runtime
         interface = results.interface