feat: Switch to pre-defined variables (rather then link)

Author: Alan Christie

tests/workflow-definitions/simple-python-split-combine.yaml

@@ -54,6 +54,7 @@ steps:
       name: parallel
       variable: outputFile
   - variable: inputDirPrefix
-    from-link-prefix:
+    from-predefined:
+      variable: link-prefix
   - variable: outputFile
     to-project:

workflow/decoder.py

@@ -144,6 +144,23 @@ def get_step_workflow_variable_connections(
     return connections
 
 
+def get_step_predefined_variable_connections(
+    *, step_definition: dict[str, Any]
+) -> list[Connector]:
+    """Returns the set of connections of pre-defined variables (in) to
+    step variables (out)."""
+    connections: list[Connector] = []
+    if "plumbing" in step_definition:
+        for v_map in step_definition["plumbing"]:
+            if "from-predefined" in v_map:
+                connections.append(
+                    Connector(
+                        in_=v_map["from-predefined"]["variable"], out=v_map["variable"]
+                    )
+                )
+    return connections
+
+
 def get_step_prior_step_connections(
     *, step_definition: dict[str, Any]
 ) -> dict[str, list[Connector]]:
@@ -166,14 +183,3 @@ def get_step_prior_step_connections(
                         Connector(in_=step_variable, out=v_map["variable"])
                     ]
     return plumbing
-
-
-def get_step_link_prefix_variables(*, step_definition: dict[str, Any]) -> set[str]:
-    """Returns the set of variables expected to be set to the value
-    of the instance directory prefix."""
-    variables: set[str] = set()
-    if "plumbing" in step_definition:
-        for v_map in step_definition["plumbing"]:
-            if "from-link-prefix" in v_map:
-                variables.add(v_map["variable"])
-    return variables

workflow/workflow-schema.yaml

@@ -57,7 +57,7 @@ definitions:
   # Stuff like 'candidateMolecules' or 'clustered_molecules'
   variable-name:
     type: string
-    pattern: ^[a-zA-Z_][a-zA-Z0-9_]*$
+    pattern: ^[a-zA-Z_][a-zA-Z0-9_-]*$
 
   # A Step variable
   # (whose value is derived from a variable used in a prior step)
@@ -103,27 +103,24 @@ definitions:
     - from-workflow
 
   # A Step variable
-  # (whose value is set to the value of a directory prefix used when the DM
-  # links the instance directories of prior step instances into this
-  # step's instance directory)
-  #
-  # This _must_ be treated by the step's job as a directory prefix,
-  # typiclaly '.instance-', that can be used to identify directories in this step's
-  # execution directory where the execution directory of prior steps
-  # are hard-linked by the DM. A job can find all the prior step directory names
-  # using the selected variable (e.g. inspect any directory name
-  # that starts with "{variable}").
-  step-variable-from-link-prefix:
+  # (whose value is set to the value of a pre-defined engine variable)
+  step-variable-from-predefined:
     type: object
     additionalProperties: false
     properties:
       variable:
         $ref: '#/definitions/variable-name'
-      from-link-prefix:
-        type: 'null'
+      from-predefined:
+        type: object
+        additionalProperties: false
+        properties:
+          variable:
+            $ref: '#/definitions/variable-name'
+        required:
+        - variable
     required:
     - variable
-    - from-link-prefix
+    - from-predefined
 
   # A Step variable
   # (whose value (a file) is to be copied to the project directory)
@@ -204,7 +201,7 @@ definitions:
           anyOf:
           - $ref: "#/definitions/step-variable-from-step"
           - $ref: "#/definitions/step-variable-from-workflow"
-          - $ref: "#/definitions/step-variable-from-link-prefix"
+          - $ref: "#/definitions/step-variable-from-predefined"
           - $ref: "#/definitions/step-variable-to-project"
         minItems: 1
     required:

workflow/workflow_engine.py

@@ -43,7 +43,7 @@
 from .decoder import (
     Connector,
     get_step,
-    get_step_link_prefix_variables,
+    get_step_predefined_variable_connections,
     get_step_prior_step_connections,
     get_step_workflow_variable_connections,
 )
@@ -84,9 +84,11 @@ def __init__(
         and a step (directory) link prefix (the directory prefix the DM uses to hard-link
         prior step instanes into the next step, typically '.instance-')"""
         # Keep the dependent objects
-        self._wapi_adapter = wapi_adapter
-        self._instance_launcher = instance_launcher
-        self._step_link_prefix = step_link_prefix
+        self._wapi_adapter: WorkflowAPIAdapter = wapi_adapter
+        self._instance_launcher: InstanceLauncher = instance_launcher
+        self._step_link_prefix: str = step_link_prefix
+
+        self._predefined_variables: dict[str, Any] = {"link-prefix": step_link_prefix}
 
     def handle_message(self, msg: Message) -> None:
         """Expect Workflow and Pod messages.
@@ -489,14 +491,12 @@ def _prepare_step(
             assert connector.in_ in rwf_variables
             variables[connector.out] = rwf_variables[connector.in_]
 
-        # Process the step's "plumbing" relating to link-prefix variables.
-        #
-        # This will be a set of variable names. We just set each one
-        # to the built-in step link prefix.
-        for link_variable in get_step_link_prefix_variables(
+        # Process the step's "plumbing" relating to pre-defined variables.
+        for connector in get_step_predefined_variable_connections(
             step_definition=step_definition
         ):
-            variables[link_variable] = self._step_link_prefix
+            assert connector.in_ in self._predefined_variables
+            variables[connector.out] = self._predefined_variables[connector.in_]
 
         # Now process variables (in the "plumbing" block)
         # that relate to values used in prior steps.