maintenance: refactor code in provenance

psafont · psafont · commit 4e6103891c94 · 2018-11-15T18:24:52.000+01:00
Remove make_fs_access from ProvenanceProfile
Decouple adding start and stop times to profiles from the actual time it
happens
Decouple adding prospective provenance from the start of a process
diff --git a/README.rst b/README.rst
@@ -698,6 +698,6 @@ Workflow.make_workflow_step
   ::
 
     make_workflow_step(toolpath_object, pos, loadingContext, parentworkflowProv)
-      (Dict[Text, Any], int, LoadingContext, Optional[CreateProvProfile]) -> WorkflowStep
+      (Dict[Text, Any], int, LoadingContext, Optional[ProvenanceProfile]) -> WorkflowStep
 
   Create and return a workflow step object.
diff --git a/cwltool/builder.py b/cwltool/builder.py
@@ -28,7 +28,7 @@
 
 
 if TYPE_CHECKING:
-    from .provenance import CreateProvProfile  # pylint: disable=unused-import
+    from .provenance import ProvenanceProfile  # pylint: disable=unused-import
 CONTENT_LIMIT = 64 * 1024
 
 
@@ -162,7 +162,7 @@ def __init__(self,
         self.stagedir = stagedir
 
         self.pathmapper = None  # type: Optional[PathMapper]
-        self.prov_obj = None  # type: Optional[CreateProvProfile]
+        self.prov_obj = None  # type: Optional[ProvenanceProfile]
         self.find_default_container = None  # type: Optional[Callable[[], Text]]
 
     def build_job_script(self, commands):
diff --git a/cwltool/command_line_tool.py b/cwltool/command_line_tool.py
@@ -49,7 +49,7 @@
                     docker_windows_path_adjust, json_dumps, onWindows,
                     random_outdir, windows_default_container_id)
 if TYPE_CHECKING:
-    from .provenance import CreateProvProfile  # pylint: disable=unused-import
+    from .provenance import ProvenanceProfile  # pylint: disable=unused-import
 
 ACCEPTLIST_EN_STRICT_RE = re.compile(r"^[a-zA-Z0-9._+-]+$")
 ACCEPTLIST_EN_RELAXED_RE = re.compile(r".*")  # Accept anything
@@ -89,7 +89,7 @@ def __init__(self,
             self.outdir = outdir
             self.tmpdir = tmpdir
             self.script = script
-            self.prov_obj = None  # type: Optional[CreateProvProfile]
+            self.prov_obj = None  # type: Optional[ProvenanceProfile]
 
         def run(self, runtimeContext):  # type: (RuntimeContext) -> None
             try:
@@ -179,7 +179,7 @@ def __init__(self, job, output_callback, cachebuilder, jobcache):
         self.output_callback = output_callback
         self.cachebuilder = cachebuilder
         self.outdir = jobcache
-        self.prov_obj = None  # type: Optional[CreateProvProfile]
+        self.prov_obj = None  # type: Optional[ProvenanceProfile]
 
     def run(self, runtimeContext):
         # type: (RuntimeContext) -> None
diff --git a/cwltool/context.py b/cwltool/context.py
@@ -21,7 +21,7 @@
 if TYPE_CHECKING:
     from .process import Process
     from .provenance import (ResearchObject,  # pylint: disable=unused-import
-                             CreateProvProfile)
+                             ProvenanceProfile)
 
 class ContextBase(object):
     def __init__(self, kwargs=None):
@@ -63,7 +63,7 @@ def __init__(self, kwargs=None):
         self.cwl_full_name = ""            # type: str
         self.host_provenance = False       # type: bool
         self.user_provenance = False       # type: bool
-        self.prov_obj = None               # type: Optional[CreateProvProfile]
+        self.prov_obj = None               # type: Optional[ProvenanceProfile]
 
         super(LoadingContext, self).__init__(kwargs)
 
@@ -131,7 +131,7 @@ def __init__(self, kwargs=None):
         self.orcid = ''                 # type: str
         self.cwl_full_name = ""         # type: str
         self.process_run_id = None      # type: Optional[str]
-        self.prov_obj = None            # type: Optional[CreateProvProfile]
+        self.prov_obj = None            # type: Optional[ProvenanceProfile]
         super(RuntimeContext, self).__init__(kwargs)
 
 
diff --git a/cwltool/executors.py b/cwltool/executors.py
@@ -21,7 +21,7 @@
 from .mutation import MutationManager
 from .process import Process  # pylint: disable=unused-import
 from .process import cleanIntermediate, relocateOutputs
-from .provenance import CreateProvProfile
+from .provenance import ProvenanceProfile
 from .utils import DEFAULT_TMP_PREFIX
 from .workflow import Workflow, WorkflowJob, WorkflowJobStep
 
@@ -134,7 +134,7 @@ def run_jobs(self,
         # define provenance profile for single commandline tool
         if not isinstance(process, Workflow) \
                 and runtime_context.research_obj is not None:
-            process.provenance_object = CreateProvProfile(
+            process.provenance_object = ProvenanceProfile(
                 runtime_context.research_obj,
                 full_name=runtime_context.cwl_full_name,
                 host_provenance=False,
@@ -159,11 +159,12 @@ def run_jobs(self,
                         else:
                             runtime_context.prov_obj = job.prov_obj
                         assert runtime_context.prov_obj
-                        process_run_id = \
-                            runtime_context.prov_obj.evaluate(
-                                process, job, job_order_object,
-                                runtime_context.make_fs_access,
-                                runtime_context.research_obj)
+                        runtime_context.prov_obj.evaluate(
+                            process, job, job_order_object,
+                            runtime_context.research_obj)
+                        process_run_id =\
+                            runtime_context.prov_obj.record_process_start(
+                                process, job)
                         runtime_context = runtime_context.copy()
                         runtime_context.process_run_id = process_run_id
                     job.run(runtime_context)
diff --git a/cwltool/job.py b/cwltool/job.py
@@ -40,7 +40,7 @@
                     subprocess)
 
 if TYPE_CHECKING:
-    from .provenance import CreateProvProfile  # pylint: disable=unused-import
+    from .provenance import ProvenanceProfile  # pylint: disable=unused-import
 needs_shell_quoting_re = re.compile(r"""(^$|[\s|&;()<>\'"$@])""")
 
 FORCE_SHELLED_POPEN = os.getenv("CWLTOOL_FORCE_SHELL_POPEN", "0") == "1"
@@ -194,8 +194,8 @@ def __init__(self,
         self.generatefiles = {"class": "Directory", "listing": [], "basename": ""}  # type: Directory
         self.stagedir = None  # type: Optional[Text]
         self.inplace_update = False
-        self.prov_obj = None   # type: Optional[CreateProvProfile]
-        self.parent_wf = None  # type: Optional[CreateProvProfile]
+        self.prov_obj = None   # type: Optional[ProvenanceProfile]
+        self.parent_wf = None  # type: Optional[ProvenanceProfile]
         self.timelimit = None  # type: Optional[int]
         self.networkaccess = False  # type: bool
 
@@ -344,11 +344,8 @@ def _execute(self,
         if runtimeContext.research_obj is not None and self.prov_obj is not None and \
                 runtimeContext.process_run_id is not None:
             #creating entities for the outputs produced by each step (in the provenance document)
-            self.prov_obj.generate_output_prov(
-                outputs, runtimeContext.process_run_id, str(self.name))
-            self.prov_obj.document.wasEndedBy(
-                runtimeContext.process_run_id, None, self.prov_obj.workflow_run_uri,
-                datetime.datetime.now())
+            self.prov_obj.record_process_end(str(self.name), runtimeContext.process_run_id,
+                                      outputs, datetime.datetime.now())
         if processStatus != "success":
             _logger.warning(u"[job %s] completed %s", self.name, processStatus)
         else:
diff --git a/cwltool/main.py b/cwltool/main.py
@@ -583,7 +583,7 @@ def main(argsl=None,                   # type: List[str]
             if not args.compute_checksum:
                 _logger.error("--provenance incompatible with --no-compute-checksum")
                 return 1
-            ro = ResearchObject(
+            ro = ResearchObject(runtimeContext.make_fs_access,
                 temp_prefix_ro=args.tmpdir_prefix, orcid=args.orcid,
                 full_name=args.cwl_full_name)
             runtimeContext.research_obj = ro
@@ -808,7 +808,8 @@ def my_represent_none(self, data):  # pylint: disable=unused-argument
 
             if out is not None:
                 if runtimeContext.research_obj is not None:
-                    runtimeContext.research_obj.create_job(out, None, True)
+                    runtimeContext.research_obj.create_job(
+                        out, None, True)
 
                 def loc_to_path(obj):
                     for field in ("path", "nameext", "nameroot", "dirname"):
diff --git a/cwltool/process.py b/cwltool/process.py
@@ -52,7 +52,7 @@
 except ImportError:
     from scandir import scandir  # type: ignore
 if TYPE_CHECKING:
-    from .provenance import CreateProvProfile  # pylint: disable=unused-import
+    from .provenance import ProvenanceProfile  # pylint: disable=unused-import
 
 
 class LogAsDebugFilter(logging.Filter):
@@ -453,8 +453,8 @@ def __init__(self,
                  loadingContext        # type: LoadingContext
                 ):  # type: (...) -> None
         self.metadata = getdefault(loadingContext.metadata, {})  # type: Dict[Text,Any]
-        self.provenance_object = None  # type: Optional[CreateProvProfile]
-        self.parent_wf = None          # type: Optional[CreateProvProfile]
+        self.provenance_object = None  # type: Optional[ProvenanceProfile]
+        self.parent_wf = None          # type: Optional[ProvenanceProfile]
         global SCHEMA_FILE, SCHEMA_DIR, SCHEMA_ANY  # pylint: disable=global-statement
         if SCHEMA_FILE is None or SCHEMA_ANY is None or SCHEMA_DIR is None:
             get_schema("v1.0")
diff --git a/cwltool/provenance.py b/cwltool/provenance.py
@@ -301,7 +301,7 @@ def _valid_orcid(orcid):  # type: (Optional[Text]) -> Text
     return u"https://orcid.org/%s" % orcid_num
 
 
-class CreateProvProfile():
+class ProvenanceProfile():
     """
     Provenance profile.
 
@@ -338,7 +338,7 @@ def __init__(self,
         self.generate_prov_doc()
 
     def __str__(self):
-        return "CreateProvProfile <%s> in <%s>" % (
+        return "ProvenanceProfile <%s> in <%s>" % (
             self.workflow_run_uri, self.research_object)
 
     def generate_prov_doc(self):
@@ -448,41 +448,34 @@ def evaluate(self,
                  process,           # type: Process
                  job,               # type: Any
                  job_order_object,  # type: Dict[Text, Text]
-                 make_fs_access,    # type: Callable[[Text], StdFsAccess]
                  research_obj       # type: ResearchObject
-                ):  # type: (...) -> Optional[str]
-        """Evaluate the nature of job and initialize the activity start."""
-        process_run_id = None
-        research_obj.make_fs_access = make_fs_access
+                ):  # type: (...) -> None
+        """Evaluate the nature of job"""
         if not hasattr(process, "steps"):
-            # record provenance of an independent commandline tool execution
+            # record provenance of independent commandline tool executions
             self.prospective_prov(job)
             customised_job = copy_job_order(job, job_order_object)
             self.used_artefacts(customised_job, self.workflow_run_uri)
             research_obj.create_job(customised_job, job)
-            # self.used_artefacts(inputs, self.workflow_run_uri)
-            name = ""
-            if hasattr(job, "name"):
-                name = str(job.name)
-            process_name = urllib.parse.quote(name, safe=":/,#")
-            process_run_id = self.workflow_run_uri
         elif hasattr(job, "workflow"):
-            # record provenance for the workflow execution
+            # record provenance of workflow executions
             self.prospective_prov(job)
             customised_job = copy_job_order(job, job_order_object)
             self.used_artefacts(customised_job, self.workflow_run_uri)
-            # inputs = research_obj.create_job(customised_job)
-            # self.used_artefacts(inputs, self.workflow_run_uri)
-        else:  # in case of commandline tool execution as part of workflow
-            name = ""
-            if hasattr(job, "name"):
-                name = str(job.name)
+
+    def record_process_start(self, process, job, process_run_id=None):
+        # type: (Process, Any, str) -> Optional[str]
+        if not hasattr(process, 'steps'):
+            process_run_id = self.workflow_run_uri
+        elif not hasattr(job, 'workflow'):
+            # commandline tool execution as part of workflow
+            name = str(job.name) if hasattr(job, 'name') else ''
             process_name = urllib.parse.quote(name, safe=":/,#")
-            process_run_id = self.start_process(process_name)
+            process_run_id = self.start_process(process_name, datetime.datetime.now())
         return process_run_id
 
-    def start_process(self, process_name, process_run_id=None):
-        # type: (Any, str, str) -> str
+    def start_process(self, process_name, when, process_run_id=None):
+        # type: (Any, Text, datetime.datetime, Optional[str]) -> str
         """Record the start of each Process."""
         if process_run_id is None:
             process_run_id = uuid.uuid4().urn
@@ -495,11 +488,16 @@ def start_process(self, process_name, process_run_id=None):
             process_run_id, self.engine_uuid, str("wf:main/" + process_name))
         self.document.wasStartedBy(
             process_run_id, None, self.workflow_run_uri,
-            datetime.datetime.now(), None, None)
+            when, None, None)
         return process_run_id
 
+    def record_process_end(self, process_name, process_run_id, outputs, when):
+        # type: (Text, str, Any, datetime.datetime) -> None
+        self.generate_output_prov(outputs, process_run_id, process_name)
+        self.document.wasEndedBy(process_run_id, None, self.workflow_run_uri, when)
+
     def declare_file(self, value):
-        # type: (MutableMapping) -> Tuple[ProvEntity,ProvEntity,str]
+        # type: (MutableMapping) -> Tuple[ProvEntity, ProvEntity, str]
         if value["class"] != "File":
             raise ValueError("Must have class:File: %s" % value)
         # Need to determine file hash aka RO filename
@@ -856,11 +854,9 @@ def prospective_prov(self, job):
                         "prov:type": PROV["Plan"],
                         "prov:label":"Prospective provenance"})
 
-        steps = []
         for step in job.steps:
-            stepnametemp = "wf:main/"+str(step.name)[5:]
+            stepnametemp = "wf:main/" + str(step.name)[5:]
             stepname = urllib.parse.quote(stepnametemp, safe=":/,#")
-            steps.append(stepname)
             step = self.document.entity(
                 stepname, {provM.PROV_TYPE: WFDESC["Process"],
                            "prov:type": PROV["Plan"]})
@@ -947,9 +943,10 @@ def finalize_prov_profile(self, name):
 class ResearchObject():
     """CWLProv Research Object."""
 
-    def __init__(self, temp_prefix_ro="tmp", orcid='', full_name=''):
-        # type: (str, Text, Text) -> None
+    def __init__(self, make_fs_access, temp_prefix_ro="tmp", orcid='', full_name=''):
+        # type: (Callable[[Text], StdFsAccess], str, Text, Text) -> None
 
+        self.make_fs_access = make_fs_access
         self.temp_prefix = temp_prefix_ro
         self.orcid = '' if not orcid else _valid_orcid(orcid)
         self.full_name = full_name
@@ -969,8 +966,6 @@ def __init__(self, temp_prefix_ro="tmp", orcid='', full_name=''):
         self.base_uri = "arcp://uuid,%s/" % self.ro_uuid
         self.cwltool_version = "cwltool %s" % versionstring().split()[-1]
         ##
-        # This function will be added by create_job()
-        self.make_fs_access = None  # type: Optional[Callable[[Text], StdFsAccess]]
         self.relativised_input_object = {}  # type: Dict[Any, Any]
 
         self._initialize()
@@ -1499,8 +1494,8 @@ def _add_to_bagit(self, rel_path, **checksums):
         self.add_to_manifest(rel_path, checksums)
 
     def create_job(self,
-                   builder_job,  # type: Dict[Text, Any]
-                   wf_job=None,  # type: Callable[[Dict[Text, Text], Callable[[Any, Any], Any], RuntimeContext], Generator[Any, None, None]]
+                   builder_job,    # type: Dict[Text, Any]
+                   wf_job=None,    # type: Callable[[Dict[Text, Text], Callable[[Any, Any], Any], RuntimeContext], Generator[Any, None, None]]
                    is_output=False
                   ):  # type: (...) -> Dict
         #TODO customise the file
diff --git a/cwltool/workflow.py b/cwltool/workflow.py
diff --git a/tests/test_provenance.py b/tests/test_provenance.py
