Merge pull request #186 from fair-workflows/issue185_stepretroprov_workflowretroprov

raar1 · web-flow · commit cd880441673a · 2021-03-05T16:53:55.000+01:00
Issue185 stepretroprov workflowretroprov
diff --git a/CITATION.cff b/CITATION.cff
@@ -27,4 +27,4 @@ keywords:
 license: "Apache-2.0"
 message: "If you use this software, please cite it using these metadata."
 title: "fairworkflows: A python library for handling semantically described scientific workflows"
-version: "0.2.4"
+version: "0.2.5"
diff --git a/examples/noodles_fw.ipynb b/examples/noodles_fw.ipynb
diff --git a/fairworkflows/_version.py b/fairworkflows/_version.py
@@ -1 +1 @@
-__version__ = "0.2.4"
+__version__ = "0.2.5"
diff --git a/fairworkflows/fairstep.py b/fairworkflows/fairstep.py
@@ -5,6 +5,7 @@
 from copy import deepcopy
 from typing import Callable, get_type_hints, List, Union
 from urllib.parse import urldefrag
+from datetime import datetime
 
 import noodles
 import rdflib
@@ -38,13 +39,23 @@ class FairVariable:
                             become mapped to e.g. XSD types.
         semantic_types: One or more URIs that describe the semantic type(s) of this FairVariable.
     """
-    def __init__(self, name: str = None, computational_type: str = None, semantic_types = None, uri: str = None):
+    def __init__(self, name: str = None, computational_type: str = None, semantic_types = None, uri: str = None, stepuri: str = None):
         if uri and name is None:
             # Get the name from the uri (i.e. 'input1' from http://example.org#input1)
             _, name = urldefrag(uri)
+
+        if name is None and uri is None:
+            raise ValueError('Both name and uri cannot both be None when constructing a FairVariable.')
+
         self.name = name
         self.computational_type = computational_type
 
+        if uri:
+            self._uri = rdflib.URIRef(uri)
+        else:
+            step_base_uri, _ = urldefrag(stepuri)
+            self._uri = rdflib.Namespace(step_base_uri)['#' + name]
+
         if semantic_types is None:
             self.semantic_types = []
         else:
@@ -53,14 +64,20 @@ def __init__(self, name: str = None, computational_type: str = None, semantic_ty
             else:
                 self.semantic_types = [rdflib.URIRef(t) for t in semantic_types]
 
+    @property
+    def uri(self):
+        if self._uri:
+            return self._uri
+
     def __eq__(self, other):
         return self.name == other.name and self.computational_type == other.computational_type
 
     def __hash__(self):
         return hash(str(self))
 
     def __str__(self):
-        return f'FairVariable {self.name} of computational type: {self.computational_type} and semantic types: {self.semantic_types}'
+        return (f'FairVariable {self.name} of computational type: {self.computational_type},'
+                f'semantic types: {self.semantic_types}.\nHas URI {self.uri}.\n')
 
 
 class FairStep(RdfWrapper):
@@ -237,6 +254,10 @@ def is_script_task(self, value: bool):
 
     def _get_variable(self, var_ref: Union[rdflib.term.BNode, rdflib.URIRef]) -> FairVariable:
         """Retrieve a specific FairVariable from the RDF triples."""
+
+        if var_ref is None:
+            raise ValueError('Variable reference var_ref cannot be None.')
+
         rdfs_comment_objs = list(self._rdf.objects(var_ref, RDFS.comment))
         computational_type = str(rdfs_comment_objs[0])
 
@@ -245,9 +266,9 @@ def _get_variable(self, var_ref: Union[rdflib.term.BNode, rdflib.URIRef]) -> Fai
         sem_types = [sem_type for sem_type in sem_type_objs if sem_type != namespaces.PPLAN.Variable]
 
         if isinstance(var_ref, rdflib.term.BNode):
-            return FairVariable(name=str(var_ref), computational_type=computational_type, semantic_types=sem_types)
+            return FairVariable(name=str(var_ref), computational_type=computational_type, semantic_types=sem_types, stepuri=self._uri)
         else:
-            return FairVariable(uri=str(var_ref), computational_type=computational_type, semantic_types=sem_types)
+            return FairVariable(uri=str(var_ref), computational_type=computational_type, semantic_types=sem_types, stepuri=self._uri)
 
     def _add_variable(self, variable: FairVariable, relation_to_step):
         """Add triples describing FairVariable to rdf."""
@@ -360,12 +381,15 @@ def publish_as_nanopub(self, use_test_server=False, **kwargs):
                            newvalue=rdflib.URIRef(self.uri))
 
             # Similarly replace old URIs for variable name bindings
+            # in both this step and any workflow objects that use it.
             published_step_uri_defrag, _ = urldefrag(self.uri)
             for var_name in var_names:
                 old_var_uri = old_uri + '#' + var_name
                 new_var_uri = published_step_uri_defrag + '#' + var_name
                 replace_in_rdf(self.rdf, oldvalue=rdflib.URIRef(old_var_uri),
                                newvalue=rdflib.URIRef(new_var_uri))
+                replace_in_rdf(workflow.rdf, oldvalue=rdflib.URIRef(old_var_uri),
+                               newvalue=rdflib.URIRef(new_var_uri))
             del workflow._steps[old_uri]
             workflow._steps[self.uri] = self
 
@@ -442,8 +466,21 @@ def _modify_function(func):
         def _add_logging(func):
             @functools.wraps(func)
             def _wrapper(*func_args, **func_kwargs):
-                prov_logger.add(StepRetroProv(step=fairstep))
-                return func(*func_args, **func_kwargs)
+
+                # Get the arg label/value pairs as a dict (for both args and kwargs)
+                func_args_dict = dict(zip(inspect.getfullargspec(func).args, func_args))
+                all_args = {**func_args_dict, **func_kwargs}
+
+                # Execute step (with timing)
+                t0 = datetime.now()
+                execution_result = func(*func_args, **func_kwargs)
+                t1 = datetime.now()
+
+                # Log step execution
+                prov_logger.add(StepRetroProv(step=fairstep, step_args=all_args, output=execution_result, time_start=t0, time_end=t1))
+
+                return execution_result
+
             return _wrapper
         func._fairstep = fairstep
         return noodles.schedule(_add_logging(func))
diff --git a/fairworkflows/prov.py b/fairworkflows/prov.py
@@ -1,6 +1,6 @@
 import threading
 from datetime import datetime
-from typing import List, Iterator
+from typing import List, Iterator, Dict
 
 import rdflib
 
@@ -29,46 +29,102 @@ def empty(self):
 prov_logger = ProvLogger()
 
 
-class RetroProv(RdfWrapper):
-    def __init__(self):
-        super().__init__(uri=None, ref_name='retroprov')
-        self.timestamp = datetime.now()
-
-
-class StepRetroProv(RetroProv):
-    def __init__(self, step):
-        super().__init__()
+class StepRetroProv(RdfWrapper):
+    def __init__(self, step=None, step_args:Dict = None, time_start:datetime = None, time_end:datetime = None, output=None):
+        super().__init__(uri=None, ref_name='fairstepprov')
         self.set_attribute(rdflib.RDF.type, namespaces.PPLAN.Activity)
         self.step = step
         self.step_uri = step.uri
 
+        stepbase = rdflib.Namespace(step.uri)
+
+        # Bind inputs
+        for inputvar in step.inputs:
+            if inputvar.name in step_args:
+                retrovar = rdflib.BNode(inputvar.name)
+                self._rdf.add( (self.self_ref, namespaces.PROV.used, retrovar) )
+                self._rdf.add( (retrovar, rdflib.RDF.type, namespaces.PPLAN.Entity) )
+                self._rdf.add( (retrovar, rdflib.RDFS.label, rdflib.Literal(inputvar.name)) )
+                self._rdf.add( (retrovar, rdflib.RDF.value, rdflib.Literal(step_args[inputvar.name])) )
+
+                if inputvar.uri:
+                    self._rdf.add( (retrovar, namespaces.PPLAN.correspondsToVariable, inputvar.uri) )
+
+        # Bind outputs
+        num_outputs = len(list(step.outputs))
+        if num_outputs == 1:
+            outvardict = {'out1': output}
+        else:
+            outvardict = {('out' + str(i)): outval for i, outval in enumerate(output) }
+
+        for outputvar in step.outputs:
+            retrovar = rdflib.BNode(outputvar.name)
+            if outputvar.name in outvardict:
+                self._rdf.add( (self.self_ref, namespaces.PROV.used, retrovar) )
+                self._rdf.add( (retrovar, rdflib.RDF.type, namespaces.PPLAN.Entity) )
+                self._rdf.add( (retrovar, rdflib.RDFS.label, rdflib.Literal(outputvar.name)) )
+                self._rdf.add( (retrovar, rdflib.RDF.value, rdflib.Literal(outvardict[outputvar.name])) )
+
+                if outputvar.uri:
+                    self._rdf.add( (retrovar, namespaces.PPLAN.correspondsToVariable, outputvar.uri) )
+
+        # Add times to RDF (if available)
+        if time_start:
+            self.set_attribute(namespaces.PROV.startedAtTime, rdflib.Literal(time_start, datatype=rdflib.XSD.dateTime))
+        if time_end:
+            self.set_attribute(namespaces.PROV.endedAtTime, rdflib.Literal(time_end, datatype=rdflib.XSD.dateTime))
+
     @property
     def step_uri(self):
         """Refers to URI of step associated to this provenance.
 
-        Matches the predicate prov:wasDerivedFrom associated to this retrospective provenance
+        Matches the predicate pplan:correspondsToStep associated to this retrospective provenance
         """
-        return self.get_attribute(namespaces.PROV.wasDerivedFrom)
+        return self.get_attribute(namespaces.PPLAN.correspondsToStep)
 
     @step_uri.setter
     def step_uri(self, value):
         self.set_attribute(namespaces.PPLAN.correspondsToStep, rdflib.URIRef(value), overwrite=True)
 
+    def publish_as_nanopub(self, use_test_server=False, **kwargs):
+        """
+        Publish this rdf as a nanopublication.
+
+        Args:
+            use_test_server (bool): Toggle using the test nanopub server.
+            kwargs: Keyword arguments to be passed to [nanopub.Publication.from_assertion](
+                https://nanopub.readthedocs.io/en/latest/reference/publication.html#
+                nanopub.publication.Publication.from_assertion).
+                This allows for more control over the nanopublication RDF.
+
+        Returns:
+            a dictionary with publication info, including 'nanopub_uri', and 'concept_uri'
+        """
+        return self._publish_as_nanopub(use_test_server=use_test_server, **kwargs)
+
     def __str__(self):
         """String representation."""
         s = f'Step retrospective provenance.\n'
-        s += self._rdf.serialize(format='trig').decode('utf-8')
+        s += self._rdf.serialize(format='turtle').decode('utf-8')
         return s
 
 
-class WorkflowRetroProv(RetroProv):
+class WorkflowRetroProv(RdfWrapper):
     def __init__(self, workflow, workflow_uri, step_provs: List[StepRetroProv]):
-        super().__init__()
-        self.set_attribute(rdflib.RDF.type, namespaces.PPLAN.Bundle)
+        super().__init__(uri=None, ref_name='fairworkflowprov')
+        self._rdf.add((self.self_ref, rdflib.RDF.type, namespaces.PPLAN.Bundle))
+        self._rdf.add((self.self_ref, rdflib.RDF.type, namespaces.PROV.Collection))
         self.workflow = workflow
         self.workflow_uri = workflow_uri
         self._step_provs = step_provs
 
+        # Add the Entity links for now (dummy links, if unpublished)
+        for stepprov in self._step_provs:
+            if stepprov.uri:
+                self._rdf.add((self.self_ref, namespaces.PROV.hasMember, rdflib.URIRef(stepprov.uri)))
+            else:
+                self._rdf.add((self.self_ref, namespaces.PROV.hasMember, rdflib.URIRef('http://www.example.org/unpublished-entity-' + str(hash(stepprov)))))
+
     @property
     def workflow_uri(self):
         """Refers to URI of step associated to this provenance.
@@ -88,8 +144,32 @@ def __iter__(self) -> Iterator[StepRetroProv]:
     def __len__(self) -> int:
         return len(self._step_provs)
 
+    def publish_as_nanopub(self, use_test_server=False, **kwargs):
+        """
+        Publish this rdf as a nanopublication.
+
+        Args:
+            use_test_server (bool): Toggle using the test nanopub server.
+            kwargs: Keyword arguments to be passed to [nanopub.Publication.from_assertion](
+                https://nanopub.readthedocs.io/en/latest/reference/publication.html#
+                nanopub.publication.Publication.from_assertion).
+                This allows for more control over the nanopublication RDF.
+
+        Returns:
+            a dictionary with publication info, including 'nanopub_uri', and 'concept_uri'
+        """
+
+        # Clear existing members of this entity (to be replaced with newly published links)
+        self.remove_attribute(namespaces.PROV.hasMember)
+
+        for stepprov in self._step_provs:
+            stepprov.publish_as_nanopub(use_test_server=use_test_server, **kwargs)
+            self._rdf.add((self.self_ref, namespaces.PROV.hasMember, rdflib.URIRef(stepprov.uri)))
+
+        return self._publish_as_nanopub(use_test_server=use_test_server, **kwargs)
+
     def __str__(self):
         """String representation."""
         s = f'Workflow retrospective provenance.\n'
-        s += self._rdf.serialize(format='trig').decode('utf-8')
+        s += self._rdf.serialize(format='turtle').decode('utf-8')
         return s
diff --git a/fairworkflows/rdf_wrapper.py b/fairworkflows/rdf_wrapper.py
@@ -19,7 +19,10 @@ class RdfWrapper:
     def __init__(self, uri, ref_name='fairobject', derived_from: List[str] = None,
                  language: LinguisticSystem = None ):
         self._rdf = rdflib.Graph()
-        self._uri = str(uri)
+        if uri:
+            self._uri = str(uri)
+        else:
+            self._uri = None
         self.self_ref = rdflib.term.BNode(ref_name)
         self._is_modified = False
         self._is_published = False
@@ -30,7 +33,6 @@ def __init__(self, uri, ref_name='fairobject', derived_from: List[str] = None,
         # A blank node to which triples about the linguistic
         # system for this FAIR object can be added
         self.lingsys_ref = rdflib.BNode('LinguisticSystem')
-        self._rdf.add((self.self_ref, DCTERMS.language, self.lingsys_ref))
 
         if language is not None:
             self.language = language
@@ -42,6 +44,7 @@ def _bind_namespaces(self):
         """
         self.rdf.bind("npx", namespaces.NPX)
         self.rdf.bind("pplan", namespaces.PPLAN)
+        self.rdf.bind("prov", namespaces.PROV)
         self.rdf.bind("dul", namespaces.DUL)
         self.rdf.bind("bpmn", namespaces.BPMN)
         self.rdf.bind("pwo", namespaces.PWO)
@@ -163,16 +166,23 @@ def language(self):
         """Returns the language for this fair objects's description (could be code).
            Returns a LinguisticSystem object.
         """
-        lingsys_rdf = rdflib.Graph()
-        for t in self._rdf.triples((self.lingsys_ref, None, None)):
-            lingsys_rdf.add(t)
-        return LinguisticSystem.from_rdf(lingsys_rdf)
+        if (None, DCTERMS.language, self.lingsys_ref) in self._rdf:
+            lingsys_rdf = rdflib.Graph()
+            for t in self._rdf.triples((self.lingsys_ref, None, None)):
+                lingsys_rdf.add(t)
+            return LinguisticSystem.from_rdf(lingsys_rdf)
+        else:
+            return None
 
     @language.setter
     def language(self, value: LinguisticSystem):
         """Sets the language for this fair object's code (takes a LinguisticSystem).
            Removes the existing linguistic system triples from the RDF decription
            and replaces them with the new linguistic system."""
+
+        if (None, DCTERMS.language, self.lingsys_ref) not in self._rdf:
+            self._rdf.add((self.self_ref, DCTERMS.language, self.lingsys_ref))
+
         lingsys_triples = list(self._rdf.triples( (self.lingsys_ref, None, None) ))
         if len(lingsys_triples) > 0:
             self._rdf.remove(lingsys_triples)
@@ -299,6 +309,7 @@ def _publish_as_nanopub(self, use_test_server=False, **kwargs):
                                              introduces_concept=self.self_ref,
                                              derived_from=self._derived_from,
                                              **kwargs)
+
         client = NanopubClient(use_test_server=use_test_server)
         publication_info = client.publish(nanopub)
 
diff --git a/requirements.txt b/requirements.txt
@@ -1,4 +1,4 @@
-nanopub==1.2.3
+nanopub==1.2.5
 networkx~=2.5
 pytest
 pyyaml
diff --git a/tests/test_fairstep.py b/tests/test_fairstep.py
@@ -246,8 +246,8 @@ def test_shacl_does_not_validate(self):
 
         assert len(step.rdf) == n_triples_before, 'shacl_validate mutated RDF'
 
-
-def test_is_fairstep_decorator():
+@patch('fairworkflows.rdf_wrapper.NanopubClient.publish')
+def test_is_fairstep_decorator(mock_publish):
     @is_fairstep(label='test_label')
     def add(a: int, b: int) -> int:
         """
@@ -257,6 +257,9 @@ def add(a: int, b: int) -> int:
 
     assert hasattr(add(1,2), '_fairstep')
 
+    add._fairstep.publish_as_nanopub()
+    assert mock_publish.call_count == 1
+
 def test_decorator_semantic_types():
     test_types_a = ['http://www.example.org/distance', 'http://www.example.org/number']
     test_type_output = 'http://www.example.org/walrus'
diff --git a/tests/test_fairworkflow.py b/tests/test_fairworkflow.py

Original file line number	Diff line number	Diff line change
`@@ -1 +1 @@`
`1`		`-__version__ = "0.2.4"`
	`1`	`+__version__ = "0.2.5"`
Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-nanopub==1.2.3`
	`1`	`+nanopub==1.2.5`
`2`	`2`	`networkx~=2.5`
`3`	`3`	`pytest`
`4`	`4`	`pyyaml`