This document is for an old release 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.deps.requirements

import copy
import os
from typing import (

from typing_extensions import (

from galaxy.util import (
from galaxy.util.oset import OrderedSet


[docs]class ToolRequirement: """ Represents an external requirement that must be available for the tool to run (for example, a program, package, or library). Requirements can optionally assert a specific version. """
[docs] def __init__(self, name=None, type=None, version=None, specs=None): if specs is None: specs = [] self.name = name self.type = type self.version = version self.specs = specs
[docs] def to_dict(self): specs = [s.to_dict() for s in self.specs] return dict(name=self.name, type=self.type, version=self.version, specs=specs)
[docs] def copy(self): return copy.deepcopy(self)
[docs] @staticmethod def from_dict(d: Dict[str, Any]) -> "ToolRequirement": version = d.get("version") name = d.get("name") type = d.get("type") specs = [RequirementSpecification.from_dict(s) for s in d.get("specs", [])] return ToolRequirement(name=name, type=type, version=version, specs=specs)
def __eq__(self, other): return ( self.name == other.name and self.type == other.type and self.version == other.version and self.specs == other.specs ) def __ne__(self, other): return not self.__eq__(other) def __hash__(self): return hash((self.name, self.type, self.version, frozenset(self.specs))) def __str__(self): return f"ToolRequirement[{self.name},version={self.version},type={self.type},specs={self.specs}]" __repr__ = __str__
[docs]class RequirementSpecification: """Refine a requirement using a URI."""
[docs] def __init__(self, uri, version=None): self.uri = uri self.version = version
@property def specifies_version(self): return self.version is not None @property def short_name(self): return self.uri.split("/")[-1]
[docs] def to_dict(self): return dict(uri=self.uri, version=self.version)
[docs] @staticmethod def from_dict(dict): uri = dict.get("uri") version = dict.get("version", None) return RequirementSpecification(uri=uri, version=version)
def __eq__(self, other): return self.uri == other.uri and self.version == other.version def __ne__(self, other): return not self.__eq__(other) def __hash__(self): return hash((self.uri, self.version))
[docs]class ToolRequirements: """ Represents all requirements (packages, env vars) needed to run a tool. """
[docs] def __init__(self, tool_requirements=None): if tool_requirements: if not isinstance(tool_requirements, list): raise ToolRequirementsException("ToolRequirements Constructor expects a list") self.tool_requirements = OrderedSet( [r if isinstance(r, ToolRequirement) else ToolRequirement.from_dict(r) for r in tool_requirements] ) else: self.tool_requirements = OrderedSet()
[docs] @staticmethod def from_list(requirements: Union[List[ToolRequirement], Dict[str, Any]]) -> "ToolRequirements": return ToolRequirements(requirements)
@property def resolvable(self): return ToolRequirements([r for r in self.tool_requirements if r.type in {"package", "set_environment"}]) @property def packages(self): return ToolRequirements([r for r in self.tool_requirements if r.type == "package"])
[docs] def to_list(self): return [r.to_dict() for r in self.tool_requirements]
[docs] def append(self, requirement): if not isinstance(requirement, ToolRequirement): requirement = ToolRequirement.from_dict(requirement) self.tool_requirements.add(requirement)
def __eq__(self, other): return ( len(self.tool_requirements & other.tool_requirements) == len(self.tool_requirements) == len(other.tool_requirements) ) def __ne__(self, other): return not self.__eq__(other) def __iter__(self): yield from self.tool_requirements def __getitem__(self, ii): return list(self.tool_requirements)[ii] def __len__(self): return len(self.tool_requirements) def __hash__(self): return sum(r.__hash__() for r in self.tool_requirements)
[docs] def to_dict(self): return [r.to_dict() for r in self.tool_requirements]
[docs]class ToolRequirementsException(Exception): pass
DEFAULT_CONTAINER_TYPE = "docker" DEFAULT_CONTAINER_RESOLVE_DEPENDENCIES = False DEFAULT_CONTAINER_SHELL = "/bin/sh" # Galaxy assumes bash, but containers are usually thinner.
[docs]class ContainerDescription:
[docs] def __init__( self, identifier: Optional[str] = None, type: str = DEFAULT_CONTAINER_TYPE, resolve_dependencies: bool = DEFAULT_CONTAINER_RESOLVE_DEPENDENCIES, shell: str = DEFAULT_CONTAINER_SHELL, ) -> None: # Force to lowercase because container image names must be lowercase. # Cached singularity images include the path on disk, so only lowercase # the image identifier portion. self.identifier = None if identifier: parts = identifier.rsplit(os.sep, 1) parts[-1] = parts[-1].lower() self.identifier = os.sep.join(parts) self.type = type self.resolve_dependencies = resolve_dependencies self.shell = shell self.explicit = False
[docs] def to_dict(self, *args, **kwds): return dict( identifier=self.identifier, type=self.type, resolve_dependencies=self.resolve_dependencies, shell=self.shell, )
[docs] @staticmethod def from_dict(dict): identifier = dict["identifier"] type = dict.get("type", DEFAULT_CONTAINER_TYPE) resolve_dependencies = dict.get("resolve_dependencies", DEFAULT_CONTAINER_RESOLVE_DEPENDENCIES) shell = dict.get("shell", DEFAULT_CONTAINER_SHELL) return ContainerDescription( identifier=identifier, type=type, resolve_dependencies=resolve_dependencies, shell=shell, )
def __str__(self): return f"ContainerDescription[identifier={self.identifier},type={self.type}]"
ResourceType = Literal[ "cores_min", "cores_max", "ram_min", "ram_max", "tmpdir_min", "tmpdir_max", ] VALID_RESOURCE_TYPES = get_args(ResourceType)
[docs]class ResourceRequirement:
[docs] def __init__(self, value_or_expression: Union[int, float, str], resource_type: ResourceType): self.value_or_expression = value_or_expression if not resource_type: raise ValueError("Missing resource requirement type") if resource_type not in VALID_RESOURCE_TYPES: raise ValueError(f"Invalid resource requirement type '{resource_type}'") self.resource_type = resource_type try: float(self.value_or_expression) self.runtime_required = False except ValueError: self.runtime_required = True
[docs] @staticmethod def from_dict(resource_dict): resource_type = next(iter(resource_dict.keys())) value_or_expression = resource_dict[resource_type] return ResourceRequirement(value_or_expression=value_or_expression, resource_type=resource_type)
[docs] def get_value(self, runtime: Optional[Dict] = None, js_evaluator: Optional[Callable] = None): if self.runtime_required: # TODO: hook up evaluator # return js_evaluator(self.value_or_expression, runtime) raise NotImplementedError( f"{self.value_or_expression} is not an integer or float value, expressions currently not implemented" ) return float(self.value_or_expression)
[docs]def resource_requirements_from_list(requirements) -> List[ResourceRequirement]: cwl_to_galaxy = { "coresMin": "cores_min", "coresMax": "cores_max", "ramMin": "ram_min", "ramMax": "ram_max", "tmpdirMin": "tmpdir_min", "tmpdirMax": "tmpdir_max", } rr = [] for r in requirements: if r.get("class") == "ResourceRequirement": valid_key_set = set(cwl_to_galaxy.keys()) elif r.get("type") == "resource": valid_key_set = set(cwl_to_galaxy.values()) else: continue for key in valid_key_set.intersection(set(r.keys())): value = r[key] key = cast(ResourceType, cwl_to_galaxy.get(key, key)) rr.append(ResourceRequirement(value_or_expression=value, resource_type=key)) return rr
[docs]def parse_requirements_from_dict(root_dict): requirements = root_dict.get("requirements", []) resource_requirements = resource_requirements_from_list(requirements) containers = root_dict.get("containers", []) return ( ToolRequirements.from_list(requirements), [ContainerDescription.from_dict(c) for c in containers], resource_requirements, )
[docs]def parse_requirements_from_xml(xml_root, parse_resources=False): """ Parses requirements, containers and optionally resource requirements from Xml tree. >>> from galaxy.util import parse_xml_string >>> def load_requirements(contents, parse_resources=False): ... contents_document = '''<tool><requirements>%s</requirements></tool>''' ... root = parse_xml_string(contents_document % contents) ... return parse_requirements_from_xml(root, parse_resources=parse_resources) >>> reqs, containers = load_requirements('''<requirement>bwa</requirement>''') >>> reqs[0].name 'bwa' >>> reqs[0].version is None True >>> reqs[0].type 'package' >>> reqs, containers = load_requirements('''<requirement type="binary" version="1.3.3">cufflinks</requirement>''') >>> reqs[0].name 'cufflinks' >>> reqs[0].version '1.3.3' >>> reqs[0].type 'binary' """ requirements_elem = xml_root.find("requirements") requirement_elems = [] if requirements_elem is not None: requirement_elems = requirements_elem.findall("requirement") requirements = ToolRequirements() for requirement_elem in requirement_elems: name = xml_text(requirement_elem) type = requirement_elem.get("type", DEFAULT_REQUIREMENT_TYPE) version = requirement_elem.get("version", DEFAULT_REQUIREMENT_VERSION) requirement = ToolRequirement(name=name, type=type, version=version) requirements.append(requirement) container_elems = [] if requirements_elem is not None: container_elems = requirements_elem.findall("container") containers = [container_from_element(c) for c in container_elems] if parse_resources: resource_elems = requirements_elem.findall("resource") if requirements_elem is not None else [] resources = [resource_from_element(r) for r in resource_elems] return requirements, containers, resources return requirements, containers
[docs]def resource_from_element(resource_elem): value_or_expression = xml_text(resource_elem) resource_type = resource_elem.get("type") return ResourceRequirement(value_or_expression=value_or_expression, resource_type=resource_type)
[docs]def container_from_element(container_elem): identifier = xml_text(container_elem) type = container_elem.get("type", DEFAULT_CONTAINER_TYPE) resolve_dependencies = asbool(container_elem.get("resolve_dependencies", DEFAULT_CONTAINER_RESOLVE_DEPENDENCIES)) shell = container_elem.get("shell", DEFAULT_CONTAINER_SHELL) container = ContainerDescription( identifier=identifier, type=type, resolve_dependencies=resolve_dependencies, shell=shell, ) return container