import copy
import os
from typing import (
Any,
Callable,
cast,
Dict,
List,
Optional,
Union,
)
from typing_extensions import (
get_args,
Literal,
)
from galaxy.util import (
asbool,
xml_text,
)
from galaxy.util.oset import OrderedSet
DEFAULT_REQUIREMENT_TYPE = "package"
DEFAULT_REQUIREMENT_VERSION = None
[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))
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