Implementat a small subset of the Common Workflow Language tool format.

CWL_TODO

@@ -0,0 +1,8 @@
+ - Sniff the outputs so Galaxy can generate the correct dataytpes.
+ - More parameter types and constructs (e.g. tmap implementation).
+
+ - Explore Galaxy extensions to CWL - 
+    - Annotate datatypes. 
+    - Annotate parameter labels and help.
+    - Annotate tool help.
+    - Annotate requirements.

lib/galaxy/jobs/command_factory.py

@@ -1,7 +1,9 @@
 from os import getcwd
 from os import chmod
+from os import fdopen
 from os.path import join
 from os.path import abspath
+import tempfile
 
 CAPTURE_RETURN_CODE = "return_code=$?"
 YIELD_CAPTURED_CODE = 'sh -c "exit $return_code"'
@@ -114,6 +116,10 @@ def __handle_work_dir_outputs(commands_builder, job_wrapper, runner, remote_comm
     work_dir_outputs = runner.get_work_dir_outputs( job_wrapper, **work_dir_outputs_kwds )
     if work_dir_outputs:
         commands_builder.capture_return_code()
+        fd, fp = tempfile.mkstemp( suffix='.py', dir=job_wrapper.working_directory, prefix="relocate_dynamic_outputs_" )
+        metadata_script_file = abspath( fp )
+        fdopen( fd, 'w' ).write( 'from galaxy_ext.metadata.set_metadata import relocate_dynamic_outputs; relocate_dynamic_outputs()' )
+        commands_builder.append_command("python %s" % metadata_script_file)
         copy_commands = map(__copy_if_exists_command, work_dir_outputs)
         commands_builder.append_commands(copy_commands)
 

lib/galaxy/tools/__init__.py

@@ -2100,7 +2100,7 @@ def call_hook( self, hook_name, *args, **kwargs ):
     def exec_before_job( self, app, inp_data, out_data, param_dict={} ):
         pass
 
-    def exec_after_process( self, app, inp_data, out_data, param_dict, job=None ):
+    def exec_after_process( self, app, inp_data, out_data, param_dict, job=None, **kwds ):
         pass
 
     def job_failed( self, job_wrapper, message, exception=False ):
@@ -2836,7 +2836,7 @@ class SetMetadataTool( Tool ):
     tool_type = 'set_metadata'
     requires_setting_metadata = False
 
-    def exec_after_process( self, app, inp_data, out_data, param_dict, job=None ):
+    def exec_after_process( self, app, inp_data, out_data, param_dict, job=None, **kwds ):
         for name, dataset in inp_data.iteritems():
             external_metadata = JobExternalOutputMetadataWrapper( job )
             if external_metadata.external_metadata_set_successfully( dataset, app.model.context ):
@@ -2881,6 +2881,62 @@ class GenomeIndexTool( Tool ):
     tool_type = 'index_genome'
 
 
+class CwlTool( Tool ):
+    tool_type = 'cwl'
+
+    def exec_before_job( self, app, inp_data, out_data, param_dict=None ):
+        super( CwlTool, self ).exec_before_job( app, inp_data, out_data, param_dict=param_dict )
+        # Working directory on Galaxy server (instead of remote compute).
+        local_working_directory = param_dict["__local_working_directory__"]
+        log.info("exec_before_job for CWL tool")
+        input_json = {}
+
+        def simple_value(input, param_dict_value):
+            if isinstance(input, DataToolParameter):
+                return {"path": str(param_dict_value)}
+            elif isinstance(input, IntegerToolParameter):
+                return int(str(param_dict_value))
+            else:
+                return str(param_dict_value)
+
+        for input_name, input in self.inputs.iteritems():
+            if isinstance(input, Repeat):
+                only_input = input.inputs.values()[0]
+                array_value = []
+                for instance in param_dict[input_name]:
+                    array_value.append(simple_value(only_input, instance[input_name[:-len("_repeat")]]))
+                input_json[input_name[:-len("_repeat")]] = array_value
+            else:
+                input_json[input_name] = simple_value(input, param_dict[input_name])
+
+        log.info("input_json is %s" % input_json)
+        input_json["allocatedResources"] = {
+            "cpu": "$GALAXY_SLOTS",
+        }
+        if param_dict is None:
+            raise Exception("Internal error - param_dict is empty.")
+        cwl_job_proxy = self._cwl_tool_proxy.job_proxy(input_json, local_working_directory)
+        # Write representation to disk that can be reloaded at runtime
+        # and outputs collected before Galaxy metadata is gathered.
+        cwl_job_proxy.save_job()
+        cwl_job = cwl_job_proxy.cwl_job()
+        command_line = " ".join(cwl_job.command_line)
+        if cwl_job.stdin:
+            command_line += '< "' + cwl_job.stdin + '"'
+        if cwl_job.stdout:
+            command_line += '> "' + cwl_job.stdout + '"'
+        # TODO: handle generatefiles
+        param_dict["__cwl_command"] = command_line
+        log.info("CwlTool.exec_before_job() generated command_line %s" % command_line)
+
+    def parse( self, tool_source, guid=None ):
+        super( CwlTool, self ).parse( tool_source, guid=guid )
+        cwl_tool_proxy = getattr( tool_source, 'tool_proxy', None )
+        if cwl_tool_proxy is None:
+            raise Exception("CwlTool.parse() called on tool source not defining a proxy object to underlying CWL tool.")
+        self._cwl_tool_proxy = cwl_tool_proxy
+
+
 class DataManagerTool( OutputParameterJSONTool ):
     tool_type = 'manage_data'
     default_tool_action = DataManagerToolAction
@@ -2952,7 +3008,7 @@ def allow_user_access( self, user ):
 tool_types = {}
 for tool_class in [ Tool, SetMetadataTool, OutputParameterJSONTool,
                     DataManagerTool, DataSourceTool, AsyncDataSourceTool,
-                    DataDestinationTool ]:
+                    DataDestinationTool, CwlTool ]:
     tool_types[ tool_class.tool_type ] = tool_class
 
 

lib/galaxy/tools/cwl/__init__.py

@@ -0,0 +1,7 @@
+from .parser import tool_proxy
+from .runtime_actions import handle_outputs
+
+__all__ = [
+    'tool_proxy',
+    'handle_outputs',
+]

lib/galaxy/tools/cwl/cwltool_deps.py

@@ -0,0 +1,55 @@
+""" This module contains logic for dealing with cwltool as an optional
+dependency for Galaxy and/or applications which use Galaxy as a library.
+"""
+
+try:
+    from galaxy import eggs
+    eggs.require("requests")
+except Exception:
+    pass
+
+try:
+    import requests
+except ImportError:
+    requests = None
+
+try:
+    from cwltool import (
+        ref_resolver,
+        draft1tool,
+        draft2tool,
+    )
+except ImportError as e:
+    ref_resolver = None
+    draft1tool = None
+    draft2tool = None
+
+try:
+    import jsonschema
+except ImportError:
+    jsonschema = None
+
+try:
+    import avro
+except ImportError:
+    avro = None
+
+
+def ensure_cwltool_available():
+    if ref_resolver is None:
+        message = "This feature requires cwltool and dependencies to be available, they are not."
+        if avro is None:
+            message += " Library avro unavailable."
+        if jsonschema is None:
+            message += " Library jsonschema unavailable."
+        if requests is None:
+            message += " Library requests unavailable."
+        raise ImportError(message)
+
+
+__all__ = [
+    'ref_resolver',
+    'draft1tool',
+    'draft2tool',
+    'ensure_cwltool_available',
+]