Source code for galaxy.model.dataset_collections.matching

from typing import Optional

from galaxy import exceptions
from galaxy.util import bunch
from .structure import (
    get_structure,
    leaf,
)

CANNOT_MATCH_ERROR_MESSAGE = "Cannot match collection types."


[docs]class CollectionsToMatch: """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: """Structure holding the result of matching a list of collections together. This class being different than the class above and being created in the DatasetCollectionManager 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 service - hence the complexity now. """
[docs] def __init__(self): self.linked_structure = None self.unlinked_structures = [] self.collections = {} self.subcollection_types = {} self.action_tuples = {} self.when_values = None
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): self.linked_structure.when_values = self.when_values 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) effective_structure.when_values = self.when_values return None if effective_structure.is_leaf else effective_structure
[docs] def map_over_action_tuples(self, input_name): if input_name not in self.action_tuples: collection_instance = self.collections[input_name] self.action_tuples[input_name] = collection_instance.collection.dataset_action_tuples return self.action_tuples[input_name]
[docs] def is_mapped_over(self, input_name): return input_name in self.collections
[docs] @staticmethod def for_collections(collections_to_match, collection_type_descriptions) -> Optional["MatchingCollections"]: 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