Fix

README.md

@@ -12,5 +12,24 @@ short_description: Application for Automatic Crop Type Mapping
 
 Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
 
-# License
-This project is licensed under the terms of the MIT License. See the LICENSE file for details.
+## Technologies Used
+This project builds upon several open-source research initiatives:
+
+*   **[ESA WorldCereal](https://esa-worldcereal.org/):** Global crop mapping system funded by the European Space Agency.
+*   **[NASA Harvest Presto](https://github.com/nasaharvest/presto):** A lightweight, pre-trained transformer model for remote sensing timeseries.
+*   **[openEO GFMap](https://github.com/Open-EO/openeo-gfmap):** A generic framework for Earth Observation mapping applications.
+*   **Streamlit:** For the interactive web interface.
+
+## Citations & References
+If you use this tool or the underlying models, please consider citing the original authors:
+
+**Presto Model:**
+> Tseng, G., Zvonkov, I., Purohit, M., Rolnick, D., & Kerner, H. (2023). Lightweight, Pre-trained Transformers for Remote Sensing Timeseries. *arXiv preprint arXiv:2304.14065*.
+
+**WorldCereal System:**
+> Van Tricht, K., Gobin, A., Gilliams, S., & Piccard, I. (2023). WorldCereal: a dynamic open-source system for global-scale crop mapping. *Earth System Science Data, 15*(12), 5491-5515.
+
+## License
+This project is licensed under the **MIT License**.
+*   **WorldCereal** code components are Apache-2.0 / MIT compatible.
+*   **Presto** model weights and code are released under MIT License.

openeo_gfmap/fetching/__init__.py

@@ -0,0 +1,33 @@
+"""Extraction sub-module.
+
+Logic behind the extraction of training or inference data. Different backends
+are supported in order to obtain a very similar result at the end of this
+component.
+"""
+
+import logging
+
+_log = logging.getLogger(__name__)
+_log.setLevel(logging.INFO)
+
+ch = logging.StreamHandler()
+ch.setLevel(logging.INFO)
+
+formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+ch.setFormatter(formatter)
+
+_log.addHandler(ch)
+
+__all__ = [
+    "build_sentinel2_l2a_extractor",
+    "CollectionFetcher",
+    "FetchType",
+    "build_sentinel1_grd_extractor",
+    "build_generic_extractor",
+    "build_generic_extractor_stac",
+]
+
+from .fetching import CollectionFetcher, FetchType  # noqa: E402
+from .generic import build_generic_extractor, build_generic_extractor_stac  # noqa: E402
+from .s1 import build_sentinel1_grd_extractor  # noqa: E402
+from .s2 import build_sentinel2_l2a_extractor  # noqa: E402

openeo_gfmap/fetching/commons.py

@@ -0,0 +1,189 @@
+"""
+Common internal operations within collection extraction logic, such as reprojection.
+"""
+
+from functools import partial
+from typing import Dict, Optional, Sequence, Union
+
+import openeo
+from geojson import GeoJSON
+from openeo.api.process import Parameter
+from openeo.rest.connection import InputDate
+from pyproj.crs import CRS
+from pyproj.exceptions import CRSError
+
+from openeo_gfmap.spatial import BoundingBoxExtent, SpatialContext
+from openeo_gfmap.temporal import TemporalContext
+
+from .fetching import FetchType
+
+
+def convert_band_names(desired_bands: list, band_dict: dict) -> list:
+    """Renames the desired bands to the band names of the collection specified
+    in the backend.
+
+    Parameters
+    ----------
+    desired_bands: list
+        List of bands that are desired by the user, written in the OpenEO-GFMAP
+        harmonized names convention.
+    band_dict: dict
+        Dictionnary mapping for a backend the collection band names to the
+        OpenEO-GFMAP harmonized names. This dictionnary will be reversed to be
+        used within this function.
+    Returns
+    -------
+    backend_band_list: list
+        List of band names within the backend collection names.
+    """
+    # Reverse the dictionarry
+    band_dict = {v: k for k, v in band_dict.items()}
+    return [band_dict[band] for band in desired_bands]
+
+
+def resample_reproject(
+    datacube: openeo.DataCube,
+    resolution: float,
+    epsg_code: Optional[Union[str, int]] = None,
+    method: str = "near",
+) -> openeo.DataCube:
+    """Reprojects the given datacube to the target epsg code, if the provided
+    epsg code is not None. Also performs checks on the give code to check
+    its validity.
+    """
+    if epsg_code is not None:
+        # Checks that the code is valid
+        try:
+            CRS.from_epsg(int(epsg_code))
+        except (CRSError, ValueError) as exc:
+            raise ValueError(
+                f"Specified target_crs: {epsg_code} is not a valid EPSG code."
+            ) from exc
+        return datacube.resample_spatial(
+            resolution=resolution, projection=epsg_code, method=method
+        )
+    return datacube.resample_spatial(resolution=resolution, method=method)
+
+
+def rename_bands(datacube: openeo.DataCube, mapping: dict) -> openeo.DataCube:
+    """Rename the bands from the given mapping scheme"""
+
+    # Filter out bands that are not part of the datacube
+    def filter_condition(band_name, _):
+        return band_name in datacube.metadata.band_names
+
+    mapping = {k: v for k, v in mapping.items() if filter_condition(k, v)}
+
+    return datacube.rename_labels(
+        dimension="bands", target=list(mapping.values()), source=list(mapping.keys())
+    )
+
+
+def _load_collection_hybrid(
+    connection: openeo.Connection,
+    is_stac: bool,
+    collection_id_or_url: str,
+    bands: list,
+    spatial_extent: Union[Dict[str, float], Parameter, None] = None,
+    temporal_extent: Union[Sequence[InputDate], Parameter, str, None] = None,
+    properties: Optional[dict] = None,
+):
+    """Wrapper around the load_collection, or load_stac method of the openeo connection."""
+    if not is_stac:
+        return connection.load_collection(
+            collection_id=collection_id_or_url,
+            spatial_extent=spatial_extent,
+            temporal_extent=temporal_extent,
+            bands=bands,
+            properties=properties,
+        )
+    cube = connection.load_stac(
+        url=collection_id_or_url,
+        spatial_extent=spatial_extent,
+        temporal_extent=temporal_extent,
+        bands=bands,
+        properties=properties,
+    )
+    cube = cube.rename_labels(dimension="bands", target=bands)
+    return cube
+
+
+def _load_collection(
+    connection: openeo.Connection,
+    bands: list,
+    collection_name: str,
+    spatial_extent: SpatialContext,
+    temporal_extent: Optional[TemporalContext],
+    fetch_type: FetchType,
+    is_stac: bool = False,
+    **params,
+):
+    """Loads a collection from the openeo backend, acting differently depending
+    on the fetch type.
+    """
+    load_collection_parameters = params.get("load_collection", {})
+    load_collection_method = partial(
+        _load_collection_hybrid, is_stac=is_stac, collection_id_or_url=collection_name
+    )
+
+    if (
+        temporal_extent is not None
+    ):  # Can be ignored for intemporal collections such as DEM
+        temporal_extent = [temporal_extent.start_date, temporal_extent.end_date]
+
+    if fetch_type == FetchType.TILE:
+        if isinstance(spatial_extent, BoundingBoxExtent):
+            spatial_extent = dict(spatial_extent)
+        elif spatial_extent is not None:
+            raise ValueError(
+                "`spatial_extent` should be either None or an instance of BoundingBoxExtent for tile-based fetching."
+            )
+        cube = load_collection_method(
+            connection=connection,
+            bands=bands,
+            spatial_extent=spatial_extent,
+            temporal_extent=temporal_extent,
+            properties=load_collection_parameters,
+        )
+    elif fetch_type == FetchType.POINT or fetch_type == FetchType.POLYGON:
+        cube = load_collection_method(
+            connection=connection,
+            bands=bands,
+            temporal_extent=temporal_extent,
+            properties=load_collection_parameters,
+        )
+
+    # Adding the process graph updates for experimental features
+    if params.get("update_arguments") is not None:
+        cube.result_node().update_arguments(**params["update_arguments"])
+
+    # Peforming pre-mask optimization
+    pre_mask = params.get("pre_mask", None)
+    if pre_mask is not None:
+        assert isinstance(pre_mask, openeo.DataCube), (
+            f"The 'pre_mask' parameter must be an openeo datacube, got {pre_mask}."
+        )
+        cube = cube.mask(pre_mask)
+
+    # Merges additional bands continuing the operations.
+    pre_merge_cube = params.get("pre_merge", None)
+    if pre_merge_cube is not None:
+        assert isinstance(pre_merge_cube, openeo.DataCube), (
+            f"The 'pre_merge' parameter value must be an openeo datacube, "
+            f"got {pre_merge_cube}."
+        )
+        if pre_mask is not None:
+            pre_merge_cube = pre_merge_cube.mask(pre_mask)
+        cube = cube.merge_cubes(pre_merge_cube)
+
+    if fetch_type == FetchType.POLYGON and spatial_extent is not None:
+        if isinstance(spatial_extent, str):
+            geometry = connection.load_url(
+                spatial_extent,
+                format="Parquet" if ".parquet" in spatial_extent else "GeoJSON",
+            )
+            cube = cube.filter_spatial(geometry)
+        else:
+            cube = cube.filter_spatial(spatial_extent)
+
+    return cube

openeo_gfmap/fetching/fetching.py

@@ -0,0 +1,98 @@
+""" Main file for extractions and pre-processing of data through OpenEO
+"""
+
+from enum import Enum
+from typing import Callable
+
+import openeo
+
+from openeo_gfmap import BackendContext
+from openeo_gfmap.spatial import SpatialContext
+from openeo_gfmap.temporal import TemporalContext
+
+
+class FetchType(Enum):
+    """Enumerates the different types of extraction. There are three types of
+    enumerations.
+
+    * TILE: Tile based extractions, getting the data for a dense part. The
+    output of such fetching process in a dense DataCube.
+    * POINT: Point based extractions. From a datasets of polygons, gets sparse
+    extractions and performs spatial aggregation on the selected polygons. The
+    output of such fetching process is a VectorCube, that can be used to get
+    a pandas.DataFrame
+    * POLYGON: Patch based extractions, returning a VectorCube of sparsed
+    patches. This can be retrieved as multiple NetCDF files from one job.
+    """
+
+    TILE = "tile"
+    POINT = "point"
+    POLYGON = "polygon"
+
+
+class CollectionFetcher:
+    """Base class to fetch a particular collection.
+
+    Parameters
+    ----------
+    backend_context: BackendContext
+        Information about the backend in use, useful in certain cases.
+    bands: list
+        List of band names to load from that collection.
+    collection_fetch: Callable
+        Function defining how to fetch a collection for a specific backend,
+        the function accepts the following parameters: connection,
+        spatial extent, temporal extent, bands and additional parameters.
+    collection_preprocessing: Callable
+        Function defining how to harmonize the data of a collection in a
+        backend. For example, this function could rename the bands as they
+        can be different for every backend/collection (SENTINEL2_L2A or
+        SENTINEL2_L2A_SENTINELHUB). Accepts the following parameters:
+        datacube (of pre-fetched collection) and additional parameters.
+    colection_params: dict
+        Additional parameters encoded within a dictionnary that will be
+        passed in the fetch and preprocessing function.
+    """
+
+    def __init__(
+        self,
+        backend_context: BackendContext,
+        bands: list,
+        collection_fetch: Callable,
+        collection_preprocessing: Callable,
+        **collection_params,
+    ):
+        self.backend_contect = backend_context
+        self.bands = bands
+        self.fetcher = collection_fetch
+        self.processing = collection_preprocessing
+        self.params = collection_params
+
+    def get_cube(
+        self,
+        connection: openeo.Connection,
+        spatial_context: SpatialContext,
+        temporal_context: TemporalContext,
+    ) -> openeo.DataCube:
+        """Retrieve a data cube from the given spatial and temporal context.
+
+        Parameters
+        ----------
+        connection: openeo.Connection
+            A connection to an OpenEO backend. The backend provided must be the
+            same as the one this extractor class is configured for.
+        spatial_extent: SpatialContext
+            Either a GeoJSON collection on which spatial filtering will be
+            applied or a bounding box with an EPSG code. If a bounding box is
+            provided, no filtering is applied and the entirety of the data is
+            fetched for that region.
+        temporal_extent: TemporalContext
+            The begin and end date of the extraction.
+        """
+        collection_data = self.fetcher(
+            connection, spatial_context, temporal_context, self.bands, **self.params
+        )
+
+        preprocessed_data = self.processing(collection_data, **self.params)
+
+        return preprocessed_data

openeo_gfmap/fetching/generic.py

@@ -0,0 +1,173 @@
+""" Generic extraction of features, supporting VITO backend.
+"""
+
+from typing import Callable, Optional
+
+import openeo
+from openeo.rest import OpenEoApiError
+
+from openeo_gfmap.backend import Backend, BackendContext
+from openeo_gfmap.fetching import CollectionFetcher, FetchType, _log
+from openeo_gfmap.fetching.commons import (
+    _load_collection,
+    convert_band_names,
+    rename_bands,
+    resample_reproject,
+)
+from openeo_gfmap.spatial import SpatialContext
+from openeo_gfmap.temporal import TemporalContext
+
+BASE_DEM_MAPPING = {"DEM": "COP-DEM"}
+BASE_WEATHER_MAPPING = {
+    "dewpoint-temperature": "AGERA5-DEWTEMP",
+    "precipitation-flux": "AGERA5-PRECIP",
+    "solar-radiation-flux": "AGERA5-SOLRAD",
+    "temperature-max": "AGERA5-TMAX",
+    "temperature-mean": "AGERA5-TMEAN",
+    "temperature-min": "AGERA5-TMIN",
+    "vapour-pressure": "AGERA5-VAPOUR",
+    "wind-speed": "AGERA5-WIND",
+}
+AGERA5_STAC_MAPPING = {
+    "dewpoint_temperature_mean": "AGERA5-DEWTEMP",
+    "total_precipitation": "AGERA5-PRECIP",
+    "solar_radiation_flux": "AGERA5-SOLRAD",
+    "2m_temperature_max": "AGERA5-TMAX",
+    "2m_temperature_mean": "AGERA5-TMEAN",
+    "2m_temperature_min": "AGERA5-TMIN",
+    "vapour_pressure": "AGERA5-VAPOUR",
+    "wind_speed": "AGERA5-WIND",
+}
+KNOWN_UNTEMPORAL_COLLECTIONS = ["COPERNICUS_30"]
+
+AGERA5_TERRASCOPE_STAC = "https://stac.openeo.vito.be/collections/agera5_daily"
+
+
+def _get_generic_fetcher(
+    collection_name: str, fetch_type: FetchType, backend: Backend, is_stac: bool
+) -> Callable:
+    band_mapping: Optional[dict] = None
+
+    if collection_name == "COPERNICUS_30":
+        band_mapping = BASE_DEM_MAPPING
+    elif collection_name == "AGERA5":
+        band_mapping = BASE_WEATHER_MAPPING
+    elif is_stac and (AGERA5_TERRASCOPE_STAC in collection_name):
+        band_mapping = AGERA5_STAC_MAPPING
+
+    def generic_default_fetcher(
+        connection: openeo.Connection,
+        spatial_extent: SpatialContext,
+        temporal_extent: TemporalContext,
+        bands: list,
+        **params,
+    ) -> openeo.DataCube:
+        if band_mapping is not None:
+            bands = convert_band_names(bands, band_mapping)
+
+        if (collection_name in KNOWN_UNTEMPORAL_COLLECTIONS) and (
+            temporal_extent is not None
+        ):
+            _log.warning(
+                "Ignoring the temporal extent provided by the user as the collection %s is known to be untemporal.",
+                collection_name,
+            )
+            temporal_extent = None
+
+        try:
+            cube = _load_collection(
+                connection,
+                bands,
+                collection_name,
+                spatial_extent,
+                temporal_extent,
+                fetch_type,
+                is_stac=is_stac,
+                **params,
+            )
+        except OpenEoApiError as e:
+            if "CollectionNotFound" in str(e):
+                raise ValueError(
+                    f"Collection {collection_name} not found in the selected backend {backend.value}."
+                ) from e
+            raise e
+
+        # # Apply if the collection is a GeoJSON Feature collection
+        # if isinstance(spatial_extent, GeoJSON):
+        #     cube = cube.filter_spatial(spatial_extent)
+
+        return cube
+
+    return generic_default_fetcher
+
+
+def _get_generic_processor(
+    collection_name: str, fetch_type: FetchType, is_stac: bool
+) -> Callable:
+    """Builds the preprocessing function from the collection name as it stored
+    in the target backend.
+    """
+    band_mapping: Optional[dict] = None
+    if collection_name == "COPERNICUS_30":
+        band_mapping = BASE_DEM_MAPPING
+    elif collection_name == "AGERA5":
+        band_mapping = BASE_WEATHER_MAPPING
+    elif is_stac and (AGERA5_TERRASCOPE_STAC in collection_name):
+        band_mapping = AGERA5_STAC_MAPPING
+
+    def generic_default_processor(cube: openeo.DataCube, **params):
+        """Default collection preprocessing method for generic datasets.
+        This method renames bands and removes the time dimension in case the
+        requested dataset is DEM
+        """
+        if params.get("target_resolution", None) is not None:
+            cube = resample_reproject(
+                cube,
+                params.get("target_resolution", 10.0),
+                params.get("target_crs", None),
+                method=params.get("resampling_method", "near"),
+            )
+
+        if collection_name == "COPERNICUS_30":
+            cube = cube.min_time()
+
+        if band_mapping is not None:
+            cube = rename_bands(cube, band_mapping)
+
+        return cube
+
+    return generic_default_processor
+
+
+def build_generic_extractor(
+    backend_context: BackendContext,
+    bands: list,
+    fetch_type: FetchType,
+    collection_name: str,
+    **params,
+) -> CollectionFetcher:
+    """Creates a generic extractor adapted to the given backend. Provides band mappings for known
+    collections, such as AGERA5 available on Terrascope/FED and Copernicus 30m DEM in all backends.
+    """
+    fetcher = _get_generic_fetcher(
+        collection_name, fetch_type, backend_context.backend, False
+    )
+    preprocessor = _get_generic_processor(collection_name, fetch_type, False)
+
+    return CollectionFetcher(backend_context, bands, fetcher, preprocessor, **params)
+
+
+def build_generic_extractor_stac(
+    backend_context: BackendContext,
+    bands: list,
+    fetch_type: FetchType,
+    collection_url: str,
+    **params,
+) -> CollectionFetcher:
+    """Creates a generic extractor adapted to the given backend. Currently only tested with VITO backend"""
+    fetcher = _get_generic_fetcher(
+        collection_url, fetch_type, backend_context.backend, True
+    )
+    preprocessor = _get_generic_processor(collection_url, fetch_type, True)
+
+    return CollectionFetcher(backend_context, bands, fetcher, preprocessor, **params)

openeo_gfmap/fetching/s1.py

@@ -0,0 +1,193 @@
+""" Collection fetching of S1 features, supporting different backends.
+"""
+
+from functools import partial
+from typing import Callable
+
+import openeo
+from geojson import GeoJSON
+
+from openeo_gfmap.backend import Backend, BackendContext
+from openeo_gfmap.spatial import SpatialContext
+from openeo_gfmap.temporal import TemporalContext
+
+from .commons import (
+    _load_collection,
+    convert_band_names,
+    rename_bands,
+    resample_reproject,
+)
+from .fetching import CollectionFetcher, FetchType
+
+BASE_SENTINEL1_GRD_MAPPING = {
+    "VH": "S1-SIGMA0-VH",
+    "HH": "S1-SIGMA0-HH",
+    "HV": "S1-SIGMA0-HV",
+    "VV": "S1-SIGMA0-VV",
+}
+
+
+def _get_s1_grd_default_fetcher(
+    collection_name: str, fetch_type: FetchType
+) -> Callable:
+    """Return a default fetcher for Sentinel-1 GRD data.
+
+    Parameters
+    ----------
+    collection_name : str
+        The name of the sentinel1 collection to fetch as named in the backend.
+    fetch_type : FetchType
+        The type of fetching: TILE, POINT and POLYGON.
+    """
+
+    def s1_grd_fetch_default(
+        connection: openeo.Connection,
+        spatial_extent: SpatialContext,
+        temporal_extent: TemporalContext,
+        bands: list,
+        **params,
+    ) -> openeo.DataCube:
+        """Default collection fetcher for Sentinel-1 GRD collections. The
+        collection values are expected to be expressed in power values.
+        Parameters
+        ----------
+        connection: openeo.Connection
+            Connection to a general backend.
+        spatial_extent: SpatialContext
+            Either a GeoJSON collection or a bounding box of locations.
+            Performs spatial filtering if the spatial context is a GeoJSON
+            collection, as it implies sparse data.
+        temporal_extent: TemporalContexct
+            A time range, defined by a start and end date.
+        bands: list
+            The name of the bands to load from that collection
+        Returns
+        ------
+        openeo.DataCube: a datacube containing the collection raw products.
+        """
+        bands = convert_band_names(bands, BASE_SENTINEL1_GRD_MAPPING)
+
+        cube = _load_collection(
+            connection,
+            bands,
+            collection_name,
+            spatial_extent,
+            temporal_extent,
+            fetch_type,
+            **params,
+        )
+
+        if fetch_type is not FetchType.POINT and isinstance(spatial_extent, GeoJSON):
+            cube = cube.filter_spatial(spatial_extent)
+
+        return cube
+
+    return s1_grd_fetch_default
+
+
+def _get_s1_grd_default_processor(
+    collection_name: str, fetch_type: FetchType, backend: Backend
+) -> Callable:
+    """Builds the preprocessing function from the collection name as it is stored
+    in the target backend.
+    """
+
+    def s1_grd_default_processor(cube: openeo.DataCube, **params):
+        """Default collection preprocessing method.
+        This method performs:
+
+        * Compute the backscatter of all the S1 products. By default, the
+        "sigma0-ellipsoid" method is used with "COPERNICUS_30" DEM, but those
+        can be changed by specifying "coefficient" and "elevation_model" in
+        params.
+        * Resampling to 10m resolution.
+        * Reprojection if a "target_crs" key is specified in `params`.
+        * Performs value rescaling to uint16.
+        """
+        elevation_model = params.get("elevation_model", "COPERNICUS_30")
+        coefficient = params.get("coefficient", "sigma0-ellipsoid")
+
+        cube = cube.sar_backscatter(
+            elevation_model=elevation_model,
+            coefficient=coefficient,
+            local_incidence_angle=False,
+        )
+
+        # Reproject collection data to target CRS and resolution, if specified so.
+        # Can be disabled by setting target_resolution=None in the parameters
+        if params.get("target_resolution", True) is not None:
+            cube = resample_reproject(
+                cube,
+                params.get("target_resolution", 10.0),
+                params.get("target_crs", None),
+                method=params.get("resampling_method", "near"),
+            )
+        elif params.get("target_crs") is not None:
+            raise ValueError(
+                "In fetching parameters: `target_crs` specified but not `target_resolution`, which is required to perform reprojection."
+            )
+
+        # Harmonizing the collection band names to the default GFMAP band names
+        cube = rename_bands(cube, BASE_SENTINEL1_GRD_MAPPING)
+
+        return cube
+
+    return s1_grd_default_processor
+
+
+SENTINEL1_GRD_BACKEND_MAP = {
+    Backend.TERRASCOPE: {
+        "default": partial(
+            _get_s1_grd_default_fetcher, collection_name="SENTINEL1_GRD"
+        ),
+        "preprocessor": partial(
+            _get_s1_grd_default_processor,
+            collection_name="SENTINEL1_GRD",
+            backend=Backend.TERRASCOPE,
+        ),
+    },
+    Backend.CDSE: {
+        "default": partial(
+            _get_s1_grd_default_fetcher, collection_name="SENTINEL1_GRD"
+        ),
+        "preprocessor": partial(
+            _get_s1_grd_default_processor,
+            collection_name="SENTINEL1_GRD",
+            backend=Backend.CDSE,
+        ),
+    },
+    Backend.CDSE_STAGING: {
+        "default": partial(
+            _get_s1_grd_default_fetcher, collection_name="SENTINEL1_GRD"
+        ),
+        "preprocessor": partial(
+            _get_s1_grd_default_processor,
+            collection_name="SENTINEL1_GRD",
+            backend=Backend.CDSE_STAGING,
+        ),
+    },
+    Backend.FED: {
+        "default": partial(
+            _get_s1_grd_default_fetcher, collection_name="SENTINEL1_GRD"
+        ),
+        "preprocessor": partial(
+            _get_s1_grd_default_processor,
+            collection_name="SENTINEL1_GRD",
+            backend=Backend.FED,
+        ),
+    },
+}
+
+
+def build_sentinel1_grd_extractor(
+    backend_context: BackendContext, bands: list, fetch_type: FetchType, **params
+) -> CollectionFetcher:
+    """Creates a S1 GRD collection extractor for the given backend."""
+    backend_functions = SENTINEL1_GRD_BACKEND_MAP.get(backend_context.backend)
+
+    fetcher, preprocessor = (
+        backend_functions["default"](fetch_type=fetch_type),
+        backend_functions["preprocessor"](fetch_type=fetch_type),
+    )
+
+    return CollectionFetcher(backend_context, bands, fetcher, preprocessor, **params)

openeo_gfmap/fetching/s2.py

@@ -0,0 +1,237 @@
+""" Extraction of S2 features, supporting different backends.
+"""
+
+from functools import partial
+from typing import Callable
+
+import openeo
+from geojson import GeoJSON
+
+from openeo_gfmap.backend import Backend, BackendContext
+from openeo_gfmap.metadata import FakeMetadata
+from openeo_gfmap.spatial import BoundingBoxExtent, SpatialContext
+from openeo_gfmap.temporal import TemporalContext
+
+from .commons import (
+    _load_collection,
+    convert_band_names,
+    rename_bands,
+    resample_reproject,
+)
+from .fetching import CollectionFetcher, FetchType
+
+BASE_SENTINEL2_L2A_MAPPING = {
+    "B01": "S2-L2A-B01",
+    "B02": "S2-L2A-B02",
+    "B03": "S2-L2A-B03",
+    "B04": "S2-L2A-B04",
+    "B05": "S2-L2A-B05",
+    "B06": "S2-L2A-B06",
+    "B07": "S2-L2A-B07",
+    "B08": "S2-L2A-B08",
+    "B8A": "S2-L2A-B8A",
+    "B09": "S2-L2A-B09",
+    "B11": "S2-L2A-B11",
+    "B12": "S2-L2A-B12",
+    "AOT": "S2-L2A-AOT",
+    "SCL": "S2-L2A-SCL",
+    "SNW": "S2-L2A-SNW",
+}
+
+ELEMENT84_SENTINEL2_L2A_MAPPING = {
+    "coastal": "S2-L2A-B01",
+    "blue": "S2-L2A-B02",
+    "green": "S2-L2A-B03",
+    "red": "S2-L2A-B04",
+    "rededge1": "S2-L2A-B05",
+    "rededge2": "S2-L2A-B06",
+    "rededge3": "S2-L2A-B07",
+    "nir": "S2-L2A-B08",
+    "nir08": "S2-L2A-B8A",
+    "nir09": "S2-L2A-B09",
+    "cirrus": "S2-L2A-B10",
+    "swir16": "S2-L2A-B11",
+    "swir22": "S2-L2A-B12",
+    "scl": "S2-L2A-SCL",
+    "aot": "S2-L2A-AOT",
+}
+
+
+def _get_s2_l2a_default_fetcher(
+    collection_name: str, fetch_type: FetchType
+) -> Callable:
+    """Builds the fetch function from the collection name as it stored in the
+    target backend.
+
+    Parameters
+    ----------
+    collection_name: str
+        The name of the sentinel2 collection as named in the backend
+    point_based: bool
+        The type of fetching: TILE, POINT and POLYGON.
+    """
+
+    def s2_l2a_fetch_default(
+        connection: openeo.Connection,
+        spatial_extent: SpatialContext,
+        temporal_extent: TemporalContext,
+        bands: list,
+        **params,
+    ) -> openeo.DataCube:
+        """Default collection fetcher for Sentinel_L2A collections.
+        Parameters
+        ----------
+        connection: openeo.Connection
+            Connection to a general backend.
+        spatial_extent: SpatialContext
+            Either a GeoJSON collection or a bounding box of locations.
+            Performs spatial filtering if the spatial context is a GeoJSON
+            collection, as it implies sparse data.
+        temporal_extent: TemporalContexct
+            A time range, defined by a start and end date.
+        bands: list
+            The name of the bands to load from that collection
+        Returns
+        -------
+        openeo.DataCube: a datacube containing the collection raw products.
+        """
+        # Rename the bands to the backend collection names
+        bands = convert_band_names(bands, BASE_SENTINEL2_L2A_MAPPING)
+
+        cube = _load_collection(
+            connection,
+            bands,
+            collection_name,
+            spatial_extent,
+            temporal_extent,
+            fetch_type,
+            **params,
+        )
+
+        return cube
+
+    return s2_l2a_fetch_default
+
+
+# TODO deprecated?
+def _get_s2_l2a_element84_fetcher(
+    collection_name: str, fetch_type: FetchType
+) -> Callable:
+    """Fetches the collections from the Sentinel-2 Cloud-Optimized GeoTIFFs
+    bucket provided by Amazon and managed by Element84.
+    """
+
+    def s2_l2a_element84_fetcher(
+        connection: openeo.Connection,
+        spatial_extent: SpatialContext,
+        temporal_extent: TemporalContext,
+        bands: list,
+        **params,
+    ) -> openeo.DataCube:
+        """Collection fetcher on the element84 collection."""
+        bands = convert_band_names(bands, ELEMENT84_SENTINEL2_L2A_MAPPING)
+
+        if isinstance(spatial_extent, BoundingBoxExtent):
+            spatial_extent = dict(spatial_extent)
+        elif isinstance(spatial_extent, GeoJSON):
+            assert (
+                spatial_extent.get("crs", None) is not None
+            ), "CRS not defined within GeoJSON collection."
+            spatial_extent = dict(spatial_extent)
+
+        cube = connection.load_stac(
+            "https://earth-search.aws.element84.com/v1/collections/sentinel-2-l2a",
+            spatial_extent,
+            temporal_extent,
+            bands,
+        )
+
+        cube.metadata = FakeMetadata(band_names=bands)
+
+        # Apply if the collection is a GeoJSON Feature collection
+        if isinstance(spatial_extent, GeoJSON):
+            cube = cube.filter_spatial(spatial_extent)
+
+        return cube
+
+    return s2_l2a_element84_fetcher
+
+
+def _get_s2_l2a_default_processor(
+    collection_name: str, fetch_type: FetchType
+) -> Callable:
+    """Builds the preprocessing function from the collection name as it stored
+    in the target backend.
+    """
+
+    def s2_l2a_default_processor(cube: openeo.DataCube, **params):
+        """Default collection preprocessing method.
+        This method performs reprojection if specified, upsampling of bands
+        at 10m resolution as well as band reprojection. Finally, it converts
+        the type of the cube values to uint16
+        """
+        # Reproject collection data to target CRS and resolution, if specified so.
+        # Can be disabled by setting target_resolution=None in the parameters
+        if params.get("target_resolution", True) is not None:
+            cube = resample_reproject(
+                cube,
+                params.get("target_resolution", 10.0),
+                params.get("target_crs", None),
+                method=params.get("resampling_method", "near"),
+            )
+        elif params.get("target_crs") is not None:
+            raise ValueError(
+                "In fetching parameters: `target_crs` specified but not `target_resolution`, which is required to perform reprojection."
+            )
+
+        # Harmonizing the collection band names to the default GFMAP band names
+        cube = rename_bands(cube, BASE_SENTINEL2_L2A_MAPPING)
+
+        # Change the data type to uint16 for optimization purposes
+        cube = cube.linear_scale_range(0, 65534, 0, 65534)
+
+        return cube
+
+    return s2_l2a_default_processor
+
+
+SENTINEL2_L2A_BACKEND_MAP = {
+    Backend.TERRASCOPE: {
+        "fetch": partial(_get_s2_l2a_default_fetcher, collection_name="SENTINEL2_L2A"),
+        "preprocessor": partial(
+            _get_s2_l2a_default_processor, collection_name="SENTINEL2_L2A"
+        ),
+    },
+    Backend.CDSE: {
+        "fetch": partial(_get_s2_l2a_default_fetcher, collection_name="SENTINEL2_L2A"),
+        "preprocessor": partial(
+            _get_s2_l2a_default_processor, collection_name="SENTINEL2_L2A"
+        ),
+    },
+    Backend.CDSE_STAGING: {
+        "fetch": partial(_get_s2_l2a_default_fetcher, collection_name="SENTINEL2_L2A"),
+        "preprocessor": partial(
+            _get_s2_l2a_default_processor, collection_name="SENTINEL2_L2A"
+        ),
+    },
+    Backend.FED: {
+        "fetch": partial(_get_s2_l2a_default_fetcher, collection_name="SENTINEL2_L2A"),
+        "preprocessor": partial(
+            _get_s2_l2a_default_processor, collection_name="SENTINEL2_L2A"
+        ),
+    },
+}
+
+
+def build_sentinel2_l2a_extractor(
+    backend_context: BackendContext, bands: list, fetch_type: FetchType, **params
+) -> CollectionFetcher:
+    """Creates a S2 L2A extractor adapted to the given backend."""
+    backend_functions = SENTINEL2_L2A_BACKEND_MAP.get(backend_context.backend)
+
+    fetcher, preprocessor = (
+        backend_functions["fetch"](fetch_type=fetch_type),
+        backend_functions["preprocessor"](fetch_type=fetch_type),
+    )
+
+    return CollectionFetcher(backend_context, bands, fetcher, preprocessor, **params)

openeo_gfmap/preprocessing/compositing.py

@@ -0,0 +1,74 @@
+"""Temporal compositing, or temporal aggregation, is a method to increase the
+quality of data within timesteps by reducing the temporal resolution of a time
+series of satellite images.
+"""
+
+from typing import Union
+
+import openeo
+
+
+def median_compositing(
+    cube: openeo.DataCube, period: Union[str, list]
+) -> openeo.DataCube:
+    """Perfrom median compositing on the given datacube."""
+    if isinstance(period, str):
+        return cube.aggregate_temporal_period(
+            period=period, reducer="median", dimension="t"
+        )
+    elif isinstance(period, list):
+        return cube.aggregate_temporal(
+            intervals=period, reducer="median", dimension="t"
+        )
+
+
+def mean_compositing(
+    cube: openeo.DataCube, period: Union[str, list]
+) -> openeo.DataCube:
+    """Perfrom mean compositing on the given datacube."""
+    if isinstance(period, str):
+        return cube.aggregate_temporal_period(
+            period=period, reducer="mean", dimension="t"
+        )
+    elif isinstance(period, list):
+        return cube.aggregate_temporal(intervals=period, reducer="mean", dimension="t")
+
+
+def sum_compositing(cube: openeo.DataCube, period: Union[str, list]) -> openeo.DataCube:
+    """Perform sum compositing on the given datacube."""
+    if isinstance(period, str):
+        return cube.aggregate_temporal_period(
+            period=period, reducer="sum", dimension="t"
+        )
+    elif isinstance(period, list):
+        return cube.aggregate_temporal(intervals=period, reducer="sum", dimension="t")
+
+
+def max_ndvi_compositing(cube: openeo.DataCube, period: str) -> openeo.DataCube:
+    """Perform compositing by selecting the observation with the highest NDVI value over the
+    given compositing window."""
+
+    def max_ndvi_selection(ndvi: openeo.DataCube):
+        max_ndvi = ndvi.max()
+        return ndvi.array_apply(lambda x: x != max_ndvi)
+
+    if isinstance(period, str):
+        ndvi = cube.ndvi(nir="S2-L2A-B08", red="S2-L2A-B04")
+
+        rank_mask = ndvi.apply_neighborhood(
+            max_ndvi_selection,
+            size=[
+                {"dimension": "x", "unit": "px", "value": 1},
+                {"dimension": "y", "unit": "px", "value": 1},
+                {"dimension": "t", "value": period},
+            ],
+            overlap=[],
+        )
+
+        cube = cube.mask(mask=rank_mask).aggregate_temporal_period(period, "first")
+
+    else:
+        raise ValueError(
+            "Custom temporal intervals are not yet supported for max NDVI compositing."
+        )
+    return cube

openeo_gfmap/preprocessing/sar.py

@@ -0,0 +1,74 @@
+"""Routines to pre-process sar signals."""
+
+import openeo
+from openeo.processes import array_create, power
+
+from openeo_gfmap import BackendContext
+
+
+def compress_backscatter_uint16(
+    backend_context: BackendContext, cube: openeo.DataCube
+) -> openeo.DataCube:
+    """
+    Scaling the bands from float32 power values to uint16 for memory optimization. The scaling
+    casts the values from power to decibels and applies a linear scaling from 0 to 65534.
+
+    The resulting datacube has a uint16 memory representation which makes an optimization
+    before passing through any UDFs.
+
+    Parameters
+    ----------
+    backend_context : BackendContext
+        The backend context to fetch the backend name.
+    cube : openeo.DataCube
+        The datacube to compress the backscatter values.
+    Returns
+    -------
+    openeo.DataCube
+        The datacube with the backscatter values compressed to uint16.
+    """
+
+    # Apply rescaling of power values in a logarithmic way
+    cube = cube.apply_dimension(
+        dimension="bands",
+        process=lambda x: array_create(
+            [
+                power(base=10, p=(10.0 * x[0].log(base=10) + 83.0) / 20.0),
+                power(base=10, p=(10.0 * x[1].log(base=10) + 83.0) / 20.0),
+            ]
+        ),
+    )
+
+    # Change the data type to uint16 for optimization purposes
+    return cube.linear_scale_range(1, 65534, 1, 65534)
+
+
+def decompress_backscatter_uint16(
+    backend_context: BackendContext, cube: openeo.DataCube
+) -> openeo.DataCube:
+    """
+    Decompresing the bands from uint16 to their original float32 values.
+
+    Parameters
+    ----------
+    backend_context : BackendContext
+        The backend context to fetch the backend name.
+    cube : openeo.DataCube
+                The datacube to decompress the backscatter values.
+    Returns
+    -------
+    openeo.DataCube
+        The datacube with the backscatter values in their original float32 values.
+    """
+
+    cube = cube.apply_dimension(
+        dimension="bands",
+        process=lambda x: array_create(
+            [
+                power(base=10, p=(20.0 * x[0].log(base=10) - 83.0) / 10.0),
+                power(base=10, p=(20.0 * x[1].log(base=10) - 83.0) / 10.0),
+            ]
+        ),
+    )
+
+    return cube

openeo_gfmap/preprocessing/scaling.py

@@ -0,0 +1,65 @@
+"""Scaling and compressing methods for datacubes."""
+
+import openeo
+
+
+def _compress(
+    cube: openeo.DataCube,
+    min_val: int,
+    max_val: int,
+    alpha: float,
+    beta: float,
+):
+    if (
+        alpha != 1.0 or beta != 0.0
+    ):  # Avoid adding a node in the computing graph if scaling is not necessary
+        cube = (cube * alpha) + beta
+
+    return cube.linear_scale_range(min_val, max_val, min_val, max_val)
+
+
+def compress_uint16(
+    cube: openeo.DataCube, alpha: float = 1.0, beta: float = 0.0
+) -> openeo.DataCube:
+    """Scales the data linearly using the formula `output = (input * a) + b` and compresses values
+    from float32 to uint16 for memory optimization.
+
+    Parameters
+    ----------
+    cube : openeo.DataCube
+        The input datacube to compress, only meteo data should be present.
+    alpha : float, optional (default=1.0)
+        The scaling factor. Values in the input datacube are multiplied by this coefficient.
+    beta : float, optional (default=0.0)
+        The offset. Values in the input datacube are added by this value.
+
+    Returns
+    -------
+    cube : openeo.DataCube
+        The datacube with the data linearly scaled and compressed to uint16 and rescaled frome.
+    """
+    return _compress(cube, 0, 65534, alpha, beta)
+
+
+def compress_uint8(
+    cube: openeo.DataCube, alpha: float = 1.0, beta: float = 0.0
+) -> openeo.DataCube:
+    """
+    Scales the data linearly using the formula `output = (input * a) + b` and compresses values
+    from float32 to uint8 for memory optimization.
+
+    Parameters
+    ----------
+    cube : openeo.DataCube
+        The input datacube to compress, only meteo data should be present.
+    alpha : float, optional (default=1.0)
+        The scaling factor. Values in the input datacube are multiplied by this coefficient.
+    beta : float, optional (default=0.0)
+        The offset. Values in the input datacube are added by this value.
+
+    Returns
+    -------
+    cube : openeo.DataCube
+        The datacube with the data linearly scaled and compressed to uint8 and rescaled frome.
+    """
+    return _compress(cube, 0, 253, alpha, beta)

openeo_gfmap/utils/__init__.py

@@ -0,0 +1,39 @@
+"""This sub-module contains utilitary function and tools for OpenEO-GFMap"""
+
+import logging
+
+from openeo_gfmap.utils.build_df import load_json
+from openeo_gfmap.utils.intervals import quintad_intervals
+from openeo_gfmap.utils.netcdf import update_nc_attributes
+from openeo_gfmap.utils.split_stac import split_collection_by_epsg
+from openeo_gfmap.utils.tile_processing import (
+    array_bounds,
+    arrays_cosine_similarity,
+    normalize_array,
+    select_optical_bands,
+    select_sar_bands,
+)
+
+_log = logging.getLogger(__name__)
+_log.setLevel(logging.INFO)
+
+ch = logging.StreamHandler()
+ch.setLevel(logging.INFO)
+
+formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+ch.setFormatter(formatter)
+
+_log.addHandler(ch)
+
+
+__all__ = [
+    "load_json",
+    "normalize_array",
+    "select_optical_bands",
+    "array_bounds",
+    "select_sar_bands",
+    "arrays_cosine_similarity",
+    "quintad_intervals",
+    "split_collection_by_epsg",
+    "update_nc_attributes",
+]

openeo_gfmap/utils/build_df.py

@@ -0,0 +1,48 @@
+"""Utilities to build a `pandas.DataFrame` from the output of a VectorCube
+based job. Usefull to collect the output of point based extraction.
+"""
+
+from pathlib import Path
+
+import pandas as pd
+
+VECTORCUBE_TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S%z"
+TIMESTAMP_FORMAT = "%Y-%m-%d"
+
+
+def load_json(input_file: Path, bands: list) -> pd.DataFrame:
+    """Reads a json file and outputs it as a proper pandas dataframe.
+
+    Parameters
+    ----------
+    input_file: PathLike
+        The path of the JSON file to read.
+    bands: list
+        The name of the bands that will be used in the columns names. The band
+        names must be the same as the vector cube that resulted into the parsed
+        JSON file.
+    Returns
+    -------
+    df: pd.DataFrame
+        A `pandas.DataFrame` containing a combination of the band names and the
+        timestamps as column names.
+        For example, the Sentinel-2 green band on the 1st October 2020 is will
+        have the column name `S2-L2A-B02:2020-10-01`
+    """
+
+    df = pd.read_json(input_file)
+
+    target_timestamps = list(
+        map(lambda date: date.strftime(TIMESTAMP_FORMAT), df.columns.to_pydatetime())
+    )
+
+    df = df.rename(dict(zip(df.columns, target_timestamps)), axis=1)
+
+    expanded_df = pd.DataFrame()
+    for col in df.columns:
+        expanded_col = pd.DataFrame(
+            df[col].to_list(), columns=[f"{feature}:{col}" for feature in bands]
+        )
+        expanded_df = pd.concat([expanded_df, expanded_col], axis=1)
+
+    return expanded_df

openeo_gfmap/utils/catalogue.py

@@ -0,0 +1,387 @@
+"""Functionalities to interract with product catalogues."""
+
+from typing import Optional
+
+import geojson
+import pandas as pd
+import requests
+from pyproj.crs import CRS
+from rasterio.warp import transform_bounds
+from requests import adapters
+from shapely.geometry import Point, box, shape
+from shapely.ops import unary_union
+
+from openeo_gfmap import (
+    Backend,
+    BackendContext,
+    BoundingBoxExtent,
+    SpatialContext,
+    TemporalContext,
+)
+from openeo_gfmap.utils import _log
+
+request_sessions: Optional[requests.Session] = None
+
+
+def _request_session() -> requests.Session:
+    global request_sessions
+
+    if request_sessions is None:
+        request_sessions = requests.Session()
+        retries = adapters.Retry(
+            total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504]
+        )
+        request_sessions.mount("https://", adapters.HTTPAdapter(max_retries=retries))
+    return request_sessions
+
+
+class UncoveredS1Exception(Exception):
+    """Exception raised when there is no product available to fully cover spatially a given
+    spatio-temporal context for the Sentinel-1 collection."""
+
+    pass
+
+
+def _parse_cdse_products(response: dict):
+    """Parses the geometry and timestamps of products from the CDSE catalogue."""
+    geometries = []
+    timestamps = []
+    products = response["features"]
+
+    for product in products:
+        if "geometry" in product and "startDate" in product["properties"]:
+            geometries.append(shape(product["geometry"]))
+            timestamps.append(pd.to_datetime(product["properties"]["startDate"]))
+        else:
+            _log.warning(
+                "Cannot parse product %s does not have a geometry or timestamp.",
+                product["properties"]["id"],
+            )
+    return geometries, timestamps
+
+
+def _query_cdse_catalogue(
+    collection: str,
+    bounds: list,
+    temporal_extent: TemporalContext,
+    **additional_parameters: dict,
+) -> dict:
+    """
+    Queries the CDSE catalogue for a given collection, spatio-temporal context and additional
+    parameters.
+
+    Params
+    ------
+
+    """
+    minx, miny, maxx, maxy = bounds
+
+    # The date format should be YYYY-MM-DD
+    start_date = f"{temporal_extent.start_date}T00:00:00Z"
+    end_date = f"{temporal_extent.end_date}T00:00:00Z"
+
+    url = (
+        f"https://catalogue.dataspace.copernicus.eu/resto/api/collections/"
+        f"{collection}/search.json?box={minx},{miny},{maxx},{maxy}"
+        f"&sortParam=startDate&maxRecords=1000&dataset=ESA-DATASET"
+        f"&startDate={start_date}&completionDate={end_date}"
+    )
+    for key, value in additional_parameters.items():
+        url += f"&{key}={value}"
+
+    session = _request_session()
+    response = session.get(url, timeout=60)
+
+    if response.status_code != 200:
+        raise Exception(
+            f"Cannot check S1 catalogue on CDSE: Request to {url} failed with "
+            f"status code {response.status_code}"
+        )
+
+    return response.json()
+
+
+def _check_cdse_catalogue(
+    collection: str,
+    bounds: list,
+    temporal_extent: TemporalContext,
+    **additional_parameters: dict,
+) -> bool:
+    """Checks if there is at least one product available in the
+    given spatio-temporal context for a collection in the CDSE catalogue,
+    as there might be issues in the API that sometimes returns empty results
+    for a valid query.
+
+    Parameters
+    ----------
+    collection : str
+        The collection name to be checked. (For example: Sentinel1 or Sentinel2)
+    spatial_extent : SpatialContext
+        The spatial extent to be checked, it will check within its bounding box.
+    temporal_extent : TemporalContext
+        The temporal period to be checked.
+    additional_parameters : Optional[dict], optional
+        Additional parameters to be passed to the catalogue, by default empty.
+        Parameters (key, value) will be passed as "&key=value" in the query,
+        for example: {"sortOrder": "ascending"} will be passed as "&ascendingOrder=True"
+
+    Returns
+    -------
+    True if there is at least one product, False otherwise.
+    """
+    body = _query_cdse_catalogue(
+        collection, bounds, temporal_extent, **additional_parameters
+    )
+
+    grd_tiles = list(
+        filter(
+            lambda feature: feature["properties"]["productType"].contains("GRD"),
+            body["features"],
+        )
+    )
+
+    return len(grd_tiles) > 0
+
+
+def _compute_max_gap_days(
+    temporal_extent: TemporalContext, timestamps: list[pd.DatetimeIndex]
+) -> int:
+    """Computes the maximum temporal gap in days from the timestamps parsed from the catalogue.
+    Requires the start and end date to be included in the timestamps to compute the gap before
+    and after the first and last observation.
+
+    Parameters
+    ----------
+    temporal_extent : TemporalContext
+        The temporal extent to be checked. Same as used to query the catalogue.
+    timestamps : list[pd.DatetimeIndex]
+        The list of timestamps parsed from the catalogue and to compute the gap from.
+
+    Returns
+    -------
+    days : int
+        The maximum temporal gap in days.
+    """
+    # Computes max temporal gap. Include requested start and end date so we dont miss
+    # any start or end gap before first/last observation
+    timestamps = pd.DatetimeIndex(
+        sorted(
+            [pd.to_datetime(temporal_extent.start_date, utc=True)]
+            + timestamps
+            + [pd.to_datetime(temporal_extent.end_date, utc=True)]
+        )
+    )
+    return timestamps.to_series().diff().max().days
+
+
+def s1_area_per_orbitstate_vvvh(
+    backend: BackendContext,
+    spatial_extent: SpatialContext,
+    temporal_extent: TemporalContext,
+) -> dict:
+    """
+    Evaluates for both the ascending and descending state orbits the area of interesection and
+    maximum temporal gap for the available products with a VV&VH polarisation.
+
+    Parameters
+    ----------
+    backend : BackendContext
+        The backend to be within, as each backend might use different catalogues. Only the CDSE,
+        CDSE_STAGING and FED backends are supported.
+    spatial_extent : SpatialContext
+        The spatial extent to be checked, it will check within its bounding box.
+    temporal_extent : TemporalContext
+        The temporal period to be checked.
+
+    Returns
+    ------
+    dict
+        Keys containing the orbit state and values containing the total area of intersection and
+        in km^2 and maximum temporal gap in days.
+    """
+    if isinstance(spatial_extent, geojson.FeatureCollection):
+        # Transform geojson into shapely geometry and compute bounds
+        shapely_geometries = [
+            shape(feature["geometry"]) for feature in spatial_extent["features"]
+        ]
+        if len(shapely_geometries) == 1 and isinstance(shapely_geometries[0], Point):
+            point = shapely_geometries[0]
+            buffer_size = 0.0001
+            buffered_geometry = point.buffer(buffer_size)
+            bounds = buffered_geometry.bounds
+        else:
+            geometry = unary_union(shapely_geometries)
+            bounds = geometry.bounds
+        epsg = 4326
+    elif isinstance(spatial_extent, BoundingBoxExtent):
+        bounds = [
+            spatial_extent.west,
+            spatial_extent.south,
+            spatial_extent.east,
+            spatial_extent.north,
+        ]
+        epsg = spatial_extent.epsg
+    else:
+        raise ValueError(
+            "Provided spatial extent is not a valid GeoJSON or SpatialContext object."
+        )
+    # Warp the bounds if  the epsg is different from 4326
+    if epsg != 4326:
+        bounds = transform_bounds(CRS.from_epsg(epsg), CRS.from_epsg(4326), *bounds)
+
+    # Queries the products in the catalogues
+    if backend.backend in [Backend.CDSE, Backend.CDSE_STAGING, Backend.FED]:
+        ascending_products, ascending_timestamps = _parse_cdse_products(
+            _query_cdse_catalogue(
+                "Sentinel1",
+                bounds,
+                temporal_extent,
+                orbitDirection="ASCENDING",
+                polarisation="VV%26VH",
+                productType="IW_GRDH_1S-COG",
+            )
+        )
+        descending_products, descending_timestamps = _parse_cdse_products(
+            _query_cdse_catalogue(
+                "Sentinel1",
+                bounds,
+                temporal_extent,
+                orbitDirection="DESCENDING",
+                polarisation="VV%26VH",
+                productType="IW_GRDH_1S-COG",
+            )
+        )
+    else:
+        raise NotImplementedError(
+            f"This feature is not supported for backend: {backend.backend}."
+        )
+
+    # Builds the shape of the spatial extent and computes the area
+    spatial_extent = box(*bounds)
+
+    # Computes if there is the full overlap for each of those states
+    union_ascending = unary_union(ascending_products)
+    union_descending = unary_union(descending_products)
+
+    ascending_covers = union_ascending.contains(spatial_extent)
+    descending_covers = union_descending.contains(spatial_extent)
+
+    # Computes the area of intersection
+    return {
+        "ASCENDING": {
+            "full_overlap": ascending_covers,
+            "max_temporal_gap": _compute_max_gap_days(
+                temporal_extent, ascending_timestamps
+            ),
+            "area": sum(
+                product.intersection(spatial_extent).area
+                for product in ascending_products
+            ),
+        },
+        "DESCENDING": {
+            "full_overlap": descending_covers,
+            "max_temporal_gap": _compute_max_gap_days(
+                temporal_extent, descending_timestamps
+            ),
+            "area": sum(
+                product.intersection(spatial_extent).area
+                for product in descending_products
+            ),
+        },
+    }
+
+
+def select_s1_orbitstate_vvvh(
+    backend: BackendContext,
+    spatial_extent: SpatialContext,
+    temporal_extent: TemporalContext,
+    max_temporal_gap: int = 60,
+) -> str:
+    """Selects the orbit state based on some predefined rules that
+    are checked in sequential order:
+    1. prefer an orbit with full coverage over the requested bounds
+    2. prefer an orbit with a maximum temporal gap under a
+        predefined threshold
+    3. prefer the orbit that covers the most area of intersection
+        for the available products
+
+    Parameters
+    ----------
+    backend : BackendContext
+        The backend to be within, as each backend might use different catalogues. Only the CDSE,
+        CDSE_STAGING and FED backends are supported.
+    spatial_extent : SpatialContext
+        The spatial extent to be checked, it will check within its bounding box.
+    temporal_extent : TemporalContext
+        The temporal period to be checked.
+    max_temporal_gap: int, optional, default: 30
+        The maximum temporal gap in days to be considered for the orbit state.
+
+    Returns
+    ------
+    str
+        The orbit state that covers the most area of the given spatio-temporal context
+    """
+
+    # Queries the products in the catalogues
+    areas = s1_area_per_orbitstate_vvvh(backend, spatial_extent, temporal_extent)
+
+    ascending_overlap = areas["ASCENDING"]["full_overlap"]
+    descending_overlap = areas["DESCENDING"]["full_overlap"]
+    ascending_gap_too_large = areas["ASCENDING"]["max_temporal_gap"] > max_temporal_gap
+    descending_gap_too_large = (
+        areas["DESCENDING"]["max_temporal_gap"] > max_temporal_gap
+    )
+
+    orbit_choice = None
+
+    if not ascending_overlap and not descending_overlap:
+        raise UncoveredS1Exception(
+            "No product available to fully cover the requested area in both orbit states."
+        )
+
+    # Rule 1: Prefer an orbit with full coverage over the requested bounds
+    if ascending_overlap and not descending_overlap:
+        orbit_choice = "ASCENDING"
+        reason = "Only orbit fully covering the requested area."
+    elif descending_overlap and not ascending_overlap:
+        orbit_choice = "DESCENDING"
+        reason = "Only orbit fully covering the requested area."
+
+    # Rule 2: Prefer an orbit with a maximum temporal gap under a predefined threshold
+    elif ascending_gap_too_large and not descending_gap_too_large:
+        orbit_choice = "DESCENDING"
+        reason = (
+            "Only orbit with temporal gap under the threshold. "
+            f"{areas['DESCENDING']['max_temporal_gap']} days < {max_temporal_gap} days"
+        )
+    elif descending_gap_too_large and not ascending_gap_too_large:
+        orbit_choice = "ASCENDING"
+        reason = (
+            "Only orbit with temporal gap under the threshold. "
+            f"{areas['ASCENDING']['max_temporal_gap']} days < {max_temporal_gap} days"
+        )
+    # Rule 3: Prefer the orbit that covers the most area of intersection
+    # for the available products
+    elif ascending_overlap and descending_overlap:
+        ascending_cover_area = areas["ASCENDING"]["area"]
+        descending_cover_area = areas["DESCENDING"]["area"]
+
+        # Selects the orbit state that covers the most area
+        if ascending_cover_area > descending_cover_area:
+            orbit_choice = "ASCENDING"
+            reason = (
+                "Orbit has more cumulative intersected area. "
+                f"{ascending_cover_area} > {descending_cover_area}"
+            )
+        else:
+            reason = (
+                "Orbit has more cumulative intersected area. "
+                f"{descending_cover_area} > {ascending_cover_area}"
+            )
+            orbit_choice = "DESCENDING"
+
+    if orbit_choice is not None:
+        _log.info(f"Selected orbit state: {orbit_choice}. Reason: {reason}")
+        return orbit_choice
+    raise UncoveredS1Exception("Failed to select suitable Sentinel-1 orbit.")

openeo_gfmap/utils/intervals.py

@@ -0,0 +1,64 @@
+"""Utilitary function for intervals, useful for temporal aggregation
+methods.
+"""
+
+from datetime import timedelta
+
+from openeo_gfmap import TemporalContext
+
+
+def quintad_intervals(temporal_extent: TemporalContext) -> list:
+    """Returns a list of tuples (start_date, end_date) of quintad intervals
+    from the input temporal extent. Quintad intervals are intervals of
+    generally 5 days, that never overlap two months.
+
+    All months are divided in 6 quintads, where the 6th quintad might
+    contain 6 days for months of 31 days.
+    For the month of February, the 6th quintad is only of three days, or
+    four days for the leap year.
+    """
+    start_date, end_date = temporal_extent.to_datetime()
+    quintads = []
+
+    current_date = start_date
+
+    # Compute the offset of the first day on the start of the last quintad
+    if start_date.day != 1:
+        offset = (start_date - timedelta(days=1)).day % 5
+        current_date = current_date - timedelta(days=offset)
+    else:
+        offset = 0
+
+    while current_date <= end_date:
+        # Get the last day of the current month
+        last_day = current_date.replace(day=28) + timedelta(days=4)
+        last_day = last_day - timedelta(days=last_day.day)
+
+        # Get the last day of the current quintad
+        last_quintad = current_date + timedelta(days=4)
+
+        # Add a day if the day is the 30th and there is the 31th in the current month
+        if last_quintad.day == 30 and last_day.day == 31:
+            last_quintad = last_quintad + timedelta(days=1)
+
+        # If the last quintad is after the last day of the month, then
+        # set it to the last day of the month
+        if last_quintad > last_day:
+            last_quintad = last_day
+        # In the case the last quintad is after the end date, then set it to the end date
+        elif last_quintad > end_date:
+            last_quintad = end_date
+
+        quintads.append((current_date, last_quintad))
+
+        # Set the current date to the next quintad
+        current_date = last_quintad + timedelta(days=1)
+
+    # Fixing the offset issue for intervals starting in the middle of a quintad
+    quintads[0] = (quintads[0][0] + timedelta(days=offset), quintads[0][1])
+
+    # Returns to string with the YYYY-mm-dd format
+    return [
+        (start_date.strftime("%Y-%m-%d"), end_date.strftime("%Y-%m-%d"))
+        for start_date, end_date in quintads
+    ]

openeo_gfmap/utils/split_stac.py

@@ -0,0 +1,125 @@
+"""Utility function to split a STAC collection into multiple STAC collections based on CRS.
+Requires the "proj:epsg" property to be present in all the STAC items.
+"""
+
+import os
+from pathlib import Path
+from typing import Iterator, Union
+
+import pystac
+
+
+def _extract_epsg_from_stac_item(stac_item: pystac.Item) -> int:
+    """
+    Extract the EPSG code from a STAC item.
+
+    Parameters:
+    stac_item (pystac.Item): The STAC item.
+
+    Returns:
+    int: The EPSG code.
+
+    Raises:
+    KeyError: If the "proj:epsg" property is missing from the STAC item.
+    """
+
+    try:
+        epsg_code = stac_item.properties["proj:epsg"]
+        return epsg_code
+    except KeyError:
+        raise KeyError("The 'proj:epsg' property is missing from the STAC item.")
+
+
+def _get_items_by_epsg(
+    collection: pystac.Collection,
+) -> Iterator[tuple[int, pystac.Item]]:
+    """
+    Generator function that yields items grouped by their EPSG code.
+
+    Parameters:
+    collection (pystac.Collection): The STAC collection.
+
+    Yields:
+    tuple[int, pystac.Item]: EPSG code and corresponding STAC item.
+    """
+    for item in collection.get_all_items():
+        epsg = _extract_epsg_from_stac_item(item)
+        yield epsg, item
+
+
+def _create_collection_skeleton(
+    collection: pystac.Collection, epsg: int
+) -> pystac.Collection:
+    """
+    Create a skeleton for a new STAC collection with a given EPSG code.
+
+    Parameters:
+    collection (pystac.Collection): The original STAC collection.
+    epsg (int): The EPSG code.
+
+    Returns:
+    pystac.Collection: The skeleton of the new STAC collection.
+    """
+    new_collection = pystac.Collection(
+        id=f"{collection.id}_{epsg}",
+        description=f"{collection.description} Containing only items with EPSG code {epsg}",
+        extent=collection.extent.clone(),
+        summaries=collection.summaries,
+        license=collection.license,
+        stac_extensions=collection.stac_extensions,
+    )
+    if "item_assets" in collection.extra_fields:
+        item_assets_extension = pystac.extensions.item_assets.ItemAssetsExtension.ext(
+            collection
+        )
+
+        new_item_assets_extension = (
+            pystac.extensions.item_assets.ItemAssetsExtension.ext(
+                new_collection, add_if_missing=True
+            )
+        )
+
+        new_item_assets_extension.item_assets = item_assets_extension.item_assets
+    return new_collection
+
+
+def split_collection_by_epsg(
+    collection: Union[str, Path, pystac.Collection], output_dir: Union[str, Path]
+):
+    """
+    Split a STAC collection into multiple STAC collections based on EPSG code.
+
+    Parameters
+    ----------
+    collection: Union[str, Path, pystac.Collection]
+        A collection of STAC items or a path to a STAC collection.
+    output_dir: Union[str, Path]
+        The directory where the split STAC collections will be saved.
+    """
+
+    if not isinstance(collection, pystac.Collection):
+        collection = Path(collection)
+        output_dir = Path(output_dir)
+        os.makedirs(output_dir, exist_ok=True)
+
+        try:
+            collection = pystac.read_file(collection)
+        except pystac.STACError:
+            print("Please provide a path to a valid STAC collection.")
+            return
+
+    collections_by_epsg = {}
+
+    for epsg, item in _get_items_by_epsg(collection):
+        if epsg not in collections_by_epsg:
+            collections_by_epsg[epsg] = _create_collection_skeleton(collection, epsg)
+
+        # Add item to the corresponding collection
+        collections_by_epsg[epsg].add_item(item)
+
+    # Write each collection to disk
+    for epsg, new_collection in collections_by_epsg.items():
+        new_collection.update_extent_from_items()  # Update extent based on added items
+        collection_path = output_dir / f"collection-{epsg}"
+        new_collection.normalize_hrefs(str(collection_path))
+        new_collection.save()

openeo_gfmap/utils/tile_processing.py

@@ -0,0 +1,64 @@
+"""Utilitaries to process data tiles."""
+
+import numpy as np
+import xarray as xr
+
+
+def normalize_array(inarr: xr.DataArray, percentile: float = 0.99) -> xr.DataArray:
+    """Performs normalization between 0.0 and 1.0 using the given
+    percentile.
+    """
+    quantile_value = inarr.quantile(percentile, dim=["x", "y", "t"])
+    minimum = inarr.min(dim=["x", "y", "t"])
+
+    inarr = (inarr - minimum) / (quantile_value - minimum)
+
+    # Perform clipping on values that are higher than the computed quantile
+    return inarr.where(inarr < 1.0, 1.0)
+
+
+def select_optical_bands(inarr: xr.DataArray) -> xr.DataArray:
+    """Filters and keep only the optical bands for a given array."""
+    return inarr.sel(
+        bands=[
+            band
+            for band in inarr.coords["bands"].to_numpy()
+            if band.startswith("S2-L2A-B")
+        ]
+    )
+
+
+def select_sar_bands(inarr: xr.DataArray) -> xr.DataArray:
+    """Filters and keep only the SAR bands for a given array."""
+    return inarr.sel(
+        bands=[
+            band
+            for band in inarr.coords["bands"].to_numpy()
+            if band in ["S1-SIGMA0-VV", "S1-SIGMA0-VH", "S1-SIGMA0-HH", "S1-SIGMA0-HV"]
+        ]
+    )
+
+
+def array_bounds(inarr: xr.DataArray) -> tuple:
+    """Returns the 4 bounds values for the x and y coordinates of the tile"""
+    return (
+        inarr.coords["x"].min().item(),
+        inarr.coords["y"].min().item(),
+        inarr.coords["x"].max().item(),
+        inarr.coords["y"].max().item(),
+    )
+
+
+def arrays_cosine_similarity(
+    first_array: xr.DataArray, second_array: xr.DataArray
+) -> float:
+    """Returns a similarity score based on normalized cosine distance. The
+    input arrays must have similar ranges to obtain a valid score.
+    1.0 represents the best score (same tiles), while 0.0 is the worst score.
+    """
+    dot_product = np.sum(first_array * second_array)
+    first_norm = np.linalg.norm(first_array)
+    second_norm = np.linalg.norm(second_array)
+    similarity = (dot_product / (first_norm * second_norm)).item()
+
+    return similarity

pyproject.toml

@@ -8,9 +8,17 @@ dependencies = [
     "geojson>=3.2.0",
     "geopandas>=1.1.2",
     "joblib>=1.5.3",
-    "matplotlib>=3.10.8",
+    "matplotlib==3.7.5",
+    "numpy<2.0.0",
     "openeo>=0.47.0",
-    "rasterio>=1.5.0",
+    "prometheo",
+    "pydantic>=2.12.5",
+    "rasterio<1.4.0",
+    "scipy>=1.17.0",
     "streamlit>=1.53.1",
     "streamlit-folium>=0.26.1",
+    "torch==2.3.1",
 ]
+
+[tool.uv.sources]
+prometheo = { git = "https://github.com/WorldCereal/prometheo.git" }