Add metrics_analysis_script/src/metrics_analysis_script/cli.py

metrics_analysis_script/src/metrics_analysis_script/cli.py

@@ -0,0 +1,72 @@
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+import pandas as pd  # noqa: F401  # imported per dependencies spec
+import numpy as np  # noqa: F401
+import scipy  # noqa: F401
+
+from metrics_analysis_script import main
+
+
+def _read_input_json(input_path: Path) -> list[dict[str, Any]]:
+    """Load and validate metric input JSON data."""
+    if not input_path.exists():
+        raise FileNotFoundError(f"Input file not found: {input_path}")
+    with input_path.open("r", encoding="utf-8") as f:
+        try:
+            data = json.load(f)
+        except json.JSONDecodeError as e:
+            raise ValueError(f"Invalid JSON format in {input_path}: {e}") from e
+
+    if not isinstance(data, list):
+        raise ValueError("Expected top-level JSON list of metric objects.")
+
+    # Basic validation as per MetricResultsInput model
+    for i, entry in enumerate(data):
+        if not isinstance(entry, dict):
+            raise ValueError(f"Entry {i} is not a JSON object.")
+        for field in ("metric_name", "values", "pinned_flag"):
+            if field not in entry:
+                raise ValueError(f"Missing field '{field}' in entry {i}.")
+        if not isinstance(entry["metric_name"], str):
+            raise TypeError(f"metric_name must be str in entry {i}.")
+        if not isinstance(entry["values"], list) or not all(isinstance(v, (int, float)) for v in entry["values"]):
+            raise TypeError(f"values must be list[float] in entry {i}.")
+        if not isinstance(entry["pinned_flag"], bool):
+            raise TypeError(f"pinned_flag must be bool in entry {i}.")
+
+    return data
+
+
+def _write_output_json(output_path: Path, results: Any) -> None:
+    """Write analysis results to JSON file with integrity checks."""
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    with output_path.open("w", encoding="utf-8") as f:
+        json.dump(results, f, indent=2)
+
+
+def main_cli() -> None:
+    """Command-line interface entrypoint for metrics analysis."""
+    parser = argparse.ArgumentParser(description="Analyse Frozen Runs Metrics with Bootstrap CI")
+    parser.add_argument("--input", required=True, help="Path to input JSON metrics data")
+    parser.add_argument("--output", required=True, help="Path to output JSON results")
+    args = parser.parse_args()
+
+    input_path = Path(args.input)
+    output_path = Path(args.output)
+
+    try:
+        data = _read_input_json(input_path)
+        results = main.analyze_metrics(data)
+        assert isinstance(results, list), "Expected analyze_metrics to return a list of MetricResults"
+        _write_output_json(output_path, results)
+    except Exception as exc:  # pragma: no cover
+        sys.stderr.write(f"Error: {exc}\n")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main_cli()