Initial crack at implementing StepInputExpressionRequirements.

scatter-stepfrom-wf test case passes - need to get integer inputs to workflows fixed up in order to test scatter-stepfrom2-wf.

Author: jmchilton

lib/galaxy/managers/workflows.py

@@ -881,6 +881,7 @@ def __module_from_dict( self, trans, steps, steps_by_external_id, step_dict, **k
                 step_input.name = input_dict["name"]
                 step_input.merge_type = input_dict.get("merge_type", step_input.default_merge_type)
                 step_input.scatter_type = input_dict.get("scatter_type", step_input.default_scatter_type)
+                step_input.value_from = input_dict.get("value_from", None)
                 step.inputs.append(step_input)
 
         # Create the model class for the step

lib/galaxy/model/__init__.py

@@ -3231,6 +3231,10 @@ def __init__(
         if not populated:
             self.populated_state = DatasetCollection.populated_states.NEW
 
+    @property
+    def allow_implicit_mapping(self):
+        return self.collection_type != "record"
+
     @property
     def populated( self ):
         top_level_populated = self.populated_state == DatasetCollection.populated_states.OK

lib/galaxy/model/mapping.py

@@ -849,6 +849,8 @@
     Column( "name", Unicode( 255 ) ),
     Column( "merge_type", TEXT ),
     Column( "scatter_type", TEXT ),
+    Column( "value_from", JSONType ),
+    Column( "value_from_type", TEXT ),
     Column( "default_value", JSONType ) )
 
 

lib/galaxy/model/migrate/versions/0137_worklfow_step_input_definitions.py

@@ -20,6 +20,8 @@
     Column( "name", Unicode( 255 ) ),
     Column( "merge_type", TEXT ),
     Column( "scatter_type", TEXT ),
+    Column( "value_from", JSONType ),
+    Column( "value_from_type", TEXT ),
     Column( "default_value", JSONType ),
 )
 

lib/galaxy/tools/cwl/parser.py

@@ -43,6 +43,7 @@
     "ShellCommandRequirement",
     "ScatterFeatureRequirement",
     "SubworkflowFeatureRequirement",
+    "StepInputExpressionRequirement",
     "MultipleInputFeatureRequirement",
 ]
 
@@ -599,7 +600,7 @@ def cwl_input_to_galaxy_step(self, input, i):
             input_as_dict["type"] = "data_collection_input"
             input_as_dict["collection_type"] = "record"
         else:
-            raise NotImplementedError()
+            input_as_dict["type"] = "parameter_input"
 
         return input_as_dict
 
@@ -750,6 +751,9 @@ def to_dict(self):
             as_dict["merge_type"] = self._cwl_input["linkMerge"]
         if "scatterMethod" in self.step_proxy._step.tool:
             as_dict["scatter_method"] = self.step_proxy._step.tool["scatterMethod"]
+        if "valueFrom" in self._cwl_input:
+            # TODO: Add a table for expressions - mark the type as CWL 1.0 JavaScript.
+            as_dict["value_from"] = self._cwl_input["valueFrom"]
         return as_dict
 
 

lib/galaxy/workflow/modules.py

@@ -863,6 +863,64 @@ def decode_runtime_state( self, runtime_state ):
         else:
             raise ToolMissingException( "Tool %s missing. Cannot recover runtime state." % self.tool_id )
 
+    def evaluate_value_from_expressions(self, step, execution_state):
+        value_from_expressions = {}
+
+        for key, value in execution_state.inputs.items():
+            step_input = step.inputs_by_name.get(key, None)
+            if step_input and step_input.value_from is not None:
+                value_from_expressions[key] = step_input.value_from
+
+        if not value_from_expressions:
+            return
+
+        hda_references = []
+
+        def to_cwl(value):
+            if isinstance(value, model.HistoryDatasetAssociation):
+                hda_references.append(value)
+                return {
+                    "class": "File",
+                    "location": "step_input://%d" % len(hda_references),
+                }
+            elif hasattr(value, "collection"):
+                collection = value.collection
+                if collection.collection_type == "list":
+                    return map(to_cwl, collection.dataset_instances)
+                elif collection.collection_type == "record":
+                    rval = {}
+                    for element in collection.elements:
+                        rval[element.element_identifier] = to_cwl(element.element_object)
+                    return rval
+            else:
+                return value
+
+        def from_cwl(value):
+            if isinstance(value, dict) and "class" in value and "location" in value:
+                assert value["location"].startswith("step_input://")
+                return hda_references[int(value["location"][len("step_input://"):])-1]
+            elif isinstance(value, dict):
+                raise NotImplementedError()
+            else:
+                return value
+
+        step_state = {}
+        for key, value in execution_state.inputs.items():
+            step_state[key] = to_cwl(value)
+
+        for key, value_from in value_from_expressions.items():
+            from cwltool.expression import do_eval
+            as_cwl_value = do_eval(
+                value_from,
+                step_state,
+                [],
+                None,
+                None,
+                {},
+                context=step_state[key],
+            )
+            execution_state.inputs[key] = from_cwl(as_cwl_value)
+
     def execute( self, trans, progress, invocation, step ):
         tool = trans.app.toolbox.get_tool( step.tool_id, tool_version=step.tool_version, tool_hash=step.tool_hash )
         tool_state = step.state
@@ -925,6 +983,10 @@ def callback( input, prefixed_name, **kwargs ):
                 message = message_template % (tool.name, k.message)
                 raise exceptions.MessageException( message )
 
+            self.evaluate_value_from_expressions(
+                step, execution_state
+            )
+
             unmatched_input_connections = expected_replacement_keys - found_replacement_keys
             if unmatched_input_connections:
                 log.warn("Failed to use input connections for inputs [%s]" % unmatched_input_connections)
@@ -997,22 +1059,22 @@ def callback( input, prefixed_name, **kwargs ):
             is_data_param = isinstance( input, DataToolParameter )
             if is_data_param and not input.multiple:
                 data = progress.replacement_for_tool_input( step, input, prefixed_name )
-                if hasattr( data, "collection" ):
+                if hasattr( data, "collection" ) and data.collection.allow_implicit_mapping:
                     collections_to_match.add( prefixed_name, data )
                 return
             is_data_collection_param = isinstance( input, DataCollectionToolParameter )
             if is_data_collection_param and not input.multiple:
                 data = progress.replacement_for_tool_input( step, input, prefixed_name )
                 history_query = input._history_query( self.trans )
-                if hasattr( data, "collection" ):
+                if hasattr( data, "collection" ) and data.collection.allow_implicit_mapping:
                     subcollection_type_description = history_query.can_map_over( data )
                     if subcollection_type_description:
                         collections_to_match.add( prefixed_name, data, subcollection_type=subcollection_type_description.collection_type )
                 return
 
             data = progress.replacement_for_tool_input( step, input, prefixed_name )
             if data is not NO_REPLACEMENT:
-                if hasattr( data, "collection" ):
+                if hasattr( data, "collection" ) and data.collection.allow_implicit_mapping:
                     collections_to_match.add( prefixed_name, data )
 
         visit_input_values( tool.inputs, step.state.inputs, callback )

lib/galaxy/workflow/run.py

@@ -295,6 +295,7 @@ def replacement_for_tool_input( self, step, input, prefixed_name ):
                         replacement = replacement[ 0 ]
             else:
                 step_input = step.inputs_by_name.get(prefixed_name, None)
+
                 merge_type = model.WorkflowStepInput.default_merge_type
                 if step_input:
                     merge_type = step_input.merge_type
@@ -328,7 +329,6 @@ def replacement_for_tool_input( self, step, input, prefixed_name ):
 
                         inputs.append(input_from_connection)
 
-
                     if input.type == "data_collection":
                         # TODO: Implement more nested types here...
                         assert input.collection_types == ["list"], input.collection_types
@@ -384,7 +384,7 @@ def replacement_for_tool_input( self, step, input, prefixed_name ):
                         collection=collection,
                         history=self.workflow_invocation.history,
                     )
-                    return ephemeral_collection
+                    replacement = ephemeral_collection
 
         return replacement
 

test/unit/tools/test_cwl.py

@@ -200,6 +200,20 @@ def test_workflow_multiple_input_merge_flattened():
     assert input["merge_type"] == "merge_flattened"
 
 
+def test_workflow_step_value_from():
+    version = "v1.0"
+    proxy = workflow_proxy(_cwl_tool_path("%s/step-valuefrom-wf.cwl" % version))
+
+    galaxy_workflow_dict = proxy.to_dict()
+    assert len(galaxy_workflow_dict["steps"]) == 3
+
+    tool_step = galaxy_workflow_dict["steps"][2]
+    assert "inputs" in tool_step
+    inputs = tool_step["inputs"]
+    assert len(inputs) == 1
+    assert "value_from" in inputs[0]
+
+
 def test_load_proxy_simple():
     cat3 = _cwl_tool_path("draft3/cat3-tool.cwl")
     tool_source = get_tool_source(cat3)