Repository: fair-workflows/fairworkflows

Issue:
Inputs should not be bound to workflow instantiated with is_fairworkflow 
Now we do this, we already bind input variable values to the workflow when we instantiate it.
```python
@is_fairworkflow(label='My Workflow')
def my_workflow(in1, in2):
    """
    A simple addition, subtraction workflow
    """
    return add(in1, sub(in2, in2))
 fw = my_workflow(1, 2)
```
And we want it to be more like a 'template' workflow, where we pass input variable values to the execute method. We can probably achieve this by making the inputs noodles Promises.

Discussion:

Patch:
diff --git a/fairworkflows/fairstep.py b/fairworkflows/fairstep.py
index b42e21d..2d7fff0 100644
--- a/fairworkflows/fairstep.py
+++ b/fairworkflows/fairstep.py
@@ -169,14 +169,13 @@ class FairStep(RdfWrapper):
     @classmethod
     def from_function(cls, func: Callable):
         """
-        Generates a plex rdf decription for the given python function,
-        and makes this FairStep object a bpmn:ScriptTask.
+        Return a FairStep object for a function decorated with is_fairstep decorator
         """
         try:
             return func._fairstep
         except AttributeError:
             raise ValueError('The function was not marked as a fair step,'
-                             'use mark_as_fairstep decorator to mark it.')
+                             'use is_fairworkflow decorator to mark it.')
 
     @property
     def is_pplan_step(self):
diff --git a/fairworkflows/fairworkflow.py b/fairworkflows/fairworkflow.py
index f3c8dc7..540498c 100644
--- a/fairworkflows/fairworkflow.py
+++ b/fairworkflows/fairworkflow.py
@@ -1,11 +1,15 @@
 import inspect
+import io
+import logging
 import warnings
 from copy import deepcopy
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from typing import Iterator, Optional
+from typing import Iterator, Optional, Callable
 
+import nanopub
 import networkx as nx
+import noodles
 import rdflib
 from noodles.interface import PromisedObject
 from rdflib import RDF, RDFS, DCTERMS
@@ -15,7 +19,7 @@ from requests import HTTPError
 from fairworkflows import namespaces
 from fairworkflows.fairstep import FairStep
 from fairworkflows.rdf_wrapper import RdfWrapper
-import nanopub
+
 
 class FairWorkflow(RdfWrapper):
 
@@ -69,23 +73,35 @@ class FairWorkflow(RdfWrapper):
         self.anonymise_rdf()
         return self
 
+    @classmethod
+    def from_function(cls, func: Callable):
+        """
+        Return a FairWorkflow object for a function decorated with is_fairworkflow decorator
+        """
+        try:
+            return func._fairworkflow
+        except AttributeError:
+            raise ValueError('The function was not marked as a fair workflow,'
+                             'use is_fairworkflow decorator to mark it.')
+
     @classmethod
     def from_noodles_promise(cls, noodles_promise, description: str = None, label: str = None,
                  is_pplan_plan: bool = True, derived_from=None):
         self = cls(description=description, label=label, is_pplan_plan=is_pplan_plan, derived_from=derived_from)
         self.noodles_promise = noodles_promise
 
-        from noodles import get_workflow
-        workflow = get_workflow(self.noodles_promise)
+        workflow = noodles.get_workflow(self.noodles_promise)
 
-        steps_dict = {}
-        for i, n in workflow.nodes.items():
-            steps_dict[i] = n.foo._fairstep
+        # Exclude the root, because that is the workflow definition itself and not a step
+        steps_dict = {i: n.foo._fairstep for i, n in workflow.nodes.items() if i != workflow.root}
 
         for i, step in steps_dict.items():
             self._add_step(step)
 
         for i in workflow.links:
+            # Exclude the root, because that is the workflow definition itself and not a step
+            if i == workflow.root:
+                continue
             current_step = steps_dict[i]
             from_uri = rdflib.URIRef(steps_dict[i].uri + '#' + current_step.outputs[0].name)
             for j in workflow.links[i]:
@@ -362,7 +378,7 @@ class FairWorkflow(RdfWrapper):
                 rdf2dot(self._rdf, f)
             return graphviz.Source.from_file(filename)
 
-    def execute(self, num_threads=1):
+    def execute(self, *args, **kwargs):
         """
         Executes the workflow on the specified number of threads. Noodles is used as the execution
         engine. If a noodles workflow has not been generated for this fairworkflow object, then
@@ -375,11 +391,6 @@ class FairWorkflow(RdfWrapper):
         if not hasattr(self, 'noodles_promise'):
             raise ValueError('Cannot execute workflow as no noodles promise has been constructed.')
 
-        from noodles.run.threading.sqlite3 import run_parallel
-        from noodles import serial
-        import io
-        import logging
-
         log = io.StringIO()
         log_handler = logging.StreamHandler(log)
         formatter = logging.Formatter('%(asctime)s - %(message)s')
@@ -388,29 +399,37 @@ class FairWorkflow(RdfWrapper):
         logger = logging.getLogger('noodles')
         logger.setLevel(logging.INFO)
         logger.handlers = [log_handler]
-
-        # A local function defining the serializers in the registry
-        # If we want to allow remote execution with noodles, we will
-        # need to define this function differently, but for multithreaded
-        # execution (most likely what we'll stick with) this is fine.
-        def registry():
-            return serial.pickle() + serial.base()
-
-        # TODO: Currently we are dumping a lot to an sqlite db, that will slow the
-        # execution down a bit. In future we will either write our own version of
-        # the runner, to grab that extra prov without any db writing, or we will
-        # read it from the database and convert it to RDF then.
-        with TemporaryDirectory() as td:
-            temp_db_fname = Path(td) / 'retro_prov.db'
-            result = run_parallel(self.noodles_promise, n_threads=num_threads,
-                                  registry=registry, db_file=temp_db_fname,
-                                  always_cache=True, echo_log=False)
+        self.noodles_promise = self._replace_input_arguments(self.noodles_promise, args, kwargs)
+        result = noodles.run_single(self.noodles_promise)
 
         # Generate the retrospective provenance as a (nano-) Publication object
         retroprov = self._generate_retrospective_prov_publication(log.getvalue())
 
         return result, retroprov
 
+    def _replace_input_arguments(self, promise: noodles.interface.PromisedObject, args, kwargs):
+        """
+        Replace the input arguments of the promise so we can run the workflow with the right
+        inputs. This goes into the guts of noodles, doing something noodles was not intended to be
+        used for.
+        TODO: find a better solution for this
+        """
+        workflow = noodles.get_workflow(promise)
+        signature = inspect.signature(workflow.root_node.foo)
+        arguments_dict = self._get_arguments_dict(args, kwargs, signature)
+        workflow.root_node.bound_args = inspect.BoundArguments(signature, arguments_dict)
+        return promise
+
+    @staticmethod
+    def _get_arguments_dict(args, kwargs, signature):
+        """
+        Create dictionary of keyword arguments from positional and keyword arguments and the
+        signature of the function.
+        """
+        arguments_dict = {key: arg for arg, key in zip(args, signature.parameters.keys())}
+        arguments_dict.update(kwargs)
+        return arguments_dict
+
     def _generate_retrospective_prov_publication(self, log:str) -> nanopub.Publication:
         """
         Utility method for generating a Publication object for the retrospective
@@ -507,19 +526,24 @@ def is_fairworkflow(label: str = None, is_pplan_plan: bool = True):
     """
     def _modify_function(func):
         """
-        Store FairStep object as _fairstep attribute of the function. Use inspection to get the
-        description, inputs, and outputs of the step based on the function specification.
-        """
-        def _wrapper(*args, **kwargs):
-            promise = func(*args, **kwargs)
-            if not isinstance(promise, PromisedObject):
-                raise TypeError("The workflow does not return a 'promise'. Did you use the "
-                                "is_fairstep decorator on all the steps?")
-
-            # Description of workflow is the raw function code
-            description = inspect.getsource(func)
-
-            return FairWorkflow.from_noodles_promise(promise, description=description, label=label,
-                 is_pplan_plan=is_pplan_plan, derived_from=None)
-        return _wrapper
+        Store FairWorkflow as _fairworkflow attribute of the function. Use inspection to get the
+        description, inputs, and outputs of the workflow based on the function specification.
+        Call the scheduled_workflow with empty arguments so we get a PromisedObject. These empty
+        arguments will be replaced upon execution with the input arguments that the .execute()
+        method is called with.
+        """
+        scheduled_workflow = noodles.schedule(func)
+        num_params = len(inspect.signature(func).parameters)
+        empty_args = ([inspect.Parameter.empty()] * num_params)
+        promise = scheduled_workflow(*empty_args)
+        if not isinstance(func(*empty_args), PromisedObject):
+            raise TypeError("The workflow does not return a 'promise'. Did you use the "
+                            "is_fairstep decorator on all the steps?")
+
+        # Description of workflow is the raw function code
+        description = inspect.getsource(func)
+        promise._fairworkflow = FairWorkflow.from_noodles_promise(
+            promise, description=description, label=label,
+            is_pplan_plan=is_pplan_plan, derived_from=None)
+        return promise
     return _modify_function


Test patch:
diff --git a/tests/test_fairworkflow.py b/tests/test_fairworkflow.py
index 63bbe9f..442bb27 100644
--- a/tests/test_fairworkflow.py
+++ b/tests/test_fairworkflow.py
@@ -1,3 +1,4 @@
+import inspect
 import warnings
 from unittest import mock
 
@@ -10,6 +11,7 @@ from fairworkflows import FairWorkflow, FairStep, namespaces, FairVariable, is_f
 from fairworkflows.rdf_wrapper import replace_in_rdf
 from nanopub import Publication
 
+
 class TestFairWorkflow:
     test_description = 'This is a test workflow.'
     test_label = 'Test'
@@ -336,18 +338,19 @@ class TestFairWorkflow:
             """
             A simple addition, subtraction, multiplication workflow
             """
-            t1 = add(in1, in2)
-            t2 = sub(in1, in2)
-            t3 = mul(weird(t1, in3), t2)
-            return t3
+            t1 = add(in1, in2)  # 5
+            t2 = sub(in1, in2)  # -3
+            t3 = weird(t1, in3)  # 10 + 12 = 22
+            t4 = mul(t3, t2)  # 22 * -3 = 66
+            return t4
+
+        fw = FairWorkflow.from_function(my_workflow)
 
-        fw = my_workflow(1, 4, 3)
         assert isinstance(fw, FairWorkflow)
 
-        # TODO: Find out why execution hangs only when running in pytest. May be to do with the threading.
-        #result, prov = fw.execute(num_threads=1)
-        #assert result == -66
-        #assert isinstance(prov, Publication)
+        result, prov = fw.execute(1, 4, 3)
+        assert result == -66
+        assert isinstance(prov, Publication)
 
     def test_workflow_complex_serialization(self):
         class OtherType:
@@ -363,25 +366,32 @@ class TestFairWorkflow:
             return return_that_object(im)
 
         obj = OtherType('I could be e.g. a PIL Image')
-        fw = process_image(obj)
+        fw = FairWorkflow.from_function(process_image)
+        result, prov = fw.execute(obj)
+        assert isinstance(result, type(obj))
+        assert result.message == obj.message
+        assert isinstance(prov, Publication)
+
+    def test_get_arguments_dict(self):
+        args = (1, 2)
+        kwargs = {'c': 3, 'd': 4}
+
+        def func(a, b, c, d):
+            return
 
-        # TODO: Find out why execution hangs only when running in pytest. May be to do with the threading.
-        #result, prov = fw.execute()
-        #assert isinstance(result, type(obj))
-        #assert result.message == obj.message
-        #assert isinstance(prov, Publication)
+        result = FairWorkflow._get_arguments_dict(args, kwargs, inspect.signature(func))
+        assert result == {'a': 1, 'b': 2, 'c': 3, 'd': 4}
 
     def test_workflow_non_decorated_step(self):
-        def add(a: float, b: float) -> float:
-            """Adding up numbers. NB: no is_fairstep decorator!"""
-            return a + b
+        def return_value(a: float) -> float:
+            """Return the input value. NB: no is_fairstep decorator!"""
+            return a
 
-        @is_fairworkflow(label='My Workflow')
-        def my_workflow(in1, in2):
-            """
-            A simple addition workflow
-            """
-            return add(in1, in2)
         with pytest.raises(TypeError) as e:
-            my_workflow(1, 2)
+            @is_fairworkflow(label='My Workflow')
+            def my_workflow(in1):
+                """
+                A simple workflow
+                """
+                return return_value(in1)
         assert "The workflow does not return a 'promise'" in str(e.value)