import itertools
import sys
from types import MappingProxyType
from typing import TYPE_CHECKING, Union
from typing_extensions import Literal, get_args
from weaver.base import Constants
if TYPE_CHECKING:
from typing import Dict
from weaver.typedefs import CWL_NamespaceDefinition
[docs]
IO_SelectOutput_Type = Literal["output"]
[docs]
IO_Select_Type = Literal[IO_SelectInput_Type, IO_SelectOutput_Type]
[docs]
IO_OUTPUT = get_args(IO_SelectOutput_Type)[0]
[docs]
WPS_Literal_Type = Literal["literal"]
[docs]
WPS_Reference_Type = Literal["reference"]
[docs]
WPS_Complex_Type = Literal["complex"]
[docs]
WPS_COMPLEX = get_args(WPS_Complex_Type)[0]
[docs]
WPS_ComplexData_Type = Literal["ComplexData"]
[docs]
WPS_BoundingBoxData_Type = Literal["BoundingBoxData"]
[docs]
WPS_BoundingBox_Type = Literal["bbox"]
[docs]
WPS_BOUNDINGBOX = get_args(WPS_BoundingBox_Type)[0]
[docs]
WPS_CategoryType = Union[
WPS_Literal_Type,
WPS_Reference_Type,
WPS_ComplexData_Type,
WPS_BoundingBoxData_Type,
]
[docs]
WPS_LITERAL = get_args(WPS_Literal_Type)[0]
[docs]
WPS_REFERENCE = get_args(WPS_Reference_Type)[0]
[docs]
WPS_COMPLEX_DATA = get_args(WPS_ComplexData_Type)[0]
[docs]
WPS_BOUNDINGBOX_DATA = get_args(WPS_BoundingBoxData_Type)[0]
[docs]
WPS_LiteralDataBoolean_Type = Literal["bool", "boolean"]
[docs]
WPS_LITERAL_DATA_BOOLEAN = frozenset(get_args(WPS_LiteralDataBoolean_Type))
[docs]
WPS_LiteralDataDateTime_Type = Literal["date", "time", "dateTime"]
[docs]
WPS_LITERAL_DATA_DATETIME = frozenset(get_args(WPS_LiteralDataDateTime_Type))
[docs]
WPS_LiteralDataFloat_Type = Literal["scale", "angle", "float", "double"]
[docs]
WPS_LITERAL_DATA_FLOAT = frozenset(get_args(WPS_LiteralDataFloat_Type))
[docs]
WPS_LiteralDataInteger_Type = Literal[
"int", "integer", "long", "positiveInteger", "nonNegativeInteger"
]
[docs]
WPS_LITERAL_DATA_INTEGER = frozenset(get_args(WPS_LiteralDataInteger_Type))
[docs]
WPS_LiteralDataString_Type = Literal["anyURI", "string"]
[docs]
WPS_LITERAL_DATA_STRING = frozenset(get_args(WPS_LiteralDataString_Type))
[docs]
WPS_LiteralData_Type = Literal[
WPS_LiteralDataBoolean_Type,
WPS_LiteralDataDateTime_Type,
WPS_LiteralDataFloat_Type,
WPS_LiteralDataInteger_Type,
WPS_LiteralDataString_Type,
]
[docs]
WPS_LITERAL_DATA_TYPES = frozenset(get_args(WPS_LiteralData_Type))
# WPS 'type' string variations employed to indicate a Complex (file) I/O by different libraries
# for literal types, see 'any2cwl_literal_datatype' and 'any2wps_literal_datatype' functions
[docs]
WPS_ComplexType = Literal[WPS_Complex_Type, WPS_ComplexData_Type, WPS_Reference_Type]
[docs]
WPS_COMPLEX_TYPES = frozenset(get_args(WPS_ComplexType))
# WPS 'type' string of all combinations (type of data / library implementation)
[docs]
WPS_DataType = Literal[WPS_Literal_Type, WPS_BoundingBox_Type, WPS_ComplexType]
[docs]
WPS_DATA_TYPES = frozenset(get_args(WPS_DataType))
[docs]
class OpenSearchField(Constants):
[docs]
START_DATE = "StartDate"
[docs]
COLLECTION = "collection"
# data source cache
[docs]
LOCAL_FILE_SCHEME = "opensearchfile" # must be a valid url scheme parsable by urlparse
[docs]
CWL_NAMESPACE_CWL_SPEC_ID = "cwl"
[docs]
CWL_NAMESPACE_CWL_SPEC_URL = "https://w3id.org/cwl/cwl#"
[docs]
CWL_NAMESPACE_CWL_SPEC_DEFINITION = MappingProxyType({
CWL_NAMESPACE_CWL_SPEC_ID: CWL_NAMESPACE_CWL_SPEC_URL
}) # type: CWL_NamespaceDefinition
"""
Namespace used to reference :term:`CWL` definitions provided the common specification.
"""
"""
Namespace used to reference :term:`CWL` definitions provided by mod:`cwltool`.
"""
[docs]
CWL_NAMESPACE_SCHEMA_ID = "s"
[docs]
CWL_NAMESPACE_SCHEMA_URL = "https://schema.org/"
[docs]
CWL_NAMESPACE_SCHEMA_DEFINITION = MappingProxyType({
CWL_NAMESPACE_SCHEMA_ID: CWL_NAMESPACE_SCHEMA_URL
}) # type: CWL_NamespaceDefinition
"""
Namespace used to reference :term:`CWL` definitions provided by ``schema.org`` typically used for additional metadata.
"""
"""
Fields that can be directly in the :term:`CWL` contents.
"""
[docs]
CWL_NAMESPACE_OGC_API_PROC_PART1_ID = "ogcapi-processes-1"
[docs]
CWL_NAMESPACE_OGC_API_PROC_PART1_URL = "https://schemas.opengis.net/ogcapi/processes/part1/1.0/openapi/"
# CWL_NAMESPACE_OGC_API_PROC_PART2_ID = "ogcapi-processes-2"
# CWL_NAMESPACE_OGC_API_PROC_PART2_URL = "https://schemas.opengis.net/ogcapi/processes/part2/1.0/openapi/"
[docs]
CWL_NAMESPACE_OGC_API_PROC_DEFINITION = MappingProxyType({
CWL_NAMESPACE_OGC_API_PROC_PART1_ID: CWL_NAMESPACE_OGC_API_PROC_PART1_URL,
# CWL_NAMESPACE_OGC_API_PROC_PART2_ID: CWL_NAMESPACE_OGC_API_PROC_PART2_URL,
}) # type: CWL_NamespaceDefinition
"""
Namespaces used to reference :term:`OGC API - Processes` schema definitions.
"""
# weaver-specific requirements, but non-namespaced for backward support
CWL_NAMESPACES = {} # type: Dict[str, str]
CWL_NAMESPACES.update(CWL_NAMESPACE_CWL_SPEC_DEFINITION)
CWL_NAMESPACES.update(CWL_NAMESPACE_CWLTOOL_DEFINITION)
CWL_NAMESPACES.update(CWL_NAMESPACE_SCHEMA_DEFINITION)
CWL_NAMESPACES.update(CWL_NAMESPACE_OGC_API_PROC_DEFINITION)
[docs]
CWL_NAMESPACES = MappingProxyType(CWL_NAMESPACES) # type: CWL_NamespaceDefinition
[docs]
CWL_NAMESPACES_REVERSED = MappingProxyType({
_urn: _ns for _ns, _urn in CWL_NAMESPACES.items()
}) # type: CWL_NamespaceDefinition
[docs]
CWL_RequirementBuiltinType = Literal["BuiltinRequirement"]
[docs]
CWL_RequirementESGFCWTType = Literal["ESGF-CWTRequirement"]
[docs]
CWL_RequirementOGCAPIType = Literal["OGCAPIRequirement"]
[docs]
CWL_RequirementWPS1Type = Literal["WPS1Requirement"]
# weaver-specific requirements with namespace
[docs]
CWL_RequirementWeaverBuiltinType = Literal["weaver:BuiltinRequirement"]
[docs]
CWL_RequirementWeaverESGFCWTType = Literal["weaver:ESGF-CWTRequirement"]
[docs]
CWL_RequirementWeaverOGCAPIType = Literal["weaver:OGCAPIRequirement"]
[docs]
CWL_RequirementWeaverWPS1Type = Literal["weaver:WPS1Requirement"]
[docs]
CWL_RequirementDockerType = Literal["DockerRequirement"]
[docs]
CWL_RequirementDockerGpuType = Literal["DockerGpuRequirement"]
[docs]
CWL_RequirementCUDANameType = Literal["CUDARequirement"]
[docs]
CWL_RequirementCUDAType = Literal["cwltool:CUDARequirement"]
[docs]
CWL_RequirementEnvVarType = Literal["EnvVarRequirement"]
[docs]
CWL_RequirementInitialWorkDirType = Literal["InitialWorkDirRequirement"]
[docs]
CWL_RequirementInlineJavascriptType = Literal["InlineJavascriptRequirement"]
[docs]
CWL_RequirementInplaceUpdateType = Literal["InplaceUpdateRequirement"]
[docs]
CWL_RequirementLoadListingType = Literal["LoadListingRequirement"]
[docs]
CWL_RequirementMPIType = Literal["MPIRequirement"]
[docs]
CWL_RequirementNetworkAccessType = Literal["NetworkAccess"]
[docs]
CWL_RequirementProcessGeneratorType = Literal["ProcessGenerator"]
[docs]
CWL_RequirementResourceType = Literal["ResourceRequirement"]
[docs]
CWL_RequirementScatterFeatureType = Literal["ScatterFeatureRequirement"]
[docs]
CWL_RequirementSecretsType = Literal["cwltool:Secrets"]
[docs]
CWL_RequirementSubworkflowFeatureType = Literal["SubworkflowFeatureRequirement"]
[docs]
CWL_RequirementWorkReuseType = Literal["WorkReuse"]
# FIXME: convert to 'Constants' class
# CWL package (requirements/hints) corresponding to `ProcessType.APPLICATION`
[docs]
CWL_REQUIREMENT_APP_BUILTIN = get_args(CWL_RequirementBuiltinType)[0]
[docs]
CWL_REQUIREMENT_APP_DOCKER = get_args(CWL_RequirementDockerType)[0]
# backward compatibility, instead use ('DockerRequirement' + 'cwltool:CUDARequirement')
[docs]
CWL_REQUIREMENT_APP_DOCKER_GPU = get_args(CWL_RequirementDockerGpuType)[0]
[docs]
CWL_REQUIREMENT_APP_ESGF_CWT = get_args(CWL_RequirementESGFCWTType)[0]
[docs]
CWL_REQUIREMENT_APP_OGC_API = get_args(CWL_RequirementOGCAPIType)[0]
[docs]
CWL_REQUIREMENT_APP_WPS1 = get_args(CWL_RequirementWPS1Type)[0]
[docs]
CWL_REQUIREMENT_APP_WEAVER_CLASSES = frozenset([
CWL_REQUIREMENT_APP_BUILTIN,
CWL_REQUIREMENT_APP_ESGF_CWT,
CWL_REQUIREMENT_APP_OGC_API,
CWL_REQUIREMENT_APP_WPS1,
])
"""
Set of :term:`CWL` requirements defined by `Weaver` for an :term:`Application Package` implementation.
"""
[docs]
CWL_NAMESPACE_WEAVER_ID = "weaver"
[docs]
CWL_NAMESPACE_WEAVER_URL = "https://schemas.crim.ca/cwl/weaver#"
[docs]
CWL_NAMESPACE_WEAVER_DEFINITION = MappingProxyType({CWL_NAMESPACE_WEAVER_ID: CWL_NAMESPACE_WEAVER_URL})
"""
Namespace used to reference :term:`CWL` definitions provided by `Weaver`.
"""
[docs]
CWL_REQUIREMENT_APP_WEAVER_DEFINITION = MappingProxyType({
f"{CWL_NAMESPACE_WEAVER_ID}:{_req}": f"{CWL_NAMESPACE_WEAVER_URL}{_req}"
for _req in CWL_REQUIREMENT_APP_WEAVER_CLASSES
})
[docs]
CWL_RequirementAppTypes = Literal[
CWL_RequirementBuiltinType,
CWL_RequirementDockerType,
CWL_RequirementDockerGpuType,
CWL_RequirementESGFCWTType,
CWL_RequirementOGCAPIType,
CWL_RequirementWPS1Type,
]
[docs]
CWL_REQUIREMENT_APP_TYPES = frozenset(
list(get_args(CWL_RequirementAppTypes))
+ list(CWL_REQUIREMENT_APP_WEAVER_DEFINITION)
)
"""
Set of :term:`CWL` requirements consisting of known :term:`Application Package` by this `Weaver` instance.
"""
[docs]
CWL_REQUIREMENT_APP_LOCAL = frozenset([
CWL_REQUIREMENT_APP_BUILTIN,
CWL_REQUIREMENT_APP_DOCKER,
CWL_REQUIREMENT_APP_DOCKER_GPU,
])
"""
Set of :term:`CWL` requirements that correspond to local execution of an :term:`Application Package`.
"""
[docs]
CWL_REQUIREMENT_APP_REMOTE = frozenset(
f"{_prefix}{_req}"
for _prefix, _req in itertools.product(
["", f"{CWL_NAMESPACE_WEAVER_ID}:", CWL_NAMESPACE_WEAVER_URL],
[
CWL_REQUIREMENT_APP_ESGF_CWT,
CWL_REQUIREMENT_APP_OGC_API,
CWL_REQUIREMENT_APP_WPS1,
]
)
)
"""
Set of :term:`CWL` requirements that correspond to remote execution of an :term:`Application Package`.
"""
[docs]
CWL_REQUIREMENT_CUDA_DEFAULT_PARAMETERS = MappingProxyType({
# use older minimal version/capability to allow more chances to match any available GPU
# if this causes an issue for an actual application, it must provide it explicitly anyway
"cudaVersionMin": "10.0",
"cudaComputeCapability": "3.0",
# use minimum defaults, single GPU
"cudaDeviceCountMin": 1,
"cudaDeviceCountMax": 1,
})
"""
Parameters employed by default for updating :data:`CWL_REQUIREMENT_APP_DOCKER_GPU` into :data:`CWL_REQUIREMENT_CUDA`.
"""
# FIXME: convert to 'Constants' class
# NOTE: depending on the 'cwlVersion' of the document, some items are extensions or native to the standard specification
[docs]
CWL_REQUIREMENT_CUDA = get_args(CWL_RequirementCUDAType)[0]
[docs]
CWL_REQUIREMENT_CUDA_NAME = get_args(CWL_RequirementCUDANameType)[0]
[docs]
CWL_REQUIREMENT_CUDA_NAMESPACE = CWL_NAMESPACE_CWLTOOL_DEFINITION
[docs]
CWL_REQUIREMENT_ENV_VAR = get_args(CWL_RequirementEnvVarType)[0]
[docs]
CWL_REQUIREMENT_INIT_WORKDIR = get_args(CWL_RequirementInitialWorkDirType)[0]
[docs]
CWL_REQUIREMENT_INLINE_JAVASCRIPT = get_args(CWL_RequirementInlineJavascriptType)[0]
[docs]
CWL_REQUIREMENT_INPLACE_UPDATE = get_args(CWL_RequirementInplaceUpdateType)[0]
[docs]
CWL_REQUIREMENT_LOAD_LISTING = get_args(CWL_RequirementLoadListingType)[0]
[docs]
CWL_REQUIREMENT_MPI = get_args(CWL_RequirementMPIType)[0] # no implication yet
[docs]
CWL_REQUIREMENT_NETWORK_ACCESS = get_args(CWL_RequirementNetworkAccessType)[0]
[docs]
CWL_REQUIREMENT_PROCESS_GENERATOR = get_args(CWL_RequirementProcessGeneratorType)[0]
[docs]
CWL_REQUIREMENT_RESOURCE = get_args(CWL_RequirementResourceType)[0]
[docs]
CWL_REQUIREMENT_SCATTER = get_args(CWL_RequirementScatterFeatureType)[0]
[docs]
CWL_REQUIREMENT_SECRETS = get_args(CWL_RequirementSecretsType)[0]
[docs]
CWL_REQUIREMENT_SUBWORKFLOW = get_args(CWL_RequirementSubworkflowFeatureType)[0]
[docs]
CWL_REQUIREMENT_TIME_LIMIT = get_args(CWL_RequirementToolTimeLimitType)[0]
# default is to reuse, employed to explicitly disable
[docs]
CWL_REQUIREMENT_WORK_REUSE = get_args(CWL_RequirementWorkReuseType)[0]
[docs]
CWL_REQUIREMENT_FEATURES = frozenset([
CWL_REQUIREMENT_CUDA, # note: only allowed in 'hints' because of 'cwltool:' namespace
CWL_REQUIREMENT_CUDA_NAME, # extension import does not have namespace, but it requires it during execution
CWL_REQUIREMENT_ENV_VAR,
CWL_REQUIREMENT_INIT_WORKDIR,
CWL_REQUIREMENT_INPLACE_UPDATE,
CWL_REQUIREMENT_INLINE_JAVASCRIPT,
CWL_REQUIREMENT_LOAD_LISTING,
# CWL_REQUIREMENT_MPI, # no implication yet
CWL_REQUIREMENT_MULTIPLE_INPUT,
CWL_REQUIREMENT_NETWORK_ACCESS,
# CWL_REQUIREMENT_PROCESS_GENERATOR, # explicitly unsupported, works against Weaver's behavior
CWL_REQUIREMENT_RESOURCE, # FIXME: perform pre-check on job submit? (https://github.com/crim-ca/weaver/issues/138)
CWL_REQUIREMENT_SCATTER,
CWL_REQUIREMENT_STEP_INPUT_EXPRESSION,
CWL_REQUIREMENT_SECRETS, # note: only allowed in 'hints' because of 'cwltool:' namespace
CWL_REQUIREMENT_SUBWORKFLOW,
CWL_REQUIREMENT_TIME_LIMIT,
CWL_REQUIREMENT_WORK_REUSE, # allow it, but makes sense only for Workflow steps if cwltool handles it by itself
])
"""
Set of :term:`CWL` requirements that corresponds to extra functionalities.
An :term:`Application Package` that only contains these requirements by themselves would not be considered complete.
These extra requirements must be accompanied by another one from :data:`CWL_REQUIREMENT_APP_TYPES` to be considered
a complete definition.
"""
[docs]
CWL_REQUIREMENTS_SUPPORTED = frozenset(
CWL_REQUIREMENT_APP_TYPES |
CWL_REQUIREMENT_FEATURES
)
"""
Set of all :term:`CWL` requirements or hints that are supported for deployment of valid :term:`Application Package`.
"""
# CWL package types and extensions
[docs]
PACKAGE_EXTENSIONS = frozenset(["yaml", "yml", "json", "cwl", "job"])
[docs]
PACKAGE_INTEGER_TYPES = frozenset(["int", "integer", "long"])
[docs]
PACKAGE_FLOATING_TYPES = frozenset(["float", "double"])
[docs]
PACKAGE_NUMERIC_TYPES = frozenset(PACKAGE_INTEGER_TYPES | PACKAGE_FLOATING_TYPES)
[docs]
PACKAGE_BASIC_TYPES = frozenset({"string", "boolean"} | PACKAGE_NUMERIC_TYPES)
[docs]
PACKAGE_LITERAL_TYPES = frozenset(PACKAGE_BASIC_TYPES | {"null", "Any"})
[docs]
PACKAGE_FILE_TYPE = "File"
[docs]
PACKAGE_DIRECTORY_TYPE = "Directory"
[docs]
PACKAGE_COMPLEX_TYPES = frozenset([PACKAGE_FILE_TYPE, PACKAGE_DIRECTORY_TYPE])
[docs]
PACKAGE_ENUM_BASE = "enum"
[docs]
PACKAGE_CUSTOM_TYPES = frozenset([PACKAGE_ENUM_BASE]) # can be anything, but support "enum" which is more common
[docs]
PACKAGE_ARRAY_BASE = "array"
[docs]
PACKAGE_ARRAY_MAX_SIZE = sys.maxsize # pywps doesn't allow None, so use max size # FIXME: unbounded (weaver #165)
[docs]
PACKAGE_ARRAY_ITEMS = frozenset(PACKAGE_BASIC_TYPES | PACKAGE_CUSTOM_TYPES | PACKAGE_COMPLEX_TYPES)
[docs]
PACKAGE_ARRAY_TYPES = frozenset([f"{item}[]" for item in PACKAGE_ARRAY_ITEMS])
# string values the lowest 'type' field can have by itself (as simple mapping {type: <type-string>})
[docs]
PACKAGE_TYPE_NULLABLE = frozenset(PACKAGE_BASIC_TYPES | PACKAGE_CUSTOM_TYPES | PACKAGE_COMPLEX_TYPES)
# shortcut notations that can be employed to convert basic types into corresponding array or nullable variants
[docs]
PACKAGE_SHORTCUTS = frozenset(
{f"{typ}?" for typ in PACKAGE_TYPE_NULLABLE} |
PACKAGE_ARRAY_TYPES |
{f"{typ}?" for typ in PACKAGE_ARRAY_TYPES}
)
[docs]
PACKAGE_TYPE_POSSIBLE_VALUES = frozenset(
PACKAGE_LITERAL_TYPES |
PACKAGE_COMPLEX_TYPES |
PACKAGE_SHORTCUTS
)
# OpenAPI definitions
[docs]
OAS_COMPLEX_TYPES = frozenset(["object"])
[docs]
OAS_ARRAY_TYPES = frozenset(["array"])
[docs]
OAS_LITERAL_TYPES = frozenset(["boolean", "integer", "number", "string"])
[docs]
OAS_LITERAL_NUMERIC = frozenset(["integer", "number"])
[docs]
OAS_KEYWORD_TYPES = frozenset(["allOf", "anyOf", "oneOf", "not"])
[docs]
OAS_DATA_TYPES = frozenset(
OAS_COMPLEX_TYPES |
OAS_ARRAY_TYPES |
OAS_LITERAL_TYPES
)
# pylint: disable=C0103,invalid-name
[docs]
ProcessSchemaOGCType = Literal["OGC", "ogc"]
[docs]
ProcessSchemaOLDType = Literal["OLD", "old"]
[docs]
ProcessSchemaWPSType = Literal["WPS", "wps"]
[docs]
ProcessSchemaType = Union[ProcessSchemaOGCType, ProcessSchemaOLDType, ProcessSchemaWPSType]
[docs]
JobStatusSchemaType_OGC = Literal["OGC", "ogc"]
[docs]
JobStatusSchemaType_OpenEO = Literal["OPENEO", "openeo", "openEO", "OpenEO"]
[docs]
JobStatusSchemaType = Union[JobStatusSchemaType_OGC, JobStatusSchemaType_OpenEO]
[docs]
class ProcessSchema(Constants):
"""
Schema selector to represent a :term:`Process` description.
"""
[docs]
OGC = "OGC" # type: ProcessSchemaOGCType
[docs]
OLD = "OLD" # type: ProcessSchemaOLDType
[docs]
WPS = "WPS" # type: ProcessSchemaWPSType
[docs]
class JobStatusSchema(Constants):
"""
Schema selector to represent a :term:`Job` status response.
"""
[docs]
OGC = "ogc" # type: JobStatusSchemaType_OGC
[docs]
OPENEO = "openeo" # type: JobStatusSchemaType_OpenEO
if TYPE_CHECKING:
# pylint: disable=invalid-name
[docs]
CWL_RequirementNames = Literal[
CWL_RequirementBuiltinType,
CWL_RequirementESGFCWTType,
CWL_RequirementOGCAPIType,
CWL_RequirementWPS1Type,
CWL_RequirementWeaverBuiltinType,
CWL_RequirementWeaverESGFCWTType,
CWL_RequirementWeaverOGCAPIType,
CWL_RequirementWeaverWPS1Type,
CWL_RequirementDockerType,
CWL_RequirementDockerGpuType,
CWL_RequirementCUDAType,
CWL_RequirementEnvVarType,
CWL_RequirementInitialWorkDirType,
CWL_RequirementInlineJavascriptType,
CWL_RequirementInplaceUpdateType,
CWL_RequirementLoadListingType,
CWL_RequirementMPIType,
CWL_RequirementMultipleInputFeatureType,
CWL_RequirementNetworkAccessType,
CWL_RequirementResourceType,
CWL_RequirementScatterFeatureType,
CWL_RequirementSecretsType,
CWL_RequirementStepInputExpressionType,
CWL_RequirementSubworkflowFeatureType,
CWL_RequirementToolTimeLimitType,
CWL_RequirementWorkReuseType,
]