12 job splitters

pyproject.toml

@@ -34,7 +34,8 @@ dependencies = [
     "openeo[localprocessing]",
     "cftime",
     "pyarrow",
-    "fastparquet"
+    "fastparquet",
+    "h3",
 ]
 
 [project.urls]

src/openeo_gfmap/backend.py

@@ -19,6 +19,7 @@ class Backend(Enum):
     TERRASCOPE = "terrascope"
     EODC = "eodc"  # Dask implementation. Do not test on this yet.
     CDSE = "cdse"  # Terrascope implementation (pyspark) #URL: openeo.dataspace.copernicus.eu (need to register)
+    CDSE_STAGING = "cdse-staging"
     LOCAL = "local"  # Based on the same components of EODc
 
 
@@ -82,6 +83,14 @@ def cdse_connection() -> openeo.Connection:
     )
 
 
+def cdse_staging_connection() -> openeo.Connection:
+    """Performs a connection to the CDSE backend using oidc authentication."""
+    return _create_connection(
+        url="openeo-staging.dataspace.copernicus.eu",
+        env_var_suffix="CDSE_STAGING",
+    )
+
+
 def eodc_connection() -> openeo.Connection:
     """Perfroms a connection to the EODC backend using the oidc authentication."""
     return _create_connection(
@@ -93,4 +102,5 @@ def eodc_connection() -> openeo.Connection:
 BACKEND_CONNECTIONS: Dict[Backend, Callable] = {
     Backend.TERRASCOPE: vito_connection,
     Backend.CDSE: cdse_connection,
+    Backend.CDSE_STAGING: cdse_staging_connection,
 }

src/openeo_gfmap/fetching/commons.py

@@ -111,12 +111,16 @@ def load_collection(
             properties=load_collection_parameters,
         )
     elif fetch_type == FetchType.POLYGON:
-        assert isinstance(
-            spatial_extent, GeoJSON
-        ), "Please provide only a GeoJSON FeatureCollection for point based fetching."
-        assert (
-            spatial_extent["type"] == "FeatureCollection"
-        ), "Please provide a FeatureCollection type of GeoJSON"
+        if isinstance(spatial_extent, GeoJSON):
+            assert (
+                spatial_extent["type"] == "FeatureCollection"
+            ), "Please provide a FeatureCollection type of GeoJSON"
+        elif isinstance(spatial_extent, str):
+            assert spatial_extent.startswith("https://") or spatial_extent.startswith(
+                "http://"
+            ), "Please provide a valid URL or a path to a GeoJSON file."
+        else:
+            raise ValueError("Please provide a valid URL to a GeoParquet or GeoJSON file.")
         cube = connection.load_collection(
             collection_id=collection_name,
             temporal_extent=[temporal_extent.start_date, temporal_extent.end_date],
@@ -133,6 +137,13 @@ def load_collection(
         cube = cube.mask(pre_mask.resample_cube_spatial(cube))
 
     if fetch_type == FetchType.POLYGON:
-        cube = cube.filter_spatial(spatial_extent)
+        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

src/openeo_gfmap/fetching/s1.py

@@ -130,6 +130,10 @@ def s1_grd_default_processor(cube: openeo.DataCube, **params):
         "default": partial(get_s1_grd_default_fetcher, collection_name="SENTINEL1_GRD"),
         "preprocessor": partial(get_s1_grd_default_processor, collection_name="SENTINEL1_GRD"),
     },
+    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"),
+    },
 }
 
 

src/openeo_gfmap/fetching/s2.py

@@ -105,10 +105,6 @@ def s2_l2a_fetch_default(
             **params,
         )
 
-        # 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_fetch_default
@@ -190,6 +186,10 @@ def s2_l2a_default_processor(cube: openeo.DataCube, **params):
         "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"),
+    },
 }
 
 

src/openeo_gfmap/manager/__init__.py

@@ -0,0 +1,6 @@
+"""OpenEO GFMAP Manager submodule. Implements the logic of splitting the jobs into subjobs and
+managing the subjobs.
+"""
+import logging
+
+_log = logging.getLogger(__name__)