fix: ensure --crs-groups flag set for S2 conversions

The --crs-groups flag triggers prepare_dataset_with_crs_info() in data-model,
which writes CRS metadata via ds.rio.write_crs() and creates the spatial_ref
coordinate variable required by TiTiler validation.

Restores working configuration from commit 21ea009.

fix: update sentinel-2 conversion parameters for accuracy

feat: add enable-sharding parameter support

- Add enable_sharding to mission configs (S2: true, S1: false)
- Pass enable_sharding to create_geozarr_dataset kwargs
- Add shell/JSON output support for new parameter
- Align with data-model launch.json reference config
- Remove Dask cluster parameter (use ENABLE_DASK_CLUSTER env var instead)

Refs: data-model/.vscode/launch.json L2A config

Author: wietzesuijker

scripts/augment_stac_item.py

@@ -66,9 +66,8 @@ def add_visualization(item: Item, raster_base: str, collection_id: str) -> None:
             _add_tile_links(item, base_url, query, "Sentinel-1 GRD VH")
 
     elif coll_lower.startswith(("sentinel-2", "sentinel2")):
-        # S2: Point to overview level 0 for quicklook TCI
-        # Use /r10m/0/tci path to access the overview array with spatial_ref
-        var_path = "/quality/l2a_quicklook/r10m/0/tci"
+        # S2: Use colon separator for TiTiler variable path
+        var_path = "/quality/l2a_quicklook/r10m:tci"
         query = (
             f"variables={urllib.parse.quote(var_path, safe='')}&bidx=1&bidx=2&bidx=3&assets=TCI_10m"
         )

scripts/convert.py

@@ -77,10 +77,11 @@ def run_conversion(
     # Get conversion parameters from collection config
     logger.debug(f"Getting conversion parameters for {collection}...")
     params = get_conversion_params(collection)
-    logger.debug(f"  Groups:      {params['groups']}")
-    logger.debug(f"  Chunk:       {params['spatial_chunk']}")
-    logger.debug(f"  Tile width:  {params['tile_width']}")
-    logger.debug(f"  Extra flags: {params['extra_flags']}")
+    logger.debug(f"  Groups:          {params['groups']}")
+    logger.debug(f"  Chunk:           {params['spatial_chunk']}")
+    logger.debug(f"  Tile width:      {params['tile_width']}")
+    logger.debug(f"  Extra flags:     {params['extra_flags']}")
+    logger.debug(f"  Enable sharding: {params['enable_sharding']}")
 
     # Construct output path
     output_url = f"s3://{s3_output_bucket}/{s3_output_prefix}/{collection}/{item_id}.zarr"
@@ -98,46 +99,60 @@ def run_conversion(
     logger.info(f"  Source:      {zarr_url}")
     logger.info(f"  Destination: {output_url}")
 
-    # Set up Dask cluster for parallel processing
-    from dask.distributed import Client
-
-    with Client() as client:
-        logger.info(f"🚀 Dask cluster started: {client.dashboard_link}")
-
-        # Load source dataset
-        logger.info("Loading source dataset...")
-        storage_options = get_storage_options(zarr_url)
-        dt = xr.open_datatree(
-            zarr_url,
-            engine="zarr",
-            chunks="auto",
-            storage_options=storage_options,
-        )
-        logger.info(f"Loaded DataTree with {len(dt.children)} groups")
-
-        # Convert to GeoZarr
-        logger.info("Converting to GeoZarr format...")
-
-        # Parse extra flags for optional parameters
-        kwargs = {}
-        if params["extra_flags"] and "--crs-groups" in params["extra_flags"]:
-            crs_groups_str = params["extra_flags"].split("--crs-groups")[1].strip().split()[0]
-            kwargs["crs_groups"] = [crs_groups_str]
-
-        # groups parameter must be a list
-        groups_list = [params["groups"]] if isinstance(params["groups"], str) else params["groups"]
-
-        create_geozarr_dataset(
-            dt_input=dt,
-            groups=groups_list,
-            output_path=output_url,
-            spatial_chunk=params["spatial_chunk"],
-            tile_width=params["tile_width"],
-            **kwargs,
-        )
-
-        logger.info("✅ Conversion completed successfully!")
-        logger.info(f"Output: {output_url}")
+    # Optional: Set up Dask cluster if enabled via environment variable
+    # Note: eopf-geozarr handles its own Dask setup when using create_geozarr_dataset
+    # This is here only for future compatibility if we need external cluster management
+    use_dask = os.getenv("ENABLE_DASK_CLUSTER", "").lower() in ("true", "1", "yes")
+    if use_dask:
+        logger.info("🚀 Dask cluster enabled via ENABLE_DASK_CLUSTER env var")
+        # Future: Could connect to external cluster here if needed
+        # from dask.distributed import Client
+        # dask_address = os.getenv("DASK_SCHEDULER_ADDRESS")
+        # client = Client(dask_address) if dask_address else Client()
+
+    # Load source dataset
+    logger.info("Loading source dataset...")
+    storage_options = get_storage_options(zarr_url)
+    dt = xr.open_datatree(
+        zarr_url,
+        engine="zarr",
+        chunks="auto",
+        storage_options=storage_options,
+    )
+    logger.info(f"Loaded DataTree with {len(dt.children)} groups")
+
+    # Convert to GeoZarr
+    logger.info("Converting to GeoZarr format...")
+
+    # Parse extra flags for optional parameters
+    kwargs = {}
+    if params["extra_flags"] and "--crs-groups" in params["extra_flags"]:
+        crs_groups_str = params["extra_flags"].split("--crs-groups")[1].strip().split()[0]
+        kwargs["crs_groups"] = [crs_groups_str]
+
+    # Add sharding if enabled
+    if params.get("enable_sharding", False):
+        kwargs["enable_sharding"] = True
+
+    # groups parameter must be a list
+    groups_param = params["groups"]
+    if isinstance(groups_param, str):
+        groups_list: list[str] = [groups_param]
+    else:
+        # groups_param is list[str] in mission configs
+        groups_list = list(groups_param) if groups_param else []
+
+    create_geozarr_dataset(
+        dt_input=dt,
+        groups=groups_list,
+        output_path=output_url,
+        spatial_chunk=params["spatial_chunk"],
+        tile_width=params["tile_width"],
+        **kwargs,
+    )
+
+    logger.info("✅ Conversion completed successfully!")
+    logger.info(f"Output: {output_url}")
 
     return output_url
 

scripts/create_geozarr_item.py

@@ -5,7 +5,6 @@
 
 import argparse
 import logging
-import re
 from typing import Any
 from urllib.parse import urlparse
 
@@ -31,30 +30,6 @@ def s3_to_https(s3_url: str, endpoint: str) -> str:
     return f"https://{bucket}.{host}/{path}"
 
 
-def normalize_r60m_href(href: str) -> str:
-    """Add /0/ subdirectory to r10m/r20m/r60m paths to match GeoZarr output structure.
-
-    GeoZarr conversion creates /0/ subdirectories (overview level 0) for all
-    resolution bands. This normalizes asset hrefs accordingly.
-
-    Example: .../r10m/tci → .../r10m/0/tci
-             .../r60m/b09 → .../r60m/0/b09
-    """
-    # Check for any resolution level pattern
-    for res in ["r10m", "r20m", "r60m"]:
-        if f"/{res}/" not in href:
-            continue
-
-        # If already has /0/ or other digit subdirectory, don't modify
-        if re.search(rf"/{res}/\d+/", href):
-            continue
-
-        # Insert /0/ after /{res}/
-        href = re.sub(rf"/({res})/", r"/\1/0/", href)
-
-    return href
-
-
 def find_source_zarr_base(source_item: dict) -> str | None:
     """Find the base Zarr URL from source item assets."""
     for asset in source_item.get("assets", {}).values():
@@ -112,9 +87,6 @@ def create_geozarr_item(
                     subpath = old_href[len(source_zarr_base) :]
                     new_href = output_zarr_base + subpath
 
-                    # Normalize r60m paths to include /0/ subdirectory (GeoZarr structure)
-                    new_href = normalize_r60m_href(new_href)
-
                     # Convert to https if needed
                     if new_href.startswith("s3://"):
                         new_href = s3_to_https(new_href, s3_endpoint)

scripts/get_conversion_params.py

@@ -25,12 +25,19 @@
         "extra_flags": "--gcp-group /conditions/gcp",
         "spatial_chunk": 4096,
         "tile_width": 512,
+        "enable_sharding": False,
     },
     "sentinel-2": {
-        "groups": "/quality/l2a_quicklook/r10m",
-        "extra_flags": "--crs-groups /quality/l2a_quicklook/r10m",
-        "spatial_chunk": 4096,
-        "tile_width": 512,
+        "groups": [
+            "/measurements/reflectance/r10m",
+            "/measurements/reflectance/r20m",
+            "/measurements/reflectance/r60m",
+            "/quality/l2a_quicklook/r10m",
+        ],
+        "extra_flags": "--crs-groups /conditions/geometry",
+        "spatial_chunk": 1024,
+        "tile_width": 256,
+        "enable_sharding": True,
     },
 }
 
@@ -42,7 +49,7 @@ def get_conversion_params(collection_id: str) -> dict[str, Any]:
         collection_id: Collection identifier (e.g., sentinel-1-l1-grd, sentinel-2-l2a-dp-test)
 
     Returns:
-        Dict of conversion parameters (groups, extra_flags, spatial_chunk, tile_width)
+        Dict of conversion parameters (groups, extra_flags, spatial_chunk, tile_width, enable_sharding)
     """
     # Extract mission prefix (sentinel-1 or sentinel-2)
     parts = collection_id.lower().split("-")
@@ -73,7 +80,7 @@ def main(argv: list[str] | None = None) -> int:
     )
     parser.add_argument(
         "--param",
-        choices=["groups", "extra_flags", "spatial_chunk", "tile_width"],
+        choices=["groups", "extra_flags", "spatial_chunk", "tile_width", "enable_sharding"],
         help="Get single parameter (for shell scripts)",
     )
 
@@ -82,7 +89,12 @@ def main(argv: list[str] | None = None) -> int:
 
     if args.param:
         # Output single parameter (for shell variable assignment)
-        print(params.get(args.param, ""))
+        value = params.get(args.param, "")
+        # Convert boolean to shell-friendly format
+        if isinstance(value, bool):
+            print("true" if value else "false")
+        else:
+            print(value if value is not None else "")
     elif args.format == "json":
         # Output JSON (for parsing with jq)
         print(json.dumps(params, indent=2))
@@ -92,6 +104,7 @@ def main(argv: list[str] | None = None) -> int:
         print(f"EXTRA_FLAGS='{params['extra_flags']}'")
         print(f"CHUNK={params['spatial_chunk']}")
         print(f"TILE_WIDTH={params['tile_width']}")
+        print(f"ENABLE_SHARDING={'true' if params['enable_sharding'] else 'false'}")
 
     return 0
 

scripts/get_zarr_url.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python3
+"""Extract Zarr URL from STAC item - standalone script for workflow templates."""
+
+import sys
+
+import httpx
+
+
+def get_zarr_url(stac_item_url: str) -> str:
+    """Get Zarr asset URL from STAC item."""
+    r = httpx.get(stac_item_url, timeout=30.0, follow_redirects=True)
+    r.raise_for_status()
+    assets = r.json().get("assets", {})
+
+    # Priority: product, zarr, then any .zarr asset
+    for key in ["product", "zarr"]:
+        if key in assets and (href := assets[key].get("href")):
+            return str(href)
+
+    # Fallback: any asset with .zarr in href
+    for asset in assets.values():
+        if ".zarr" in asset.get("href", ""):
+            return str(asset["href"])
+
+    raise RuntimeError("No Zarr asset found in STAC item")
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 2:
+        print("Usage: get_zarr_url.py <stac_item_url>", file=sys.stderr)
+        sys.exit(1)
+
+    try:
+        zarr_url = get_zarr_url(sys.argv[1])
+        print(zarr_url)
+    except Exception as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)