chore: remove unused type: ignore on zarr import

Author: wietzesuijker

docs/pipeline-integration-recommendations.md

@@ -0,0 +1,74 @@
+# Pipeline Integration Recommendations
+
+This document lists enhancements observed while integrating `eopf-geozarr` into an Argo-based batch conversion pipeline (data-model-pipeline). They are candidates for upstream inclusion or API refinement.
+
+## 1. Output Prefix Expansion
+**Current (pipeline)**: Wrapper detects `output_zarr` ending with `/` and appends `<item_id>_geozarr.zarr` derived from the input STAC/Zarr URL.
+**Recommendation**: Support `--output-prefix` OR accept a trailing slash on positional `output_path` and perform the expansion internally (emitting the resolved final path). Add a log line: `Resolved output store: s3://.../S2A_..._geozarr.zarr`.
+
+## 2. Group Existence Validation (Pre-flight)
+**Current (pipeline)**: `validate_params_groups.py` inspects filesystem structure (presence of `.zarray` / `.zgroup`).
+**Recommendation**: Native CLI flag `--validate-groups` to prune or fail fast when groups don’t exist. Modes:
+- `--validate-groups=warn` (default): drop missing, report.
+- `--validate-groups=error`: abort if any missing.
+Emit JSON or structured summary when `--verbose`.
+
+## 3. Profiles (WOZ Profiles)
+**Current (pipeline)**: External JSON profile expansion before calling CLI.
+**Recommendation**: Provide `--profile <name>` in CLI mapping to preset groups + chunk params. Add `eopf-geozarr profile list` / `profile show <name>` subcommands. Keep external mechanism as fallback.
+
+## 4. Compressor Handling
+**Current**: Template attempted to pass `--compressor`; CLI does not expose codec choice.
+**Recommendation**: If codec selection is desired, add `--compressor <name>` now (zstd, lz4, blosc) with validation; else document fixed default explicitly to avoid confusion.
+
+## 5. CRS Groups Convenience
+**Current**: `--crs-groups` optional list.
+**Recommendation**: Discover candidate groups automatically (search for geometry-like datasets) unless `--crs-groups` provided (override). Provide `eopf-geozarr info --crs-scan` to preview.
+
+## 6. Dask Cluster Ergonomics
+**Current**: `--dask-cluster` toggles local cluster with no feedback.
+**Recommendation**: Print cluster dashboard URL (if available) and add `--dask-workers N` for quick scaling.
+
+## 7. Structured Logging / Run Metadata
+**Current**: Plain prints; pipeline scrapes logs.
+**Recommendation**: Optional `--run-metadata <path.json>` to write machine-readable summary: inputs, resolved groups, timings, warnings. Eases automation and reproducibility.
+
+## 8. Validation Command Enhancements
+**Current**: `validate` skeleton present but incomplete.
+**Recommendation**:
+- Implement spec checks (multiscales, attributes, chunk shape policy).
+- Exit code non-zero on *hard* failures, zero with warnings for soft issues.
+- `--format json` for programmatic consumption.
+
+## 9. HTML Tree Generation
+**Current**: `_generate_html_output` scaffold incomplete.
+**Recommendation**: Finish implementation; integrate with `info --html-output`. Provide minimal inline CSS (already drafted) and optional `--open-browser` flag.
+
+## 10. Progress Reporting
+**Recommendation**: Emit periodic progress per group: `group=/measurements/r10m scale=2 written=...MB elapsed=...s` to assist monitoring in batch workflows.
+
+## 11. Retry / Resumability
+**Recommendation**: Add `--resume` to skip already existing multiscale levels if output store partially present.
+
+## 12. Exit Codes (Contract)
+Document exit code meanings:
+- 0 success
+- 2 validation (input) error
+- 3 group resolution failure
+- 4 conversion runtime error
+
+## 13. Environment Variable Overrides
+Allow `EOZ_DEFAULT_PROFILE`, `EOZ_OUTPUT_PREFIX` env vars as implicit defaults (still overridden by flags).
+
+## 14. Example Invocation Block in README
+Provide ready-to-copy examples for Sentinel-2 & Sentinel-1 including polarization groups when Phase 1 logic is public.
+
+---
+**Next Steps (Suggested Order)**
+1. Implement output prefix expansion (low risk, high UX win)
+2. Group validation flag (prevents silent empty writes)
+3. Finish validate + info HTML features
+4. Add structured run metadata output
+5. Introduce profiles subcommands then deprecate external expansion path over time
+
+Feedback welcome—pipeline experience will continue surfacing actionable deltas.

src/eopf_geozarr/cli.py

@@ -21,6 +21,7 @@
     is_s3_path,
     validate_s3_access,
 )
+from .metrics import MetricsRecorder
 
 
 def setup_dask_cluster(
@@ -196,6 +197,7 @@ def convert_command(args: argparse.Namespace) -> None:
     run_id = datetime.now().strftime("%Y%m%d-%H%M%S")
     status = "unknown"
     error_msg = None
+    metrics = None
 
     try:
         # Validate input path (handle both local paths and URLs)
@@ -266,12 +268,31 @@ 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)
-        dt = xr.open_datatree(
-            str(input_path),
-            engine="zarr",
-            chunks="auto",
-            storage_options=storage_options,
-        )
+        # Metrics setup
+        if getattr(args, "metrics", True):
+            metrics = MetricsRecorder()
+            metrics.set_environment()
+            metrics.set_input(
+                source_uri=str(input_path),
+                profile=None,
+                groups=args.groups,
+                dask={
+                    "enabled": bool(dask_client is not None),
+                    "mode": getattr(args, "dask_mode", None),
+                    "workers": getattr(args, "dask_workers", None),
+                    "threads_per_worker": getattr(
+                        args, "dask_threads_per_worker", 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")
@@ -306,6 +327,21 @@ def convert_command(args: argparse.Namespace) -> None:
                 from dask.distributed import get_task_stream
 
                 with get_task_stream(client=dask_client, plot=False) as ts:
+                    with metrics.time_step("convert") if metrics else nullcontext():
+                        dt_geozarr = create_geozarr_dataset(
+                            dt_input=dt,
+                            groups=args.groups,
+                            output_path=output_path,
+                            spatial_chunk=args.spatial_chunk,
+                            min_dimension=args.min_dimension,
+                            tile_width=args.tile_width,
+                            max_retries=args.max_retries,
+                            crs_groups=args.crs_groups,
+                        )
+                task_events = ts.data
+        else:
+            with perf_ctx:
+                with metrics.time_step("convert") if metrics else nullcontext():
                     dt_geozarr = create_geozarr_dataset(
                         dt_input=dt,
                         groups=args.groups,
@@ -316,19 +352,6 @@ def convert_command(args: argparse.Namespace) -> None:
                         max_retries=args.max_retries,
                         crs_groups=args.crs_groups,
                     )
-                task_events = ts.data
-        else:
-            with perf_ctx:
-                dt_geozarr = create_geozarr_dataset(
-                    dt_input=dt,
-                    groups=args.groups,
-                    output_path=output_path,
-                    spatial_chunk=args.spatial_chunk,
-                    min_dimension=args.min_dimension,
-                    tile_width=args.tile_width,
-                    max_retries=args.max_retries,
-                    crs_groups=args.crs_groups,
-                )
 
         wall_clock = perf_counter() - t0
         if (
@@ -366,6 +389,16 @@ def convert_command(args: argparse.Namespace) -> None:
                 summary_path.write_text(_dumps(summary, indent=2))
                 if args.verbose:
                     print(f"🧾 Wrote run summary: {summary_path}")
+            # New metrics: run_summary.json and chunk_report.json
+            if metrics and debug_dir is not None and getattr(args, "metrics", True):
+                # Build output summary best-effort
+                with metrics.time_step("summarize_output"):
+                    metrics.build_output_summary(output_path)
+                payload = metrics.finalize(status="ok")
+                run_summary = Path(output_path) / "debug" / "run_summary.json"
+                MetricsRecorder.write_json(run_summary, payload)
+                if args.verbose:
+                    print(f"🧾 Wrote metrics: {run_summary}")
         except Exception as _exc:
             if args.verbose:
                 print(f"(debug) could not write run summary: {_exc}")
@@ -387,6 +420,16 @@ def convert_command(args: argparse.Namespace) -> None:
             import traceback
 
             traceback.print_exc()
+        # Write failure metrics if possible
+        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}")
+        except Exception:
+            pass
         sys.exit(1)
     finally:
         # Clean up dask client if it was created
@@ -1359,6 +1402,18 @@ def create_parser() -> argparse.ArgumentParser:
         type=str,
         help="Write a Dask performance report HTML to this path (local only).",
     )
+    convert_parser.add_argument(
+        "--metrics",
+        action="store_true",
+        default=True,
+        help="Emit structured conversion metrics JSON (default: on). Use --no-metrics to disable.",
+    )
+    convert_parser.add_argument(
+        "--no-metrics",
+        dest="metrics",
+        action="store_false",
+        help="Disable metrics emission.",
+    )
     convert_parser.set_defaults(func=convert_command)
 
     # Info command