Implement CWL valueFrom for multiple inputs and non-File inputs.

Author: jmchilton

lib/galaxy/tools/__init__.py

@@ -2368,9 +2368,9 @@ def exec_before_job( self, app, inp_data, out_data, param_dict=None ):
 
         command_line = " ".join([shellescape.quote(arg) if needs_shell_quoting(arg) else arg for arg in cwl_command_line])
         if cwl_stdin:
-            command_line += '< "' + cwl_stdin + '"'
+            command_line += ' < "' + cwl_stdin + '"'
         if cwl_stdout:
-            command_line += '> "' + cwl_stdout + '"'
+            command_line += ' > "' + cwl_stdout + '"'
         cwl_job_state = {
             'args': cwl_command_line,
             'stdin': cwl_stdin,

lib/galaxy/tools/cwl/parser.py

@@ -734,16 +734,30 @@ def __init__(self, step_proxy, cwl_input):
         self._cwl_input = cwl_input
         self.step_proxy = step_proxy
         cwl_input_id = cwl_input["id"]
-        cwl_source_id = cwl_input["source"]
-        step_name, input_name = split_step_references(
-            cwl_input_id,
-            multiple=False,
-            workflow_id=step_proxy.cwl_workflow_id
-        )
+        cwl_source_id = cwl_input.get("source", None)
+        if cwl_source_id is None:
+            if "valueFrom" not in cwl_input:
+                raise NotImplementedError("Workflow step input must define a source or a valueFrom value.")
+
+        if cwl_source_id is not None:
+            step_name, input_name = split_step_references(
+                cwl_input_id,
+                multiple=False,
+                workflow_id=step_proxy.cwl_workflow_id
+            )
+            self.step_name = step_name
+            self.input_name = input_name
+        else:
+            self.step_name = None
+            self.input_name = None
+
         self.cwl_input_id = cwl_input_id
         self.cwl_source_id = cwl_source_id
-        self.step_name = step_name
-        self.input_name = input_name
+
+        scatter = False
+        if self.input_name in listify(step_proxy._step.tool.get("scatter", [])):
+            scatter = True
+        self.scatter = scatter
 
     def to_dict(self):
         as_dict = {
@@ -752,7 +766,9 @@ def to_dict(self):
         if "linkMerge" in self._cwl_input:
             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"]
+            as_dict["scatter_type"] = self.step_proxy._step.tool.get("scatterMethod", "dotproduct")
+        else:
+            as_dict["scatter_type"] = "dotproduct" if self.scatter else "disabled"
         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"]

lib/galaxy/workflow/modules.py

@@ -879,10 +879,14 @@ def evaluate_value_from_expressions(self, step, execution_state):
         def to_cwl(value):
             if isinstance(value, model.HistoryDatasetAssociation):
                 hda_references.append(value)
-                return {
-                    "class": "File",
-                    "location": "step_input://%d" % len(hda_references),
-                }
+                if value.ext == "expression.json":
+                    with open( value.file_name, "r" ) as f:
+                        return loads(f.read())
+                else:
+                    return {
+                        "class": "File",
+                        "location": "step_input://%d" % len(hda_references),
+                    }
             elif hasattr(value, "collection"):
                 collection = value.collection
                 if collection.collection_type == "list":
@@ -983,14 +987,20 @@ 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
-            )
+            extra_step_state = {}
+            for step_input in step.inputs:
+                if step_input.name not in execution_state.inputs:
+                    value = progress.replacment_for_step_input( step, step_input )
+                    extra_step_state[step_input.name] = value
 
             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)
 
+            self.evaluate_value_from_expressions(
+                step, execution_state
+            )
+
             param_combinations.append( execution_state.inputs )
 
         try:
@@ -1050,11 +1060,15 @@ def _find_collections_to_match( self, tool, progress, step ):
         collections_to_match = matching.CollectionsToMatch()
 
         def callback( input, prefixed_name, **kwargs ):
-            step_input = step.inputs_by_name[ prefixed_name ]
+            step_input = step.inputs_by_name.get( prefixed_name, None )
             scatter_type = "dotproduct"
             if step_input:
                 scatter_type = step_input.scatter_type
-            assert scatter_type == "dotproduct"
+            assert scatter_type in ["dotproduct", "disabled"]
+
+            log.info("SCATTER_TYPE IS %s" % scatter_type)
+            if scatter_type == "disabled":
+                return
 
             is_data_param = isinstance( input, DataToolParameter )
             if is_data_param and not input.multiple:

test/unit/tools/test_cwl.py

@@ -214,6 +214,20 @@ def test_workflow_step_value_from():
     assert "value_from" in inputs[0]
 
 
+def test_workflow_input_without_source():
+    version = "v1.0"
+    proxy = workflow_proxy(_cwl_tool_path("%s/step-valuefrom3-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) == 3
+    assert inputs[2].get("value_from")
+
+
 def test_load_proxy_simple():
     cat3 = _cwl_tool_path("draft3/cat3-tool.cwl")
     tool_source = get_tool_source(cat3)