weaver.processes.opensearch =========================== .. py:module:: weaver.processes.opensearch Module Contents --------------- .. py:data:: LOGGER .. py:function:: alter_payload_after_query(payload: weaver.typedefs.JSON) -> weaver.typedefs.JSON When redeploying the package on :term:`ADES`, strip out any :term:`EOImage` parameter. .. py:data:: WKT_BBOX_BOUNDS .. py:function:: load_wkt_bbox_bounds(wkt: str) -> str .. py:function:: validate_bbox(bbox: str) -> None Validate bounding box formatted as ``x1, y1, x2, y2`` string composed of floating point numbers. .. py:function:: query_eo_images_from_wps_inputs(wps_inputs: Dict[str, Deque], eoimage_source_info: Dict[str, weaver.typedefs.DataSourceOpenSearch], accept_mime_types: Dict[str, List[str]], settings: Optional[weaver.typedefs.AnySettingsContainer] = None) -> Dict[str, Deque[weaver.processes.convert.WPS_Input_Type]] Query `OpenSearch` using parameters in inputs and return file links. :param wps_inputs: inputs containing info to query :param eoimage_source_info: data source info of eoimages :param accept_mime_types: dict of list of accepted mime types, ordered by preference :param settings: application settings to retrieve request options as necessary. .. py:function:: replace_with_opensearch_scheme(link: str) -> str Replaces ``file://`` scheme by ``opensearch://`` scheme. .. py:class:: OpenSearchQuery(collection_identifier: str, osdd_url: str, catalog_search_field: str = 'parentIdentifier', settings: Optional[weaver.typedefs.AnySettingsContainer] = None) Container to handle `OpenSearch` queries. :param collection_identifier: Collection ID to query :param osdd_url: Global OSDD url for `OpenSearch` queries. :param catalog_search_field: Name of the field for the collection identifier. :param settings: application settings to retrieve request options as necessary. .. py:attribute:: DEFAULT_MAX_QUERY_RESULTS :value: 5 .. py:attribute:: settings :value: None .. py:attribute:: collection_identifier .. py:attribute:: osdd_url .. py:attribute:: params .. py:method:: get_template_url() -> str .. py:method:: _prepare_query_url(template_url: str, params: Dict) -> Tuple[str, weaver.typedefs.JSON] Prepare the URL for the `OpenSearch` query. :param template_url: url containing query parameters. :param params: parameters to insert in formatted URL. .. py:method:: _fetch_datatsets_from_alternates_links(alternate_links) .. py:method:: _query_features_paginated(params: weaver.typedefs.JSON) -> Iterator[weaver.typedefs.JSON, str] Iterates over paginated results until all features are retrieved. :param params: query parameters .. py:method:: query_datasets(params: weaver.typedefs.JSON, accept_schemes: List[str], accept_mime_types: List[str]) -> Iterator[str] Query the specified datasets. Loop on every `OpenSearch` result feature and yield URL matching required mime-type and scheme. Log a warning if a feature cannot yield a valid URL (either no compatible mime-type or scheme) :param params: query parameters :param accept_schemes: only return links of this scheme :param accept_mime_types: list of accepted mime types, ordered by preference :raise KeyError: If the feature doesn't contain a json data section or an atom alternative link .. py:function:: get_additional_parameters(input_data: weaver.typedefs.JSON) -> List[Tuple[str, str]] Retrieve the values from the ``additionalParameters`` of the input. :param input_data: Dict containing or not the "additionalParameters" key .. py:class:: EOImageDescribeProcessHandler(inputs: List[weaver.processes.convert.JSON_IO_Type]) .. py:attribute:: eoimage_inputs .. py:attribute:: other_inputs .. py:method:: is_eoimage_input(input_data: weaver.typedefs.JSON) -> bool :staticmethod: .. py:method:: get_allowed_collections(input_data: weaver.typedefs.JSON) -> List[str] :staticmethod: .. py:method:: make_aoi(id_: str) -> weaver.typedefs.JSON :staticmethod: .. py:method:: make_collection(identifier: str, allowed_values: List[str]) -> weaver.typedefs.JSON :staticmethod: .. py:method:: make_toi(id_: str, start_date: bool = True) -> weaver.typedefs.JSON :staticmethod: Generate the Time-Of-Interest definition. :param id_: ID of the input. :param start_date: (Default value = True) .. py:method:: to_opensearch(unique_aoi: bool, unique_toi: bool) -> List[weaver.typedefs.JSON] Convert the inputs with `OpenSearch` request parameters considering Area-Of-Interest and Time-Of-Interest. :param unique_aoi: indicate if a single/global AOI must be applied or individual ones for each input. :param unique_toi: indicate if a single/global TOI must be applied or individual ones for each input. .. py:function:: get_eo_images_inputs_from_payload(payload: weaver.typedefs.JSON) -> List[weaver.typedefs.JSON] Extracts only inputs that correspond to an :term:`EOImage` based on provided ``additionalParameters``. .. py:function:: get_original_collection_id(payload: weaver.typedefs.JSON, wps_inputs: Dict[str, collections.deque[weaver.processes.convert.WPS_Input_Type]]) -> Dict[str, collections.deque[weaver.processes.convert.WPS_Input_Type]] Obtains modified WPS inputs considering mapping to known `OpenSearch` collection IDs. When we deploy a `Process` that contains `OpenSearch` parameters, the collection identifier is modified:: Ex: files -> collection Ex: s2 -> collection_s2, probav -> collection_probav This function changes the ID in the execute request to the one from the deployed `Process` description. :param payload: :param wps_inputs: :return: .. py:function:: get_eo_images_data_sources(payload: Dict, wps_inputs: Dict[str, collections.deque[weaver.processes.convert.WPS_Input_Type]]) -> Dict[str, weaver.typedefs.DataSourceOpenSearch] Resolve the data source of an ``EOImage`` input reference. :param payload: Deploy payload :param wps_inputs: Execute inputs :returns: Data source of the ``EOImage``. .. py:function:: get_eo_images_mime_types(payload: weaver.typedefs.JSON) -> Dict[str, List[str]] Get the accepted media-types from the deployment payload. :param payload: Deploy payload. :returns: Accepted media-type. .. py:function:: insert_max_occurs(payload: weaver.typedefs.JSON, wps_inputs: Dict[str, Deque[weaver.processes.convert.WPS_Input_Type]]) -> None Insert maxOccurs value in wps inputs using the deploy payload. :param payload: Deploy payload. :param wps_inputs: WPS inputs. .. py:function:: modified_collection_identifiers(eo_image_identifiers: List[str]) -> List[str] .. py:function:: get_data_source(collection_id: str) -> weaver.typedefs.DataSourceOpenSearch Obtain the applicable :term:`Data Source` based on the provided collection identifier. .. seealso:: - :ref:`conf_data_sources` - :ref:`opensearch_data_source` .. py:function:: get_eo_images_ids_from_payload(payload: weaver.typedefs.JSON) -> List[str] .. py:function:: replace_inputs_describe_process(inputs: List[weaver.typedefs.JSON], payload: weaver.typedefs.JSON) -> List[weaver.typedefs.JSON] Replace ``EOImage`` inputs (if ``additionalParameter -> EOImage -> true``) with `OpenSearch` query parameters. .. py:function:: make_param_id(param_name: str, identifier: str) -> str Only adds an underscore between the parameters.