Infrastructure for parsing CWL tools.

This commit is contained in:
John Chilton
2016-01-03 13:10:08 +00:00
parent a9d4783f2c
commit 2054f538be
7 changed files with 736 additions and 0 deletions
+18
View File
@@ -0,0 +1,18 @@
from .parser import tool_proxy
from .runtime_actions import handle_outputs
from .representation import to_cwl_job, to_galaxy_parameters
from .cwltool_deps import (
needs_shell_quoting,
shellescape,
)
__all__ = [
'tool_proxy',
'handle_outputs',
'to_cwl_job',
'to_galaxy_parameters',
'needs_shell_quoting',
'shellescape',
]
+48
View File
@@ -0,0 +1,48 @@
""" This module contains logic for dealing with cwltool as an optional
dependency for Galaxy and/or applications which use Galaxy as a library.
"""
try:
import requests
except ImportError:
requests = None
try:
from cwltool import (
main,
workflow,
job,
)
except ImportError:
main = None
workflow = None
job = None
try:
import shellescape
except ImportError:
shellescape = None
if job is not None:
needs_shell_quoting = job.needs_shell_quoting
else:
needs_shell_quoting = None
def ensure_cwltool_available():
if main is None or workflow is None or shellescape is None:
message = "This feature requires cwltool and dependencies to be available, they are not."
if requests is None:
message += " Library 'requests' unavailable."
if shellescape is None:
message += " Library 'shellescape' unavailable."
raise ImportError(message)
__all__ = [
'main',
'workflow',
'ensure_cwltool_available',
'shellescape',
'needs_shell_quoting',
]
+392
View File
@@ -0,0 +1,392 @@
""" This module provides proxy objects around objects from the common
workflow language reference implementation library cwltool. These proxies
adapt cwltool to Galaxy features and abstract the library away from the rest
of the framework.
"""
from __future__ import absolute_import
from abc import ABCMeta, abstractmethod
from galaxy.util.odict import odict
import json
import os
from .cwltool_deps import (
main,
workflow,
ensure_cwltool_available,
)
from galaxy.util.bunch import Bunch
JOB_JSON_FILE = ".cwl_job.json"
def tool_proxy(tool_path):
""" Provide a proxy object to cwltool data structures to just
grab relevant data.
"""
ensure_cwltool_available()
tool = to_cwl_tool_object(tool_path)
return tool
def load_job_proxy(job_directory):
ensure_cwltool_available()
job_objects_path = os.path.join(job_directory, JOB_JSON_FILE)
job_objects = json.load(open(job_objects_path, "r"))
tool_path = job_objects["tool_path"]
job_inputs = job_objects["job_inputs"]
cwl_tool = tool_proxy(tool_path)
cwl_job = cwl_tool.job_proxy(job_inputs, job_directory=job_directory)
return cwl_job
def to_cwl_tool_object(tool_path):
if workflow is None:
raise Exception("Using CWL tools requires cwltool module.")
proxy_class = None
cwl_tool = None
make_tool = workflow.defaultMakeTool
cwl_tool = main.load_tool(tool_path, False, False, make_tool, False)
proxy_class = Draft2ToolProxy
if proxy_class is None:
raise Exception("Unsupported CWL object encountered.")
proxy = proxy_class(cwl_tool, tool_path)
return proxy
class ToolProxy( object ):
__metaclass__ = ABCMeta
def __init__(self, tool, tool_path):
self._tool = tool
self._tool_path = tool_path
def job_proxy(self, input_dict, job_directory="."):
""" Build a cwltool.job.Job describing computation using a input_json
Galaxy will generate mapping the Galaxy description of the inputs into
a cwltool compatible variant.
"""
return JobProxy(self, input_dict, job_directory=job_directory)
@abstractmethod
def input_instances(self):
""" Return InputInstance objects describing mapping to Galaxy inputs. """
@abstractmethod
def output_instances(self):
""" Return OutputInstance objects describing mapping to Galaxy inputs. """
@abstractmethod
def docker_identifier(self):
""" Return docker identifier for embedding in tool description. """
@abstractmethod
def description(self):
""" Return description to tool. """
class Draft2ToolProxy(ToolProxy):
def description(self):
# Feels like I should be getting some abstract namespaced thing
# not actually description, is this correct?
return self._tool.tool.get('description')
def input_instances(self):
return self._find_inputs(self._tool.inputs_record_schema)
def _find_inputs(self, schema):
schema_type = schema["type"]
if isinstance(schema_type, list):
raise Exception("Union types not yet implemented.")
elif isinstance(schema_type, dict):
return self._find_inputs(schema_type)
else:
if schema_type in self._tool.schemaDefs:
schema = self._tool.schemaDefs[schema_type]
if schema["type"] == "record":
return map(_simple_field_to_input, schema["fields"])
def output_instances(self):
outputs_schema = self._tool.outputs_record_schema
return self._find_outputs(outputs_schema)
def _find_outputs(self, schema):
rval = []
if not rval and schema["type"] == "record":
for output in schema["fields"]:
# TODO: Handle non-files differently? Make them
# JSON maybe?.
# output_type = output.get("type", None)
# if output_type != "File":
# template = "Unhandled output type [%s] encountered."
# raise Exception(template % output_type)
rval.append(_simple_field_to_output(output))
return rval
def docker_identifier(self):
tool = self._tool.tool
reqs_and_hints = tool.get("requirements", []) + tool.get("hints", [])
for hint in reqs_and_hints:
if hint["class"] == "DockerRequirement":
if "dockerImageId" in hint:
return hint["dockerImageId"]
else:
return hint["dockerPull"]
return None
class JobProxy(object):
def __init__(self, tool_proxy, input_dict, job_directory):
self._tool_proxy = tool_proxy
self._input_dict = input_dict
self._job_directory = job_directory
def cwl_job(self):
return self._tool_proxy._tool.job(
self._input_dict,
self._job_directory,
self._output_callback,
use_container=False
).next()
def _output_callback(self, out):
pass
def save_job(self):
job_file = JobProxy._job_file(self._job_directory)
job_objects = {
"tool_path": os.path.abspath(self._tool_proxy._tool_path),
"job_inputs": self._input_dict,
}
json.dump(job_objects, open(job_file, "w"))
# @staticmethod
# def load_job(tool_proxy, job_directory):
# job_file = JobProxy._job_file(job_directory)
# input_dict = json.load(open(job_file, "r"))
# return JobProxy(tool_proxy, input_dict, job_directory)
@staticmethod
def _job_file(job_directory):
return os.path.join(job_directory, JOB_JSON_FILE)
def _simple_field_union(field):
field_type = _field_to_field_type(field)
non_null_type = None
if len(field_type) == 2:
if "null" == field_type[0]:
non_null_type = field_type[1]
elif "null" == field_type[1]:
non_null_type = field_type[1]
if non_null_type is None:
raise Exception("General union types not yet implemented (simple).")
name, label, description = _field_metadata(field)
case_name = "_cwl__type_"
case_label = "Specify %s" % label
options = [
{"value": "null", "label": "No", "selected": True},
{"value": "%s" % non_null_type, "label": "Yes", "selected": False},
]
case_input = SelectInputInstance(
name=case_name,
label=case_label,
description=False,
options=options,
)
non_null_type_kwds = _simple_field_to_input_type_kwds(field, field_type=non_null_type)
non_null_name = "_cwl__value_"
non_null_label = label
non_null_description = description
non_null_input = InputInstance(
non_null_name,
non_null_label,
non_null_description,
**non_null_type_kwds
)
options = [
("null", []),
(non_null_type, [non_null_input]),
]
return ConditionalInstance(name, case_input, options)
def _simple_field_to_input(field):
field_type = _field_to_field_type(field)
if isinstance(field_type, list):
# Length must be greater than 1...
return _simple_field_union(field)
name, label, description = _field_metadata(field)
type_kwds = _simple_field_to_input_type_kwds(field)
return InputInstance(name, label, description, **type_kwds)
def _simple_field_to_input_type_kwds(field, field_type=None):
simple_map_type_map = {
"File": INPUT_TYPE.DATA,
"int": INPUT_TYPE.INTEGER,
"string": INPUT_TYPE.TEXT,
"boolean": INPUT_TYPE.BOOLEAN,
}
if field_type is None:
field_type = _field_to_field_type(field)
if field_type in simple_map_type_map.keys():
input_type = simple_map_type_map[field_type]
return {"input_type": input_type, "array": False}
elif field_type == "array":
if isinstance(field["type"], dict):
array_type = field["type"]["items"]
else:
array_type = field["items"]
if array_type in simple_map_type_map.keys():
input_type = simple_map_type_map[array_type]
return {"input_type": input_type, "array": True}
else:
raise Exception("Unhandled simple field type encountered - [%s]." % field_type)
def _field_to_field_type(field):
field_type = field["type"]
if isinstance(field_type, dict):
field_type = field_type["type"]
if isinstance(field_type, list):
field_type_length = len(field_type)
if field_type_length == 0:
raise Exception("Zero-length type list encountered, invalid CWL?")
elif len(field_type) == 1:
field_type = field_type[0]
return field_type
def _field_metadata(field):
name = field["name"]
label = field.get("label", None)
description = field.get("description", None)
return name, label, description
def _simple_field_to_output(field):
name = field["name"]
output_instance = OutputInstance(name, output_type=OUTPUT_TYPE.GLOB)
return output_instance
INPUT_TYPE = Bunch(
DATA="data",
INTEGER="integer",
TEXT="text",
BOOLEAN="boolean",
SELECT="select",
CONDITIONAL="conditional",
)
class ConditionalInstance(object):
def __init__(self, name, case, whens):
self.input_type = INPUT_TYPE.CONDITIONAL
self.name = name
self.case = case
self.whens = whens
def to_dict(self):
as_dict = dict(
name=self.name,
type=INPUT_TYPE.CONDITIONAL,
test=self.case.to_dict(),
when=odict(),
)
for value, block in self.whens:
as_dict["when"][value] = map(lambda i: i.to_dict(), block)
return as_dict
class SelectInputInstance(object):
def __init__(self, name, label, description, options):
self.input_type = INPUT_TYPE.SELECT
self.name = name
self.label = label
self.description = description
self.options = options
def to_dict(self):
# TODO: serialize options...
as_dict = dict(
name=self.name,
label=self.label or self.name,
help=self.description,
type=self.input_type,
options=self.options,
)
return as_dict
class InputInstance(object):
def __init__(self, name, label, description, input_type, array=False):
self.input_type = input_type
self.name = name
self.label = label
self.description = description
self.required = True
self.array = array
def to_dict(self, itemwise=True):
if itemwise and self.array:
as_dict = dict(
type="repeat",
name="%s_repeat" % self.name,
title="%s" % self.name,
blocks=[
self.to_dict(itemwise=False)
]
)
else:
as_dict = dict(
name=self.name,
label=self.label or self.name,
help=self.description,
type=self.input_type,
optional=not self.required,
)
if self.input_type == INPUT_TYPE.INTEGER:
as_dict["value"] = "0"
return as_dict
OUTPUT_TYPE = Bunch(
GLOB="glob",
STDOUT="stdout",
)
class OutputInstance(object):
def __init__(self, name, output_type, path=None):
self.name = name
self.output_type = output_type
self.path = path
__all__ = [
'tool_proxy',
'load_job_proxy',
]
+100
View File
@@ -0,0 +1,100 @@
""" This module is responsible for converting between Galaxy's tool
input description and the CWL description for a job json. """
import logging
from galaxy.util import string_as_bool
log = logging.getLogger(__name__)
NOT_PRESENT = object()
def to_cwl_job(tool, param_dict):
""" tool is Galaxy's representation of the tool and param_dict is the
parameter dictionary with wrapped values.
"""
inputs = tool.inputs
input_json = {}
def simple_value(input, param_dict_value):
if input.type == "data":
return {"path": str(param_dict_value), "class": "File"}
elif input.type == "integer":
return int(str(param_dict_value))
elif input.type == "boolean":
return string_as_bool(param_dict_value)
else:
return str(param_dict_value)
for input_name, input in inputs.iteritems():
if input.type == "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
elif input.type == "conditional":
assert input_name in param_dict, "No value for %s in %s" % (input_name, param_dict)
current_case = param_dict[input_name]["_cwl__type_"]
if current_case != "null":
# TODO: make sure trans is not needed here...
case_index = input.get_current_case( current_case, None )
case_input = input.cases[ case_index ].inputs[ "_cwl__value_"]
case_value = param_dict[input_name]["_cwl__value_"]
input_json[input_name] = simple_value(case_input, case_value)
else:
input_json[input_name] = simple_value(input, param_dict[input_name])
input_json["allocatedResources"] = {
"cpu": "$GALAXY_SLOTS",
}
return input_json
def to_galaxy_parameters(tool, as_dict):
""" Tool is Galaxy's representation of the tool and as_dict is a Galaxified
representation of the input json (no paths, HDA references for instance).
"""
inputs = tool.inputs
galaxy_request = {}
def from_simple_value(input, param_dict_value):
return param_dict_value
for input_name, input in inputs.iteritems():
as_dict_value = as_dict.get(input_name, NOT_PRESENT)
if input.type == "repeat":
if input_name not in as_dict:
continue
only_input = input.inputs.values()[0]
for index, value in enumerate(as_dict_value):
key = "%s_repeat_0|%s" % (input_name, only_input.name)
galaxy_value = from_simple_value(only_input, value)
galaxy_request[key] = galaxy_value
elif input.type == "conditional":
# TODO: less crazy handling of defaults...
if as_dict_value is NOT_PRESENT:
cwl_type = "null"
elif isinstance(as_dict_value, bool):
cwl_type = "boolean"
elif isinstance(as_dict_value, int):
cwl_type = "integer"
else:
cwl_type = "string"
galaxy_request["%s|_cwl__type_" % input_name] = cwl_type
if cwl_type != "null":
current_case_index = input.get_current_case(cwl_type, None)
current_case_input = input.cases[ current_case_index ].inputs[ "_cwl__value_" ]
galaxy_value = from_simple_value(current_case_input, as_dict_value)
galaxy_request["%s|_cwl__value_" % input_name] = galaxy_value
elif as_dict_value is NOT_PRESENT:
continue
else:
galaxy_value = from_simple_value(input, as_dict_value)
galaxy_request[input_name] = galaxy_value
log.info("Converted galaxy_request is %s" % galaxy_request)
return galaxy_request
+29
View File
@@ -0,0 +1,29 @@
import os
import shutil
from .parser import (
JOB_JSON_FILE,
load_job_proxy,
)
def handle_outputs(job_directory=None):
# Relocate dynamically collected files to pre-determined locations
# registered with ToolOutput objects via from_work_dir handling.
if job_directory is None:
job_directory = os.getcwd()
cwl_job_file = os.path.join(job_directory, JOB_JSON_FILE)
if not os.path.exists(cwl_job_file):
# Not a CWL job, just continue
return
job_proxy = load_job_proxy(job_directory)
cwl_job = job_proxy.cwl_job()
outputs = cwl_job.collect_outputs(job_directory)
for output_name, output in outputs.iteritems():
target_path = os.path.join(job_directory, "__cwl_output_%s" % output_name)
shutil.move(output["path"], target_path)
__all__ = [
'handle_outputs',
]
+145
View File
@@ -0,0 +1,145 @@
import logging
import os
from .interface import ToolSource
from .interface import PagesSource
from .interface import PageSource
from .interface import ToolStdioExitCode
from .yaml import YamlInputSource
from .output_actions import ToolOutputActionGroup
from .output_objects import ToolOutput
from galaxy.tools.deps import requirements
from galaxy.tools.cwl import tool_proxy
from galaxy.util.odict import odict
log = logging.getLogger(__name__)
class CwlToolSource(ToolSource):
def __init__(self, tool_file):
self._cwl_tool_file = tool_file
self._id, _ = os.path.splitext(os.path.basename(tool_file))
self._tool_proxy = tool_proxy(tool_file)
@property
def tool_proxy(self):
return self._tool_proxy
def parse_tool_type(self):
return 'cwl'
def parse_id(self):
log.warn("TOOL ID is %s" % self._id)
return self._id
def parse_name(self):
return self._id
def parse_command(self):
return "$__cwl_command"
def parse_environment_variables(self):
environment_variables = []
# TODO: Is this even possible from here, should instead this be moved
# into the job.
# for environment_variable_el in environment_variables_el.findall("environment_variable"):
# definition = {
# "name": environment_variable_el.get("name"),
# "template": environment_variable_el.text,
# }
# environment_variables.append(
# definition
# )
return environment_variables
def parse_help(self):
return ""
def parse_stdio(self):
# TODO: remove duplication with YAML
from galaxy.jobs.error_level import StdioErrorLevel
# New format - starting out just using exit code.
exit_code_lower = ToolStdioExitCode()
exit_code_lower.range_start = float("-inf")
exit_code_lower.range_end = -1
exit_code_lower.error_level = StdioErrorLevel.FATAL
exit_code_high = ToolStdioExitCode()
exit_code_high.range_start = 1
exit_code_high.range_end = float("inf")
exit_code_lower.error_level = StdioErrorLevel.FATAL
return [exit_code_lower, exit_code_high], []
def parse_interpreter(self):
return None
def parse_version(self):
return "0.0.1"
def parse_description(self):
return self._tool_proxy.description() or ""
def parse_input_pages(self):
page_source = CwlPageSource(self._tool_proxy)
return PagesSource([page_source])
def parse_outputs(self, tool):
output_instances = self._tool_proxy.output_instances()
outputs = 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()
def _parse_output(self, tool, output_instance):
name = output_instance.name
# TODO: handle filters, actions, change_format
output = ToolOutput( name )
output.format = "_sniff_"
output.change_format = []
output.format_source = None
output.metadata_source = ""
output.parent = None
output.label = None
output.count = None
output.filters = []
output.tool = tool
output.from_work_dir = "__cwl_output_%s" % name
output.hidden = ""
output.dataset_collectors = []
output.actions = ToolOutputActionGroup( output, None )
return output
def parse_requirements_and_containers(self):
containers = []
docker_identifier = self._tool_proxy.docker_identifier()
if docker_identifier:
containers.append({"type": "docker",
"identifier": docker_identifier})
return requirements.parse_requirements_from_dict(dict(
requirements=[], # TODO: enable via extensions
containers=containers,
))
class CwlPageSource(PageSource):
def __init__(self, tool_proxy):
cwl_instances = tool_proxy.input_instances()
self._input_list = map(self._to_input_source, cwl_instances)
def _to_input_source(self, input_instance):
as_dict = input_instance.to_dict()
return YamlInputSource(as_dict)
def parse_input_sources(self):
return self._input_list
+4
View File
@@ -5,6 +5,7 @@ import yaml
from .yaml import YamlToolSource
from .xml import XmlToolSource
from .xml import XmlInputSource
from .cwl import CwlToolSource
from .interface import InputSource
@@ -27,6 +28,9 @@ def get_tool_source(config_file, enable_beta_formats=True):
with open(config_file, "r") as f:
as_dict = ordered_load(f)
return YamlToolSource(as_dict)
elif config_file.endswith(".json") or config_file.endswith(".cwl"):
log.info("Loading CWL tool - this is experimental - tool likely will not function in future at least in same way.")
return CwlToolSource(config_file)
else:
tree = load_tool_xml(config_file)
root = tree.getroot()