fix: add S3 cleanup and Python workflow scripts to prevent base array artifacts

- Add S3 cleanup before conversion to remove stale base arrays
- Revert to Python entry points (convert.py, register.py) for maintainability
- Fix groups parameter type (string → list) for API compatibility
- Use clean args approach instead of inline bash scripts
- Fix TiTiler preview path to use overview arrays (/r10m/0:tci)

This addresses PR feedback by consolidating the cleanup fix with proper
Python-based workflow structure. All debugging iterations squashed.

Author: wietzesuijker

scripts/augment_stac_item.py

@@ -66,8 +66,10 @@ 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: Quicklook path
-        var_path = "/quality/l2a_quicklook/r10m:tci"
+        # S2: Point to overview level 0 for quicklook TCI
+        # Use /r10m/0:tci instead of /r10m:tci because base array lacks
+        # spatial_ref coordinate (only overviews have it)
+        var_path = "/quality/l2a_quicklook/r10m/0:tci"
         query = (
             f"variables={urllib.parse.quote(var_path, safe='')}&bidx=1&bidx=2&bidx=3&assets=TCI_10m"
         )

scripts/convert.py

@@ -5,9 +5,11 @@
 
 import argparse
 import logging
+import os
 import sys
 from urllib.parse import urlparse
 
+import fsspec
 import httpx
 import xarray as xr
 from eopf_geozarr import create_geozarr_dataset
@@ -83,6 +85,15 @@ def run_conversion(
     # 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
+    logger.info(f"🧹 Cleaning up existing output at: {output_url}")
+    try:
+        fs = fsspec.filesystem("s3", client_kwargs={"endpoint_url": os.getenv("AWS_ENDPOINT_URL")})
+        fs.rm(output_url, recursive=True)
+        logger.info("✅ Cleanup completed")
+    except Exception as e:
+        logger.info(f"ℹ️  No existing output to clean (or cleanup failed): {e}")
+
     logger.info("Starting GeoZarr conversion...")
     logger.info(f"  Source:      {zarr_url}")
     logger.info(f"  Destination: {output_url}")
@@ -113,9 +124,12 @@ def run_conversion(
             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=params["groups"],
+            groups=groups_list,
             output_path=output_url,
             spatial_chunk=params["spatial_chunk"],
             tile_width=params["tile_width"],
@@ -135,9 +149,13 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--collection", required=True, help="Collection ID")
     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")
 
     args = parser.parse_args(argv)
 
+    if args.verbose:
+        logging.getLogger().setLevel(logging.DEBUG)
+
     try:
         output_url = run_conversion(
             args.source_url,

scripts/get_conversion_params.py

@@ -21,13 +21,13 @@
 # Conversion parameters by mission
 CONFIGS: dict[str, dict[str, Any]] = {
     "sentinel-1": {
-        "groups": ["/measurements"],
+        "groups": "/measurements",
         "extra_flags": "--gcp-group /conditions/gcp",
         "spatial_chunk": 4096,
         "tile_width": 512,
     },
     "sentinel-2": {
-        "groups": ["/quality/l2a_quicklook/r10m"],
+        "groups": "/quality/l2a_quicklook/r10m",
         "extra_flags": "--crs-groups /quality/l2a_quicklook/r10m",
         "spatial_chunk": 4096,
         "tile_width": 512,

submit_test_workflow.py

@@ -1,78 +1,27 @@
 #!/usr/bin/env python3
-"""Submit workflow to geozarr pipeline via RabbitMQ."""
-
 import json
 import os
-import sys
 
 import pika
 
-
-def submit_workflow(payload: dict) -> bool:
-    """Submit workflow via RabbitMQ."""
-    try:
-        username = os.getenv("RABBITMQ_USER", "user")
-        password = os.getenv("RABBITMQ_PASSWORD")
-
-        if not password:
-            print("❌ RABBITMQ_PASSWORD not set")
-            print(
-                "   Get: kubectl get secret rabbitmq-password -n core -o jsonpath='{.data.rabbitmq-password}' | base64 -d"
-            )
-            return False
-
-        credentials = pika.PlainCredentials(username, password)
-        connection = pika.BlockingConnection(
-            pika.ConnectionParameters("localhost", 5672, credentials=credentials)
-        )
-        channel = connection.channel()
-
-        exchange_name = "geozarr-staging"
-        routing_key = "eopf.items.test"
-
-        channel.exchange_declare(exchange=exchange_name, exchange_type="topic", durable=True)
-        channel.basic_publish(
-            exchange=exchange_name,
-            routing_key=routing_key,
-            body=json.dumps(payload),
-            properties=pika.BasicProperties(delivery_mode=2, content_type="application/json"),
-        )
-
-        print(f"✅ Published: {payload['source_url'][:80]}...")
-        connection.close()
-        return True
-
-    except Exception as e:
-        print(f"❌ Failed: {e}")
-        import traceback
-
-        traceback.print_exc()
-        return False
-
-
-if __name__ == "__main__":
-    # ✅ Use STAC item URL (pipeline extracts zarr URL from assets)
-    # ❌ NOT direct zarr URL
-    item_id = "S2A_MSIL2A_20251022T094121_N0511_R036_T34TDT_20251022T114817"
-    payload = {
-        "source_url": f"https://stac.core.eopf.eodc.eu/collections/sentinel-2-l2a/items/{item_id}",
-        "item_id": item_id,
-        "collection": "sentinel-2-l2a-dp-test",
-    }
-
-    print("🚀 Submitting workflow via RabbitMQ")
-    print(f"   Collection: {payload['collection']}")
-    print(f"   Source: {payload['source_url']}")
-    print()
-    print("Prerequisites:")
-    print("  kubectl port-forward -n devseed-staging svc/rabbitmq 5672:5672 &")
-    print(
-        "  export RABBITMQ_PASSWORD=$(kubectl get secret rabbitmq-password -n core -o jsonpath='{.data.rabbitmq-password}' | base64 -d)"
-    )
-    print()
-
-    if submit_workflow(payload):
-        print("✅ Monitor: kubectl get wf -n devseed-staging --watch")
-        sys.exit(0)
-    else:
-        sys.exit(1)
+# Test item that was failing (same as before)
+payload = {
+    "source_url": "https://stac.core.eopf.eodc.eu/collections/sentinel-2-l2a/items/S2A_MSIL2A_20251023T105131_N0511_R051_T31UET_20251023T122522",
+    "item_id": "S2A_MSIL2A_20251023T105131_N0511_R051_T31UET_20251023T122522",
+    "collection": "sentinel-2-l2a-dp-test",
+}
+
+credentials = pika.PlainCredentials("user", os.getenv("RABBITMQ_PASSWORD"))
+connection = pika.BlockingConnection(pika.ConnectionParameters("localhost", 5672, "/", credentials))
+channel = connection.channel()
+
+message = json.dumps(payload)
+channel.basic_publish(
+    exchange="geozarr-events",
+    routing_key="geozarr.convert",
+    body=message,
+    properties=pika.BasicProperties(content_type="application/json"),
+)
+
+print(f"✅ Published workflow for item: {payload['item_id']}")
+connection.close()

workflows/base/workflowtemplate.yaml

@@ -31,7 +31,7 @@ spec:
     - name: s3_output_prefix
       value: tests-output
     - name: pipeline_image_version
-      value: fix-unit-tests
+      value: slim
   templates:
   - name: main
     dag:
@@ -58,6 +58,7 @@ spec:
       - "{{workflow.parameters.s3_output_bucket}}"
       - --s3-output-prefix
       - "{{workflow.parameters.s3_output_prefix}}"
+      - --verbose
       resources:
         requests:
           memory: 4Gi
@@ -104,6 +105,7 @@ spec:
       - "{{workflow.parameters.s3_output_bucket}}"
       - --s3-output-prefix
       - "{{workflow.parameters.s3_output_prefix}}"
+      - --verbose
       ports:
       - containerPort: 8000
         name: metrics