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.tool_util.parser.output_objects

from typing import (
    Any,
    Dict,
    List,
    Optional,
    Type,
    Union,
)

from typing_extensions import TypedDict

from galaxy.util import Element
from galaxy.util.dictifiable import Dictifiable
from .output_actions import ToolOutputActionGroup
from .output_collection_def import (
    dataset_collector_descriptions_from_output_dict,
    DatasetCollectionDescription,
)
from .output_models import (
    ToolOutputBoolean as ToolOutputBooleanModel,
    ToolOutputCollection as ToolOutputCollectionModel,
    ToolOutputCollectionStructure as ToolOutputCollectionStructureModel,
    ToolOutputDataset as ToolOutputDataModel,
    ToolOutputFloat as ToolOutputFloatModel,
    ToolOutputInteger as ToolOutputIntegerModel,
    ToolOutputT as ToolOutputModel,
    ToolOutputText as ToolOutputTextModel,
)


[docs]class ChangeFormatModel(TypedDict): value: Optional[str] format: Optional[str] input: Optional[str] input_dataset: Optional[str] check_attribute: Optional[str]
[docs]class ToolOutputBase(Dictifiable): name: str label: Optional[str] hidden: bool
[docs] def __init__( self, name: str, label: Optional[str] = None, filters: Optional[List[Element]] = None, hidden: bool = False, from_expression: Optional[str] = None, ) -> None: super().__init__() self.name = name self.label = label self.filters = filters or [] self.hidden = hidden self.collection = False self.from_expression = from_expression
[docs] def to_dict(self, view="collection", value_mapper=None, app=None): return super().to_dict(view=view, value_mapper=value_mapper)
@property def output_discover_patterns(self) -> List[str]: return []
[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", "output_type", "format_source", "default_identifier_source", "metadata_source", "parent", "count", "from_work_dir", ]
[docs] def __init__( self, name: str, format: Optional[str] = None, format_source: Optional[str] = None, metadata_source: Optional[str] = None, parent: Optional[str] = None, label: Optional[str] = None, filters: Optional[List[Element]] = None, actions: Optional[ToolOutputActionGroup] = None, hidden: bool = False, implicit: bool = False, from_expression: Optional[str] = None, ) -> None: super().__init__(name, label=label, filters=filters, hidden=hidden, from_expression=from_expression) self.output_type = "data" 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: List[ChangeFormatModel] = [] self.implicit = implicit self.from_work_dir: Optional[str] = None self.dataset_collector_descriptions: List[DatasetCollectionDescription] = [] self.default_identifier_source: Optional[str] = None self.count: Optional[int] = None self.tool: Optional[Any]
# Tuple emulation def __len__(self) -> int: 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().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 if self.dataset_collector_descriptions: as_dict["discover_datasets"] = [d.to_dict() for d in self.dataset_collector_descriptions] else: as_dict["discover_datasets"] = [] return as_dict
[docs] def to_model(self) -> ToolOutputDataModel: return ToolOutputDataModel( type="data", name=self.name, label=self.label, hidden=self.hidden, format=self.format, format_source=self.format_source, metadata_source=self.metadata_source or None, # model is decorated as Optional discover_datasets=[d.to_model() for d in self.dataset_collector_descriptions], )
[docs] @staticmethod def from_dict(name: str, output_dict: Dict[str, Any], tool: Optional[object] = None) -> "ToolOutput": output = ToolOutput(name) output.format = output_dict.get("format", "data") output.change_format = [] output.format_source = output_dict.get("format_source", None) output.default_identifier_source = output_dict.get("default_identifier_source", None) output.metadata_source = output_dict.get("metadata_source", "") output.parent = output_dict.get("parent", None) output.label = output_dict.get("label", None) output.count = output_dict.get("count", 1) output.filters = [] output.tool = tool output.from_work_dir = output_dict.get("from_work_dir", None) output.hidden = output_dict.get("hidden", False) # TODO: implement tool output action group fixes output.actions = ToolOutputActionGroup(output, None) output.dataset_collector_descriptions = dataset_collector_descriptions_from_output_dict(output_dict) return output
@property def output_discover_patterns(self) -> List[str]: return _merge_dataset_collector_descriptions_patterns(self.dataset_collector_descriptions)
[docs]class ToolExpressionOutput(ToolOutputBase): dict_collection_visible_keys = ("name", "format", "label", "hidden", "output_type")
[docs] def __init__(self, name, output_type, from_expression, label=None, filters=None, actions=None, hidden=False): super().__init__(name, label=label, filters=filters, hidden=hidden) self.output_type = output_type # JSON type... self.from_expression = from_expression self.format = "expression.json" # galaxy.datatypes.text.ExpressionJson.file_ext self.format_source = None self.metadata_source = None self.parent = None self.actions = actions # Initialize default values self.change_format = [] self.implicit = False self.from_work_dir = None
[docs] def to_model(self) -> ToolOutputModel: model_class: Union[ Type[ToolOutputIntegerModel], Type[ToolOutputFloatModel], Type[ToolOutputBooleanModel], Type[ToolOutputTextModel], ] model_type = self.output_type if self.output_type == "integer": model_class = ToolOutputIntegerModel elif self.output_type == "float": model_class = ToolOutputFloatModel elif self.output_type == "bool": model_class = ToolOutputBooleanModel model_type = "boolean" elif self.output_type == "text": model_class = ToolOutputTextModel assert model_class return model_class( type=model_type, name=self.name, label=self.label, hidden=self.hidden, )
[docs]class ToolOutputCollection(ToolOutputBase): """ Represents a HistoryDatasetCollectionAssociation of output datasets produced by a tool. .. code-block:: <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", "format", "label", "hidden", "output_type", "default_format", "default_format_source", "default_metadata_source", "inherit_format", "inherit_metadata", ]
[docs] def __init__( self, name: str, structure: "ToolOutputCollectionStructure", label: Optional[str] = None, filters: Optional[List[Element]] = None, hidden: bool = False, default_format: str = "data", default_format_source: Optional[str] = None, default_metadata_source: Optional[str] = None, inherit_format: bool = False, inherit_metadata: bool = False, ) -> None: super().__init__(name, label=label, filters=filters, hidden=hidden) self.output_type = "collection" self.collection = True self.default_format = default_format self.structure = structure self.outputs: Dict[str, str] = {} 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: List = [] # TODO: not implemented
[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=None): parent_ids = parent_ids or [] 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=None): parent_ids = parent_ids or [] 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] def to_dict(self, view="collection", value_mapper=None, app=None): as_dict = super().to_dict(view=view, value_mapper=value_mapper, app=app) as_dict["structure"] = self.structure.to_dict() return as_dict
[docs] def to_model(self) -> ToolOutputCollectionModel: return ToolOutputCollectionModel( type="collection", name=self.name, label=self.label, hidden=self.hidden, structure=self.structure.to_model(), )
[docs] @staticmethod def from_dict(name, output_dict, tool=None) -> "ToolOutputCollection": structure = ToolOutputCollectionStructure.from_dict(output_dict["structure"]) rval = ToolOutputCollection( name, structure=structure, label=output_dict.get("label", None), filters=[], hidden=output_dict.get("hidden", False), default_format=output_dict.get("default_format", "data"), default_format_source=output_dict.get("default_format_source", None), default_metadata_source=output_dict.get("default_metadata_source", None), inherit_format=output_dict.get("inherit_format", False), inherit_metadata=output_dict.get("inherit_metadata", False), ) return rval
@property def output_discover_patterns(self) -> List[str]: return self.structure.output_discover_patterns
[docs]class ToolOutputCollectionStructure: collection_type: Optional[str] collection_type_source: Optional[str] collection_type_from_rules: Optional[str] structured_like: Optional[str] dataset_collector_descriptions: Optional[List[DatasetCollectionDescription]] dynamic: bool
[docs] def __init__( self, collection_type: Optional[str], collection_type_source: Optional[str] = None, collection_type_from_rules: Optional[str] = None, structured_like: Optional[str] = None, dataset_collector_descriptions: Optional[List[DatasetCollectionDescription]] = None, ) -> 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 or [] 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 (discover_datasets) and collection type attributes structured_like or collection_type_from_rules." ) self.dynamic = bool(dataset_collector_descriptions)
[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] def to_dict(self): return self.to_model().dict()
[docs] def to_model(self) -> ToolOutputCollectionStructureModel: discover_datasets = [] if self.dataset_collector_descriptions: discover_datasets = [d.to_model() for d in self.dataset_collector_descriptions] return ToolOutputCollectionStructureModel( collection_type=self.collection_type, collection_type_source=self.collection_type_source, collection_type_from_rules=self.collection_type_from_rules, structured_like=self.structured_like, discover_datasets=discover_datasets, )
[docs] @staticmethod def from_dict(as_dict) -> "ToolOutputCollectionStructure": structure = ToolOutputCollectionStructure( collection_type=as_dict["collection_type"], collection_type_source=as_dict["collection_type_source"], collection_type_from_rules=as_dict["collection_type_from_rules"], structured_like=as_dict["structured_like"], dataset_collector_descriptions=dataset_collector_descriptions_from_output_dict(as_dict), ) return structure
@property def output_discover_patterns(self) -> List[str]: if not self.dataset_collector_descriptions: return [] else: return _merge_dataset_collector_descriptions_patterns(self.dataset_collector_descriptions)
def _merge_dataset_collector_descriptions_patterns( dataset_collector_descriptions: List[DatasetCollectionDescription], ) -> List[str]: patterns = [] for description in dataset_collector_descriptions: patterns.extend(description.discover_patterns) return patterns
[docs]class ToolOutputCollectionPart:
[docs] def __init__(self, output_collection_def, element_identifier, output_def, parent_ids=None): parent_ids = parent_ids or [] 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 = f"{name}|__part__|{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__|")