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]
def get_child_collection(item):
    # item could be HDCA or DCE
    return getattr(item, "child_collection", item.collection)




[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] = get_child_collection(collection_instance).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(
                get_child_collection(hdca).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