# HG changeset patch
# User fubar
# Date 1618383922 0
# Node ID 1dba6c7687a96f6ff83a1f1106bf1fb9503056a9
# Parent 63d15caea378f1fecb7a648295c00eb92da33cc6
Uploaded
diff -r 63d15caea378 -r 1dba6c7687a9 toolfactory/galaxyxml/__init__.py
--- a/toolfactory/galaxyxml/__init__.py Tue Apr 13 03:06:27 2021 +0000
+++ /dev/null Thu Jan 01 00:00:00 1970 +0000
@@ -1,71 +0,0 @@
-from builtins import (
- object,
- 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
-
- - recurse into lists, dicts, OrderedDict
- - remove dict/OrderedDict entries with None value
- - kill_lists: True -> replace lists by their first element
- """
- 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 63d15caea378 -r 1dba6c7687a9 toolfactory/galaxyxml/tool/__init__.py
--- a/toolfactory/galaxyxml/tool/__init__.py Tue Apr 13 03:06:27 2021 +0000
+++ /dev/null Thu Jan 01 00:00:00 1970 +0000
@@ -1,277 +0,0 @@
-import copy
-import logging
-
-from galaxyxml import GalaxyXML, Util
-from galaxyxml.tool.parameters import (
- Expand,
- Import,
- Inputs,
- Macro,
- Macros,
- Outputs,
- 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,
- macros=[],
- ):
-
- self.id = id
- 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
- if len(macros) > 0:
- self.macros = Macros()
- for m in macros:
- self.macros.append(Import(m))
- self.inputs = Inputs()
- self.outputs = Outputs()
-
- 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):
- # see lib/galaxy/tool_util/linters/xml_order.py
- export_xml = copy.deepcopy(self)
- try:
- export_xml.append(export_xml.macros)
- except Exception:
- pass
-
- 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:
- export_xml.append(Expand(macro="requirements"))
-
- # Add stdio section - now an XMLParameter
- try:
- stdio_element = export_xml.stdios
- except Exception:
- stdio_element = None
- if not stdio_element:
- stdio_element = etree.SubElement(export_xml.root, "stdio")
- etree.SubElement(stdio_element, "exit_code", range="1:", level="fatal")
- try:
- export_xml.append(stdio_element)
- except Exception:
- export_xml.append(Expand(macro="stdio"))
-
- # Append version command
- export_xml.append_version_command()
-
- 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))
- raise
- try:
- command_line.append(export_xml.outputs.cli())
- except Exception:
- pass
- # 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())
- export_xml.append(command_node)
-
- try:
- export_xml.append(export_xml.configfiles)
- except Exception:
- pass
-
- 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:
- export_xml.append(Expand(macro="%s_tests" % self.id))
-
- help_element = etree.SubElement(export_xml.root, "help")
- help_element.text = etree.CDATA(export_xml.help)
- export_xml.append(help_element)
-
- try:
- export_xml.append(export_xml.citations)
- except Exception:
- export_xml.append(Expand(macro="citations"))
-
- return super(Tool, export_xml).export()
-
-
-class MacrosTool(Tool):
- """
- creates a tag containing macros and tokens
- for the inputs and outputs:
-
- for the inputs
-
- - a macro `` containing all the inputs
- - a token `` containing the CLI for the inputs
-
- where ID is the id used in initialization.
-
- analogously for the outputs `ID_outmacro` and `ID_OUTMACRO`
- are created.
-
- TODO all other elements, like requirements are currently ignored
- """
- def __init__(self, *args, **kwargs):
- super(MacrosTool, self).__init__(*args, **kwargs)
- self.root = etree.Element('macros')
- self.inputs = Macro("%s_inmacro" % self.id)
- self.outputs = Macro("%s_outmacro" % self.id)
-
-
- def export(self, keep_old_command=False): # noqa
-
- export_xml = copy.deepcopy(self)
-
- try:
- for child in export_xml.macros:
- export_xml.append(child)
- except Exception:
- pass
-
- command_line = []
- try:
- command_line.append(export_xml.inputs.cli())
- except Exception as e:
- logger.warning(str(e))
- raise
-
- # Add command section
- command_node = etree.SubElement(export_xml.root, 'token', {"name": "%s_INMACRO" % self.id.upper()})
- actual_cli = "%s" % (export_xml.clean_command_string(command_line))
- command_node.text = etree.CDATA(actual_cli.strip())
-
- command_line = []
- try:
- command_line.append(export_xml.outputs.cli())
- except Exception:
- pass
- command_node = etree.SubElement(export_xml.root, 'token', {"name": "%s_OUTMACRO" % self.id.upper()})
- actual_cli = "%s" % (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
-
- return super(Tool, export_xml).export()
diff -r 63d15caea378 -r 1dba6c7687a9 toolfactory/galaxyxml/tool/import_xml.py
--- a/toolfactory/galaxyxml/tool/import_xml.py Tue Apr 13 03:06:27 2021 +0000
+++ /dev/null Thu Jan 01 00:00:00 1970 +0000
@@ -1,785 +0,0 @@
-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):
- """
-
- :param tool: root to append to.
- :param stdio_root: root of tag.
- :param stdio_root: :class:`xml.etree._Element`
- """
- tool.stdios = gxtp.Stdios()
- for std in stdio_root:
- slevel = std.attrib['level']
- srange = std.attrib['range']
- tool.stdios.append(gxtp.Stdio(level=slevel, range=srange))
- logger.info(" loaded.")
-
- 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
-
+ galaxyxml
bioblend
ephemeris
planemo