# HG changeset patch # User fubar # Date 1605830390 0 # Node ID 6ce360759c284827d31e3f8d5cc9dd719152a6ef # Parent 7176af503cdd33b3a7469093b4b6644cb7fa49ff Uploaded diff -r 7176af503cdd -r 6ce360759c28 toolfactory/galaxyxml/__init__.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/toolfactory/galaxyxml/__init__.py Thu Nov 19 23:59:50 2020 +0000 @@ -0,0 +1,64 @@ +from builtins import object +from builtins import str + +from lxml import etree + + +class GalaxyXML(object): + def __init__(self): + self.root = etree.Element("root") + + def export(self): + return etree.tostring(self.root, pretty_print=True, encoding="unicode") + + +class Util(object): + @classmethod + def coerce(cls, data, kill_lists=False): + """Recursive data sanitisation + """ + if isinstance(data, dict): + return {k: cls.coerce(v, kill_lists=kill_lists) for k, v in list(data.items()) if v is not None} + elif isinstance(data, list): + if kill_lists: + return cls.coerce(data[0]) + else: + return [cls.coerce(v, kill_lists=kill_lists) for v in data] + else: + return cls.coerce_value(data) + + @classmethod + def coerce_value(cls, obj): + """Make everything a string! + """ + if isinstance(obj, bool): + if obj: + return "true" + else: + return "false" + elif isinstance(obj, str): + return obj + else: + return str(obj) + + @classmethod + def clean_kwargs(cls, params, final=False): + if "kwargs" in params: + kwargs = params["kwargs"] + for k in kwargs: + params[k] = kwargs[k] + del params["kwargs"] + if "self" in params: + del params["self"] + + if "__class__" in params: + del params["__class__"] + + # There will be more params, it would be NICE to use a whitelist + # instead of a blacklist, but until we have more data let's just + # blacklist stuff we see commonly. + if final: + for blacklist in ("positional",): + if blacklist in params: + del params[blacklist] + return params diff -r 7176af503cdd -r 6ce360759c28 toolfactory/galaxyxml/tool/__init__.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/toolfactory/galaxyxml/tool/__init__.py Thu Nov 19 23:59:50 2020 +0000 @@ -0,0 +1,184 @@ +import copy +import logging + +from galaxyxml import GalaxyXML, Util +from galaxyxml.tool.parameters import XMLParam + +from lxml import etree + +VALID_TOOL_TYPES = ("data_source", "data_source_async") +VALID_URL_METHODS = ("get", "post") + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +class Tool(GalaxyXML): + + def __init__( + self, + name, + id, + version, + description, + executable, + hidden=False, + tool_type=None, + URL_method=None, + workflow_compatible=True, + interpreter=None, + version_command="interpreter filename.exe --version", + command_override=None, + ): + + self.executable = executable + self.interpreter = interpreter + self.command_override = command_override + kwargs = { + "name": name, + "id": id, + "version": version, + "hidden": hidden, + "workflow_compatible": workflow_compatible, + } + self.version_command = version_command + + # Remove some of the default values to make tools look a bit nicer + if not hidden: + del kwargs["hidden"] + if workflow_compatible: + del kwargs["workflow_compatible"] + + kwargs = Util.coerce(kwargs) + self.root = etree.Element("tool", **kwargs) + + if tool_type is not None: + if tool_type not in VALID_TOOL_TYPES: + raise Exception("Tool type must be one of %s" % ",".join(VALID_TOOL_TYPES)) + else: + kwargs["tool_type"] = tool_type + + if URL_method is not None: + if URL_method in VALID_URL_METHODS: + kwargs["URL_method"] = URL_method + else: + raise Exception("URL_method must be one of %s" % ",".join(VALID_URL_METHODS)) + + description_node = etree.SubElement(self.root, "description") + description_node.text = description + + def add_comment(self, comment_txt): + comment = etree.Comment(comment_txt) + self.root.insert(0, comment) + + def append_version_command(self): + version_command = etree.SubElement(self.root, "version_command") + try: + version_command.text = etree.CDATA(self.version_command) + except Exception: + pass + + def append(self, sub_node): + if issubclass(type(sub_node), XMLParam): + self.root.append(sub_node.node) + else: + self.root.append(sub_node) + + def clean_command_string(self, command_line): + clean = [] + for x in command_line: + if x is not [] and x is not [""]: + clean.append(x) + + return "\n".join(clean) + + def export(self, keep_old_command=False): # noqa + + export_xml = copy.deepcopy(self) + + try: + export_xml.append(export_xml.edam_operations) + except Exception: + pass + + try: + export_xml.append(export_xml.edam_topics) + except Exception: + pass + + try: + export_xml.append(export_xml.requirements) + except Exception: + pass + + try: + export_xml.append(export_xml.configfiles) + except Exception: + pass + + if self.command_override: + command_line = self.command_override + else: + command_line = [] + try: + command_line.append(export_xml.inputs.cli()) + except Exception as e: + logger.warning(str(e)) + + try: + command_line.append(export_xml.outputs.cli()) + except Exception: + pass + + # Add stdio section + stdio = etree.SubElement(export_xml.root, "stdio") + etree.SubElement(stdio, "exit_code", range="1:", level="fatal") + + # Append version command + export_xml.append_version_command() + + # Steal interpreter from kwargs + command_kwargs = {} + if export_xml.interpreter is not None: + command_kwargs["interpreter"] = export_xml.interpreter + + # Add command section + command_node = etree.SubElement(export_xml.root, "command", **command_kwargs) + + if keep_old_command: + if getattr(self, "command", None): + command_node.text = etree.CDATA(export_xml.command) + else: + logger.warning("The tool does not have any old command stored. " + "Only the command line is written.") + command_node.text = export_xml.executable + else: + if self.command_override: + actual_cli = export_xml.clean_command_string(command_line) + else: + actual_cli = "%s %s" % (export_xml.executable, export_xml.clean_command_string(command_line)) + command_node.text = etree.CDATA(actual_cli.strip()) + + try: + export_xml.append(export_xml.inputs) + except Exception: + pass + + try: + export_xml.append(export_xml.outputs) + except Exception: + pass + + try: + export_xml.append(export_xml.tests) + except Exception: + pass + + help_element = etree.SubElement(export_xml.root, "help") + help_element.text = etree.CDATA(export_xml.help) + + try: + export_xml.append(export_xml.citations) + except Exception: + pass + + return super(Tool, export_xml).export() diff -r 7176af503cdd -r 6ce360759c28 toolfactory/galaxyxml/tool/__pycache__/__init__.cpython-36.pyc Binary file toolfactory/galaxyxml/tool/__pycache__/__init__.cpython-36.pyc has changed diff -r 7176af503cdd -r 6ce360759c28 toolfactory/galaxyxml/tool/__pycache__/import_xml.cpython-36.pyc Binary file toolfactory/galaxyxml/tool/__pycache__/import_xml.cpython-36.pyc has changed diff -r 7176af503cdd -r 6ce360759c28 toolfactory/galaxyxml/tool/import_xml.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/toolfactory/galaxyxml/tool/import_xml.py Thu Nov 19 23:59:50 2020 +0000 @@ -0,0 +1,713 @@ +import logging +import xml.etree.ElementTree as ET + +import galaxyxml.tool as gxt +import galaxyxml.tool.parameters as gxtp + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +class GalaxyXmlParser(object): + """ + Class to import content from an existing Galaxy XML wrapper. + """ + + def _init_tool(self, xml_root): + """ + Init tool from existing xml tool. + + :param xml_root: root of the galaxy xml file. + :type xml_root: :class:`xml.etree._Element` + """ + version_cmd = None + description = None + for child in xml_root: + if child.tag == "description": + description = child.text + elif child.tag == "command": + executable = child.text.split()[0] + command = child.text + elif child.tag == "version_command": + version_cmd = child.text + + tool = gxt.Tool( + xml_root.attrib["name"], + xml_root.attrib["id"], + xml_root.attrib.get("version", None), + description, + executable, + hidden=xml_root.attrib.get("hidden", False), + tool_type=xml_root.attrib.get("tool_type", None), + URL_method=xml_root.attrib.get("URL_method", None), + workflow_compatible=xml_root.attrib.get("workflow_compatible", True), + version_command=version_cmd, + ) + tool.command = command + return tool + + def _load_description(self, tool, desc_root): + """ + is already loaded during initiation. + + :param tool: Tool object from galaxyxml. + :type tool: :class:`galaxyxml.tool.Tool` + :param desc_root: root of tag. + :type desc_root: :class:`xml.etree._Element` + """ + logger.info(" is loaded during initiation of the object.") + + def _load_version_command(self, tool, vers_root): + """ + is already loaded during initiation. + + :param tool: Tool object from galaxyxml. + :type tool: :class:`galaxyxml.tool.Tool` + :param vers_root: root of tag. + :type vers_root: :class:`xml.etree._Element` + """ + logger.info(" is loaded during initiation of the object.") + + def _load_stdio(self, tool, stdio_root): + """ + So far, is automatically generated by galaxyxml. + + :param tool: Tool object from galaxyxml. + :type tool: :class:`galaxyxml.tool.Tool` + :param desc_root: root of tag. + :type desc_root: :class:`xml.etree._Element` + """ + logger.info(" is not loaded but automatically generated by galaxyxml.") + + def _load_command(self, tool, desc_root): + """ + is already loaded during initiation. + + :param tool: Tool object from galaxyxml. + :type tool: :class:`galaxyxml.tool.Tool` + :param desc_root: root of tag. + :type desc_root: :class:`xml.etree._Element` + """ + logger.info(" is loaded during initiation of the object.") + + def _load_help(self, tool, help_root): + """ + Load the content of the into the tool. + + :param tool: Tool object from galaxyxml. + :type tool: :class:`galaxyxml.tool.Tool` + :param requirements_root: root of tag. + :type requirements_root: :class:`xml.etree._Element` + """ + tool.help = help_root.text + + def _load_requirements(self, tool, requirements_root): + """ + Add to the tool. + + :param tool: Tool object from galaxyxml. + :type tool: :class:`galaxyxml.tool.Tool` + :param requirements_root: root of tag. + :type requirements_root: :class:`xml.etree._Element` + """ + tool.requirements = gxtp.Requirements() + for req in requirements_root: + req_type = req.attrib["type"] + value = req.text + if req.tag == "requirement": + version = req.attrib.get("version", None) + tool.requirements.append(gxtp.Requirement(req_type, value, version=version)) + elif req.tag == "container": + tool.requirements.append(gxtp.Container(req_type, value)) + else: + logger.warning(req.tag + " is not a valid tag for requirements child") + + def _load_edam_topics(self, tool, topics_root): + """ + Add to the tool. + + :param tool: Tool object from galaxyxml. + :type tool: :class:`galaxyxml.tool.Tool` + :param topics_root: root of tag. + :type topics_root: :class:`xml.etree._Element` + """ + tool.edam_topics = gxtp.EdamTopics() + for edam_topic in topics_root: + tool.edam_topics.append(gxtp.EdamTopic(edam_topic.text)) + + def _load_edam_operations(self, tool, operations_root): + """ + Add to the tool. + + :param tool: Tool object from galaxyxml. + :type tool: :class:`galaxyxml.tool.Tool` + :param operations_root: root of tag. + :type operations_root: :class:`xml.etree._Element` + """ + tool.edam_operations = gxtp.EdamOperations() + for edam_op in operations_root: + tool.edam_operations.append(gxtp.EdamOperation(edam_op.text)) + + def _load_configfiles(self, tool, configfiles_root): + """ + Add to the tool. + + :param tool: Tool object from galaxyxml. + :type tool: :class:`galaxyxml.tool.Tool` + :param configfiles_root: root of tag. + :type configfiles_root: :class:`xml.etree._Element` + """ + tool.configfiles = gxtp.Configfiles() + for conf in configfiles_root: + name = conf.attrib["name"] + value = conf.text + tool.configfiles.append(gxtp.Configfile(name, value)) + + def _load_citations(self, tool, citations_root): + """ + Add to the tool. + + :param tool: Tool object from galaxyxml. + :type tool: :class:`galaxyxml.tool.Tool` + :param citations_root: root of tag. + :type citations_root: :class:`xml.etree._Element` + """ + tool.citations = gxtp.Citations() + for cit in citations_root: + cit_type = cit.attrib["type"] + value = cit.text + tool.citations.append(gxtp.Citation(cit_type, value)) + + def _load_inputs(self, tool, inputs_root): + """ + Add to the tool using the :class:`galaxyxml.tool.import_xml.InputsParser` object. + + :param tool: Tool object from galaxyxml. + :type tool: :class:`galaxyxml.tool.Tool` + :param inputs_root: root of tag. + :type inputs_root: :class:`xml.etree._Element` + """ + tool.inputs = gxtp.Inputs() + inp_parser = InputsParser() + inp_parser.load_inputs(tool.inputs, inputs_root) + + def _load_outputs(self, tool, outputs_root): + """ + Add to the tool using the :class:`galaxyxml.tool.import_xml.OutputsParser` object. + + :param tool: Tool object from galaxyxml. + :type tool: :class:`galaxyxml.tool.Tool` + :param outputs_root: root of tag. + :type outputs_root: :class:`xml.etree._Element` + """ + tool.outputs = gxtp.Outputs() + out_parser = OutputsParser() + out_parser.load_outputs(tool.outputs, outputs_root) + + def _load_tests(self, tool, tests_root): + """ + Add to the tool using the :class:`galaxyxml.tool.import_xml.TestsParser` object. + + :param tool: Tool object from galaxyxml. + :type tool: :class:`galaxyxml.tool.Tool` + :param tests_root: root of tag. + :type tests_root: :class:`xml.etree._Element` + """ + tool.tests = gxtp.Tests() + tests_parser = TestsParser() + tests_parser.load_tests(tool.tests, tests_root) + + def import_xml(self, xml_path): + """ + Load existing xml into the :class:`galaxyxml.tool.Tool` object. + + :param xml_path: Path of the XML to be loaded. + :type xml_path: STRING + :return: XML content in the galaxyxml model. + :rtype: :class:`galaxyxml.tool.Tool` + """ + xml_root = ET.parse(xml_path).getroot() + tool = self._init_tool(xml_root) + # Now we import each tag's field + for child in xml_root: + try: + getattr(self, "_load_{}".format(child.tag))(tool, child) + except AttributeError: + logger.warning(child.tag + " tag is not processed.") + return tool + + +class InputsParser(object): + """ + Class to parse content of the tag from a Galaxy XML wrapper. + """ + + def _load_text_param(self, root, text_param): + """ + Add to the root. + + :param root: root to append the param to. + :param text_param: root of tag. + :type text_param: :class:`xml.etree._Element` + """ + root.append( + gxtp.TextParam( + text_param.attrib["name"], + optional=text_param.get("optional", None), + label=text_param.get("label", None), + help=text_param.get("help", None), + value=text_param.get("value", None), + ) + ) + + def _load_data_param(self, root, data_param): + """ + Add to the root. + + :param root: root to append the param to. + :param data_param: root of tag. + :type data_param: :class:`xml.etree._Element` + """ + root.append( + gxtp.DataParam( + data_param.attrib["name"], + optional=data_param.attrib.get("optional", None), + label=data_param.attrib.get("label", None), + help=data_param.attrib.get("help", None), + format=data_param.attrib.get("format", None), + multiple=data_param.attrib.get("multiple", None), + ) + ) + + def _load_boolean_param(self, root, bool_param): + """ + Add to the root. + + :param root: root to append the param to. + :param bool_param: root of tag. + :type bool_param: :class:`xml.etree._Element` + """ + root.append( + gxtp.BooleanParam( + bool_param.attrib["name"], + optional=bool_param.attrib.get("optional", None), + label=bool_param.attrib.get("label", None), + help=bool_param.attrib.get("help", None), + checked=bool_param.attrib.get("checked", False), + truevalue=bool_param.attrib.get("truevalue", None), + falsevalue=bool_param.attrib.get("falsevalue", None), + ) + ) + + def _load_integer_param(self, root, int_param): + """ + Add to the root. + + :param root: root to append the param to. + :param int_param: root of tag. + :type int_param: :class:`xml.etree._Element` + """ + root.append( + gxtp.IntegerParam( + int_param.attrib["name"], + int_param.attrib.get("value", None), + optional=int_param.attrib.get("optional", None), + label=int_param.attrib.get("label", None), + help=int_param.attrib.get("help", None), + min=int_param.attrib.get("min", None), + max=int_param.attrib.get("max", None), + ) + ) + + def _load_float_param(self, root, float_param): + """ + Add to the root. + + :param root: root to append the param to. + :param float_param: root of tag. + :type float_param: :class:`xml.etree._Element` + """ + root.append( + gxtp.FloatParam( + float_param.attrib["name"], + float_param.attrib.get("value", None), + optional=float_param.attrib.get("optional", None), + label=float_param.attrib.get("label", None), + help=float_param.attrib.get("help", None), + min=float_param.attrib.get("min", None), + max=float_param.attrib.get("max", None), + ) + ) + + def _load_option_select(self, root, option): + """ + Add