Module Contents

weaver.processes.convert.WPS_FIELD_FORMAT = ['formats', 'supported_formats', 'supported_values', 'default'][source]
weaver.processes.convert.DEFAULT_FORMAT_MISSING = __DEFAULT_FORMAT_MISSING__[source]
weaver.processes.convert.complex2json(data: Union[owslib.wps.ComplexData, Any])Union[weaver.typedefs.JSON, Any][source]

Obtains the JSON representation of a ComplexData or simply return the unmatched type.

weaver.processes.convert.metadata2json(meta: Union[ANY_Metadata_Type, Any], force: bool = False)Union[weaver.typedefs.JSON, Any][source]

Retrieve metadata information and generate its JSON representation.

Obtains the JSON representation of a OWS_Metadata or pywps.app.Common.Metadata. Otherwise, simply return the unmatched type. If requested, can enforce parsing a dictionary for the corresponding keys.

weaver.processes.convert.ows2json_field(ows_field: Union[owslib.wps.ComplexData, owslib.wps.Metadata, weaver.typedefs.AnyValueType])Union[weaver.typedefs.JSON, weaver.typedefs.AnyValueType][source]

Obtains the JSON or raw value from an owslib.wps I/O field.

weaver.processes.convert.ows2json_io(ows_io: OWS_IO_Type)JSON_IO_Type[source]

Converts I/O definition from owslib.wps to JSON.

weaver.processes.convert.ows2json_output_data(output: OWS_Output_Type, process_description: owslib.wps.Process, container: Optional[weaver.typedefs.AnySettingsContainer] = None)weaver.typedefs.JSON[source]

Utility method to convert an owslib.wps process execution output data (result) to 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 process_description is employed to retrieve the original description with expected result details.

  • output – output with data value or reference according to expected result for the corresponding process.

  • process_description – definition of the process producing the specified output following execution.

  • container – container to retrieve application settings (for request options during file retrieval as needed).


converted JSON result data and additional metadata as applicable based on data-type and content-type.

weaver.processes.convert._get_multi_json_references(output: OWS_Output_Type, container: Optional[weaver.typedefs.AnySettingsContainer])Optional[List[weaver.typedefs.JSON]][source]

Obtains the 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 JSON array containing references to these outputs.

Because the multi-output references are contained within this 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.


Array of HTTP(S) references if the specified output is effectively a JSON containing that, None otherwise.

weaver.processes.convert.any2cwl_io(wps_io: Union[JSON_IO_Type, WPS_IO_Type, OWS_IO_Type], io_select: str)Tuple[CWL_IO_Type, Dict[str, str]][source]

Converts a WPS-like I/O to CWL corresponding I/O.

Because of CWL I/O of type File with format field, the applicable namespace is also returned.


converted I/O and namespace dictionary with corresponding format references as required

weaver.processes.convert.wps2cwl_requirement(wps_service_url: Union[str, urllib.parse.ParseResult], wps_process_id: str)weaver.typedefs.JSON[source]

Obtains the CWL requirements definition needed for parsing by a remote WPS provider as an Application Package.

weaver.processes.convert.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][source]

Generates the CWL package and process definitions from a owslib.wps.Process hosted under WPS location.

weaver.processes.convert.xml_wps2cwl(wps_process_response: requests.models.Response, settings: weaver.typedefs.AnySettingsContainer)Tuple[weaver.typedefs.CWL, weaver.typedefs.JSON][source]

Obtains the CWL definition that corresponds to a XML WPS-1 process.

Converts a WPS-1 ProcessDescription XML tree structure to an equivalent WPS-3 Process JSON. and builds the associated CWL package in conformance to weaver.processes.wps_package.CWL_REQUIREMENT_APP_WPS1.

  • wps_process_response – valid response (XML, 200) from a WPS-1 ProcessDescription.

  • settings – application settings to retrieve additional request options.

weaver.processes.convert.is_cwl_file_type(io_info: CWL_IO_Type)bool[source]

Identifies if the provided CWL input/output corresponds to one, many or potentially a File type(s).

When multiple distinct atomic types are allowed for a given I/O (e.g.: [string, File]) and that one of them is a File, the result will be True even if other types are not Files. Potential File when other base type is "null" will also return True.

weaver.processes.convert.is_cwl_array_type(io_info: CWL_IO_Type)Tuple[bool, str, pywps.validator.mode.MODE, Optional[Union[Type[pywps.inout.literaltypes.AnyValue], weaver.typedefs.CWL_IO_EnumSymbols]]][source]

Verifies if the specified I/O corresponds to one of various CWL array type definitions.


tuple(is_array, io_type, io_mode, io_allow) where: - is_array: specifies if the I/O is of array type. - io_type: array element type if is_array is True, type of io_info otherwise. - io_mode: validation mode to be applied if sub-element requires it, defaults to MODE.NONE. - io_allow: validation values to be applied if sub-element requires it, defaults to AnyValue.


PackageTypeError – if the array element doesn’t have the required values and valid format.

weaver.processes.convert.is_cwl_enum_type(io_info: CWL_IO_Type)Tuple[bool, str, int, Optional[weaver.typedefs.CWL_IO_EnumSymbols]][source]

Verifies if the specified I/O corresponds to a CWL enum definition.


tuple(is_enum, io_type, io_allow) where: - is_enum: specifies if the I/O is of enum type. - io_type: enum base type if is_enum=True, type of io_info otherwise. - io_mode: validation mode to be applied if input requires it, defaults to MODE.NONE. - io_allow: validation values of the enum.


PackageTypeError – if the enum doesn’t have the required parameters and valid format.

weaver.processes.convert.get_cwl_io_type(io_info: CWL_IO_Type)Tuple[str, bool][source]

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 "<type>?" was in CWL, which can also be defined as type ["null", <type>].

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 following definition ["null", <type>, <array-type>] (same type). Whether single or array-like type, the base type can be extracted.


io_info – definition of the CWL input.


tuple of guessed base type and flag indicating if it can be null (optional input).

weaver.processes.convert.cwl2wps_io(io_info: CWL_IO_Type, io_select: str)WPS_IO_Type[source]

Converts input/output parameters from CWL types to WPS types.

  • io_info – parsed IO of a CWL file

  • io_selectWPS_INPUT or WPS_OUTPUT to specify desired WPS type conversion.


corresponding IO in WPS format

weaver.processes.convert.cwl2json_input_values(data: Dict[str, weaver.typedefs.CWL_IO_Value], schema: weaver.processes.constants.ProcessSchemaType = PROCESS_SCHEMA_OGC)weaver.typedefs.ExecutionInputs[source]

Converts CWL formatted Job inputs to corresponding OGC API - Processes format.

  • data – dictionary with inputs formatted as key-value pairs with relevant structure based on CWL types.

  • schema – either OGC or OLD format respectively for mapping/listing representations.

  • TypeError – if input data is invalid.

  • ValueError – if any input value could not be parsed with expected schema.


converted inputs for Job submission either in OGC or OLD format.

weaver.processes.convert.convert_input_values_schema(inputs: weaver.typedefs.ExecutionInputs, schema: weaver.processes.constants.ProcessSchemaType)weaver.typedefs.ExecutionInputs[source]

Convert execution input values between equivalent formats.

  • inputs – Inputs to convert.

  • schema – Desired schema.


Converted inputs.

weaver.processes.convert.repr2json_input_values(inputs: List[str])weaver.typedefs.ExecutionInputsList[source]

Converts inputs in string representation to corresponding JSON values.

Expected format is as follows:

  • 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)


inputs – list of string inputs to parse.


parsed inputs if successful.

weaver.processes.convert.any2cwl_literal_datatype(io_type: str)Union[str, Type[weaver.utils.null]][source]

Solves common literal data-type names to supported ones for CWL.

weaver.processes.convert.any2wps_literal_datatype(io_type: weaver.typedefs.AnyValueType, is_value: bool)Union[str, Type[weaver.utils.null]][source]

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.

weaver.processes.convert.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]][source]

Converts an AllowedValues definition from different packages into standardized JSON representation of OGC-API.

weaver.processes.convert.any2json_literal_data_domains(io_info: ANY_IO_Type)Union[Type[weaver.utils.null], List[weaver.typedefs.JSON]][source]

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 weaver.wps_restapi.swagger_definitions.LiteralDataDomainList with following structure.

default: bool
defaultValue: float, int, bool, str
dataType: {name: string, <reference: url: string>}
uom: string
  - string
  - url-string
  - {anyValue: bool}
  - [float, int, bool, str]
  - [{minimum, maximum, spacing, closure}]
weaver.processes.convert.json2wps_datatype(io_info: JSON_IO_Type)str[source]

Converts a JSON input definition into the corresponding pywps parameters.

Guesses the literal data-type from I/O JSON information in order to allow creation of the corresponding I/O WPS. Defaults to string if no suitable guess can be accomplished.

weaver.processes.convert.json2wps_field(field_info: weaver.typedefs.JSON, field_category: str)Any[source]

Converts an I/O field from a JSON literal data, list, or dictionary to corresponding WPS types.

  • field_info – literal data or information container describing the type to be generated.

  • field_category – one of WPS_FIELD_MAPPING keys to indicate how to parse field_info.

weaver.processes.convert.json2wps_allowed_values(io_info: JSON_IO_Type)Union[Type[weaver.utils.null], List[pywps.inout.literaltypes.AllowedValue]][source]

Obtains the allowed values constrains for the literal data type from a JSON I/O definition.

Converts the literalDataDomains definition into allowed_values understood by pywps. Handles explicit allowed_values if available and not previously defined by literalDataDomains.

See also

Function any2json_literal_data_domains() defines generated literalDataDomains JSON definition.

weaver.processes.convert.json2wps_io(io_info: JSON_IO_Type, io_select: str)WPS_IO_Type[source]

Converts an I/O from a JSON dict to PyWPS types.

  • io_info – I/O in JSON dict format.

  • io_selectWPS_INPUT or WPS_OUTPUT to specify desired WPS type conversion.


corresponding I/O in WPS format.

weaver.processes.convert.wps2json_io(io_wps: WPS_IO_Type)JSON_IO_Type[source]

Converts a PyWPS I/O into a dictionary based version with keys corresponding to standard names (WPS 2.0).

weaver.processes.convert.wps2json_job_payload(wps_request: pywps.app.WPSRequest, wps_process: pywps.Process)weaver.typedefs.JSON[source]

Converts the input and output values of a 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 pywps based on the process definition.

weaver.processes.convert.get_field(io_object: Any, field: str, search_variations: bool = False, only_variations: bool = False, pop_found: bool = False, default: Any = null)Any[source]

Gets a field by name from various I/O object types.

Default value is null used for most situations to differentiate from literal None which is often used as default for parameters. The 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 search_variation is enabled and that field could not be found within the object, field lookup will employ the values under the field entry within 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 default if no match was found.

  • io_object – Any I/O representation, either as a class instance or JSON container.

  • field – Name of the field to look for, either as property or key name based on input object type.

  • search_variations – If enabled, search for all variations to the field name to attempt search until matched.

  • only_variations – If enabled, skip the first ‘basic’ field and start search directly with field variations.

  • pop_found – If enabled, whenever a match is found by field or variations, remove that entry from the object.

  • default – Alternative default value to return if no match could be found.


Matched value (including search variations if enabled), or default.

weaver.processes.convert.set_field(io_object: Union[ANY_IO_Type, ANY_Format_Type], field: str, value: Any, force: bool = False)None[source]

Sets a field by name into various 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.

weaver.processes.convert._are_different_and_set(item1: Any, item2: Any)bool[source]

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.

weaver.processes.convert.is_equal_formats(format1: Union[pywps.inout.formats.Format, weaver.typedefs.JSON], format2: Union[pywps.inout.formats.Format, weaver.typedefs.JSON])bool[source]

Verifies for matching formats.

weaver.processes.convert.normalize_ordered_io(io_section: JSON_IO_ListOrMap, order_hints: Optional[JSON_IO_ListOrMap] = None)List[weaver.typedefs.JSON][source]

Reorders and converts I/O from any representation (dict or list) considering given ordering hints.

First, converts I/O definitions defined as dictionary to an equivalent list representation, in order to work only with a single representation method. The list is chosen over dict because sequences can enforce a specific order, while mapping have no particular order. The list representation ensures that I/O order is preserved when written to file and reloaded afterwards regardless of each server and/or library’s implementation of the mapping container.

If this function fails to correctly order any 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 I/O to avoid always shuffling their order across multiple Application Package and Process reporting formats.

The important result of this function is to provide the 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.


When defined as a dictionary, an 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 dict. In this case the order_hints is required to ensure same order.

  • io_section – Definition contained under the inputs or outputs fields.

  • order_hints – Optional/partial I/O definitions hinting an order to sort unsorted-dict I/O.


I/O specified as list of dictionary definitions with preserved order (as best as possible).

weaver.processes.convert.merge_io_formats(wps_formats: List[ANY_Format_Type], cwl_formats: List[ANY_Format_Type])List[ANY_Format_Type][source]

Merges I/O format definitions by matching mime-type field.

In case of conflict, preserve the WPS version which can be more detailed (for example, by specifying encoding).

Verifies if DEFAULT_FORMAT_MISSING was written to a single CWL format caused by a lack of any value provided as input. In this case, only WPS formats are kept.

In the event that DEFAULT_FORMAT_MISSING was written to the CWL formats and that no WPS format was specified, the DEFAULT_FORMAT is returned.


PackageTypeError – if inputs are invalid format lists

weaver.processes.convert.merge_package_io(wps_io_list: List[ANY_IO_Type], cwl_io_list: List[WPS_IO_Type], io_select: str)List[WPS_IO_Type][source]

Merges corresponding parameters of different I/O definitions from CWL/WPS sources.

Update I/O definitions to use for process creation and returned by GetCapabilities, DescribeProcess. If WPS I/O definitions where provided during deployment, update CWL-to-WPS converted I/O with the WPS I/O complementary details. Otherwise, provide minimum field requirements that can be retrieved from CWL definitions.

Removes any deployment WPS I/O definitions that don’t match any CWL I/O by ID. Adds missing deployment WPS I/O definitions using expected CWL I/O IDs.

  • wps_io_list – list of WPS I/O (as json) passed during process deployment.

  • cwl_io_list – list of CWL I/O converted to WPS-like I/O for counter-validation.

  • io_selectWPS_INPUT or WPS_OUTPUT to specify desired WPS type conversion.


list of validated/updated WPS I/O for the process matching CWL I/O requirements.