# HG changeset patch # User fubar # Date 1597273244 14400 # Node ID 2cd6555baf4445f4fbbc8269dae0fff0f2e1fffe # Parent a30536c100bf1a965190bb11c71e5ed6cd21a7d6 Uploaded diff -r a30536c100bf -r 2cd6555baf44 toolfactory/__init__.py diff -r a30536c100bf -r 2cd6555baf44 toolfactory/galaxyxml/__init__.py --- a/toolfactory/galaxyxml/__init__.py Wed Aug 12 01:43:46 2020 -0400 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,64 +0,0 @@ -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 a30536c100bf -r 2cd6555baf44 toolfactory/galaxyxml/__pycache__/__init__.cpython-36.pyc Binary file toolfactory/galaxyxml/__pycache__/__init__.cpython-36.pyc has changed diff -r a30536c100bf -r 2cd6555baf44 toolfactory/galaxyxml/tool/__init__.py --- a/toolfactory/galaxyxml/tool/__init__.py Wed Aug 12 01:43:46 2020 -0400 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,181 +0,0 @@ -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_line_override=None, - ): - - self.executable = executable - self.interpreter = interpreter - self.command_line_override = command_line_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_line_override: - command_line = self.command_line_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: - 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 a30536c100bf -r 2cd6555baf44 toolfactory/galaxyxml/tool/__pycache__/__init__.cpython-36.pyc Binary file toolfactory/galaxyxml/tool/__pycache__/__init__.cpython-36.pyc has changed diff -r a30536c100bf -r 2cd6555baf44 toolfactory/galaxyxml/tool/import_xml.py --- a/toolfactory/galaxyxml/tool/import_xml.py Wed Aug 12 01:43:46 2020 -0400 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,713 +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): - """ - 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