feat: align conversion parameters with data-model reference

- All 4 S2 groups (r10m, r20m, r60m, quicklook)
- CRS groups for spatial reference metadata
- Optimized chunk sizes (1024) and tile width (256)
- Enable-sharding parameter (S2: true, S1: false)
- Simplified parameter lookup (concise fallback logic)

Aligns with data-model/.vscode/launch.json config.

Refs: #31

Author: emmanuelmathot

.gitignore

@@ -62,3 +62,5 @@ Thumbs.db
 *.zarr
 out/
 reports/
+*.pyc
+pipeline_utils.py

scripts/convert.py

@@ -7,6 +7,7 @@
 import logging
 import os
 import sys
+from typing import Any
 from urllib.parse import urlparse
 
 import fsspec
@@ -46,6 +47,10 @@ def run_conversion(
     collection: str,
     s3_output_bucket: str,
     s3_output_prefix: str,
+    groups: str | None = None,
+    spatial_chunk: int | None = None,
+    tile_width: int | None = None,
+    enable_sharding: bool | None = None,
 ) -> str:
     """Run GeoZarr conversion workflow.
 
@@ -54,6 +59,10 @@ def run_conversion(
         collection: Collection ID for parameter lookup
         s3_output_bucket: S3 bucket for output
         s3_output_prefix: S3 prefix for output
+        groups: Override groups parameter (comma-separated if multiple)
+        spatial_chunk: Override spatial chunk size
+        tile_width: Override tile width
+        enable_sharding: Override sharding enable flag
 
     Returns:
         Output Zarr URL (s3://...)
@@ -74,15 +83,17 @@ def run_conversion(
         zarr_url = source_url
         logger.info(f"Direct Zarr URL: {zarr_url}")
 
-    # Get conversion parameters from collection config
-    logger.debug(f"Getting conversion parameters for {collection}...")
+    # Get conversion parameters (with optional overrides)
     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']}")
-
-    # Construct output path
+    overrides = {
+        "groups": groups.split(",") if groups and "," in groups else groups or None,
+        "spatial_chunk": spatial_chunk,
+        "tile_width": tile_width,
+        "enable_sharding": enable_sharding,
+    }
+    params.update({k: v for k, v in overrides.items() if v is not None})
+
+    logger.info(f"Conversion params: {params}")  # Construct output path
     output_url = f"s3://{s3_output_bucket}/{s3_output_prefix}/{collection}/{item_id}.zarr"
 
     # Clean up existing output to avoid base array artifacts
@@ -98,46 +109,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
+    # 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: dict[str, Any] = {}
+    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 []
 
-    with Client() as client:
-        logger.info(f"🚀 Dask cluster started: {client.dashboard_link}")
+    create_geozarr_dataset(
+        dt_input=dt,
+        groups=groups_list,
+        output_path=output_url,
+        spatial_chunk=params["spatial_chunk"],
+        tile_width=params["tile_width"],
+        **kwargs,
+    )
 
-        # 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}")
+    logger.info("✅ Conversion completed successfully!")
+    logger.info(f"Output: {output_url}")
 
     return output_url
 
@@ -150,18 +175,32 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--s3-output-bucket", required=True, help="S3 output bucket")
     parser.add_argument("--s3-output-prefix", required=True, help="S3 output prefix")
     parser.add_argument("--verbose", action="store_true", help="Enable verbose logging")
+    # Optional parameter overrides
+    parser.add_argument("--groups", help="Override groups (comma-separated)")
+    parser.add_argument("--spatial-chunk", help="Override spatial chunk size")
+    parser.add_argument("--tile-width", help="Override tile width")
+    parser.add_argument("--enable-sharding", help="Override sharding (true/false)")
 
     args = parser.parse_args(argv)
-
     if args.verbose:
         logging.getLogger().setLevel(logging.DEBUG)
 
+    # Parse override args (empty string → None)
+    groups = args.groups or None
+    spatial_chunk = int(args.spatial_chunk) if args.spatial_chunk else None
+    tile_width = int(args.tile_width) if args.tile_width else None
+    enable_sharding = args.enable_sharding.lower() == "true" if args.enable_sharding else None
+
     try:
         output_url = run_conversion(
             args.source_url,
             args.collection,
             args.s3_output_bucket,
             args.s3_output_prefix,
+            groups,
+            spatial_chunk,
+            tile_width,
+            enable_sharding,
         )
         logger.info(f"Success: {output_url}")
         return 0

scripts/get_conversion_params.py

@@ -25,33 +25,30 @@
         "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,
     },
 }
 
 
 def get_conversion_params(collection_id: str) -> dict[str, Any]:
-    """Get conversion parameters for collection.
-
-    Args:
-        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)
-    """
-    # Extract mission prefix (sentinel-1 or sentinel-2)
+    """Get conversion parameters for collection. Defaults to Sentinel-2 if unrecognized."""
     parts = collection_id.lower().split("-")
     if len(parts) >= 2:
-        prefix = f"{parts[0]}-{parts[1]}"  # "sentinel-1" or "sentinel-2"
+        prefix = f"{parts[0]}-{parts[1]}"
         if prefix in CONFIGS:
             return CONFIGS[prefix]
-
-    # Default to Sentinel-2 if no match
     return CONFIGS["sentinel-2"]
 
 
@@ -73,7 +70,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 +79,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 +94,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
 

workflows/README.md

@@ -114,6 +114,25 @@ Key parameters (see [../README.md](../README.md) for full reference):
 - `s3_output_bucket`: Output bucket
 - `pipeline_image_version`: Docker image tag
 
+**Override conversion parameters** (optional, for testing):
+
+```bash
+# Example: Test with different chunk size and disable sharding
+argo submit workflows/base/workflowtemplate.yaml \
+  --from workflowtemplate/geozarr-pipeline \
+  -p source_url="https://api.example.com/stac/.../items/ITEM_ID" \
+  -p override_spatial_chunk="2048" \
+  -p override_enable_sharding="false"
+```
+
+Available overrides (empty = use collection defaults):
+- `override_groups`: Comma-separated zarr groups (e.g., `/measurements/reflectance/r10m`)
+- `override_spatial_chunk`: Chunk size (e.g., `2048`)
+- `override_tile_width`: Tile width (e.g., `512`)
+- `override_enable_sharding`: Enable sharding (`true`/`false`)
+
+Defaults: S2 (1024/256/true), S1 (4096/512/false). See `scripts/get_conversion_params.py`.
+
 ### Resource Tuning
 
 Edit `workflows/base/workflowtemplate.yaml`:

workflows/base/workflowtemplate.yaml

@@ -32,6 +32,15 @@ spec:
       value: tests-output
     - name: pipeline_image_version
       value: slim
+    # Optional conversion parameter overrides (empty = use collection defaults)
+    - name: override_groups
+      value: ""
+    - name: override_spatial_chunk
+      value: ""
+    - name: override_tile_width
+      value: ""
+    - name: override_enable_sharding
+      value: ""
   templates:
   - name: main
     dag:
@@ -59,6 +68,14 @@ spec:
       - --s3-output-prefix
       - "{{workflow.parameters.s3_output_prefix}}"
       - --verbose
+      - --groups
+      - "{{workflow.parameters.override_groups}}"
+      - --spatial-chunk
+      - "{{workflow.parameters.override_spatial_chunk}}"
+      - --tile-width
+      - "{{workflow.parameters.override_tile_width}}"
+      - --enable-sharding
+      - "{{workflow.parameters.override_enable_sharding}}"
       resources:
         requests:
           memory: 4Gi