feat: Add set_variables_from_options_for_step nto decoder

Author: Alan Christie

tests/test_decoder.py

@@ -30,6 +30,19 @@
     _SIMPLE_PYTHON_MOLPROPS_WORKFLOW: Dict[str, Any] = yaml.safe_load(workflow_file)
 assert _SIMPLE_PYTHON_MOLPROPS_WORKFLOW
 
+_SIMPLE_PYTHON_MOLPROPS_WITH_OPTIONS_WORKFLOW_FILE: str = os.path.join(
+    os.path.dirname(__file__),
+    "workflow-definitions",
+    "simple-python-molprops-with-options.yaml",
+)
+with open(
+    _SIMPLE_PYTHON_MOLPROPS_WITH_OPTIONS_WORKFLOW_FILE, "r", encoding="utf8"
+) as workflow_file:
+    _SIMPLE_PYTHON_MOLPROPS_WITH_OPTIONS_WORKFLOW: Dict[str, Any] = yaml.safe_load(
+        workflow_file
+    )
+assert _SIMPLE_PYTHON_MOLPROPS_WITH_OPTIONS_WORKFLOW
+
 _DUPLICATE_WORKFLOW_VARIABLE_NAMES_WORKFLOW_FILE: str = os.path.join(
     os.path.dirname(__file__),
     "workflow-definitions",
@@ -206,3 +219,41 @@ def test_get_workflow_variables_for_duplicate_variables():
     assert len(names) == 2
     assert names[0] == "x"
     assert names[1] == "x"
+
+
+def test_get_required_variable_names_for_simnple_python_molprops_with_options():
+    # Arrange
+
+    # Act
+    rqd_variables = decoder.get_required_variable_names(
+        _SIMPLE_PYTHON_MOLPROPS_WITH_OPTIONS_WORKFLOW
+    )
+
+    # Assert
+    assert len(rqd_variables) == 2
+    assert "candidateMolecules" in rqd_variables
+    assert "rdkitPropertyValue" in rqd_variables
+
+
+@pytest.mark.skip(reason="The decoder does not currently handle options processing")
+def test_set_variables_from_options_for_step_for_simnple_python_molprops_with_options():
+    # Arrange
+    variables = {
+        "rdkitPropertyName": "propertyName",
+        "rdkitPropertyValue": "propertyValue",
+    }
+
+    # Act
+    new_variables, error = decoder.set_variables_from_options_for_step(
+        _SIMPLE_PYTHON_MOLPROPS_WITH_OPTIONS_WORKFLOW,
+        variables,
+        "step1",
+    )
+
+    # Assert
+    assert error is None
+    assert len(new_variables) == 4
+    assert "name" in new_variables
+    assert "value" in new_variables
+    assert new_variables["name"] == "propertyName"
+    assert new_variables["value"] == "propertyValue"

tests/test_workflow_engine_examples.py

@@ -314,7 +314,7 @@ def test_workflow_engine_simple_python_molprops(basic_engine):
 
 
 @pytest.mark.skip(reason="The engine does not currently create the required variables")
-def test_workflow_engine_simple_python_molprops_with_option(basic_engine):
+def test_workflow_engine_simple_python_molprops_with_options(basic_engine):
     # Arrange
     da, md = basic_engine
     # Make sure files that should be generated by the test

workflow/decoder.py

@@ -82,6 +82,42 @@ def get_variable_names(definition: dict[str, Any]) -> list[str]:
     return wf_variable_names
 
 
+def set_variables_from_options_for_step(
+    definition: dict[str, Any], variables: dict[str, Any], step_name: str
+) -> tuple[dict[str, Any], str | None]:
+    """Given a Workflow definition, an existing map of variables and values,
+    and a step name this function returns a new set of variables by adding
+    variables and values that are required for the step that have been defined in the
+    workflow's variables->options block.
+
+    As an example, the following option, which is used if the step name is 'step1',
+    expects 'rdkitPropertyName' to exist in the current set of variables,
+    and should be copied into the new set of variables using the key 'propertyName'
+    and value that is the same as the one provided in the original 'rdkitPropertyName': -
+
+        name: rdkitPropertyName
+        default: propertyName
+        as:
+        - option: propertyName
+          step: step1
+
+    And ... in the above example ... if the input variables map
+    is {"rdkitPropertyName": "rings"} then the output map would be
+    {"rdkitPropertyName": "rings", "propertyName": "rings"}
+
+    The function returns a new variable map, with and an optional error string on error.
+    """
+
+    assert isinstance(definition, dict)
+    assert isinstance(variables, dict)
+    assert step_name
+
+    new_variables: dict[str, Any] = variables.copy()
+
+    # Success...
+    return new_variables, None
+
+
 def get_required_variable_names(definition: dict[str, Any]) -> list[str]:
     """Given a Workflow definition this function returns all the names of the
     variables that are required to be defined when it is RUN - i.e.