refactor: variable-map is now 'plumbing' and Translation is a 'Connector'

Alan Christie · Alan Christie · commit 5e2c8bc7e973 · 2025-09-02T10:47:26.000Z
diff --git a/tests/test_workflow_engine_examples.py b/tests/test_workflow_engine_examples.py
@@ -398,12 +398,12 @@ def test_workflow_engine_simple_python_molprops_with_options(basic_engine):
     assert project_file_exists(output_file_2)
 
 
-def test_workflow_engine_simple_python_fanout(basic_engine):
+def test_workflow_engine_simple_python_split_combine(basic_engine):
     # Arrange
     md, da = basic_engine
 
     da.mock_get_running_workflow_step_output_values_for_output(
-        step_name="first-step",
+        step_name="split",
         output_variable="outputBase",
         output=["chunk_1.smi", "chunk_2.smi"],
     )
@@ -427,7 +427,7 @@ def test_workflow_engine_simple_python_fanout(basic_engine):
     r_wfid = start_workflow(
         md,
         da,
-        "simple-python-fanout",
+        "simple-python-split-combine",
         {"candidateMolecules": input_file_1},
     )
 
diff --git a/tests/workflow-definitions/duplicate-step-input-output-variable-names.yaml b/tests/workflow-definitions/duplicate-step-input-output-variable-names.yaml
@@ -14,7 +14,7 @@ steps:
     variables:
       name: "col1"
       value: 123
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-workflow:
       variable: candidateMolecules
@@ -34,7 +34,7 @@ steps:
     variables:
       name: "col2"
       value: "999"
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-step:
       name: step1
diff --git a/tests/workflow-definitions/example-smiles-to-file.yaml b/tests/workflow-definitions/example-smiles-to-file.yaml
@@ -13,7 +13,7 @@ steps:
     collection: workflow-engine-unit-test-jobs
     job: smiles-to-file
     version: "1.0.0"
-  variable-mapping:
+  plumbing:
   - variable: outputFile
     from-workflow:
       variable: outputFile
diff --git a/tests/workflow-definitions/replicate-using-undeclared-input.yaml b/tests/workflow-definitions/replicate-using-undeclared-input.yaml
@@ -14,7 +14,7 @@ steps:
     variables:
       name: "col1"
       value: 123
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-workflow:
       variable: candidateMolecules
@@ -33,7 +33,7 @@ steps:
   replicate:
     using:
       variable: y
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-step:
       name: step-1
diff --git a/tests/workflow-definitions/shortcut-example-1.yaml b/tests/workflow-definitions/shortcut-example-1.yaml
@@ -19,7 +19,7 @@ steps:
     collection: workflow-engine-unit-test-jobs
     job: shortcut-example-1-process-b
     version: "1.0.0"
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-step:
       name: example-1-step-1
diff --git a/tests/workflow-definitions/simple-python-molprops-with-options.yaml b/tests/workflow-definitions/simple-python-molprops-with-options.yaml
@@ -13,7 +13,7 @@ steps:
     version: "1.0.0"
     variables:
       outputFile: step1.out.smi
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-workflow:
       variable: candidateMolecules
@@ -33,7 +33,7 @@ steps:
     variables:
       name: "col2"
       value: "999"
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-step:
       name: step1
diff --git a/tests/workflow-definitions/simple-python-molprops.yaml b/tests/workflow-definitions/simple-python-molprops.yaml
@@ -15,7 +15,7 @@ steps:
       name: "col1"
       value: 123
       outputFile: "results.smi"
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-workflow:
       variable: candidateMolecules
@@ -29,7 +29,7 @@ steps:
     variables:
       name: "col2"
       value: "999"
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-step:
       name: step1
diff --git a/tests/workflow-definitions/simple-python-parallel.yaml b/tests/workflow-definitions/simple-python-parallel.yaml
@@ -14,7 +14,7 @@ steps:
     variables:
       name: "unnecessary"
       value: "0"
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-workflow:
       variable: candidateMolecules
@@ -28,7 +28,7 @@ steps:
     variables:
       name: "desc1"
       value: "777"
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-step:
       name: first-step
@@ -43,7 +43,7 @@ steps:
     variables:
       name: "desc2"
       value: "999"
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-step:
       name: first-step
diff --git a/tests/workflow-definitions/simple-python-split-combine.yaml b/tests/workflow-definitions/simple-python-split-combine.yaml
@@ -3,10 +3,12 @@ kind: DataManagerWorkflow
 kind-version: "2025.2"
 name: python-workflow
 description: >-
-  A simple parallel workflow. Input is split into N chunks and N processes of the same job is started
+  A simple parallel workflow. The input is split into chunks and a number outputFile
+  parallel steps processes these outputs. Finally a combine step concatenates the
+  files.
 
 steps:
-- name: first-step
+- name: split
   description: Split an input file
   specification:
     collection: workflow-engine-unit-test-jobs
@@ -16,7 +18,7 @@ steps:
       name: count
       value: "1"
       outputBase: chunk
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-workflow:
       variable: candidateMolecules
@@ -31,10 +33,10 @@ steps:
       name: desc1
       value: "777"
       outputFile: results.smi
-  variable-mapping:
+  plumbing:
   - variable: inputFile
     from-step:
-      name: first-step
+      name: split
       variable: outputBase
   out:
   - outputFile
diff --git a/workflow/decoder.py b/workflow/decoder.py
@@ -25,8 +25,9 @@
 
 
 @dataclass
-class Translation:
-    """A source ("in_") to destination ("out") variable map."""
+class Connector:
+    """A connection - connexts a plumbing source variable ("in_")
+    to destination variable ("out")."""
 
     in_: str
     out: str
@@ -83,13 +84,13 @@ def get_description(definition: dict[str, Any]) -> str | None:
 
 def get_workflow_variable_names(definition: dict[str, Any]) -> set[str]:
     """Given a Workflow definition this function returns all the names of the
-    variables that need to be defined at the workflow level. These are the 'variables'
-    used in every steps' variabale-mapping block.
+    variables defined in steps that need to be defined at the workflow level.
+    These are the 'variables' used in every step's 'plumbing' block.
     """
     wf_variable_names: set[str] = set()
     steps: list[dict[str, Any]] = get_steps(definition)
     for step in steps:
-        if v_map := step.get("variable-mapping"):
+        if v_map := step.get("plumbing"):
             for v in v_map:
                 if "from-workflow" in v:
                     wf_variable_names.add(v["from-workflow"]["variable"])
@@ -126,40 +127,38 @@ def get_step_input_variable_names(
     return variable_names
 
 
-def get_step_workflow_variable_mapping(*, step: dict[str, Any]) -> list[Translation]:
+def get_step_workflow_plumbing(*, step: dict[str, Any]) -> list[Connector]:
     """Returns a list of workflow vaiable name to step variable name
     Translation objects for the given step."""
-    variable_mapping: list[Translation] = []
-    if "variable-mapping" in step:
-        for v_map in step["variable-mapping"]:
+    variable_mapping: list[Connector] = []
+    if "plumbing" in step:
+        for v_map in step["plumbing"]:
             if "from-workflow" in v_map:
                 variable_mapping.append(
-                    Translation(
+                    Connector(
                         in_=v_map["from-workflow"]["variable"], out=v_map["variable"]
                     )
                 )
     return variable_mapping
 
 
-def get_step_prior_step_variable_mapping(
-    *, step: dict[str, Any]
-) -> dict[str, list[Translation]]:
+def get_step_prior_step_plumbing(*, step: dict[str, Any]) -> dict[str, list[Connector]]:
     """Returns list of Translation objects, indexed by prior step name,
     that identify source step (output) variable name to this step's (input)
     variable name."""
-    variable_mapping: dict[str, list[Translation]] = {}
-    if "variable-mapping" in step:
-        for v_map in step["variable-mapping"]:
+    variable_mapping: dict[str, list[Connector]] = {}
+    if "plumbing" in step:
+        for v_map in step["plumbing"]:
             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(
-                        Translation(in_=step_variable, out=v_map["variable"])
+                        Connector(in_=step_variable, out=v_map["variable"])
                     )
                 else:
                     variable_mapping[step_name] = [
-                        Translation(in_=step_variable, out=v_map["variable"])
+                        Connector(in_=step_variable, out=v_map["variable"])
                     ]
     return variable_mapping
diff --git a/workflow/workflow-schema.yaml b/workflow/workflow-schema.yaml
@@ -156,11 +156,11 @@ definitions:
         # The format of this is essentially idenical to the specification
         # used when a Job is launched via the DM API.
         $ref: '#/definitions/step-specification'
-      variable-mapping:
+      plumbing:
         # The map of the source of the step's variables.
-        # all variables the step needs (that aren;t already in the specification)
+        # All variables the step needs (that aren't already in the specification)
         # need to be declared here. They either come "from" a prior step
-        # or are expected in th erunning workflow variables. Here we simply
+        # or are expected to be a workflow variable. Here we simply
         # associate every required variable to a source.
         type: array
         items:
diff --git a/workflow/workflow_engine.py b/workflow/workflow_engine.py
@@ -40,10 +40,10 @@
 )
 
 from .decoder import (
-    Translation,
+    Connector,
     get_step,
-    get_step_prior_step_variable_mapping,
-    get_step_workflow_variable_mapping,
+    get_step_prior_step_plumbing,
+    get_step_workflow_plumbing,
 )
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -330,27 +330,27 @@ def _validate_step_command(
         # "in" variables are worklfow variables, and "out" variables
         # are expected Job variables. We use this to add variables
         # to the "all variables" map.
-        for tr in get_step_workflow_variable_mapping(step=step):
-            assert tr.in_ in running_workflow_variables
-            all_variables[tr.out] = running_workflow_variables[tr.in_]
+        for connector in get_step_workflow_plumbing(step=step):
+            assert connector.in_ in running_workflow_variables
+            all_variables[connector.out] = running_workflow_variables[connector.in_]
 
         # Now we apply variables from the "variable mapping" block
         # related to values used in prior steps. The decoder gives
         # us a map indexed by prior step name that's a list of "in" "out"
         # tuples as above.
-        step_prior_v_map: dict[str, list[Translation]] = (
-            get_step_prior_step_variable_mapping(step=step)
+        prior_step_plumbing: dict[str, list[Connector]] = get_step_prior_step_plumbing(
+            step=step
         )
-        for prior_step_name, v_map in step_prior_v_map.items():
+        for prior_step_name, connections in prior_step_plumbing.items():
             # Retrieve the prior "running" step
             # in order to get the variables that were set there...
             prior_step, _ = self._wapi_adapter.get_running_workflow_step_by_name(
                 name=prior_step_name, running_workflow_id=running_workflow_id
             )
             # Copy "in" value to "out"...
-            for tr in v_map:
-                assert tr.in_ in prior_step["variables"]
-                all_variables[tr.out] = prior_step["variables"][tr.in_]
+            for connector in connections:
+                assert connector.in_ in prior_step["variables"]
+                all_variables[connector.out] = prior_step["variables"][connector.in_]
 
         # Now ... can the command be compiled!?
         job: dict[str, Any] = self._get_step_job(step=step)
@@ -401,10 +401,8 @@ def _launch(
         # be more than one prior step variable that is 'files'!
         replication_values: list[str] = []
         iter_variable: str | None = None
-        tr_map: dict[str, list[Translation]] = get_step_prior_step_variable_mapping(
-            step=step
-        )
-        for p_step_name, tr_list in tr_map.items():
+        plumbing: dict[str, list[Connector]] = get_step_prior_step_plumbing(step=step)
+        for p_step_name, connections in plumbing.items():
             # We need to get the Job definition for each step
             # and then check whether the (ouptu) variable is of type 'files'...
             wf_step: dict[str, Any] = get_step(wf, p_step_name)
@@ -413,9 +411,9 @@ def _launch(
             jd_outputs: dict[str, Any] = job_defintion_decoder.get_outputs(
                 job_definition
             )
-            for tr in tr_list:
-                if jd_outputs.get(tr.in_, {}).get("type") == "files":
-                    iter_variable = tr.out
+            for connector in connections:
+                if jd_outputs.get(connector.in_, {}).get("type") == "files":
+                    iter_variable = connector.out
                     # Get the prior running step's output values
                     response, _ = self._wapi_adapter.get_running_workflow_step_by_name(
                         name=p_step_name,
@@ -426,7 +424,7 @@ def _launch(
                     result, _ = (
                         self._wapi_adapter.get_running_workflow_step_output_values_for_output(
                             running_workflow_step_id=rwfs_id,
-                            output_variable=tr.in_,
+                            output_variable=connector.in_,
                         )
                     )
                     replication_values = result["output"].copy()
