Merge pull request #743 from satra/fix/prov

Fix/prov

Author: satra

examples/fmri_spm.py

@@ -15,6 +15,9 @@
 
 Import necessary modules from nipype."""
 
+from nipype import config
+config.enable_provenance()
+
 from nipype import spm, fsl
 
 # In order to use this example with SPM's matlab common runtime
@@ -389,6 +392,4 @@ def getstripdir(subject_id):
 
 if __name__ == '__main__':
     l1pipeline.run('MultiProc')
-#    l2pipeline.run()
-#    l1pipeline.write_graph()
-
+    l2pipeline.run('MultiProc')

nipype/external/provcopy.py

@@ -14,6 +14,7 @@
 import datetime
 import json
 import re
+import dateutil.parser
 import collections
 from collections import defaultdict
 
@@ -179,31 +180,10 @@
 
 
 #  Datatypes
-_r_xsd_dateTime = re.compile(""" ^
-    (?P<year>-?[0-9]{4}) - (?P<month>[0-9]{2}) - (?P<day>[0-9]{2})
-    T (?P<hour>[0-9]{2}) : (?P<minute>[0-9]{2}) : (?P<second>[0-9]{2})
-    (?P<microsecond>\.[0-9]{1,6})?
-    (?P<tz>
-      Z | (?P<tz_hr>[-+][0-9]{2}) : (?P<tz_min>[0-9]{2})
-    )?
-    $ """, re.X)
-
 attr2rdf = lambda attr: PROV[PROV_ID_ATTRIBUTES_MAP[attr].split('prov:')[1]].rdf_representation()
 
 def _parse_xsd_dateTime(s):
-    """Returns datetime or None."""
-    m = _r_xsd_dateTime.match(s)
-    if m is not None:
-        values = m.groupdict()
-        if values["microsecond"] is None:
-            values["microsecond"] = 0
-        else:
-            values["microsecond"] = values["microsecond"][1:]
-            values["microsecond"] += "0" * (6 - len(values["microsecond"]))
-        values = dict((k, int(v)) for k, v in values.iteritems() if not k.startswith("tz"))
-        return datetime.datetime(**values)
-    else:
-        return None
+    return dateutil.parser.parse(s)
 
 
 def _ensure_datetime(time):
@@ -359,7 +339,7 @@ def provn_representation(self):
 
     def json_representation(self):
         return {'$': self._uri, 'type': u'xsd:anyURI'}
-
+    
     def rdf_representation(self):
         return URIRef(self.get_uri())
 
@@ -773,7 +753,7 @@ def rdf(self, graph=None, subj=None):
                     obj = RDFLiteral(value)
                 graph.add((subj, pred, obj))
         return graph
-
+        
     def is_asserted(self):
         return self._asserted
 
@@ -788,7 +768,7 @@ def is_relation(self):
 class ProvElement(ProvRecord):
     def is_element(self):
         return True
-
+    
     def rdf(self, graph=None):
         if graph is None:
             graph = Graph()
@@ -1666,15 +1646,15 @@ def rdf(self, graph=None):
             # graph should not None here
             uri = self.get_identifier().rdf_representation()
             graph = Graph(graph.store, uri)
-
+        
         for prefix, namespace in self._namespaces.items():
             graph.bind(prefix, namespace.get_uri())
-
+        
         for record in self._records:
             if record.is_asserted():
                 record.rdf(graph)
         return graph
-
+    
     def get_provjson(self, **kw):
         """Return the `PROV-JSON <http://www.w3.org/Submission/prov-json/>`_ representation for the bundle/document.
 

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])