WIP: CWL record inputs and outputs as well as shell tools.

Author: jmchilton

lib/galaxy/tools/cwl/parser.py

@@ -39,6 +39,7 @@
     "DockerRequirement",
     "EnvVarRequirement",
     "InlineJavascriptRequirement",
+    "ShellCommandRequirement",
 ]
 
 
@@ -703,6 +704,7 @@ def _simple_field_to_input_type_kwds(field, field_type=None):
         "double": INPUT_TYPE.INTEGER,
         "string": INPUT_TYPE.TEXT,
         "boolean": INPUT_TYPE.BOOLEAN,
+        "record": INPUT_TYPE.DATA_COLLECTON,
     }
 
     if field_type is None:
@@ -766,6 +768,7 @@ def _simple_field_to_output(field):
     BOOLEAN="boolean",
     SELECT="select",
     CONDITIONAL="conditional",
+    DATA_COLLECTON="data_collection",
 )
 
 
@@ -848,6 +851,9 @@ def to_dict(self, itemwise=True):
                 as_dict["value"] = "0"
             if self.input_type == INPUT_TYPE.FLOAT:
                 as_dict["value"] = "0.0"
+            elif self.input_type == INPUT_TYPE.DATA_COLLECTON:
+                as_dict["collection_type"] = "record"
+
         return as_dict
 
 
@@ -857,13 +863,15 @@ def to_dict(self, itemwise=True):
 )
 
 
+# TODO: Different subclasses - this is representing different types of things.
 class OutputInstance(object):
 
-    def __init__(self, name, output_data_type, output_type, path=None):
+    def __init__(self, name, output_data_type, output_type, path=None, fields=None):
         self.name = name
         self.output_data_type = output_data_type
         self.output_type = output_type
         self.path = path
+        self.fields = fields
 
 
 __all__ = (

lib/galaxy/tools/cwl/representation.py

@@ -19,7 +19,8 @@
     'float': 'float',
     'data': 'File',
     'boolean': 'boolean',
-    'text': 'text'
+    'text': 'text',
+    'data_collection': 'data_collection',
 }
 
 
@@ -35,15 +36,8 @@ def to_cwl_job(tool, param_dict, local_working_directory):
     def simple_value(input, param_dict_value, cwl_type=None):
         # Hmm... cwl_type isn't really the cwl type in every case,
         # like in the case of json for instance.
-        if cwl_type is None:
-            input_type = input.type
-            cwl_type = GALAXY_TO_CWL_TYPES[input_type]
 
-        if cwl_type == "null":
-            assert param_dict_value is None
-            return None
-        if cwl_type == "File":
-            dataset_wrapper = param_dict_value
+        def dataset_wrapper_to_file_json(dataset_wrapper):
             extra_files_path = dataset_wrapper.extra_files_path
             secondary_files_path = os.path.join(extra_files_path, "__secondary_files__")
             path = str(dataset_wrapper)
@@ -59,6 +53,17 @@ def simple_value(input, param_dict_value, cwl_type=None):
 
             return {"location": path,
                     "class": "File"}
+
+        if cwl_type is None:
+            input_type = input.type
+            cwl_type = GALAXY_TO_CWL_TYPES[input_type]
+
+        if cwl_type == "null":
+            assert param_dict_value is None
+            return None
+        if cwl_type == "File":
+            dataset_wrapper = param_dict_value
+            return dataset_wrapper_to_file_json(dataset_wrapper)
         elif cwl_type == "integer":
             return int(str(param_dict_value))
         elif cwl_type == "long":
@@ -75,6 +80,11 @@ def simple_value(input, param_dict_value, cwl_type=None):
             raw_value = param_dict_value.value
             log.info("raw_value is %s (%s)" % (raw_value, type(raw_value)))
             return json.loads(raw_value)
+        elif cwl_type == "data_collection":
+            rval = dict()  # TODO: THIS NEEDS TO BE ORDERED BUT odict not json serializable!
+            for key, value in param_dict_value.items():
+                rval[key] = dataset_wrapper_to_file_json(value)
+            return rval
         else:
             return str(param_dict_value)
 

lib/galaxy/tools/cwl/runtime_actions.py

@@ -24,30 +24,41 @@ def handle_outputs(job_directory=None):
     job_proxy = load_job_proxy(job_directory, strict_cwl_validation=False)
     tool_working_directory = os.path.join(job_directory, "working")
     outputs = job_proxy.collect_outputs(tool_working_directory)
+
+    def handle_output_location(output, target_path):
+        output_path = ref_resolver.uri_file_path(output["location"])
+        if output["class"] != "File":
+            open("galaxy.json", "w").write(json.dump({
+                "dataset_id": job_proxy.output_id(output_name),
+                "type": "dataset",
+                "ext": "expression.json",
+            }))
+        shutil.move(output_path, target_path)
+        for secondary_file in output.get("secondaryFiles", []):
+            # TODO: handle nested files...
+            secondary_file_path = ref_resolver.uri_file_path(secondary_file["location"])
+            assert secondary_file_path.startswith(output_path)
+            secondary_file_name = secondary_file_path[len(output_path):]
+            secondary_files_dir = job_proxy.output_secondary_files_dir(
+                output_name, create=True
+            )
+            extra_target = os.path.join(secondary_files_dir, secondary_file_name)
+            shutil.move(
+                secondary_file_path,
+                extra_target,
+            )
+
     for output_name, output in outputs.items():
-        target_path = job_proxy.output_path(output_name)
         if isinstance(output, dict) and "location" in output:
-            output_path = ref_resolver.uri_file_path(output["location"])
-            if output["class"] != "File":
-                open("galaxy.json", "w").write(json.dump({
-                    "dataset_id": job_proxy.output_id(output_name),
-                    "type": "dataset",
-                    "ext": "expression.json",
-                }))
-            shutil.move(output_path, target_path)
-            for secondary_file in output.get("secondaryFiles", []):
-                # TODO: handle nested files...
-                secondary_file_path = ref_resolver.uri_file_path(secondary_file["location"])
-                assert secondary_file_path.startswith(output_path)
-                secondary_file_name = secondary_file_path[len(output_path):]
-                secondary_files_dir = job_proxy.output_secondary_files_dir(
-                    output_name, create=True
-                )
-                extra_target = os.path.join(secondary_files_dir, secondary_file_name)
-                shutil.move(
-                    secondary_file_path,
-                    extra_target,
-                )
+            target_path = job_proxy.output_path(output_name)
+            handle_output_location(output, target_path)
+        elif isinstance(output, dict):
+            prefix = "%s|__part__|" % output_name
+            for record_key, record_value in output.items():
+                record_value_output_key = "%s%s" % (prefix, record_key)
+                target_path = job_proxy.output_path(record_value_output_key)
+
+                handle_output_location(record_value, target_path)
         else:
             with open(target_path, "w") as f:
                 f.write(json.dumps(output))

lib/galaxy/tools/parser/cwl.py

@@ -9,6 +9,8 @@
 from .interface import ToolSource
 from .interface import ToolStdioExitCode
 from .output_actions import ToolOutputActionGroup
+from .output_objects import ToolOutputCollection
+from .output_objects import ToolOutputCollectionStructure
 from .output_objects import ToolOutput
 from .yaml import YamlInputSource
 
@@ -106,15 +108,28 @@ def parse_input_pages(self):
     def parse_outputs(self, tool):
         output_instances = self.tool_proxy.output_instances()
         outputs = odict()
+        output_collections = odict()
         output_defs = []
         for output_instance in output_instances:
             output_defs.append(self._parse_output(tool, output_instance))
+
         # TODO: parse outputs collections
         for output_def in output_defs:
-            outputs[output_def.name] = output_def
-        return outputs, odict()
+            if isinstance(output_def, ToolOutput):
+                outputs[output_def.name] = output_def
+            else:
+                outputs[output_def.name] = output_def
+                output_collections[output_def.name] = output_def
+        return outputs, output_collections
 
     def _parse_output(self, tool, output_instance):
+        output_type = output_instance.output_data_type
+        if isinstance(output_type, dict) and output_type.get("type") == "record":
+            return self._parse_output_record(tool, output_instance)
+        else:
+            return self._parse_output_data(tool, output_instance)
+
+    def _parse_output_data(self, tool, output_instance):
         name = output_instance.name
         # TODO: handle filters, actions, change_format
         output = ToolOutput( name )
@@ -135,6 +150,20 @@ def _parse_output(self, tool, output_instance):
         output.actions = ToolOutputActionGroup( output, None )
         return output
 
+    def _parse_output_record(self, tool, output_instance):
+        name = output_instance.name
+        # TODO: clean output bindings and other non-structure information
+        # from this.
+        fields = output_instance.output_data_type.get("fields")
+        output_collection = ToolOutputCollection(
+            name,
+            ToolOutputCollectionStructure(
+                collection_type="record",
+                fields=fields,
+            ),
+        )
+        return output_collection
+
     def parse_requirements_and_containers(self):
         containers = []
         docker_identifier = self.tool_proxy.docker_identifier()

lib/galaxy/tools/parser/output_objects.py

@@ -201,6 +201,7 @@ def __init__(
         collection_type_source=None,
         structured_like=None,
         dataset_collector_descriptions=None,
+        fields=None,
     ):
         self.collection_type = collection_type
         self.collection_type_source = collection_type_source
@@ -213,6 +214,10 @@ def __init__(
             raise ValueError( "Output collection types must be specify type of structured_like" )
         if dataset_collector_descriptions and structured_like:
             raise ValueError( "Cannot specify dynamic structure (discovered_datasets) and structured_like attribute." )
+        if collection_type == "record" and fields is None:
+            raise ValueError( "If record outputs are defined, fields must be defined as well." )
+        if fields is not None and collection_type != "record":
+            raise ValueError( "If fields are specified for outputs, the collection type must be record." )
         self.dynamic = dataset_collector_descriptions is not None
 
     def collection_prototype( self, inputs, type_registry ):

lib/galaxy/tools/wrappers.py

@@ -420,6 +420,9 @@ def keys( self ):
             return []
         return self.__element_instances.keys()
 
+    def items( self ):
+        return self.__element_instances.items()
+
     @property
     def is_collection( self ):
         return True

test/api/test_tools_cwl.py

@@ -316,6 +316,18 @@ def test_parse_int_tool( self ):
         output1 = self.dataset_populator.get_history_dataset_details( run_object.history_id, hid=2 )
         self.assertEquals(output1["extension"], "expression.json")
 
+    @skip_without_tool( "record-output" )
+    def test_record_output( self ):
+        run_object = self.dataset_populator.run_cwl_tool( "record-output", "test/functional/tools/cwl_tools/v1.0/record-output-job.json")
+        result_record = run_object.output_collection(0)
+        assert result_record["collection_type"] == "record"
+        record_elements = result_record["elements"]
+        first_element = record_elements[0]
+        assert first_element["element_identifier"] == "ofoo"
+        first_hda = first_element["object"]
+        output1_content = self.dataset_populator.get_history_dataset_content( run_object.history_id, hid=first_hda["hid"] )
+        assert "Call me Ishmael." in output1_content, "Expected contents of whale.txt, got [%s]" % output1_content
+
     # def test_dynamic_tool_execution( self ):
     #     workflow_tool_json = {
     #         'inputs': [{'inputBinding': {}, 'type': 'File', 'id': 'file:///home/john/workspace/galaxy/test/unit/tools/cwl_tools/draft3/count-lines2-wf.cwl#step1/wc/wc_file1'}],