refactor: Switch away from workflow replicate property

Author: Alan Christie

poetry.lock

@@ -14,7 +14,7 @@ packages = [
 [tool.poetry.dependencies]
 python = "^3.12"
 im-protobuf = "^8.2.0"
-im-data-manager-job-decoder = "^2.4.0"
+im-data-manager-job-decoder = "^2.5.0"
 jsonschema = "^4.21.1"
 pyyaml = ">= 5.3.1, < 7.0"
 

pyproject.toml

@@ -398,6 +398,7 @@ def test_workflow_engine_simple_python_molprops_with_options(basic_engine):
     assert project_file_exists(output_file_2)
 
 
+@pytest.mark.skip(reason="WIP")
 def test_workflow_engine_simple_python_fanout(basic_engine):
     # Arrange
     md, da = basic_engine

tests/test_workflow_engine_examples.py

@@ -217,30 +217,6 @@ def test_validate_simple_python_molprops_with_missing_input():
     ]
 
 
-def test_validate_replicate_using_undeclared_input():
-    # Arrange
-    workflow_filename: str = os.path.join(
-        os.path.dirname(__file__),
-        "workflow-definitions",
-        "replicate-using-undeclared-input.yaml",
-    )
-    with open(workflow_filename, "r", encoding="utf8") as workflow_file:
-        workflow: dict[str, Any] = yaml.load(workflow_file, Loader=yaml.FullLoader)
-    assert workflow
-
-    # Act
-    error = WorkflowValidator.validate(
-        level=ValidationLevel.TAG,
-        workflow_definition=workflow,
-    )
-
-    # Assert
-    assert error.error_num == 7
-    assert error.error_msg == [
-        "Replicate input variable is not declared: y (step=step-2)"
-    ]
-
-
 def test_validate_duplicate_step_output_variable_names():
     # Arrange
     workflow_filename: str = os.path.join(

tests/test_workflow_validator_for_run_level.py

@@ -151,30 +151,6 @@ def test_validate_simple_python_molprops_with_options():
     assert error.error_msg is None
 
 
-def test_validate_replicate_using_undeclared_input():
-    # Arrange
-    workflow_filename: str = os.path.join(
-        os.path.dirname(__file__),
-        "workflow-definitions",
-        "replicate-using-undeclared-input.yaml",
-    )
-    with open(workflow_filename, "r", encoding="utf8") as workflow_file:
-        workflow: dict[str, Any] = yaml.load(workflow_file, Loader=yaml.FullLoader)
-    assert workflow
-
-    # Act
-    error = WorkflowValidator.validate(
-        level=ValidationLevel.TAG,
-        workflow_definition=workflow,
-    )
-
-    # Assert
-    assert error.error_num == 7
-    assert error.error_msg == [
-        "Replicate input variable is not declared: y (step=step-2)"
-    ]
-
-
 def test_validate_duplicate_step_output_variable_names():
     # Arrange
     workflow_filename: str = os.path.join(

tests/test_workflow_validator_for_tag_level.py

@@ -31,9 +31,6 @@ steps:
       name: desc1
       value: "777"
       outputFile: results.smi
-  replicate:
-    using:
-      variable: inputFile
   variable-mapping:
   - variable: inputFile
     from-step:

tests/workflow-definitions/simple-python-fanout.yaml

@@ -5,7 +5,6 @@
 
 import os
 from dataclasses import dataclass
-from enum import Enum
 from typing import Any
 
 import jsonschema
@@ -33,26 +32,6 @@ class Translation:
     out: str
 
 
-class ReplicationOrigin(Enum):
-    """Oirgin of a replication variable."""
-
-    STEP_VARIABLE = 1
-    WORKFLOW_VARIABLE = 2
-
-
-@dataclass
-class ReplicationDriver:
-    """A step's replication driver.
-    The 'variable' is the variable for the step-to-be-executed
-    whose value is 'driven' by the values of the 'source_variable'.
-    The source variable is either from a step (or a workflow)."""
-
-    origin: ReplicationOrigin
-    variable: str
-    source_variable: str
-    source_step_name: str | None = None
-
-
 def validate_schema(workflow: dict[str, Any]) -> str | None:
     """Checks the Workflow Definition against the built-in schema.
     If there's an error the error text is returned, otherwise None.
@@ -173,38 +152,3 @@ def get_step_prior_step_variable_mapping(
                         Translation(in_=step_variable, out=v_map["variable"])
                     ]
     return variable_mapping
-
-
-def get_step_replication_driver(*, step: dict[str, Any]) -> ReplicationDriver | None:
-    """If the step is expected to replicate we return its replication driver,
-    which consists of a (prior) step name and an (output) variable name.
-    Otherwise it returns nothing."""
-    if replicator := step.get("replicate"):
-        # We need the variable we replicate against,
-        # and the step that owns the variable.
-        #
-        # 'using' is a dict but there can be only single value for now
-        variable: str = replicator["using"]["variable"]
-        source_variable: str | None = None
-        # Is the variable from a prior step?
-        step_name: str | None = None
-        step_v_map = get_step_prior_step_variable_mapping(step=step)
-        for step_name_candidate, mappings in step_v_map.items():
-            for mapping in mappings:
-                if mapping.out == variable:
-                    step_name = step_name_candidate
-                    source_variable = mapping.in_
-                    break
-            if step_name:
-                break
-        assert step_name
-        assert source_variable
-
-        return ReplicationDriver(
-            origin=ReplicationOrigin.STEP_VARIABLE,
-            variable=variable,
-            source_step_name=step_name,
-            source_variable=source_variable,
-        )
-
-    return None

workflow/decoder.py

@@ -59,17 +59,6 @@ definitions:
     type: string
     pattern: ^[a-zA-Z_][a-zA-Z0-9_]*$
 
-  # A step replication control variable
-  # that is based on a step variable
-  replicate-using-variable:
-    type: object
-    additionalProperties: false
-    properties:
-      variable:
-        $ref: '#/definitions/variable-name'
-    required:
-    - variable
-
   # A Step variable
   # (whose value is derived from a variable used in a prior step)
   step-variable-from-step:
@@ -167,14 +156,6 @@ 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'
-      replicate:
-        # Used to indicate one input variable that is used to replicate/spawn
-        # step instances based on the number of values generated for the variable.
-        type: object
-        additionalProperties: false
-        properties:
-          using:
-            $ref: '#/definitions/replicate-using-variable'
       variable-mapping:
         # The map of the source of the step's variables.
         # all variables the step needs (that aren;t already in the specification)

workflow/workflow-schema.yaml

@@ -39,11 +39,8 @@
 )
 
 from .decoder import (
-    ReplicationDriver,
-    ReplicationOrigin,
     Translation,
     get_step_prior_step_variable_mapping,
-    get_step_replication_driver,
     get_step_workflow_variable_mapping,
 )
 
@@ -380,41 +377,20 @@ def _launch(self, *, rwf: dict[str, Any], step: dict[str, Any]) -> None:
 
         variables: dict[str, Any] = error_or_variables
 
-        # A replication number,
-        # use only for steps expected to replicate (even if just once)
+        # A step replication number,
+        # used only for steps expected to run in parallel (even if just once)
         step_replication_number: int = 0
-        # Does this step have a replicating driver?
-        r_driver: ReplicationDriver | None = get_step_replication_driver(step=step)
         replication_values: list[str] = []
-        if r_driver:
-            if r_driver.origin == ReplicationOrigin.STEP_VARIABLE:
-                # We need to get the variable values from a prior step
-                # We need the prior steps running-workflow-step-id
-                assert r_driver.source_step_name
-                response, _ = self._wapi_adapter.get_running_workflow_step_by_name(
-                    name=r_driver.source_step_name,
-                    running_workflow_id=rwf_id,
-                )
-                assert "id" in response
-                o_rwfs_id: str = response["id"]
-                response, _ = (
-                    self._wapi_adapter.get_running_workflow_step_output_values_for_output(
-                        running_workflow_step_id=o_rwfs_id,
-                        output_variable=r_driver.source_variable,
-                    )
-                )
-                assert "output" in response
-                replication_values = response["output"]
-            else:
-                assert False, "Unsupported origin"
+        source_is_splitter: bool = False
+        iter_variable: str | None = None
 
         num_step_instances: int = max(1, len(replication_values))
         for iteration in range(num_step_instances):
 
             # If we are replicating this step then we must replace the step's variable
             # with a value expected for this iteration.
-            if r_driver:
-                iter_variable: str = r_driver.variable
+            if source_is_splitter:
+                assert iter_variable
                 iter_value: str = replication_values[iteration]
                 _LOGGER.info(
                     "Replicating step: %s iteration=%s variable=%s value=%s",
@@ -424,7 +400,7 @@ def _launch(self, *, rwf: dict[str, Any], step: dict[str, Any]) -> None:
                     iter_value,
                 )
                 # Over-write the replicating variable
-                # and set the replication numebr to a unique +ve non-zero value...
+                # and set the replication number to a unique +ve non-zero value...
                 variables[iter_variable] = iter_value
                 step_replication_number = iteration + 1
 

workflow/workflow_engine.py

@@ -6,8 +6,6 @@
 
 from .decoder import (
     get_step_output_variable_names,
-    get_step_prior_step_variable_mapping,
-    get_step_workflow_variable_mapping,
     get_steps,
     get_workflow_variable_names,
     validate_schema,
@@ -113,39 +111,6 @@ def _validate_tag_level(
                 error_num=2,
                 error_msg=[f"Duplicate step names found: {', '.join(duplicate_names)}"],
             )
-        # For each 'replicating' step the replicating variable
-        # must be declared in the step - which is either a workflow variable
-        # or a prior step variable.
-        for step in get_steps(workflow_definition):
-            if (
-                replicate_using_input := step.get("replicate", {})
-                .get("using", {})
-                .get("variable")
-            ):
-                found: bool = False
-                for translation in get_step_workflow_variable_mapping(step=step):
-                    if replicate_using_input == translation.out:
-                        found = True
-                        break
-                if not found:
-                    for (
-                        step_name,
-                        translations,
-                    ) in get_step_prior_step_variable_mapping(step=step).items():
-                        for translation in translations:
-                            if replicate_using_input == translation.out:
-                                found = True
-                                break
-                        if found:
-                            break
-                if not found:
-                    return ValidationResult(
-                        error_num=7,
-                        error_msg=[
-                            "Replicate input variable is not declared:"
-                            f" {replicate_using_input} (step={step["name"]})"
-                        ],
-                    )
 
         return _VALIDATION_SUCCESS