Extend tool API to allow cross product mapping over collections.

By default if single data parameters are mapped over by collections, the collections are matched up (e.g. for two collections the items are paired off and two collections of dimension (N) will result in running N jobs and producing an (N) dimensional output for each define output). In the parameter meta value wrapper (where batch mode can be set with the ``batch`` flag), the linked flag is now respected for collection operations. If ``linked`` is ``False``, then the cross product of the inputs will be used to map over the tool. In the above simplest case, this would cause an NxN (``list:list``) collection to be created for each tool output.

This operation was previously available for individual datasets, but when supplied a collection it would not create an implicit collection pulling together all the relevant datasets with the correct structure - it would just run the jobs and leave the datasets uncollected. Now a collection with the correct structure and identifiers is created.

Limitations:
-----------------

This does not enable tool form support but the API for collections matches that for doing cross product operations over sets of individual datasets, so once support is added to the tool form for that collection support should be trivial.

Testing:
-----------------

The following test case has been extended to now ensure that an implicit collection is created and that it has the right dimensionality/structure and contents.

    ./run_tests.sh -with_framework_test_tools -api test/api/test_tools.py:ToolsTestCase.test_map_over_two_collections_unlinked
This commit is contained in:
John Chilton
2015-09-18 09:47:44 +01:00
parent 63e3596540
commit 84b5726ca9
5 changed files with 100 additions and 21 deletions
+27 -10
View File
@@ -1,6 +1,6 @@
from galaxy.util import bunch
from galaxy import exceptions
from .structure import get_structure
from .structure import get_structure, leaf
CANNOT_MATCH_ERROR_MESSAGE = "Cannot match collection types."
@@ -13,10 +13,11 @@ class CollectionsToMatch( object ):
def __init__( self ):
self.collections = {}
def add( self, input_name, hdca, subcollection_type=None ):
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,
)
def has_collections( self ):
@@ -37,21 +38,32 @@ class MatchingCollections( object ):
"""
def __init__( self ):
self.structure = None
self.linked_structure = None
self.unlinked_structures = []
self.collections = {}
def __attempt_add_to_match( self, input_name, hdca, collection_type_description, subcollection_type ):
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.structure:
self.structure = structure
if not self.linked_structure:
self.linked_structure = structure
self.collections[ input_name ] = hdca
else:
if not self.structure.can_match( structure ):
if not self.linked_structure.can_match( structure ):
raise exceptions.MessageException( CANNOT_MATCH_ERROR_MESSAGE )
self.collections[ input_name ] = hdca
def slice_collections( self ):
return self.structure.walk_collections( self.collections )
return self.linked_structure.walk_collections( self.collections )
@property
def structure( self ):
"""Yield cross product of all unlinked datasets to linked dataset."""
effective_structure = leaf
for unlinked_structure in self.unlinked_structures:
effective_structure = effective_structure.multiply( unlinked_structure )
linked_structure = self.linked_structure or leaf
effective_structure = effective_structure.multiply( linked_structure )
return None if effective_structure.is_leaf else effective_structure
@staticmethod
def for_collections( collections_to_match, collection_type_descriptions ):
@@ -61,8 +73,13 @@ class MatchingCollections( object ):
matching_collections = MatchingCollections()
for input_key, to_match in collections_to_match.iteritems():
hdca = to_match.hdca
subcollection_type = to_match.subcollection_type
collection_type_description = collection_type_descriptions.for_collection_type( hdca.collection.collection_type )
matching_collections.__attempt_add_to_match( input_key, hdca, collection_type_description, subcollection_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
+31 -5
View File
@@ -15,23 +15,34 @@ class Leaf( object ):
def is_leaf( self ):
return True
def clone( self ):
return self
def multiply( self, other_structure ):
return other_structure.clone()
leaf = Leaf()
class Tree( object ):
def __init__( self, dataset_collection, collection_type_description ):
def __init__( self, children, collection_type_description ):
self.children = children
self.collection_type_description = collection_type_description
@staticmethod
def for_dataset_collection( dataset_collection, collection_type_description ):
children = []
for element in dataset_collection.elements:
if collection_type_description.has_subcollections():
child_collection = element.child_collection
subcollection_type_description = collection_type_description.subcollection_type_description() # Type description of children
children.append( ( element.element_identifier, Tree( child_collection, collection_type_description=subcollection_type_description ) ) )
tree = Tree.for_dataset_collection( child_collection, collection_type_description=subcollection_type_description )
children.append( ( element.element_identifier, tree ) )
else:
children.append( ( element.element_identifier, leaf ) )
self.children = children
return Tree( children, collection_type_description )
def walk_collections( self, hdca_dict ):
return self._walk_collections( dict_map( lambda hdca: hdca.collection, hdca_dict ) )
@@ -96,6 +107,21 @@ class Tree( object ):
element_identifiers=element_identifiers,
)
def multiply( self, other_structure ):
if other_structure.is_leaf:
return self.clone()
new_collection_type = self.collection_type_description.multiply( other_structure.collection_type_description )
new_children = []
for (identifier, structure) in self.children:
new_children.append( (identifier, structure.multiply( other_structure ) ) )
return Tree( new_children, new_collection_type )
def clone( self ):
cloned_children = map( lambda (identifier, element): (identifier, element.clone()), self.children )
return Tree( cloned_children, self.collection_type_description )
def dict_map( func, input_dict ):
return dict( [ ( k, func(v) ) for k, v in input_dict.iteritems() ] )
@@ -105,4 +131,4 @@ def get_structure( dataset_collection_instance, collection_type_description, lea
if leaf_subcollection_type:
collection_type_description = collection_type_description.effective_collection_type_description( leaf_subcollection_type )
return Tree( dataset_collection_instance.collection, collection_type_description )
return Tree.for_dataset_collection( dataset_collection_instance.collection, collection_type_description )
@@ -102,6 +102,10 @@ class CollectionTypeDescription( object ):
def rank_type_plugin( self ):
return self.collection_type_description_factory.type_registry.get( self.rank_collection_type() )
def multiply( self, other_collection_type ):
collection_type = map_over_collection_type( self, other_collection_type )
return self.collection_type_description_factory.for_collection_type( collection_type )
def __str__( self ):
return "CollectionTypeDescription[%s]" % self.collection_type
+3 -3
View File
@@ -35,7 +35,7 @@ def expand_meta_parameters( trans, tool, incoming ):
classification = permutations.input_classification.SINGLE
if __collection_multirun_parameter( value ):
collection_value = value[ 'values' ][ 0 ]
values = __expand_collection_parameter( trans, input_key, collection_value, collections_to_match )
values = __expand_collection_parameter( trans, input_key, collection_value, collections_to_match, linked=is_linked )
else:
values = value[ 'values' ]
else:
@@ -102,7 +102,7 @@ def expand_meta_parameters( trans, tool, incoming ):
return expanded_incomings, collection_info
def __expand_collection_parameter( trans, input_key, incoming_val, collections_to_match ):
def __expand_collection_parameter( trans, input_key, incoming_val, collections_to_match, linked=False ):
# If subcollectin multirun of data_collection param - value will
# be "hdca_id|subcollection_type" else it will just be hdca_id
if "|" in incoming_val:
@@ -119,7 +119,7 @@ def __expand_collection_parameter( trans, input_key, incoming_val, collections_t
subcollection_type = None
hdc_id = trans.app.security.decode_id( encoded_hdc_id )
hdc = trans.sa_session.query( model.HistoryDatasetCollectionAssociation ).get( hdc_id )
collections_to_match.add( input_key, hdc, subcollection_type=subcollection_type )
collections_to_match.add( input_key, hdc, subcollection_type=subcollection_type, linked=linked )
if subcollection_type is not None:
from galaxy.dataset_collections import subcollections
subcollection_elements = subcollections.split_dataset_collection_instance( hdc, subcollection_type )
+35 -3
View File
@@ -804,9 +804,41 @@ class ToolsTestCase( api.ApiTestCase ):
self.assertEquals( len( outputs ), 4 )
self.assertEquals( len( response_object[ 'jobs' ] ), 4 )
# Implicit collections not created with unlinked inputs yet - this may
# be problematic.
self.assertEquals( len( response_object[ 'implicit_collections' ] ), 0 )
implicit_collections = response_object[ 'implicit_collections' ]
self.assertEquals( len( implicit_collections ), 1 )
implicit_collection = implicit_collections[ 0 ]
self.assertEquals( implicit_collection[ "collection_type" ], "paired:paired" )
outer_elements = implicit_collection[ "elements" ]
assert len( outer_elements ) == 2
element0, element1 = outer_elements
assert element0[ "element_identifier" ] == "forward"
assert element1[ "element_identifier" ] == "reverse"
elements0 = element0[ "object" ][ "elements" ]
elements1 = element1[ "object" ][ "elements" ]
assert len( elements0 ) == 2
assert len( elements1 ) == 2
element00, element01 = elements0
assert element00[ "element_identifier" ] == "forward"
assert element01[ "element_identifier" ] == "reverse"
element10, element11 = elements1
assert element10[ "element_identifier" ] == "forward"
assert element11[ "element_identifier" ] == "reverse"
expected_contents_list = [
(element00, "123\n789\n"),
(element01, "123\n0ab\n"),
(element10, "456\n789\n"),
(element11, "456\n0ab\n"),
]
for (element, expected_contents) in expected_contents_list:
dataset_id = element["object"]["id"]
contents = self.dataset_populator.get_history_dataset_content( history_id, dataset_id=dataset_id )
self.assertEquals(expected_contents, contents)
@skip_without_tool( "cat1" )
def test_map_over_collected_and_individual_datasets( self ):