Warning

This document is for an in-development version 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.parser.output_objects

from galaxy.util.dictifiable import Dictifiable
from galaxy.util.odict import odict


[docs]class ToolOutputBase(Dictifiable):
[docs] def __init__(self, name, label=None, filters=None, hidden=False): super(ToolOutputBase, self).__init__() self.name = name self.label = label self.filters = filters or [] self.hidden = hidden self.collection = False
[docs] def to_dict(self, view='collection', value_mapper=None, app=None): return super(ToolOutputBase, self).to_dict(view=view, value_mapper=value_mapper)
[docs]class ToolOutput(ToolOutputBase): """ Represents an output datasets produced by a tool. For backward compatibility this behaves as if it were the tuple:: (format, metadata_source, parent) """ dict_collection_visible_keys = ['name', 'format', 'label', 'hidden']
[docs] def __init__(self, name, format=None, format_source=None, metadata_source=None, parent=None, label=None, filters=None, actions=None, hidden=False, implicit=False): super(ToolOutput, self).__init__(name, label=label, filters=filters, hidden=hidden) self.format = format self.format_source = format_source self.metadata_source = metadata_source self.parent = parent self.actions = actions # Initialize default values self.change_format = [] self.implicit = implicit self.from_work_dir = None
# Tuple emulation def __len__(self): return 3 def __getitem__(self, index): if index == 0: return self.format elif index == 1: return self.metadata_source elif index == 2: return self.parent else: raise IndexError(index) def __iter__(self): return iter((self.format, self.metadata_source, self.parent))
[docs] def to_dict(self, view='collection', value_mapper=None, app=None): as_dict = super(ToolOutput, self).to_dict(view=view, value_mapper=value_mapper, app=app) format = self.format if format and format != "input" and app: edam_format = app.datatypes_registry.edam_formats.get(self.format) as_dict["edam_format"] = edam_format edam_data = app.datatypes_registry.edam_data.get(self.format) as_dict["edam_data"] = edam_data return as_dict
[docs]class ToolOutputCollection(ToolOutputBase): """ Represents a HistoryDatasetCollectionAssociation of output datasets produced by a tool. <outputs> <collection type="list" label="${tool.name} on ${on_string} fasta"> <discover_datasets pattern="__name__" ext="fasta" visible="True" directory="outputFiles" /> </collection> <collection type="paired" label="${tool.name} on ${on_string} paired reads"> <data name="forward" format="fastqsanger" /> <data name="reverse" format="fastqsanger"/> </collection> <outputs> """ dict_collection_visible_keys = ['name', 'default_format', 'label', 'hidden', 'inherit_format', 'inherit_metadata']
[docs] def __init__( self, name, structure, label=None, filters=None, hidden=False, default_format="data", default_format_source=None, default_metadata_source=None, inherit_format=False, inherit_metadata=False ): super(ToolOutputCollection, self).__init__(name, label=label, filters=filters, hidden=hidden) self.collection = True self.default_format = default_format self.structure = structure self.outputs = odict() self.inherit_format = inherit_format self.inherit_metadata = inherit_metadata self.metadata_source = default_metadata_source self.format_source = default_format_source self.change_format = [] # TODO
[docs] def known_outputs(self, inputs, type_registry): if self.dynamic_structure: return [] # This line is probably not right - should verify structured_like # or have outputs and all outputs have name. if len(self.outputs) > 1: output_parts = [ToolOutputCollectionPart(self, k, v) for k, v in self.outputs.items()] else: collection_prototype = self.structure.collection_prototype(inputs, type_registry) def prototype_dataset_element_to_output(element, parent_ids=[]): name = element.element_identifier format = self.default_format if self.inherit_format: format = element.dataset_instance.ext output = ToolOutput( name, format=format, format_source=self.format_source, metadata_source=self.metadata_source, implicit=True, ) if self.inherit_metadata: output.metadata_source = element.dataset_instance return ToolOutputCollectionPart( self, element.element_identifier, output, parent_ids=parent_ids, ) def prototype_collection_to_output(collection_prototype, parent_ids=[]): output_parts = [] for element in collection_prototype.elements: element_parts = [] if not element.is_collection: element_parts.append(prototype_dataset_element_to_output(element, parent_ids)) else: new_parent_ids = parent_ids[:] + [element.element_identifier] element_parts.extend(prototype_collection_to_output(element.element_object, new_parent_ids)) output_parts.extend(element_parts) return output_parts output_parts = prototype_collection_to_output(collection_prototype) return output_parts
@property def dynamic_structure(self): return self.structure.dynamic @property def dataset_collector_descriptions(self): if not self.dynamic_structure: raise Exception("dataset_collector_descriptions called for output collection with static structure") return self.structure.dataset_collector_descriptions
[docs]class ToolOutputCollectionStructure(object):
[docs] def __init__( self, collection_type, collection_type_source=None, collection_type_from_rules=None, structured_like=None, dataset_collector_descriptions=None, ): self.collection_type = collection_type self.collection_type_source = collection_type_source self.collection_type_from_rules = collection_type_from_rules self.structured_like = structured_like self.dataset_collector_descriptions = dataset_collector_descriptions if collection_type and collection_type_source: raise ValueError("Cannot set both type and type_source on collection output.") if collection_type is None and structured_like is None and dataset_collector_descriptions is None and collection_type_source is None and collection_type_from_rules is None: raise ValueError("Output collection types must specify source of collection type information (e.g. structured_like or type_source).") if dataset_collector_descriptions and (structured_like or collection_type_from_rules): raise ValueError("Cannot specify dynamic structure (discovered_datasets) and collection type attributes structured_like or collection_type_from_rules.") self.dynamic = dataset_collector_descriptions is not None
[docs] def collection_prototype(self, inputs, type_registry): # either must have specified structured_like or something worse if self.structured_like: collection_prototype = inputs[self.structured_like].collection else: collection_type = self.collection_type assert collection_type collection_prototype = type_registry.prototype(collection_type) collection_prototype.collection_type = collection_type return collection_prototype
[docs]class ToolOutputCollectionPart(object):
[docs] def __init__(self, output_collection_def, element_identifier, output_def, parent_ids=[]): self.output_collection_def = output_collection_def self.element_identifier = element_identifier self.output_def = output_def self.parent_ids = parent_ids
@property def effective_output_name(self): name = self.output_collection_def.name part_name = self.element_identifier effective_output_name = "%s|__part__|%s" % (name, part_name) return effective_output_name
[docs] @staticmethod def is_named_collection_part_name(name): return "|__part__|" in name
[docs] @staticmethod def split_output_name(name): assert ToolOutputCollectionPart.is_named_collection_part_name(name) return name.split("|__part__|")