This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.tool_util.parser.cwl

import json
import logging
import math
from typing import Optional

import packaging.version

from galaxy.tool_util.cwl.parser import (
from galaxy.tool_util.deps import requirements
from .interface import (
from .output_actions import ToolOutputActionGroup
from .output_objects import ToolOutput
from .stdio import (
from .yaml import YamlInputSource

log = logging.getLogger(__name__)

[docs]class CwlToolSource(ToolSource): language = "yaml"
[docs] def __init__(self, tool_file=None, strict_cwl_validation=True, tool_proxy: Optional[ToolProxy] = None): self._cwl_tool_file = tool_file self._tool_proxy = tool_proxy self._source_path = tool_file self._strict_cwl_validation = strict_cwl_validation
@property def source_path(self): return self._source_path @property def tool_proxy(self) -> ToolProxy: if self._tool_proxy is None: self._tool_proxy = tool_proxy(self._source_path, strict_cwl_validation=self._strict_cwl_validation) return self._tool_proxy
[docs] def parse_tool_type(self): return "cwl"
[docs] def parse_id(self): return self.tool_proxy.galaxy_id()
[docs] def parse_name(self): return self.tool_proxy.label() or self.parse_id()
[docs] def parse_command(self): return "$__cwl_command"
[docs] 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
[docs] def parse_edam_operations(self): return []
[docs] def parse_edam_topics(self): return []
[docs] def parse_help(self): return self.tool_proxy.doc()
[docs] def parse_sanitize(self): return False
[docs] def parse_strict_shell(self): return True
[docs] def parse_stdio(self): # TODO: remove duplication with YAML # New format - starting out just using exit code. exit_code_lower = ToolStdioExitCode() exit_code_lower.range_start = -math.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 = math.inf exit_code_lower.error_level = StdioErrorLevel.FATAL return [exit_code_lower, exit_code_high], []
[docs] def parse_interpreter(self): return None
[docs] def parse_version(self): return "0.0.1"
[docs] def parse_description(self): return self.tool_proxy.description()
[docs] def parse_interactivetool(self): return []
[docs] def parse_input_pages(self): page_source = CwlPageSource(self.tool_proxy) return PagesSource([page_source])
[docs] def parse_outputs(self, tool): output_instances = self.tool_proxy.output_instances() outputs = {} 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, {}
def _parse_output(self, tool, output_instance): name = output_instance.name # TODO: handle filters, actions, change_format output = ToolOutput(name) if "File" in output_instance.output_data_type: output.format = "_sniff_" else: output.format = "expression.json" 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.hidden = "" output.dataset_collector_descriptions = [] output.actions = ToolOutputActionGroup(output, None) return output
[docs] def parse_requirements_and_containers(self): containers = [] docker_identifier = self.tool_proxy.docker_identifier() if docker_identifier: containers.append({"type": "docker", "identifier": docker_identifier}) software_requirements = self.tool_proxy.software_requirements() resource_requirements = self.tool_proxy.resource_requirements() return requirements.parse_requirements_from_lists( software_requirements=[{"name": r[0], "version": r[1], "type": "package"} for r in software_requirements], containers=containers, resource_requirements=resource_requirements, )
[docs] def parse_profile(self): return "17.09"
[docs] def parse_xrefs(self): return []
[docs] def parse_license(self): return None
[docs] def parse_python_template_version(self): return packaging.version.Version("3.5")
[docs] def to_string(self): return json.dumps(self.tool_proxy.to_persistent_representation())
class CwlPageSource(PageSource): def __init__(self, tool_proxy): cwl_instances = tool_proxy.input_instances() self._input_list = 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 __all__ = ( "CwlToolSource", "tool_proxy", )