Source code for

# -*- coding: utf-8 -*-
# Copyright 2018, CS GROUP - France,
# This file is part of EODAG project
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple

import geojson

from eodag.api.search_result import SearchResult
from eodag.plugins.crunch.filter_date import FilterDate
from eodag.plugins.crunch.filter_overlap import FilterOverlap
from eodag.plugins.crunch.filter_property import FilterProperty
from import StacSearch
from eodag.utils import (
from eodag.utils.stac_reader import fetch_stac_items

    from eodag.api.product import EOProduct
    from eodag.config import PluginConfig

logger = logging.getLogger("")

[docs]class StaticStacSearch(StacSearch): """Static STAC Catalog search plugin The available configuration parameters for this plugin are (to be set in provider configuration): - **api_endpoint**: (mandatory) path to the catalog (url or local system path) - **max_connections**: (optional) Maximum number of connections for HTTP requests, defaut is 100. - **timeout**: (mandatory) Timeout in seconds for each internal HTTP request, default is 5. This plugin first loads all STAC items found in the catalog, and converts them to EOProducts using StacSearch. Then it uses crunchers to only keep products matching query parameters. :param provider: An eodag providers configuration dictionary :type provider: dict :param config: Path to the user configuration file :type config: str """
[docs] def __init__(self, provider: str, config: PluginConfig) -> None: super(StaticStacSearch, self).__init__(provider, config) self.config.__dict__.setdefault("max_connections", 100) self.config.__dict__.setdefault("timeout", HTTP_REQ_TIMEOUT)
def discover_product_types(self) -> Dict[str, Any]: """Fetch product types is disabled for `StaticStacSearch` :returns: empty dict :rtype: dict """ return {} def query( self, product_type: Optional[str] = None, items_per_page: int = DEFAULT_ITEMS_PER_PAGE, page: int = DEFAULT_PAGE, count: bool = True, **kwargs: Any, ) -> Tuple[List[EOProduct], Optional[int]]: """Perform a search on a static STAC Catalog""" features = fetch_stac_items( self.config.api_endpoint, recursive=True, max_connections=self.config.max_connections, timeout=self.config.timeout, ) nb_features = len(features) feature_collection = geojson.FeatureCollection(features) # save StaticStacSearch._request and mock it to make return loaded static results stacapi_request = self._request self._request = ( lambda url, info_message=None, exception_message=None: MockResponse( feature_collection, 200 ) ) # query on mocked StacSearch eo_products, _ = super(StaticStacSearch, self).query( items_per_page=nb_features, page=1, count=True, **kwargs ) # filter using query params search_result = SearchResult(eo_products) # Filter by date if "startTimeFromAscendingNode" in kwargs: kwargs["start"] = kwargs.pop("startTimeFromAscendingNode") if "completionTimeFromAscendingNode" in kwargs: kwargs["end"] = kwargs.pop("completionTimeFromAscendingNode") if any(k in ["start", "end"] for k in kwargs.keys()): search_result = search_result.crunch( FilterDate({k: kwargs[k] for k in ["start", "end"] if k in kwargs}) ) # Filter by geometry if "geometry" in kwargs.keys(): search_result = search_result.crunch( FilterOverlap({"intersects": True}), geometry=kwargs.pop("geometry") ) # Filter by cloudCover if "cloudCover" in kwargs.keys(): search_result = search_result.crunch( FilterProperty( {"cloudCover": kwargs.pop("cloudCover"), "operator": "lt"} ) ) # Filter by other properties skip_eodag_internal_parameters = [ "auth", "raise_errors", "productType", "locations", "start", "end", "geom", ] for property_key, property_value in kwargs.items(): if property_key not in skip_eodag_internal_parameters: search_result = search_result.crunch( FilterProperty({property_key: property_value, "operator": "eq"}) ) # restore plugin._request self._request = stacapi_request return, len(search_result)