feat: Some work on the refactored engine

Author: Alan Christie

tests/test_workflow_engine_examples.py

@@ -130,6 +130,7 @@ def wait_for_workflow(
     # are the responsibility of the caller.
     attempts = 0
     done = False
+    response = None
     while not done:
         response, _ = da.get_running_workflow(running_workflow_id=r_wfid)
         if response["done"]:
@@ -141,6 +142,7 @@ def wait_for_workflow(
             time.sleep(completion_poll_period_s)
     # When we get here the workflow must have finished (not timed-out),
     # and it must have passed (or failed) according the the caller's expectation.
+    assert response
     assert response["done"]
     assert response["success"] == expect_success
 
@@ -400,7 +402,7 @@ def test_workflow_engine_simple_python_molprops_with_options(basic_engine):
         "simple-python-molprops-with-options",
         {
             "candidateMolecules": input_file_1,
-            "outputFile": output_file_1,
+            "clusteredMolecules": output_file_2,
             "rdkitPropertyName": "prop",
             "rdkitPropertyValue": 1.2,
         },

tests/workflow-definitions/example-smiles-to-file.yaml

@@ -13,3 +13,12 @@ steps:
     collection: workflow-engine-unit-test-jobs
     job: smiles-to-file
     version: "1.0.0"
+  variable-mapping:
+  - variable: outputFile
+    from-workflow:
+      variable: outputFile
+  - variable: smiles
+    from-workflow:
+      variable: smiles
+  out:
+  - outputFile

tests/workflow-definitions/simple-python-molprops-with-options.yaml

@@ -4,62 +4,15 @@ kind-version: "2025.2"
 name: python-workflow
 description: A simple python experimental workflow
 
-# Some meaningless variables.
-# Just to make sure the decoder accepts this.
-# The Workflow engin eis not (yet) interested in this block.
-variables:
-  inputs:
-    type: object
-    required:
-    - inputFile
-    properties:
-      inputFile:
-        title: Molecules to pick from
-        mime-types:
-        - squonk/x-smiles
-        type: file
-      seeds:
-        title: Molecules that are already picked
-        mime-types:
-        - squonk/x-smiles
-        type: file
-        multiple: true
-  outputs:
-    type: object
-    properties:
-      outputFile:
-        title: Output file
-        mime-types:
-        - chemical/x-csv
-        creates: '{{ outputFile }}'
-        type: file
-  options:
-    type: object
-    required:
-    - count
-    properties:
-      outputFile:
-        title: Output file name
-        type: string
-        pattern: "^[A-Za-z0-9_/\\.\\-]+$"
-        default: diverse.smi
-      count:
-        title: Number of molecules to pick
-        type: integer
-        minimum: 1
-      threshold:
-        title: Similarity threshold
-        type: number
-        minimum: 0
-        maximum: 1
-
 steps:
 - name: step1
   description: Add column 1
   specification:
     collection: workflow-engine-unit-test-jobs
     job: rdkit-molprops
     version: "1.0.0"
+    variables:
+      outputFile: step1.out.smi
   variable-mapping:
   - variable: inputFile
     from-workflow:
@@ -70,9 +23,6 @@ steps:
   - variable: value
     from-workflow:
       variable: rdkitPropertyValue
-  - variable: outputFile
-    from-workflow:
-      variable: clusteredMolecules
 
 - name: step2
   description: Add column 2
@@ -90,6 +40,6 @@ steps:
       variable: outputFile
   - variable: outputFile
     from-workflow:
-      variable: outputFile
+      variable: clusteredMolecules
   out:
   - outputFile

workflow/decoder.py

@@ -4,7 +4,6 @@
 """
 
 import os
-from pprint import pprint
 from typing import Any
 
 import jsonschema
@@ -108,87 +107,41 @@ def get_step_input_variable_names(
     return variable_names
 
 
-def set_step_variables(
-    *,
-    workflow: dict[str, Any],
-    inputs: list[dict[str, Any]],
-    outputs: list[dict[str, Any]],
-    step_outputs: dict[str, Any],
-    previous_step_outputs: list[dict[str, Any]],
-    workflow_variables: dict[str, Any],
-    step_name: str,
-) -> dict[str, Any]:
-    """Prepare input- and output variables for the following step.
-
-    Inputs are defined in step definition but their values may
-    come from previous step outputs.
-    """
-    assert workflow
-
-    result = {}
-
-    print("ssv: wf vars:")
-    pprint(workflow_variables)
-    print("ssv: inputs:")
-    pprint(inputs)
-    print("ssv: outputs", outputs)
-    print("ssv: step_outputs", step_outputs)
-    print("ssv: prev step outputs", previous_step_outputs)
-    print("ssv: step_name", step_name)
-
-    for item in inputs:
-        p_key = item["input"]
-        p_val = ""
-        val = item["from"]
-        if "workflow-input" in val.keys():
-            p_val = workflow_variables[val["workflow-input"]]
-            result[p_key] = p_val
-        elif "step" in val.keys():
-            # this links the variable to previous step output
-            if previous_step_outputs:
-                for out in previous_step_outputs:
-                    if out["output"] == val["output"]:
-                        # p_val = out["as"]
-                        if step_outputs["output"]:
-                            p_val = step_outputs["output"]
-                            print("\n!!!!!!!!!!!!!if clause!!!!!!!!!!!!!!!!!!!!!\n")
-                            print(p_val)
-                        else:
-                            # what do I need to do here??
-                            print("\n!!!!!!!!!!!!!else clause!!!!!!!!!!!!!!!!!!!!!\n")
-                            print(out)
-                            print(val)
-
-                        # this bit handles multiple inputs: if a step
-                        # requires input from multiple steps, add them to
-                        # the list in result dict. this is the reason for
-                        # mypy ignore statements, mypy doesn't understand
-                        # redefinition
-                        if p_key in result:
-                            if not isinstance(result[p_key], set):
-                                result[p_key] = {result[p_key]}  # type: ignore [assignment]
-                            result[p_key].add(p_val)  # type: ignore [attr-defined]
-                        else:
-                            result[p_key] = p_val
-            else:
-                if val["output"] in workflow_variables:
-                    result[p_key] = workflow_variables[val["output"]]
-
-    for item in outputs:
-        p_key = item["output"]
-        # p_val = item["as"]
-        # p_val = step_outputs["output"]
-        p_val = "somefile.smi"
-        result[p_key] = p_val
-
-    #    options = set_variables_from_options_for_step(
-    #        definition=workflow,
-    #        variables=workflow_variables,
-    #        step_name=step_name,
-    #    )
-    #
-    #    result |= options
-    return result
+def get_step_workflow_variable_mapping(
+    *, step: dict[str, Any]
+) -> list[tuple[str, str]]:
+    """Returns a list of workflow vaiable name to step variable name tuples
+    for the given step."""
+    variable_mapping: list[tuple[str, str]] = []
+    if "variable-mapping" in step:
+        for v_map in step["variable-mapping"]:
+            if "from-workflow" in v_map:
+                # Tuple is "from" -> "to"
+                variable_mapping.append(
+                    (v_map["from-workflow"]["variable"], v_map["variable"])
+                )
+    return variable_mapping
+
+
+def get_step_prior_step_variable_mapping(
+    *, step: dict[str, Any]
+) -> dict[str, list[tuple[str, str]]]:
+    """Returns list of tuples, indexed by prior step name, of source step vaiable name
+    to this step's variable name."""
+    variable_mapping: dict[str, list[tuple[str, str]]] = {}
+    if "variable-mapping" in step:
+        for v_map in step["variable-mapping"]:
+            if "from-step" in v_map:
+                step_name = v_map["from-step"]["name"]
+                step_variable = v_map["from-step"]["variable"]
+                # Tuple is "from" -> "to"
+                if step_name in variable_mapping:
+                    variable_mapping[step_name].append(
+                        (step_variable, v_map["variable"])
+                    )
+                else:
+                    variable_mapping[step_name] = [(step_variable, v_map["variable"])]
+    return variable_mapping
 
 
 def get_step_replicator(*, step: dict[str, Any]) -> str | Any:

workflow/workflow_engine.py

@@ -41,8 +41,9 @@
 
 from .decoder import (
     get_step_input_variable_names,
+    get_step_prior_step_variable_mapping,
     get_step_replicator,
-    set_step_variables,
+    get_step_workflow_variable_mapping,
 )
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -485,16 +486,6 @@ def _validate_step_command(
 
         print("final prev step outputs", previous_step_outputs)
 
-        step_vars = set_step_variables(
-            workflow=workflow,
-            workflow_variables=all_variables,
-            inputs=inputs,
-            outputs=outputs,
-            step_outputs=step_outputs,
-            previous_step_outputs=previous_step_outputs,
-            step_name=running_wf_step["name"],
-        )
-        all_variables |= step_vars
         _LOGGER.debug(
             "Index %s (%s) all_variables=%s",
             our_step_index,
@@ -533,27 +524,52 @@ def _launch(
         wf_step_data, _ = self._wapi_adapter.get_workflow_steps_driving_this_step(
             running_workflow_step_id=rwfs_id,
         )
-        print("wf_step_data")
-        pprint(wf_step_data)
         assert wf_step_data["caller_step_index"] >= 0
         our_step_index: int = wf_step_data["caller_step_index"]
 
         print("step in _launch:", step_name)
         pprint(step)
 
+        # Workflow variables set by the user...
+        rwf_variables: dict[str, Any] = rwf.get("variables", {})
+
         # Now check the step command can be executed
         # (by trying to decoding the Job command).
+        # Before we do this we have to construct the variable map
+        # for this step.
         #
-        # We pass in the workflow variables (these are provided by the user
-        # when the workflow is run. All workflow variables will be present in the
-        # running workflow record)
-        running_workflow_variables: dict[str, Any] | None = rwf.get("variables")
+        # We start with any variables provided in the step specification
+        all_variables: dict[str, Any] = step["specification"].get("variables", {})
+        # We now have to iterate through the step's variable mapping block.
+        # This will name any workflow variables (from)
+        # and their corresponding step variable (to).
+        step_wf_v_map: list[tuple[str, str]] = get_step_workflow_variable_mapping(
+            step=step
+        )
+        for from_to in step_wf_v_map:
+            all_variables[from_to[1]] = rwf_variables[from_to[0]]
+        # We must now apply variables from prior steps identified in the
+        # current step's mapping block. We're given a map indexed by
+        # prior step name that's a list of tuples name the prior step's
+        # variable (from) and the curent step variable (to).
+        step_prior_v_map: dict[str, list[tuple[str, str]]] = (
+            get_step_prior_step_variable_mapping(step=step)
+        )
+        for prior_step_name, v_map in step_prior_v_map.items():
+            # Load the prior step
+            prior_step, _ = self._wapi_adapter.get_running_workflow_step_by_name(
+                name=prior_step_name, running_workflow_id=rwf_id
+            )
+            # Get its variables and copy the value
+            for from_to in v_map:
+                all_variables[from_to[1]] = prior_step["variables"][from_to[0]]
+
         error_or_variables: str | dict[str, Any] = self._validate_step_command(
             running_workflow_step_id=rwfs_id,
             step=step,
             workflow_steps=wf_step_data["steps"],
             our_step_index=our_step_index,
-            running_workflow_variables=running_workflow_variables,
+            running_workflow_variables=all_variables,
         )
         if isinstance(error_or_variables, str):
             error_msg = error_or_variables
@@ -570,13 +586,14 @@ def _launch(
 
         _LOGGER.info(
             "Launching step: RunningWorkflow=%s RunningWorkflowStep=%s step=%s"
-            " (name=%s project=%s, variables=%s)",
+            " variables=%s name=%s project=%s, (all_variables=%s)",
             rwf_id,
             rwfs_id,
             step_name,
+            variables,
             rwf["name"],
             project_id,
-            variables,
+            all_variables,
         )
 
         # When we launch a step we need to identify all the prior steps in the workflow,
@@ -593,7 +610,7 @@ def _launch(
         prior_steps: list[str] = []
         if our_step_index > 0:
             # We need the step ID of the prior step.
-            prior_step_name: str = wf_step_data["steps"][our_step_index - 1]["name"]
+            prior_step_name = wf_step_data["steps"][our_step_index - 1]["name"]
             step_response, _ = self._wapi_adapter.get_running_workflow_step_by_name(
                 name=prior_step_name,
                 running_workflow_id=rwf_id,