Introduce StepRetroProv

Author: svenvanderburg

fairworkflows/fairstep.py

@@ -13,6 +13,7 @@
 from fairworkflows import namespaces, LinguisticSystem, LINGSYS_ENGLISH, LINGSYS_PYTHON
 from fairworkflows.config import DUMMY_FAIRWORKFLOWS_URI, IS_FAIRSTEP_RETURN_VALUE_PARAMETER_NAME, \
     LOGGER
+from fairworkflows.prov import prov_logger, StepRetroProv
 from fairworkflows.rdf_wrapper import RdfWrapper, replace_in_rdf
 
 
@@ -441,7 +442,7 @@ def _modify_function(func):
         def _add_logging(func):
             @functools.wraps(func)
             def _wrapper(*func_args, **func_kwargs):
-                LOGGER.info(f'Running step: {func.__name__}')
+                prov_logger.add(StepRetroProv(step=fairstep))
                 return func(*func_args, **func_kwargs)
             return _wrapper
         func._fairstep = fairstep

fairworkflows/fairworkflow.py

@@ -18,7 +18,7 @@
 from fairworkflows import namespaces, LinguisticSystem, LINGSYS_PYTHON
 from fairworkflows.config import LOGGER
 from fairworkflows.fairstep import FairStep
-from fairworkflows.prov import WorkflowRetroProv
+from fairworkflows.prov import WorkflowRetroProv, prov_logger
 from fairworkflows.rdf_wrapper import RdfWrapper
 
 
@@ -363,38 +363,31 @@ def execute(self, *args, **kwargs):
         Returns a tuple (result, retroprov), where result is the final output of the executed
         workflow and retroprov is the retrospective provenance logged during execution.
         """
-
         if not hasattr(self, 'workflow_level_promise'):
             raise ValueError('Cannot execute workflow as no noodles step_level_promise has been constructed.')
-
-        log = io.StringIO()
-        log_handler = logging.StreamHandler(log)
-        formatter = logging.Formatter('%(asctime)s - %(message)s')
-        log_handler.setFormatter(formatter)
-
-        LOGGER.setLevel(logging.INFO)
-        LOGGER.handlers = [log_handler]
+        prov_logger.empty()
         self.workflow_level_promise = noodles.workflow.from_call(
             noodles.get_workflow(self.workflow_level_promise).root_node.foo, args, kwargs, {})
         result = noodles.run_single(self.workflow_level_promise)
 
         # Generate the retrospective provenance as a (nano-) Publication object
-        retroprov = self._generate_retrospective_prov_publication(log.getvalue())
+        retroprov = self._generate_retrospective_prov_publication()
 
         return result, retroprov
 
-    def _generate_retrospective_prov_publication(self, log: str) -> WorkflowRetroProv:
+    def _generate_retrospective_prov_publication(self) -> WorkflowRetroProv:
         """
         Utility method for generating a Publication object for the retrospective
         provenance of this workflow. Uses the given 'log' string as the actual
         provenance for now.
         """
         if self._is_published:
-            this_workflow = rdflib.URIRef(self.uri)
+            workflow_uri = rdflib.URIRef(self.uri)
         else:
-            this_workflow = rdflib.URIRef('http://www.example.org/unpublishedworkflow')
+            workflow_uri = rdflib.URIRef('http://www.example.org/unpublishedworkflow')
 
-        return WorkflowRetroProv(this_workflow, log)
+        step_provs = prov_logger.get_all()
+        return WorkflowRetroProv(self, workflow_uri, step_provs)
 
     def draw(self, filepath):
         """Visualize workflow.

fairworkflows/prov.py

@@ -1,47 +1,95 @@
+import threading
+from datetime import datetime
+from typing import List, Iterator
+
 import rdflib
 
 from fairworkflows import namespaces
 from fairworkflows.rdf_wrapper import RdfWrapper
 
 
+class ProvLogger:
+    def __init__(self):
+        self.lock = threading.Lock()
+        self.items = []
+
+    def add(self, item):
+        with self.lock:
+            self.items.append(item)
+
+    def get_all(self):
+        with self.lock:
+            items, self.items = self.items, []
+        return items
+
+    def empty(self):
+        self.items = []
+
+
+prov_logger = ProvLogger()
+
+
 class RetroProv(RdfWrapper):
-    def __init__(self, prov_was_derived_from, log_message: str):
+    def __init__(self):
         super().__init__(uri=None, ref_name='retroprov')
-        self.set_attribute(rdflib.RDF.type, namespaces.PROV.Activity)
-        self.set_attribute(namespaces.PROV.wasDerivedFrom, rdflib.URIRef(prov_was_derived_from))
-        self.rdf.add((self.self_ref, rdflib.RDFS.label, rdflib.Literal(log_message)))
+        self.timestamp = datetime.now()
+
+
+class StepRetroProv(RetroProv):
+    def __init__(self, step):
+        super().__init__()
+        self.set_attribute(rdflib.RDF.type, namespaces.PPLAN.Activity)
+        self.step = step
+        self.step_uri = step.uri
 
     @property
-    def prov_was_derived_from(self):
-        """Refers to URI of object that this provenance was derived from.
+    def step_uri(self):
+        """Refers to URI of step associated to this provenance.
 
         Matches the predicate prov:wasDerivedFrom associated to this retrospective provenance
-        object
         """
         return self.get_attribute(namespaces.PROV.wasDerivedFrom)
 
-    @prov_was_derived_from.setter
-    def prov_was_derived_from(self, value):
-        self.set_attribute(namespaces.PROV.wasDerivedFrom, rdflib.URIRef(value), overwrite=True)
-
-
-class WorkflowRetroProv(RetroProv):
-    def __init__(self, prov_was_derived_from, log_message):
-        super().__init__(prov_was_derived_from, log_message)
+    @step_uri.setter
+    def step_uri(self, value):
+        self.set_attribute(namespaces.PPLAN.correspondsToStep, rdflib.URIRef(value), overwrite=True)
 
     def __str__(self):
         """String representation."""
-        s = f'Workflow retrospective provenance.\n'
+        s = f'Step retrospective provenance.\n'
         s += self._rdf.serialize(format='trig').decode('utf-8')
         return s
 
 
-class StepRetroProv(RetroProv):
-    def __init__(self, prov_was_derived_from, log_message):
-        super().__init__(prov_was_derived_from, log_message)
+class WorkflowRetroProv(RetroProv):
+    def __init__(self, workflow, workflow_uri, step_provs: List[StepRetroProv]):
+        super().__init__()
+        self.set_attribute(rdflib.RDF.type, namespaces.PPLAN.Bundle)
+        self.workflow = workflow
+        self.workflow_uri = workflow_uri
+        self._step_provs = step_provs
+
+    @property
+    def workflow_uri(self):
+        """Refers to URI of step associated to this provenance.
+
+        Matches the predicate prov:wasDerivedFrom associated to this retrospective provenance
+        """
+        return self.get_attribute(namespaces.PROV.wasDerivedFrom)
+
+    @workflow_uri.setter
+    def workflow_uri(self, value):
+        self.set_attribute(namespaces.PROV.wasDerivedFrom, rdflib.URIRef(value), overwrite=True)
+
+    def __iter__(self) -> Iterator[StepRetroProv]:
+        """Iterate over StepRetroProv that were part of the execution of the workflow."""
+        yield from self._step_provs
+
+    def __len__(self) -> int:
+        return len(self._step_provs)
 
     def __str__(self):
         """String representation."""
-        s = f'Step retrospective provenance.\n'
+        s = f'Workflow retrospective provenance.\n'
         s += self._rdf.serialize(format='trig').decode('utf-8')
         return s

tests/test_fairworkflow.py

@@ -9,7 +9,7 @@
 
 from conftest import skip_if_nanopub_server_unavailable, read_rdf_test_resource
 from fairworkflows import FairWorkflow, FairStep, namespaces, FairVariable, is_fairstep, is_fairworkflow
-from fairworkflows.prov import WorkflowRetroProv
+from fairworkflows.prov import WorkflowRetroProv, StepRetroProv
 from fairworkflows.rdf_wrapper import replace_in_rdf
 from nanopub import Publication
 
@@ -353,8 +353,10 @@ def my_workflow(in1, in2, in3):
         assert result == -66
 
         assert isinstance(prov, WorkflowRetroProv)
-        prov_log = str(list(prov.rdf.objects(prov.self_ref, rdflib.RDFS.label))[0])
-        assert 'Running step: add' in prov_log
+        assert len(prov) == 4
+        for step_prov in prov:
+            assert isinstance(step_prov, StepRetroProv)
+            assert step_prov.step in fw._steps.values()
 
     def test_workflow_complex_serialization(self):
         class OtherType: