12 job splitters

examples/feature_extractors/quantile_feature_extractor.py

@@ -42,13 +42,13 @@ def output_labels(self) -> list:
 
     def execute(self, inarr: xr.DataArray) -> xr.DataArray:
         # compute indices
-        b03 = inarr.sel(bands="S2-B03")
-        b04 = inarr.sel(bands="S2-B04")
-        b05 = inarr.sel(bands="S2-B05")
-        b06 = inarr.sel(bands="S2-B06")
-        b08 = inarr.sel(bands="S2-B08")
-        b11 = inarr.sel(bands="S2-B11")
-        b12 = inarr.sel(bands="S2-B12")
+        b03 = inarr.sel(bands="S2-L2A-B03")
+        b04 = inarr.sel(bands="S2-L2A-B04")
+        b05 = inarr.sel(bands="S2-L2A-B05")
+        b06 = inarr.sel(bands="S2-L2A-B06")
+        b08 = inarr.sel(bands="S2-L2A-B08")
+        b11 = inarr.sel(bands="S2-L2A-B11")
+        b12 = inarr.sel(bands="S2-L2A-B12")
 
         ndvi = (b08 - b04) / (b08 + b04)
         ndwi = (b03 - b08) / (b03 + b08)
@@ -100,14 +100,14 @@ def execute(self, inarr: xr.DataArray) -> xr.DataArray:
 
     # The bands that you can extract are defined in the code openeo_gfmap.fetching.s2.BASE_SENTINEL2_L2A_MAPPING
     bands = [
-        "S2-B03",
-        "S2-B04",
-        "S2-B05",
-        "S2-B06",
-        "S2-B08",
-        "S2-B11",
-        "S2-B12",
-        "S2-SCL",
+        "S2-L2A-B03",
+        "S2-L2A-B04",
+        "S2-L2A-B05",
+        "S2-L2A-B06",
+        "S2-L2A-B08",
+        "S2-L2A-B11",
+        "S2-L2A-B12",
+        "S2-L2A-SCL",
     ]
 
     # Use the base feching

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"),
+    },
 }