Add results_analysis/src/results_analysis/cli.py

results_analysis/src/results_analysis/cli.py

@@ -0,0 +1,76 @@
+import argparse
+import json
+from pathlib import Path
+import pandas as pd
+from typing import Any
+
+from results_analysis import core
+
+
+class CLIError(Exception):
+    """Custom exception for CLI-related errors."""
+
+
+def _read_csv(path: Path) -> list[dict[str, Any]]:
+    if not path.exists() or not path.is_file():
+        raise CLIError(f"Eingabedatei nicht gefunden: {path}")
+    try:
+        df = pd.read_csv(path)
+    except Exception as e:
+        raise CLIError(f"Fehler beim Lesen der CSV-Datei {path}: {e}") from e
+
+    expected_columns = {"run_id", "decision", "unknown_rate", "warn_rate"}
+    if not expected_columns.issubset(set(df.columns)):
+        missing = expected_columns - set(df.columns)
+        raise CLIError(f"Fehlende Spalten in {path}: {', '.join(missing)}")
+
+    return df.to_dict(orient="records")
+
+
+def _write_json(data: Any, path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8") as fh:
+        json.dump(data, fh, indent=2, ensure_ascii=False)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Analyse-CLI für Policy v1.1 Ergebnisse"
+    )
+    parser.add_argument(
+        "--previous",
+        required=True,
+        help="Pfad zu den früheren Ergebnissen (CSV)",
+    )
+    parser.add_argument(
+        "--new", required=True, help="Pfad zu den neuen Ergebnissen (CSV)"
+    )
+    parser.add_argument(
+        "--out", default="output/", help="Zielverzeichnis für Ausgabe (Standard: output/)"
+    )
+
+    args = parser.parse_args()
+
+    prev_path = Path(args.previous)
+    new_path = Path(args.new)
+    out_dir = Path(args.out)
+
+    try:
+        previous_results = _read_csv(prev_path)
+        new_results = _read_csv(new_path)
+
+        matrix_dict = core.create_confusion_matrix(previous_results, new_results)
+        deltas_list = core.get_deltas(previous_results, new_results)
+
+        _write_json(matrix_dict, out_dir / "confusion_matrix.json")
+        _write_json(deltas_list, out_dir / "deltas.json")
+
+        print(f"Analyse abgeschlossen. Ergebnisse gespeichert in: {out_dir.resolve()}")
+    except CLIError as cli_err:
+        print(f"Fehler: {cli_err}")
+    except Exception as err:
+        print(f"Unerwarteter Fehler: {err}")
+
+
+if __name__ == "__main__":
+    main()