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.dataset_collections.matching

from galaxy import exceptions
from galaxy.util import bunch
from .structure import (

CANNOT_MATCH_ERROR_MESSAGE = "Cannot match collection types."

[docs]class CollectionsToMatch(object): """ Structure representing a set of collections that need to be matched up when running tools (possibly workflows in the future as well). """
[docs] def __init__(self): self.collections = {}
[docs] def add(self, input_name, hdca, subcollection_type=None, linked=True): self.collections[input_name] = bunch.Bunch( hdca=hdca, subcollection_type=subcollection_type, linked=linked, )
[docs] def has_collections(self): return len(self.collections) > 0
[docs] def items(self): return self.collections.items()
[docs]class MatchingCollections(object): """ Structure holding the result of matching a list of collections together. This class being different than the class above and being created in the dataset_collections_service layer may seem like overkill but I suspect in the future plugins will be subtypable for instance so matching collections will need to make heavy use of the dataset collection type registry managed by the dataset collections sevice - hence the complexity now. """
[docs] def __init__(self): self.linked_structure = None self.unlinked_structures = [] self.collections = {} self.subcollection_types = {}
def __attempt_add_to_linked_match(self, input_name, hdca, collection_type_description, subcollection_type): structure = get_structure(hdca, collection_type_description, leaf_subcollection_type=subcollection_type) if not self.linked_structure: self.linked_structure = structure self.collections[input_name] = hdca self.subcollection_types[input_name] = subcollection_type else: if not self.linked_structure.can_match(structure): raise exceptions.MessageException(CANNOT_MATCH_ERROR_MESSAGE) self.collections[input_name] = hdca self.subcollection_types[input_name] = subcollection_type
[docs] def slice_collections(self): return self.linked_structure.walk_collections(self.collections)
[docs] def subcollection_mapping_type(self, input_name): return self.subcollection_types[input_name]
@property def structure(self): """Yield cross product of all unlinked collections structures to linked collection structure.""" effective_structure = leaf for unlinked_structure in self.unlinked_structures: effective_structure = effective_structure.multiply(unlinked_structure) linked_structure = self.linked_structure if linked_structure is None: linked_structure = leaf effective_structure = effective_structure.multiply(linked_structure) return None if effective_structure.is_leaf else effective_structure
[docs] @staticmethod def for_collections(collections_to_match, collection_type_descriptions): if not collections_to_match.has_collections(): return None matching_collections = MatchingCollections() for input_key, to_match in sorted(collections_to_match.items()): hdca = to_match.hdca collection_type_description = collection_type_descriptions.for_collection_type(hdca.collection.collection_type) subcollection_type = to_match.subcollection_type if to_match.linked: matching_collections.__attempt_add_to_linked_match(input_key, hdca, collection_type_description, subcollection_type) else: structure = get_structure(hdca, collection_type_description, leaf_subcollection_type=subcollection_type) matching_collections.unlinked_structures.append(structure) return matching_collections