comparison toolfactory/galaxyxml/tool/import_xml.py @ 38:a30536c100bf draft

Updated history outputs

37:099047ee7094

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__)

        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")

        :param option: root of <option> tag.
        :type float_param: :class:`xml.etree._Element`
        """
        root.append(
            gxtp.SelectOption(
                option.attrib.get("value", None),
                option.text,
                selected=option.attrib.get("selected", False),
            )
        )

    def _load_column_options(self, root, column):
        """

            multiple=sel_param.attrib.get("multiple", None),
        )
        # Deal with child nodes (usually option and options)
        for sel_child in sel_param:
            try:
                getattr(self, "_load_{}_select".format(sel_child.tag))(
                    select_param, sel_child
                )
            except AttributeError:
                logger.warning(
                    sel_child.tag + " tag is not processed for <param type='select'>."
                )
        root.append(select_param)

    def _load_param(self, root, param_root):
        """
        Method to select which type of <param> is being added to the root.

        """
        for inp_child in inputs_root:
            try:
                getattr(self, "_load_{}".format(inp_child.tag))(root, inp_child)
            except AttributeError:
                logger.warning(
                    inp_child.tag
                    + " tag is not processed for <"
                    + inputs_root.tag
                    + "> tag."
                )


class OutputsParser(object):
    """
    Class to parse content of the <outputs> tag from a Galaxy XML wrapper.

        """
        change_format = gxtp.ChangeFormat()
        for chfmt_child in chfmt_root:
            change_format.append(
                gxtp.ChangeFormatWhen(
                    chfmt_child.attrib["input"],
                    chfmt_child.attrib["format"],
                    chfmt_child.attrib["value"],
                )
            )
        root.append(change_format)

    def _load_collection(self, outputs_root, coll_root):

        # Deal with child nodes
        for coll_child in coll_root:
            try:
                getattr(self, "_load_{}".format(coll_child.tag))(collection, coll_child)
            except AttributeError:
                logger.warning(
                    coll_child.tag + " tag is not processed for <collection>."
                )
        outputs_root.append(collection)

    def _load_discover_datasets(self, root, disc_root):
        """
        Add <discover_datasets> to root (<collection>).

            test = gxtp.Test()
            for test_child in test_root:
                try:
                    getattr(self, "_load_{}".format(test_child.tag))(test, test_child)
                except AttributeError:
                    logger.warning(
                        test_child.tag + " tag is not processed within <test>."
                    )
            root.append(test)

38:a30536c100bf

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__)

        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")

        :param option: root of <option> tag.
        :type float_param: :class:`xml.etree._Element`
        """
        root.append(
            gxtp.SelectOption(
                option.attrib.get("value", None), option.text, selected=option.attrib.get("selected", False)
            )
        )

    def _load_column_options(self, root, column):
        """

            multiple=sel_param.attrib.get("multiple", None),
        )
        # Deal with child nodes (usually option and options)
        for sel_child in sel_param:
            try:
                getattr(self, "_load_{}_select".format(sel_child.tag))(select_param, sel_child)
            except AttributeError:
                logger.warning(sel_child.tag + " tag is not processed for <param type='select'>.")
        root.append(select_param)

    def _load_param(self, root, param_root):
        """
        Method to select which type of <param> is being added to the root.

        """
        for inp_child in inputs_root:
            try:
                getattr(self, "_load_{}".format(inp_child.tag))(root, inp_child)
            except AttributeError:
                logger.warning(inp_child.tag + " tag is not processed for <" + inputs_root.tag + "> tag.")


class OutputsParser(object):
    """
    Class to parse content of the <outputs> tag from a Galaxy XML wrapper.

        """
        change_format = gxtp.ChangeFormat()
        for chfmt_child in chfmt_root:
            change_format.append(
                gxtp.ChangeFormatWhen(
                    chfmt_child.attrib["input"], chfmt_child.attrib["format"], chfmt_child.attrib["value"]
                )
            )
        root.append(change_format)

    def _load_collection(self, outputs_root, coll_root):

        # Deal with child nodes
        for coll_child in coll_root:
            try:
                getattr(self, "_load_{}".format(coll_child.tag))(collection, coll_child)
            except AttributeError:
                logger.warning(coll_child.tag + " tag is not processed for <collection>.")
        outputs_root.append(collection)

    def _load_discover_datasets(self, root, disc_root):
        """
        Add <discover_datasets> to root (<collection>).

            test = gxtp.Test()
            for test_child in test_root:
                try:
                    getattr(self, "_load_{}".format(test_child.tag))(test, test_child)
                except AttributeError:
                    logger.warning(test_child.tag + " tag is not processed within <test>.")
            root.append(test)