weaver.processes.convert ======================== .. py:module:: weaver.processes.convert .. autoapi-nested-parse:: Conversion functions between corresponding data structures. Module Contents --------------- .. py:data:: WPS_Input_Type .. py:data:: WPS_FIELD_MAPPING .. py:data:: WPS_FIELD_FORMAT :value: ['formats', 'supported_formats', 'supported_values', 'default'] .. py:data:: INPUT_VALUE_TYPE_MAPPING .. py:data:: UNIT_REGISTRY .. py:data:: UNIT_OGC_REVERSED :type: Dict[str, List[str]] .. py:data:: LOGGER .. py:function:: convert_unit(unit: str) -> pint.Unit Convert units considering any known registry names and :term:`OGC` :term:`URN`. Since :term:`OGC` defines units using :term:`URN`, they cannot be registered as aliases in :data:`UNIT_REGISTRY`. This is a limitation from the naming format that requires valid Python identifier names, which is not possible due to the ``:`` characters in :term:`URN` references. .. py:function:: convert_value_units(value: pint._typing.Scalar, uom: str, to: str) -> pint._typing.Scalar Converts the provided value from one :term:`UoM` to another. .. py:function:: complex2json(data: Union[owslib.wps.ComplexData, Any]) -> Union[weaver.typedefs.JSON, Any] Obtains the :term:`JSON` representation of a :class:`ComplexData` or simply return the unmatched type. .. py:function:: metadata2json(meta: Union[ANY_Metadata_Type, Any], force: bool = False) -> Union[weaver.typedefs.JSON, Any] Retrieve metadata information and generate its :term:`JSON` representation. Obtains the :term:`JSON` representation of a :class:`OWS_Metadata` or :class:`pywps.app.Common.Metadata`. Otherwise, simply return the unmatched type. If requested, can enforce parsing a dictionary for the corresponding keys. .. py:function:: uom2json(uom: Union[str, pywps.inout.basic.UOM, JSON_UoM], ref: Optional[str] = None) -> JSON_UoM Convert an :term:`UoM` definition into corresponding :term:`JSON` representation. .. py:function:: ows2json_io(ows_io: OWS_IO_Type) -> JSON_IO_Type Converts :term:`I/O` definition from :mod:`owslib.wps` to :term:`JSON`. .. py:function:: ows2json_output_data(output: OWS_Output_Type, process_description: owslib.wps.Process, container: Optional[weaver.typedefs.AnySettingsContainer] = None) -> weaver.typedefs.JSON Utility method to convert an :mod:`owslib.wps` process execution output data (result) to :term:`JSON`. In the case that a ``reference`` output of `JSON` content-type is specified and that it refers to a file that contains an array list of URL references to simulate a multiple-output, this specific output gets expanded to contain both the original URL ``reference`` field and the loaded URL list under ``data`` field for easier access from the response body. Referenced file(s) are fetched in order to store them locally if executed on a remote process, such that they can become accessible as local job result for following reporting or use by other processes in a workflow chain. If the ``dataType`` details is missing from the data output (depending on servers that might omit it), the :paramref:`process_description` is employed to retrieve the original description with expected result details. :param output: Output with data value or reference according to expected result for the corresponding process. :param process_description: Definition of the process producing the specified output following execution. :param container: Container to retrieve application settings (for request options during file retrieval as needed). :return: Converted :term:`JSON` result data and additional metadata as applicable based on data-type and content-type. .. py:function:: any2json_output_data(output_info: Union[OWS_Output_Type, weaver.typedefs.JSON], output_data: Union[weaver.typedefs.JSON, owslib.ows.BoundingBox, weaver.typedefs.AnyValueType]) -> weaver.typedefs.JSON Converts :mod:`owslib` or :mod:`pywps` :term:`JSON` equivalent output data to normalized :term:`JSON` format. Only processes a single item. Must be called iteratively for each data entry in a list. .. py:function:: any2json_literal_data(data: weaver.typedefs.AnyValueType, data_type: weaver.processes.constants.WPS_LiteralData_Type) -> weaver.typedefs.AnyValueType Converts :mod:`owslib` :term:`WPS` literal data using strings into the specific :term:`JSON` compatible type. .. py:function:: ows2json_bbox_data(bbox: owslib.ows.BoundingBox) -> weaver.typedefs.JSON Converts :mod:`owslib` :term:`WPS` Bounding Box data into a :term:`JSON` representation. .. py:function:: _get_multi_json_references(output: OWS_Output_Type, container: Optional[weaver.typedefs.AnySettingsContainer]) -> Optional[List[weaver.typedefs.JSON]] Obtains the :term:`JSON` contents of a single output corresponding to multi-file references. Since WPS standard does not allow to return multiple values for a single output, a lot of process actually return a :term:`JSON` array containing references to these outputs. Because the multi-output references are contained within this :term:`JSON` file, it is not very convenient to retrieve the list of URLs as one always needs to open and read the file to get them. This function goal is to detect this particular format and expand the references to make them quickly available in the job output response. :return: Array of HTTP(S) references if the specified output a :term:`JSON` with URL references, ``None`` otherwise. .. py:function:: get_io_type_category(io_info: ANY_IO_Type) -> weaver.processes.constants.WPS_CategoryType Guesses the applicable :term:`I/O` type with provided information from any known :term:`I/O` structure. .. py:function:: _get_cwl_fmt_details(wps_fmt: ANY_Format_Type) -> Union[Tuple[Tuple[str, str], str, str], Tuple[None, None, None]] .. py:function:: _convert_any2cwl_io_complex(cwl_io: weaver.typedefs.CWL_IO_Type, cwl_ns: Dict[str, str], wps_io: Union[JSON_IO_Type, WPS_IO_Type, OWS_IO_Type], io_select: weaver.processes.constants.IO_Select_Type) -> None Converts the :term:`WPS`-like :term:`I/O` definition and defines them inplace into the :term:`CWL` containers. .. seealso:: See :meth:`weaver.processes.wps_process_base.WpsProcessInterface.stage_results` which closely interacts with the produced ``outputBinding.glob`` patterns generated here. Methodology should align between them. :param cwl_io: Basic :term:`CWL` :term:`I/O` container (only ID needed) where to write conversion results. :param cwl_ns: Namespaces to gradually update when encountering new format Media-Type definitions. :param wps_io: Original :term:`WPS`-like :term:`I/O` to be converted. :param io_select: Context of the :term:`I/O`. :return: Nothing. Changed inplace. .. py:function:: _get_cwl_js_value_from(cwl_io_symbols: List[weaver.typedefs.AnyValueType], allow_unique: bool, allow_array: bool) -> str Obtain the JavaScript ``valueFrom`` definition for a :term:`CWL` input of non-``string`` allowed values. .. py:function:: _convert_cwl_io_enum(cwl_io_type: Union[str, Type[weaver.utils.null]], cwl_io_symbols: List[weaver.typedefs.AnyValueType], io_select: weaver.processes.constants.IO_Select_Type, allow_unique: bool, allow_array: bool) -> weaver.typedefs.CWL_IO_Type Converts the :term:`I/O` definition to a :term:`CWL` :term:`I/O` that allows ``enum``-like functionality. In the event of an explicit ``string`` as base type, :term:`CWL` directly supports ``type: enum``. Other basic types are not directly supported, and must instead perform manual validation against the set of allowed values, using JavaScript evaluation applied by ``valueFrom`` against the the submitted values to replicate the automatic behavior performed by ``enum`` type. .. seealso:: - https://github.com/common-workflow-language/cwl-v1.2/issues/267 - https://github.com/common-workflow-language/common-workflow-language/issues/764 - https://github.com/common-workflow-language/common-workflow-language/issues/907 .. warning:: Because ``valueFrom`` can only be used with ``inputBinding``, any output providing a set of allowed values that are not ``string``-based will be ignored when converted to :term:`CWL` :term:`I/O`. .. seealso:: - :func:`_get_cwl_js_value_from` defines the ``enum``-like ``valueFrom`` checks - :func:`_patch_cwl_enum_js_requirement` must be called on the entire :term:`CWL` to inject the relevant :data:`CWL_REQUIREMENT_INLINE_JAVASCRIPT`, since it might not be defined in the original :term:`CWL`. Another edge-case that happens even if the base type is a ``string`` is when the enum ``symbols`` contain an entry with a ``:`` character (e.g.: as in the case of a time ``HH:MM`` string). In such case, :mod:`schema_salad` incorrectly parses it as if a namespaced :term:`URL` (i.e.: ``:``) was specified. Because of this, the symbol strings get extended to an unresolvable namespace, which leads to a partial and incorrect enum value (e.g.: ``abc:def`` becomes ``input-id#def``). This causes the resulting ``enum`` to never accept the submitted values, or worst, causes the entire :term:`CWL` to fail validation if duplicate values are obtained (e.g.: ``12:00`` and ``24:00`` both extend to ``input-id#00``, causing a duplicate ``00``). To handle this case, the ``symbols`` can be updated with a prefixed ``#`` character, making :mod:`schema_salad` interpret it as if it was already a :term:`URI` relative to the input, which is what it aims to generate, and therefore interprets the rest of the string (including the ``:``) literally. .. seealso:: - https://github.com/common-workflow-language/cwltool/issues/2071 :param cwl_io_type: Basic type for which allowed values should apply. :param cwl_io_symbols: Allowed values to restrict the :term:`I/O` definition. :return: Converted definition as CWL Enum or with relevant value validation as applicable for the type. .. py:function:: any2cwl_io(wps_io: Union[JSON_IO_Type, WPS_IO_Type, OWS_IO_Type], io_select: weaver.processes.constants.IO_Select_Type) -> Tuple[weaver.typedefs.CWL_IO_Type, Dict[str, str]] Converts a :term:`WPS`-like :term:`I/O` from various :term:`WPS` library representations to :term:`CWL` :term:`I/O`. Conversion can be accomplished for :mod:`pywps` and :mod:`owslib` objects, as well as their :term:`JSON` equivalent. Because :term:`CWL` :term:`I/O` of type ``File`` with ``format`` field are namespaced, this is also returned if needed. :returns: converted :term:`I/O` and namespace dictionary with corresponding format references as required. .. py:function:: _patch_cwl_enum_js_requirement(cwl_package: weaver.typedefs.CWL) -> None Applies the JavaScript requirement to validate a pseudo-``Enum`` applied to a :term:`CWL` input definition. .. seealso:: - :func:`any2cwl_io` - :func:`_convert_cwl_io_enum` - :func:`_get_cwl_js_value_from` .. py:function:: wps2cwl_requirement(wps_service_url: Union[str, urllib.parse.ParseResult], wps_process_id: str) -> weaver.typedefs.JSON Obtains the `CWL` requirements definition needed for parsing by a remote `WPS` provider as an `Application Package`. .. py:function:: ows2json(wps_process: owslib.wps.Process, wps_service_name: str, wps_service_url: Union[str, urllib.parse.ParseResult], wps_provider_name: Optional[str] = None) -> Tuple[weaver.typedefs.CWL, weaver.typedefs.JSON] Generates the `CWL` package and process definitions from a :class:`owslib.wps.Process` hosted under `WPS` location. .. py:function:: xml_wps2cwl(wps_process_response: requests.models.Response, settings: weaver.typedefs.AnySettingsContainer) -> Tuple[weaver.typedefs.CWL, weaver.typedefs.JSON] Obtains the :term:`CWL` definition that corresponds to an :term:`XML` :term:`WPS` process. Converts a `WPS-1 ProcessDescription XML` tree structure to an equivalent `WPS-3 Process JSON`, and builds the associated :term:`CWL` package in conformance to :data:`weaver.processes.wps_package.CWL_REQUIREMENT_APP_WPS1`. :param wps_process_response: Valid response (XML, 200) from a `WPS-1 ProcessDescription`. :param settings: Application settings to retrieve additional request options. .. py:function:: ogcapi2cwl_process(payload: weaver.typedefs.JSON, reference: str) -> Tuple[weaver.typedefs.CWL, weaver.typedefs.JSON] Generate a :term:`CWL` for a remote :term:`OGC API - Processes` description to dispatch :term:`Process` execution. .. seealso:: - :class:`weaver.processes.wps3_process.Wps3Process` :param payload: :term:`JSON` :term:`Process` description in :term:`OGC API - Processes` format. :param reference: URL where the :term:`Process` is located. :returns: Updated :term:`CWL` package with the reference to the :term:`Process`. .. py:function:: is_cwl_complex_type(io_info: weaver.typedefs.CWL_IO_Type, complex_types: Iterable[weaver.typedefs.CWL_IO_ComplexType] = PACKAGE_COMPLEX_TYPES) -> bool Identifies if the provided :term:`CWL` input/output corresponds to one, many or a potential `Complex` type(s). When multiple distinct *atomic* types are allowed for a given I/O (e.g.: ``type: [string, File]``) and that one of them is one of the considered `Complex` type, the result will be ``True`` even if other types are not `Complex`. Similarly, optional `Complex` types combined with ``"null"`` will also return ``True``. :param io_info: :term:`I/O` to verify for complex type. :param complex_types: Complex types to consider. By default, any type between :term:`CWL` ``File`` and ``Directory`` are valid. The operation can be limited to one or the other if needed to identify a specific one. .. py:function:: parse_cwl_array_type(io_info: weaver.typedefs.CWL_IO_Type, strict: bool = True) -> CWLIODefinition Parses the specified :term:`I/O` for one of the various potential CWL array definitions. :param io_info: :term:`CWL` :term:`I/O` definition to parse. :param strict: Indicates if only pure :term:`CWL` definition is allowed, or allow implicit data-type conversions. :returns: Updated :term:`CWL` :term:`I/O` definition with applicable properties. :raises PackageTypeError: if the array element doesn't have the required values and valid format. .. py:function:: parse_cwl_enum_type(io_info: weaver.typedefs.CWL_IO_Type) -> CWLIODefinition Parses the specified I/O for potential CWL enum definition. :returns: Updated :term:`CWL` I/O definition with applicable properties. :raises PackageTypeError: if the enum doesn't have the required parameters and valid format. .. py:function:: get_cwl_io_type_name(io_type: Any) -> Any Obtain the simple type-name representation of a :term:`CWL` I/O. Depending on :mod:`cwltool` version, types are represented with or without an extended prefix, and using an explicit quoted class representation rather than plain strings. .. py:function:: resolve_cwl_io_type_schema(io_info: weaver.typedefs.CWL_IO_Type, cwl_schema_names: Optional[weaver.typedefs.CWL_SchemaNames] = None) -> weaver.typedefs.CWL_IO_Type Reverse :term:`CWL` schema references by name back to their full :term:`CWL` I/O definition. .. seealso:: - :meth:`weaver.processes.wps_package.WpsPackage.make_inputs` - :meth:`weaver.processes.wps_package.WpsPackage.update_cwl_schema_names` .. py:function:: resolve_cwl_namespaced_name(name: str) -> str Remove any :term:`URN` prefixes added by :term:`CWL` from a name. Includes removal of contextual reference of the source :term:`CWL` file that contained the name. Includes reversing :term:`CWL`-specific namespaces :term:`URN` extended to their full URL form. .. py:class:: CWLIODefinition Utility :term:`CWL` I/O definition to contain metadata from parsing results. .. seealso:: :func:`weaver.processes.convert.get_cwl_io_type` .. py:method:: keys() -> List[str] .. py:attribute:: name :type: str :value: '' Name (or identifier) or the I/O. .. py:attribute:: type :type: Union[CWL_IO_LiteralType, CWL_IO_ComplexType, CWL_IO_DataType] :value: None Type of the :term:`CWL` I/O. If :attr:`enum` is ``True``, represents the enum base type. If :attr:`array` is ``True``, represents the item type. .. note:: Before resolution with :func:`parse_cwl_array_type`, this attribute can temporarily hold any :term:`CWL` type structure (list, dict, nested types, etc.). After parsing, it will be resolved to the basic string type. .. py:attribute:: null :type: bool :value: False Indicates if the I/O is nullable. This is obtained from a type composed of ``"null"`` and something else, or using the shorthand ``{type}?`` notation. .. py:attribute:: min_occurs :type: int :value: 1 Minimum number of occurrences allowed. When :attr:`null` is ``True``, it is equal to ``0``. Otherwise, it is greater or equal to ``1``. If greater than ``1``, :attr:`array` should be ``True``. .. py:attribute:: max_occurs :type: int :value: 1 Maximum number of occurrences allowed. Applies only when :attr:`array` is ``True``. Otherwise, always equal to ``1``. Can take the value :data:`PACKAGE_ARRAY_MAX_SIZE` to represent ``"unbounded"`` occurrences. .. py:attribute:: array :type: bool :value: False Specifies if the I/O is of array type. .. py:attribute:: enum :type: bool :value: False Specifies if the I/O is of enum type. .. py:attribute:: symbols :type: Union[CWL_IO_EnumSymbols, AnyValue, Type[AnyValue]] Specifies the allowed values when the definition is marked as :attr:`enum`. When not overridden by literal values, it uses the default :class:`AnyValue`. .. py:attribute:: mode :type: pywps.validator.mode.MODE :value: 0 Validation mode to be applied if I/O requires it. Defaults to :attr:`MODE.NONE`. Indicates how strict the validation must be. Usually applies when an enum must only allow a specific set of symbols. Can also be used with Media-Types in more advanced validation use case with :mod:`pywps`. .. py:function:: get_cwl_io_type(io_info: weaver.typedefs.CWL_IO_Type, strict: bool = True, cwl_schema_names: Optional[weaver.typedefs.CWL_SchemaNames] = None) -> CWLIODefinition Obtains the basic type of the CWL input and identity if it is optional. CWL allows multiple shorthand representation or combined types definition. The *base* type must be extracted in order to identify the expected data format and supported values. Obtains real type if ``"default"`` or shorthand ``"?"`` was in CWL, which can also be defined as type ``["null", ]``. CWL allows multiple distinct types (e.g.: ``string`` and ``int`` simultaneously), but not WPS inputs. WPS allows only different amount of *same type* through ``minOccurs`` and ``maxOccurs``. Considering WPS conversion, we can also have the following definition ``["null", , ]`` will all basic types matching exactly. Whether single or array-like type, the base type can be extracted. :param io_info: :term:`CWL` definition to parse. :param strict: Indicates if only pure :term:`CWL` definition is allowed, or allow implicit data-type conversions. :param cwl_schema_names: Mapping of CWL type schema references to resolve in long form if used in a definition. :return: tuple of guessed base type and flag indicating if it can be null (optional input). .. py:function:: cwl2wps_io(io_info: weaver.typedefs.CWL_IO_Type, io_select: weaver.processes.constants.IO_Select_Type) -> WPS_IO_Type Converts input/output parameters from CWL types to WPS types. :param io_info: parsed IO of a CWL file :param io_select: :py:data:`IO_INPUT` or :py:data:`IO_OUTPUT` to specify desired WPS type conversion. :returns: corresponding IO in WPS format .. py:function:: cwl2json_input_values(data: weaver.typedefs.CWL_IO_ValueMap, schema: weaver.processes.constants.ProcessSchemaType = ProcessSchema.OGC) -> weaver.typedefs.ExecutionInputs Converts :term:`CWL` formatted :term:`Job` inputs to corresponding :term:`OGC API - Processes` format. :param data: dictionary with inputs formatted as key-value pairs with relevant structure based on :term:`CWL` types. :param schema: either ``OGC`` or ``OLD`` format respectively for mapping/listing representations. :raises TypeError: if input data is invalid. :raises ValueError: if any input value could not be parsed with expected schema. :returns: converted inputs for :term:`Job` submission either in ``OGC`` or ``OLD`` format. .. py:function:: convert_input_values_schema(inputs: weaver.typedefs.ExecutionInputs, schema: weaver.processes.constants.JobInputsOutputsSchema.OGC) -> weaver.typedefs.ExecutionInputsMap convert_input_values_schema(inputs: weaver.typedefs.ExecutionInputs, schema: weaver.processes.constants.JobInputsOutputsSchema.OLD) -> weaver.typedefs.ExecutionInputsList Convert execution input values between equivalent formats. .. seealso:: - :func:`convert_output_params_schema` - :func:`normalize_ordered_io` for I/O definitions. :param inputs: Inputs to convert. :param schema: Desired schema. :return: Converted inputs. .. py:function:: convert_output_params_schema(outputs: Optional[weaver.typedefs.ExecutionOutputs], schema: weaver.processes.constants.JobInputsOutputsSchemaAnyOGCType) -> Optional[weaver.typedefs.ExecutionOutputsMap] convert_output_params_schema(outputs: Optional[weaver.typedefs.ExecutionOutputs], schema: weaver.processes.constants.JobInputsOutputsSchemaAnyOLDType) -> Optional[weaver.typedefs.ExecutionOutputsList] convert_output_params_schema(outputs: Optional[weaver.typedefs.ExecutionOutputs], schema: weaver.processes.constants.JobInputsOutputsSchemaType) -> Optional[weaver.typedefs.ExecutionOutputs] Convert execution output parameters between equivalent formats. .. warning:: These outputs are not *values* (i.e.: *results*), but *submitted* :term:`Job` outputs for return definitions. Contents are transferred as-is without any consideration of ``value`` or ``href`` fields. .. seealso:: - :func:`convert_input_values_schema` - :func:`normalize_ordered_io` for I/O definitions. :param outputs: Outputs to convert. :param schema: Desired schema. :return: Converted outputs. .. py:function:: repr2json_input_params(value: str, converter: Optional[Callable[[str], Any]] = None) -> DataInputType Extracts and converts the value and its associated parameters from a :term:`KVP` string representation. This function only interprets a pre-extracted single-value definition (i.e.: without the input ID) from a parent :term:`KVP` string. .. seealso:: Use :func:`repr2json_input_values` For parsing multi-value arrays and the full :term:`KVP` including the ID. :param value: String representation of the value to be interpreted. :param converter: Conversion function of the value after parsing. :return: Converted value and additional parameters if applicable. .. py:function:: repr2json_input_values(inputs: List[str]) -> weaver.typedefs.ExecutionInputsList Converts inputs in string :term:`KVP` representation to corresponding :term:`JSON` values. Expected format of the input is as follows: .. code-block:: text input_id[:input_type]=input_value[@input_parameter][;input_array[@input_parameter]][;...] Where: - ``input_id`` represents the target identifier of the input - ``input_type`` represents the conversion type, as required (includes ``File`` for ``href`` instead of ``value`` key in resulting object) - ``input_value`` represents the desired value subject to conversion by ``input_type`` - ``input_array`` represents any additional values for array-like inputs (``maxOccurs > 1``) - ``input_parameter`` represents additional :term:`KVP` details associated to each ``input_value``/``input_array`` part (i.e.: per array element if applicable) The separator character for representing array-like values is ``;`` because the full :term:`KVP` (already split into a list as argument to this function), could be formed of multiple comma (``,``) or ampersand (``&``) separated input definitions, depending on where the definition came from (e.g.: URI). The ``input_parameter`` portion can combine multiple parameters each separated by ``@`` and themselves formed with :term:`KVP` representation of the corresponding parameter names and values. Parameter names do not need to be consistent between distinct array elements. For example, a multi-parameters input could be formatted as follows: .. code-block:: text input_id=item_value1@param1=value1@param2=value2;item_value2@other1=value1 .. note:: - Any character that matches one of the separators that should be interpreted literally should be URL-encoded. - Single (``'``) and double (``"``) quotes are removed if they delimit a ``File`` reference. :param inputs: list of string inputs to parse. :return: parsed inputs if successful. .. py:function:: any2cwl_literal_datatype(io_type: str) -> Union[str, Type[weaver.utils.null]] Solves common literal data-type names to supported ones for `CWL`. .. py:function:: any2wps_literal_datatype(io_type: Union[weaver.typedefs.AnyValueType, weaver.processes.constants.WPS_LiteralData_Type], is_value: bool = False, pywps: bool = False) -> Union[str, Type[weaver.utils.null]] Solves common literal data-type names to supported ones for `WPS`. Verification is accomplished by name when ``is_value=False``, otherwise with python ``type`` when ``is_value=True``. :param io_type: Type to convert to :term:`WPS` supported literal data type. :param is_value: If enabled, consider :paramref:`io_type` literal data itself to attempt detection of the type. :param pywps: If enabled, restrict only to types supported by :mod:`pywps` (subset of full :term:`WPS`). .. py:function:: any2json_literal_allowed_value(io_allow: Union[pywps.inout.literaltypes.AllowedValue, weaver.typedefs.JSON, str, float, int, bool]) -> Union[weaver.typedefs.JSON, str, str, float, int, bool, Type[weaver.utils.null]] Converts an ``AllowedValues`` definition from different packages into standardized JSON representation of `OGC-API`. .. py:function:: any2json_literal_data_domains(io_info: ANY_IO_Type) -> Union[Type[weaver.utils.null], List[weaver.typedefs.JSON]] Extracts allowed value constrains from the input definition and generate the expected literal data domains. The generated result, if applicable, corresponds to a list of a single instance of schema definition :class:`weaver.wps_restapi.swagger_definitions.LiteralDataDomainList` with following structure. .. code-block:: yaml default: bool defaultValue: float, int, bool, str dataType: {name: string, } UOMs: - default: {uom: string, reference: url-string} - supported: [{uom: string, reference: url-string}] valueDefinition: oneOf: - string - url-string - {anyValue: bool} - [float, int, bool, str] - [{minimum: number/none, maximum: number/none, spacing: number/none, closure: str open/close variations}] .. py:function:: json2oas_io_complex(io_info: JSON_IO_Type, io_hint: Union[weaver.typedefs.OpenAPISchema, Type[weaver.utils.null]] = null) -> weaver.typedefs.OpenAPISchema Convert a single-dimension complex :term:`JSON` I/O definition into corresponding :term:`OpenAPI` schema. .. py:function:: json2oas_io_bbox(io_info: JSON_IO_Type, io_hint: Union[weaver.typedefs.OpenAPISchema, Type[weaver.utils.null]] = null) -> weaver.typedefs.OpenAPISchema Convert a single-dimension bounding box :term:`JSON` I/O definition into corresponding :term:`OpenAPI` schema. .. seealso:: :data:`sd.OGC_API_BBOX_SCHEMA` .. py:function:: json2oas_io_literal_data_type(io_type: str) -> weaver.typedefs.JSON Converts various literal data types into corresponding :term:`OpenAPI` fields. .. seealso:: - https://spec.openapis.org/oas/v3.1.0#data-types - https://swagger.io/specification/#data-types .. py:function:: json2oas_io_allowed_values(io_base: weaver.typedefs.JSON, io_allowed: weaver.typedefs.JSON) -> List[weaver.typedefs.JSON] Converts literal data allowed values :term:`JSON` definitions ino :term:`OpenAPI` equivalent variations. :param io_base: Base value definitions that can be shared across variations (e.g.: default values). :param io_allowed: Allowed values definitions (enum, ranges) extracted from :term:`JSON` literal data domains. :return: List of converted :term:`OpenAPI` definitions applicable to represent the allowed values. .. py:function:: json2oas_io_literal(io_info: JSON_IO_Type, io_hint: Union[weaver.typedefs.OpenAPISchema, Type[weaver.utils.null]] = null) -> weaver.typedefs.OpenAPISchema Convert a single-dimension literal value :term:`JSON` I/O definition into corresponding :term:`OpenAPI` schema. .. py:function:: json2oas_io(io_info: JSON_IO_Type, io_hint: Union[weaver.typedefs.OpenAPISchema, Type[weaver.utils.null]] = null) -> weaver.typedefs.OpenAPISchema Converts definitions from a :term:`JSON` :term:`Process` I/O definition into corresponding :term:`OpenAPI` schema. :param io_info: :term:`WPS` I/O definition to generate a corresponding :term:`OpenAPI` schema. :param io_hint: Reference :term:`OpenAPI` definition that can improve more explicit object definitions. .. py:function:: oas2json_io_literal(io_info: weaver.typedefs.OpenAPISchemaProperty) -> Union[JSON_IO_TypedInfo, Type[weaver.utils.null]] Converts a literal value I/O definition by :term:`OpenAPI` schema into the equivalent :term:`JSON` representation. :param io_info: :term:`OpenAPI` schema of the I/O. :return: Converted :term:`JSON` I/O definition, or :data:`null` if definition could not be resolved. .. py:function:: oas2json_io_array(io_info: weaver.typedefs.OpenAPISchemaArray) -> Union[JSON_IO_TypedInfo, Type[weaver.utils.null]] Converts an array I/O definition by :term:`OpenAPI` schema into the equivalent :term:`JSON` representation. :param io_info: :term:`OpenAPI` schema of the I/O. :return: Converted :term:`JSON` I/O definition, or :data:`null` if definition could not be resolved. .. py:function:: oas2json_io_object(io_info: weaver.typedefs.OpenAPISchemaObject, io_href: str = null) -> Union[JSON_IO_TypedInfo, Type[weaver.utils.null]] Converts an object I/O definition by :term:`OpenAPI` schema into the equivalent :term:`JSON` representation. An explicit :term:`OpenAPI` schema with ``object`` type can represent any of the following I/O: - Bounding Box as GeoJSON feature - Complex JSON structure .. seealso:: :func:`oas2json_io_file` is used for file reference to be parsed as other Complex I/O. :param io_info: :term:`OpenAPI` schema of the I/O. :param io_href: Alternate schema reference for the type. :return: Converted :term:`JSON` I/O definition, or :data:`null` if definition could not be resolved. .. py:function:: oas2json_io_keyword(io_info: weaver.typedefs.OpenAPISchemaKeyword) -> Union[JSON_IO_TypedInfo, Type[weaver.utils.null]] Converts a keyword I/O definition by :term:`OpenAPI` schema into the equivalent :term:`JSON` representation. Keywords are defined as a list of combinations of :term:`OpenAPI` schema representing how to combine them according to the keyword value, being one of :data:`OAS_KEYWORD_TYPES`. :param io_info: :term:`OpenAPI` schema of the I/O. :return: Converted :term:`JSON` I/O definition, or :data:`null` if definition could not be resolved. .. py:function:: oas2json_io_file(io_info: weaver.typedefs.OpenAPISchemaObject, io_href: str = null) -> JSON_IO_TypedInfo Converts a file reference I/O definition by :term:`OpenAPI` schema into the equivalent :term:`JSON` representation. :param io_info: :term:`OpenAPI` schema of the I/O. :param io_href: Alternate schema reference for the type. :return: Converted :term:`JSON` I/O definition, or :data:`null` if definition could not be resolved. .. py:function:: oas2json_io_measure(io_info: weaver.typedefs.OpenAPISchemaObject) -> Union[JSON_IO_TypedInfo, Type[weaver.utils.null]] Convert a unit of measure (``UoM``) I/O definition by :term:`OpenAPI` schema into :term:`JSON` representation. This conversion projects an object (normally complex type) into a literal type, considering that other provided parameters are all metadata information. :param io_info: Potential :term:`OpenAPI` schema of an UoM I/O. :return: Converted I/O if it matched the UoM format, or null otherwise. .. py:function:: oas2json_io(io_info: weaver.typedefs.OpenAPISchema) -> Union[JSON_IO_TypedInfo, Type[weaver.utils.null]] Converts an :term:`I/O` definition by :term:`OpenAPI` schema into the equivalent :term:`JSON` representation. :param io_info: :term:`OpenAPI` schema of the :term:`I/O`. :return: Converted :term:`JSON` :term:`I/O` definition, or :data:`null` if definition could not be resolved. .. py:function:: oas_resolve_remote(io_info: weaver.typedefs.OpenAPISchema) -> weaver.typedefs.OpenAPISchema Perform remote :term:`OpenAPI` schema ``$ref`` resolution. Resolution is performed only sufficiently to provide enough context for following :term:`JSON` :term:`I/O` conversion. Remote references are not resolved further than required to speedup loading time and avoid recursive error on self-referring schema. Passed sufficient levels of schema definitions, the specific contents is not important nor needs to be resolved as there is they cannot be mapped to anything else than :data:`WPS_COMPLEX` :term:`I/O` type. :param io_info: :term:`I/O` :term:`OpenAPI` schema to attempt resolution as applicable. :return: Resolved :term:`I/O` schema or directly the provided schema returned unmodified if no references need resolution. .. py:function:: json2wps_datatype(io_info: JSON_IO_Type) -> str Converts a JSON input definition into the corresponding :mod:`pywps` parameters. Guesses the literal data-type from :term:`JSON` :term:`I/O` information in order to allow creation of the corresponding :term:`WPS` :term:`I/O`. Defaults to ``string`` if no suitable guess can be accomplished. .. py:function:: json2wps_field(field_info: weaver.typedefs.JSON, field_category: str) -> Any Converts an :term:`I/O` field from :term:`JSON` literal, list, or dictionary to corresponding :term:`WPS` types. :param field_info: literal data or information container describing the type to be generated. :param field_category: one of :data:`WPS_FIELD_MAPPING` keys to indicate how to parse ``field_info``. .. py:function:: json2wps_supported_uoms(io_info: JSON_IO_Type) -> Union[Type[weaver.utils.null], List[pywps.inout.basic.UOM]] Obtains instances of supported Unit of Measure (:term:`UoM`) from a :term:`JSON` :term:`I/O` definition. .. py:function:: json2wps_allowed_values(io_info: JSON_IO_Type) -> Union[Type[weaver.utils.null], List[pywps.inout.literaltypes.AllowedValue]] Obtains the allowed values constrains for the literal data type from a :term:`JSON` :term:`I/O` definition. Converts the ``literalDataDomains`` definition into ``allowed_values`` understood by :mod:`pywps`. Handles explicit ``allowed_values`` if available and not previously defined by ``literalDataDomains``. .. seealso:: Function :func:`any2json_literal_data_domains` defines generated ``literalDataDomains`` JSON definition. .. py:function:: json2wps_io(io_info: JSON_IO_Type, io_select: weaver.processes.constants.IO_Select_Type) -> WPS_IO_Type Converts an :term:`I/O` from a :term:`JSON` dict to :mod:`pywps` types. :param io_info: :term:`I/O` in :term:`JSON` dict format. :param io_select: :data:`IO_INPUT` or :data:`IO_OUTPUT` to specify desired :term:`WPS` type conversion. :return: corresponding :term:`I/O` in :term:`WPS` format. .. py:function:: wps2json_io(io_wps: WPS_IO_Type, forced_fields: bool = False) -> JSON_IO_Type Converts a :mod:`pywps` :term:`I/O` into a :term:`JSON` dictionary with corresponding standard keys names. Employs standard key names as defined by :term:`WPS` 2.0. :param io_wps: Any :mod:`pywps` :term:`I/O` definition to be converted to :term:`JSON` representation. :param forced_fields: Request transfer of additional fields normally undefined for outputs if they are available by being forcefully inserted in the objects after their creation (i.e.: using :func:`set_field`). These fields can be useful for obtaining mandatory details for further processing operations (e.g.: :term:`OpenAPI` schema conversion). .. py:function:: wps2json_job_payload(wps_request: pywps.app.WPSRequest, wps_process: pywps.Process) -> weaver.typedefs.JSON Converts the input and output values of a :mod:`pywps` WPS ``Execute`` request to corresponding WPS-REST job. The inputs and outputs must be parsed from XML POST payload or KVP GET query parameters, and converted to data container defined by :mod:`pywps` based on the process definition. .. py:function:: get_field(io_object: Union[weaver.typedefs.JSON, object], field: str, search_variations: bool = False, extra_variations: Optional[List[str]] = None, only_variations: bool = False, pop_found: bool = False, key: bool = False, default: Any = null) -> Any Gets a field by name from various :term:`I/O` object types. Default value is :py:data:`null` used for most situations to differentiate from literal ``None`` which is often used as default for parameters. The :class:`NullType` allows to explicitly tell that there was 'no field' and not 'no value' in existing field. If you provided another value, it will be returned if not found within the input object. When :paramref:`search_variation` is enabled and that :paramref:`field` could not be found within the object, field lookup will employ the values under the :paramref:`field` entry within :data:`WPS_FIELD_MAPPING` as additional field names to search for an existing property or key. Search continues until the first match is found, respecting order within the variations listing, and finally uses :paramref:`default` if no match was found. :param io_object: Any :term:`I/O` representation, either as a class instance or :term:`JSON` container. :param field: Name of the field to look for, either as property or key name based on input object type. :param search_variations: If enabled, search for all variations to the field name to attempt search until matched. :param extra_variations: Additional field names to consider as search variations, with priority over field mapping. :param only_variations: If enabled, skip the first 'basic' field and start search directly with field variations. :param pop_found: If enabled, whenever a match is found by field or variations, remove that entry from the object. :param key: If enabled, whenever a match is found by field or variations, return matched key instead of the value. :param default: Alternative default value to return if no match could be found. :returns: Matched value (including search variations if enabled), or ``default``. .. py:function:: set_field(io_object: Union[weaver.typedefs.JSON, object], field: str, value: Any, force: bool = False) -> None Sets a field by name into various :term:`I/O` object types. Field value is set only if not ``null`` to avoid inserting data considered `invalid`. If ``force=True``, verification of ``null`` value is ignored. .. py:function:: _are_different_and_set(item1: Any, item2: Any) -> bool Verifies if two items are set and are different of different "representative" value. Compares two value representations and returns ``True`` only if both are not ``null``, are of same ``type`` and of different representative value. By "representative", we consider here the visual representation of byte/unicode strings rather than literal values to support XML/JSON and Python 2/3 implementations. Other non-string-like types are verified with literal (usual) equality method. .. py:function:: is_equal_formats(format1: Union[pywps.inout.formats.Format, weaver.typedefs.JSON], format2: Union[pywps.inout.formats.Format, weaver.typedefs.JSON]) -> bool Verifies for matching formats. .. py:function:: normalize_ordered_io(io_section: JSON_IO_ListOrMap, order_hints: Optional[JSON_IO_ListOrMap] = None) -> List[weaver.typedefs.JSON] Reorders and converts :term:`I/O` from any representation (:class:`dict` or :class:`list`) using ordering hints. First, converts :term:`I/O` definitions defined as dictionary to an equivalent :class:`list` representation, in order to work only with a single representation method. The :class:`list` is chosen over :class:`dict` because sequences can enforce a specific order, while mapping (when saved as :term:`JSON` or :term:`YAML`) have no specific order. The list representation ensures that :term:`I/O` order is preserved when written to file and reloaded afterwards regardless of server and/or library's implementation of the mapping container. If this function fails to correctly order any :term:`I/O` or cannot correctly guarantee such result because of the provided parameters (e.g.: no hints given when required), the result will not break nor change the final processing behaviour of parsers. This is merely *cosmetic* adjustments to ease readability of :term:`I/O` to avoid always shuffling their order across multiple :term:`Application Package` and :term:`Process` reporting formats. The important result of this function is to provide the :term:`I/O` as a consistent list of objects, so it is less cumbersome to compare/merge/iterate over the elements with all functions that will follow. .. note:: When defined as a dictionary, an :class:`OrderedDict` is expected as input to ensure preserved field order. Prior to Python 3.7 or CPython 3.5, preserved order is not guaranteed for *builtin* :class:`dict`. In this case the :paramref:`order_hints` is required to ensure same order. This function is intended for parsing :term:`I/O` from :term:`Process` descriptions, :term:`Application Package` and other definitions that employ a ``"type"`` field. For submitted execution :term:`I/O` values, refer to other relevant functions. .. seealso:: - :func:`convert_input_values_schema` - :func:`convert_output_params_schema` :param io_section: Definition contained under the ``inputs`` or ``outputs`` fields. :param order_hints: Optional/partial :term:`I/O` definitions hinting an order to sort unsorted-dict I/O. :returns: :term:`I/O` specified as list of dictionary definitions with preserved order (as good as possible). .. py:function:: merge_io_formats(wps_formats: List[ANY_Format_Type], cwl_formats: List[ANY_Format_Type]) -> List[ANY_Format_Type] Merges :term:`I/O` format definitions by matching ``mime-type`` field. In case of conflict, preserve the :term:`WPS` version which can be more detailed (for example, by specifying ``encoding``). Verifies if :data:`DEFAULT_FORMAT_MISSING` was written to a single :term:`CWL` format caused by a lack of any value provided as input. In this case, *only* :term:`WPS` formats are kept. In the event that :data:`DEFAULT_FORMAT_MISSING` was written to the :term:`CWL` formats and that no :term:`WPS` format was specified, the :data:`DEFAULT_FORMAT` is returned. :raises PackageTypeError: if inputs are invalid format lists .. py:function:: merge_io_fields(wps_io: WPS_IO_Type, cwl_io: WPS_IO_Type) -> WPS_IO_Type Combines corresponding :term:`I/O` fields from :term:`WPS` and :term:`CWL` definitions. .. seealso:: :func:`cwl2wps_io` for conversion of :term:`CWL` to :term:`WPS` representation. :param wps_io: Original :term:`WPS` :term:`I/O` provided in the process definition during deployment. :param cwl_io: Converted :term:`CWL` :term:`I/O` into :term:`WPS` representation for matching similar details. :return: Merged :term:`I/O` definition. .. py:function:: merge_package_io(wps_io_list: List[ANY_IO_Type], cwl_io_list: List[WPS_IO_Type], io_select: weaver.processes.constants.IO_Select_Type) -> List[JSON_IO_Type] Merges corresponding parameters of different :term:`I/O` definition sources. Handled definition formats include :term:`I/O` representation for :term:`CWL`, :term:`OpenAPI`, :term:`WPS` and generic :term:`JSON` using properties resembling :mod:`pywps` or :mod:`owslib` objects. Update :term:`I/O` definitions to use for :term:`Process` creation and returned by ``GetCapabilities``/``DescribeProcess``. If :term:`WPS` :term:`I/O` definitions where provided during deployment, update `CWL-to-WPS` converted :term:`I/O` with the :term:`WPS` :term:`I/O` complementary details. If an :term:`OpenAPI` ``schema`` definition was provided to define the :term:`I/O`, infer the corresponding :term:`WPS` :term:`I/O` details. Then, considering those resolved definitions and any missing information that could be inferred, extend field requirements that can be retrieved from :term:`CWL` definitions. Removes any deployment :term:`WPS` :term:`I/O` definitions that don't match any :term:`CWL` :term:`I/O` by ID, since they will be of no use for the underlying :term:`Application Package`. Adds missing deployment :term:`WPS` :term:`I/O` definitions using expected :term:`CWL` :term:`I/O` IDs. .. seealso:: :func:`cwl2wps_io` for conversion of :term:`CWL` to :term:`WPS` representation. :param wps_io_list: List of :term:`WPS` :term:`I/O` (as json) passed during process deployment. :param cwl_io_list: List of :term:`CWL` :term:`I/O` converted to :term:`WPS`-like :term:`I/O` for counter-validation. :param io_select: :data:`IO_INPUT` or :data:`IO_OUTPUT` to specify desired WPS type conversion. :returns: List of updated :term:`JSON` :term:`I/O` combing :term:`CWL`, :term:`WPS` and :term:`OpenAPI` specifications. .. py:function:: check_io_compatible(wps_io: WPS_IO_Type, cwl_io: WPS_IO_Type, io_id: str) -> None Validate types to ensure they match categories, otherwise merging will cause more confusion. For `Literal`/`Complex` :term:`I/O` coming from :term:`WPS` side, they should be matched exactly with `Literal`/`Complex` :term:`I/O` on the :term:`CWL` side. .. note:: The :term`CWL` :term:`I/O` in this case is expected to be a :mod:`pywps` converted :term:`I/O` from the :term`CWL` definition, and not a direct :term`CWL` :term:`I/O` definition. .. warning:: When BoundingBox for :term:`WPS`, it should be mapped to ComplexInput on :term:`CWL` side (since no equivalent). :raises PackageTypeError: If :term:`I/O` are not compatible.