feat(cli): output prefix expansion, group validation, optional run-metadata writing

Author: wietzesuijker

src/eopf_geozarr/cli.py

@@ -11,6 +11,7 @@
 from pathlib import Path
 from time import perf_counter
 from typing import Any, Optional
+from urllib.parse import urlparse
 
 import xarray as xr
 
@@ -175,6 +176,44 @@ def _summarize_dask_metrics(
     return base
 
 
+def _derive_item_id_from_input_path(path: str) -> str:
+    """Best-effort item id from local path or URL."""
+    try:
+        if path.startswith(("http://", "https://", "s3://", "gs://")):
+            parsed = urlparse(path)
+            name = Path(parsed.path.rstrip("/")).name
+        else:
+            name = Path(str(path).rstrip("/")).name
+        if name.endswith(".zarr"):
+            name = name[: -len(".zarr")]
+        return name or "dataset"
+    except Exception:
+        return "dataset"
+
+
+def _resolve_output_prefix(prefix: str, item_id: str) -> str:
+    target_name = f"{item_id}_geozarr.zarr"
+    if prefix.startswith(("http://", "https://", "s3://", "gs://")):
+        if not prefix.endswith("/"):
+            prefix = prefix + "/"
+        return prefix + target_name
+    return str(Path(prefix) / target_name)
+
+
+def _has_group(tree: xr.DataTree, path: str) -> bool:
+    try:
+        parts = [p for p in str(path).strip("/").split("/") if p]
+        node = tree
+        for seg in parts:
+            children = getattr(node, "children", {}) or {}
+            if seg not in children:
+                return False
+            node = children[seg]
+        return True
+    except Exception:
+        return False
+
+
 def convert_command(args: argparse.Namespace) -> None:
     """Convert EOPF dataset to GeoZarr compliant format.
 
@@ -215,6 +254,11 @@ def convert_command(args: argparse.Namespace) -> None:
 
         # Handle output path validation
         output_path_str = args.output_path
+        # Expand trailing-slash prefix to a concrete store
+        if output_path_str.endswith("/"):
+            item_id = _derive_item_id_from_input_path(str(input_path))
+            output_path_str = _resolve_output_prefix(output_path_str, item_id)
+            print(f"Resolved output store: {output_path_str}")
         if is_s3_path(output_path_str):
             # S3 path - validate S3 access
             print("🔍 Validating S3 access...")
@@ -268,14 +312,48 @@ def convert_command(args: argparse.Namespace) -> None:
         # Load the EOPF DataTree with appropriate storage options
         print("Loading EOPF dataset...")
         storage_options = get_storage_options(input_path)
-        # Metrics setup
+        # Metrics setup (environment first; set_input after group validation)
         if getattr(args, "metrics", True):
             metrics = MetricsRecorder()
             metrics.set_environment()
+
+        with metrics.time_step("open_input") if metrics else nullcontext():
+            dt = xr.open_datatree(
+                str(input_path),
+                engine="zarr",
+                chunks="auto",
+                storage_options=storage_options,
+            )
+
+        # Validate/prune groups if requested
+        groups_effective = list(getattr(args, "groups", []) or [])
+        validate_mode = getattr(args, "validate_groups", None)
+        missing: list[str] = []
+        if validate_mode in {"warn", "error"} and groups_effective:
+            existing: list[str] = []
+            for g in groups_effective:
+                if _has_group(dt, g):
+                    existing.append(g)
+                else:
+                    missing.append(g)
+            if missing:
+                msg = f"Groups not found: {', '.join(missing)}"
+                if validate_mode == "error":
+                    print(f"❌ {msg}")
+                    sys.exit(3)
+                else:
+                    print(f"⚠️  {msg}; proceeding with remaining groups")
+                groups_effective = existing
+            if not groups_effective:
+                print("❌ No valid groups to convert after validation")
+                sys.exit(3)
+
+        # Now that groups are finalized, set input metadata for metrics
+        if metrics is not None:
             metrics.set_input(
                 source_uri=str(input_path),
                 profile=None,
-                groups=args.groups,
+                groups=groups_effective,
                 dask={
                     "enabled": bool(dask_client is not None),
                     "mode": getattr(args, "dask_mode", None),
@@ -286,19 +364,13 @@ def convert_command(args: argparse.Namespace) -> None:
                 },
             )
 
-        with metrics.time_step("open_input") if metrics else nullcontext():
-            dt = xr.open_datatree(
-                str(input_path),
-                engine="zarr",
-                chunks="auto",
-                storage_options=storage_options,
-            )
-
         if args.verbose:
             print(f"Loaded DataTree with {len(dt.children)} groups")
             print("Available groups:")
             for group_name in dt.children:
                 print(f"  - {group_name}")
+            if missing:
+                print(f"After validation, converting groups: {groups_effective}")
 
         # Convert to GeoZarr compliant format
         print("Converting to GeoZarr compliant format...")
@@ -330,7 +402,7 @@ def convert_command(args: argparse.Namespace) -> None:
                     with metrics.time_step("convert") if metrics else nullcontext():
                         dt_geozarr = create_geozarr_dataset(
                             dt_input=dt,
-                            groups=args.groups,
+                            groups=groups_effective,
                             output_path=output_path,
                             spatial_chunk=args.spatial_chunk,
                             min_dimension=args.min_dimension,
@@ -344,7 +416,7 @@ def convert_command(args: argparse.Namespace) -> None:
                 with metrics.time_step("convert") if metrics else nullcontext():
                     dt_geozarr = create_geozarr_dataset(
                         dt_input=dt,
-                        groups=args.groups,
+                        groups=groups_effective,
                         output_path=output_path,
                         spatial_chunk=args.spatial_chunk,
                         min_dimension=args.min_dimension,
@@ -381,7 +453,7 @@ def convert_command(args: argparse.Namespace) -> None:
                     ),
                     "perf_report": getattr(args, "dask_perf_html", None),
                     "wall_clock_s": wall_clock if dask_client is not None else None,
-                    "groups": args.groups,
+                    "groups": groups_effective,
                     "spatial_chunk": args.spatial_chunk,
                     "min_dimension": args.min_dimension,
                     "tile_width": args.tile_width,
@@ -399,6 +471,17 @@ def convert_command(args: argparse.Namespace) -> None:
                 MetricsRecorder.write_json(run_summary, payload)
                 if args.verbose:
                     print(f"🧾 Wrote metrics: {run_summary}")
+                # Optional external run-metadata path
+                if getattr(args, "run_metadata", None):
+                    try:
+                        outp = Path(args.run_metadata)
+                        outp.parent.mkdir(parents=True, exist_ok=True)
+                        MetricsRecorder.write_json(outp, payload)
+                        if args.verbose:
+                            print(f"🧾 Wrote run metadata: {outp}")
+                    except Exception as _e:
+                        if args.verbose:
+                            print(f"(debug) could not write run-metadata: {_e}")
         except Exception as _exc:
             if args.verbose:
                 print(f"(debug) could not write run summary: {_exc}")
@@ -424,10 +507,22 @@ def convert_command(args: argparse.Namespace) -> None:
         try:
             if metrics and debug_dir is not None and getattr(args, "metrics", True):
                 payload = metrics.finalize(status="error", exception=error_msg)
-                run_summary = Path(args.output_path) / "debug" / "run_summary.json"
-                MetricsRecorder.write_json(run_summary, payload)
-                if args.verbose:
-                    print(f"🧾 Wrote failure metrics: {run_summary}")
+                run_summary = Path(output_path) / "debug" / "run_summary.json"
+                try:
+                    MetricsRecorder.write_json(run_summary, payload)
+                    if args.verbose:
+                        print(f"🧾 Wrote failure metrics: {run_summary}")
+                except Exception:
+                    pass
+                if getattr(args, "run_metadata", None):
+                    try:
+                        outp = Path(args.run_metadata)
+                        outp.parent.mkdir(parents=True, exist_ok=True)
+                        MetricsRecorder.write_json(outp, payload)
+                        if args.verbose:
+                            print(f"🧾 Wrote run metadata (error): {outp}")
+                    except Exception:
+                        pass
         except Exception:
             pass
         sys.exit(1)
@@ -1414,6 +1509,17 @@ def create_parser() -> argparse.ArgumentParser:
         action="store_false",
         help="Disable metrics emission.",
     )
+    convert_parser.add_argument(
+        "--validate-groups",
+        type=str,
+        choices=["warn", "error"],
+        help="Validate requested groups against input; warn to prune, error to abort.",
+    )
+    convert_parser.add_argument(
+        "--run-metadata",
+        type=str,
+        help="Also write finalized metrics payload to this JSON path.",
+    )
     convert_parser.set_defaults(func=convert_command)
 
     # Info command