diff --git a/metrics_analysis_script/src/metrics_analysis_script/cli.py b/metrics_analysis_script/src/metrics_analysis_script/cli.py new file mode 100644 index 0000000..653cb99 --- /dev/null +++ b/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()