Source code for weaver.processes.constants

import sys
from types import MappingProxyType
from typing import TYPE_CHECKING, Union
from typing_extensions import Literal, get_args

from weaver.base import Constants

[docs] IO_SelectInput_Type = Literal["input"]
[docs] IO_SelectOutput_Type = Literal["output"]
[docs] IO_Select_Type = Literal[IO_SelectInput_Type, IO_SelectOutput_Type]
[docs] IO_INPUT = get_args(IO_SelectInput_Type)[0]
[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] END_DATE = "EndDate"
[docs] AOI = "aoi"
[docs] COLLECTION = "collection"
# data source cache
[docs] LOCAL_FILE_SCHEME = "opensearchfile" # must be a valid url scheme parsable by urlparse
[docs] CWL_NAMESPACE_ID = "cwl"
[docs] CWL_NAMESPACE_URL = "https://w3id.org/cwl/cwl#"
[docs] CWL_NAMESPACE = MappingProxyType({CWL_NAMESPACE_ID: CWL_NAMESPACE_URL})
[docs] CWL_TOOL_NAMESPACE_ID = "cwltool"
[docs] CWL_TOOL_NAMESPACE_URL = "http://commonwl.org/cwltool#"
[docs] CWL_TOOL_NAMESPACE = MappingProxyType({CWL_TOOL_NAMESPACE_ID: CWL_TOOL_NAMESPACE_URL})
[docs] CWL_RequirementBuiltinType = Literal["BuiltinRequirement"]
[docs] CWL_RequirementDockerType = Literal["DockerRequirement"]
[docs] CWL_RequirementDockerGpuType = Literal["DockerGpuRequirement"]
[docs] CWL_RequirementESGFCWTType = Literal["ESGF-CWTRequirement"]
[docs] CWL_RequirementOGCAPIType = Literal["OGCAPIRequirement"]
[docs] CWL_RequirementWPS1Type = Literal["WPS1Requirement"]
[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["Secrets"]
[docs] CWL_RequirementToolTimeLimitType = Literal["ToolTimeLimit"]
[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_RequirementAppTypes = Literal[ CWL_RequirementBuiltinType, CWL_RequirementDockerType, CWL_RequirementDockerGpuType, CWL_RequirementESGFCWTType, CWL_RequirementOGCAPIType, CWL_RequirementWPS1Type, ]
[docs] CWL_REQUIREMENT_APP_TYPES = frozenset(get_args(CWL_RequirementAppTypes))
""" 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([ 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_NAMESPACE = CWL_TOOL_NAMESPACE
[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_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, 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_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_SECRETS, # FIXME: support CWL Secrets (https://github.com/crim-ca/weaver/issues/511) 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 not completely defining an :term:`Application Package` by themselves. """
[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_LITERAL_FLOAT_FORMATS = frozenset(["float", "double"])
[docs] OAS_LITERAL_INTEGER_FORMATS = frozenset(["int32", "int64"])
[docs] OAS_LITERAL_NUMERIC_FORMATS = frozenset(OAS_LITERAL_FLOAT_FORMATS | OAS_LITERAL_INTEGER_FORMATS)
[docs] OAS_LITERAL_DATETIME_FORMATS = frozenset(["date", "datetime", "date-time", "full-date", "time"])
[docs] OAS_LITERAL_STRING_FORMATS = frozenset( OAS_LITERAL_DATETIME_FORMATS | {"password", "uri", "url"} )
[docs] OAS_LITERAL_BINARY_FORMATS = frozenset(["base64", "binary", "byte"])
[docs] OAS_KEYWORD_TYPES = frozenset(["allOf", "anyOf", "oneOf", "not"])
[docs] OAS_DATA_TYPES = frozenset( OAS_COMPLEX_TYPES | OAS_ARRAY_TYPES | OAS_LITERAL_TYPES )
[docs] class ProcessSchema(Constants): """ Schema selector to represent a :term:`Process` description. """
[docs] OGC = "OGC"
[docs] OLD = "OLD"
[docs] WPS = "WPS"
[docs] class JobInputsOutputsSchema(Constants): """ Schema selector to represent a :term:`Job` output results. """
[docs] OGC_STRICT = "ogc+strict"
[docs] OLD_STRICT = "old+strict"
[docs] OGC = "ogc"
[docs] OLD = "old"
if TYPE_CHECKING: # pylint: disable=invalid-name
[docs] CWL_RequirementNames = Literal[ CWL_RequirementBuiltinType, CWL_RequirementDockerType, CWL_RequirementDockerGpuType, CWL_RequirementESGFCWTType, CWL_RequirementOGCAPIType, CWL_RequirementWPS1Type, CWL_RequirementCUDAType, CWL_RequirementEnvVarType, CWL_RequirementInitialWorkDirType, CWL_RequirementInlineJavascriptType, CWL_RequirementInplaceUpdateType, CWL_RequirementLoadListingType, CWL_RequirementMPIType, CWL_RequirementNetworkAccessType, CWL_RequirementResourceType, CWL_RequirementScatterFeatureType, CWL_RequirementSecretsType, CWL_RequirementToolTimeLimitType, CWL_RequirementWorkReuseType, ]
ProcessSchemaType = Literal["OGC", "ogc", "OLD", "old", "WPS", "wps"] JobInputsOutputsSchemaType = Literal[ "ogc+strict", "OGC+STRICT", "old+strict", "OLD+STRICT", "ogc", "OGC", "old", "OLD", ]