This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.tools.parameters.dataset_matcher

from logging import getLogger

import galaxy.model

log = getLogger(__name__)

def set_dataset_matcher_factory(trans, tool):
    trans.dataset_matcher_factory = DatasetMatcherFactory(trans, tool)

def unset_dataset_matcher_factory(trans):
    trans.dataset_matcher_factory = None

def get_dataset_matcher_factory(trans):
    dataset_matcher_factory = getattr(trans, "dataset_matcher_factory", None)
    return dataset_matcher_factory or DatasetMatcherFactory(trans)

class DatasetMatcherFactory(object):

    def __init__(self, trans, tool=None):
        self._trans = trans
        self._tool = tool
        self._data_inputs = []
        self._matches_format_cache = {}
        if tool:
            valid_input_states = tool.valid_input_states
            valid_input_states = galaxy.model.Dataset.valid_input_states
        self.valid_input_states = valid_input_states
        can_process_summary = False
        if tool is not None:
            for input in tool.inputs.values():

            require_public = self._tool and self._tool.tool_type == 'data_destination'
            if not require_public and self._data_inputs:
                can_process_summary = True
                for data_input in self._data_inputs:
                    if data_input.options:
                        can_process_summary = False
        self._can_process_summary = can_process_summary

    def matches_any_format(self, hda_extension, formats):
        for format in formats:
            if self.matches_format(hda_extension, format):
                return True
        return False

    def matches_format(self, hda_extension, format):
        # cache datatype checking combinations for fast recall
        if hda_extension not in self._matches_format_cache:
            self._matches_format_cache[hda_extension] = {}

        formats = self._matches_format_cache[hda_extension]
        if format not in formats:
            datatype = galaxy.model.datatype_for_extension(hda_extension, datatypes_registry=self._trans.app.datatypes_registry)
            formats[format] = datatype.matches_any([format])

        return formats[format]

    def _collect_data_inputs(self, input):
        type_name = input.type
        if type_name == "repeat" or type_name == "upload_dataset" or type_name == "section":
            for child_input in input.inputs.values():
        elif type_name == "conditional":
            for case in input.cases:
                for child_input in case.inputs.values():
        elif type_name == "data" or type_name == "data_collection":

    def dataset_matcher(self, param, other_values):
        return DatasetMatcher(self, self._trans, param, other_values)

    def dataset_collection_matcher(self, dataset_matcher):
        if self._can_process_summary:
            return SummaryDatasetCollectionMatcher(self, dataset_matcher)
            return DatasetCollectionMatcher(dataset_matcher)

[docs]class DatasetMatcher(object): """ Utility class to aid DataToolParameter and similar classes in reasoning about what HDAs could match or are selected for a parameter and value. Goal here is to both encapsulate and reuse logic related to filtering, datatype matching, hiding errored dataset, finding implicit conversions, and permission handling. """
[docs] def __init__(self, dataset_matcher_factory, trans, param, other_values): self.dataset_matcher_factory = dataset_matcher_factory self.trans = trans self.param = param self.tool = param.tool filter_value = None if param.options and other_values: try: filter_value = param.options.get_options(trans, other_values)[0][0] except IndexError: pass # no valid options self.filter_value = filter_value
[docs] def valid_hda_match(self, hda, check_implicit_conversions=True): """ Return False of this parameter can not be matched to the supplied HDA, otherwise return a description of the match (either a HdaDirectMatch describing a direct match or a HdaImplicitMatch describing an implicit conversion.) """ rval = False formats = self.param.formats if self.dataset_matcher_factory.matches_any_format(hda.extension, formats): rval = HdaDirectMatch(hda) else: if not check_implicit_conversions: return False target_ext, converted_dataset = hda.find_conversion_destination(formats) if target_ext: original_hda = hda if converted_dataset: hda = converted_dataset rval = HdaImplicitMatch(hda, target_ext, original_hda) else: return False if self.filter(hda): return False return rval
[docs] def hda_match(self, hda, check_implicit_conversions=True, ensure_visible=True): """ If HDA is accessible, return information about whether it could match this parameter and if so how. See valid_hda_match for more information. """ dataset = hda.dataset valid_state = dataset.state in self.dataset_matcher_factory.valid_input_states if valid_state and (not ensure_visible or hda.visible): # If we are sending data to an external application, then we need to make sure there are no roles # associated with the dataset that restrict its access from "public". require_public = self.tool and self.tool.tool_type == 'data_destination' if require_public and not self.trans.app.security_agent.dataset_is_public(dataset): return False return self.valid_hda_match(hda, check_implicit_conversions=check_implicit_conversions)
[docs] def filter(self, hda): """ Filter out this value based on other values for job (if applicable). """ param = self.param return param.options and param.get_options_filter_attribute(hda) != self.filter_value
class HdaDirectMatch(object): """ Supplied HDA was a valid option directly (did not need to find implicit conversion). """ def __init__(self, hda): self.hda = hda @property def implicit_conversion(self): return False class HdaImplicitMatch(object): """ Supplied HDA was a valid option directly (did not need to find implicit conversion). """ def __init__(self, hda, target_ext, original_hda): self.original_hda = original_hda self.hda = hda self.target_ext = target_ext @property def implicit_conversion(self): return True class SummaryDatasetCollectionMatcher(object): def __init__(self, dataset_matcher_factory, dataset_matcher): self.dataset_matcher_factory = dataset_matcher_factory self.dataset_matcher = dataset_matcher def hdca_match(self, history_dataset_collection_association, reduction=False): dataset_collection = history_dataset_collection_association.collection if reduction and dataset_collection.collection_type.find(":") > 0: return False if not dataset_collection.populated_optimized: return False (states, extensions) = dataset_collection.dataset_states_and_extensions_summary for state in states: if state not in self.dataset_matcher_factory.valid_input_states: return False formats = self.dataset_matcher.param.formats for extension in extensions: if not self.dataset_matcher_factory.matches_any_format(extension, formats): return False return True
[docs]class DatasetCollectionMatcher(object):
[docs] def __init__(self, dataset_matcher): self.dataset_matcher = dataset_matcher
def __valid_element(self, element): # Simplify things for now and assume these are hdas and not implicit # converts. One could imagine handling both of those cases down the # road. if element.ldda: return False child_collection = element.child_collection if child_collection: return self.dataset_collection_match(child_collection) hda = element.hda if not hda: return False hda_match = self.dataset_matcher.hda_match(hda, ensure_visible=False) return hda_match and not hda_match.implicit_conversion
[docs] def hdca_match(self, history_dataset_collection_association, reduction=False): dataset_collection = history_dataset_collection_association.collection if reduction and dataset_collection.collection_type.find(":") > 0: return False else: return self.dataset_collection_match(dataset_collection)
[docs] def dataset_collection_match(self, dataset_collection): # If dataset collection not yet populated, cannot determine if it # would be a valid match for this parameter. if not dataset_collection.populated: return False valid = True for element in dataset_collection.elements: if not self.__valid_element(element): valid = False break return valid
__all__ = ('DatasetMatcher', 'DatasetCollectionMatcher')