Source code for

from import ExpressionValidator
from xml.etree import ElementTree
from os.path import join, abspath, dirname

from io import open

from logging import getLogger
log = getLogger(__name__)

[docs]class ToolBox(object): """ Abstraction over a tool config file largely modelled after Galaxy's shed_tool_conf.xml. Hopefully over time this toolbox schema will be a direct superset of Galaxy's with extensions to support simple, non-toolshed based tool setups. """ def __init__(self, path_string): self.tool_configs = [] paths = [path.strip() for path in path_string.split(",")] for path in paths: toolbox_tree = ElementTree.parse(path) toolbox_root = toolbox_tree.getroot() tool_path = toolbox_root.get('tool_path') self.__load_tools_from_els(toolbox_root, tool_path) def __load_tools_from_els(self, toolbox_root, tool_path): els = toolbox_root.findall('tool') for el in els: try: if 'guid' in el.attrib: tool_cls = ToolShedToolConfig else: tool_cls = SimpleToolConfig tool = tool_cls(el, tool_path) self.tool_configs.append(tool) except Exception: log.exception('Failed to load tool.')
[docs] def get_tool(self, id): # Need to handle multiple tools per id someday, but # starting simple. tools = self.__find_tools_by_id(id) if not tools: raise KeyError("Failed to find tool with id '%s'" % id) if len(tools) > 1: log.warn("Found multiple tools with id '%s', returning first." % id) return tools[0]
def __find_tools_by_id(self, id): return [tool for tool in self.tool_configs if == id]
[docs]class InputsValidator(object): def __init__(self, command_validator, config_validators): self.command_validator = command_validator self.config_validators = config_validators
[docs] def validate_command(self, job_directory, command): return self.command_validator.validate(job_directory, command)
[docs] def validate_config(self, job_directory, name, path): config_validator = self.config_validators.get(name, None) valid = True if config_validator: contents = open(path, "r", encoding="UTF-8").read() valid = config_validator.validate(job_directory, contents) return valid
[docs]class ToolConfig(object): """ Abstract description of a Galaxy tool. """ def __init__(self): super(ToolConfig, self).__init__()
[docs] def get_tool_dir(self): return abspath(dirname(self.path))
@property def inputs_validator(self): if not hasattr(self, "_inputs_validator"): command_el = self._root().find("./validators/command_validator") command_validator = ExpressionValidator(command_el) config_validators = {} for config_el in self._root().findall("./validators/configfile_validator"): name = config_el.get("name") config_validators[name] = ExpressionValidator(config_el) self._inputs_validator = InputsValidator(command_validator, config_validators) return self._inputs_validator def _root(self): return self._el().getroot() def _el(self): return ElementTree.parse(self.path)
[docs]class SimpleToolConfig(ToolConfig): """ Abstract description of a Galaxy tool loaded from a toolbox with the `tool` tag not containing a guid, i.e. one not from the toolshed. """ def __init__(self, tool_el, tool_path): super(SimpleToolConfig, self).__init__() rel_path = tool_el.get('file') assert tool_path, "tool_path not set, toolbox XML files must be configured with a tool_path directory." assert rel_path, "file not set on tool, each tool element must define a file attribute pointing to a valid tool XML file." resolved_path = join(tool_path, rel_path) self.path = resolved_path root = self._root() = root.get('id') self.version = root.get('version', '1.0.0') self.tool_dir = dirname(resolved_path)
[docs]class ToolShedToolConfig(SimpleToolConfig): """ Abstract description of a Galaxy tool loaded from a toolbox with the `tool` tag, i.e. one from the toolshed. <tool file="../shed_tools/" guid\ =""> <tool_shed></tool_shed> <repository_name>column_maker</repository_name> <repository_owner>test</repository_owner> <installed_changeset_revision>f06aa1bf1e8a</installed_changeset_revision <id></id> <version>1.1.0</version> </tool> """ def __init__(self, tool_el, tool_path): super(ToolShedToolConfig, self).__init__(tool_el, tool_path) self.guid = tool_el.get("guid") # Override id in file for tool shed tools. Use GUID instead. = self.guid