Add rerun_analysis_tool/src/rerun_analysis_tool/cli.py

rerun_analysis_tool/src/rerun_analysis_tool/cli.py

@@ -0,0 +1,78 @@
+import argparse
+import json
+import logging
+from pathlib import Path
+import pandas as pd
+from typing import Any
+
+from rerun_analysis_tool import core
+
+
+class CLIError(Exception):
+    """Custom exception for CLI related errors."""
+
+
+def configure_logging() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s [%(levelname)s] %(message)s",
+    )
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Analyse von Replay-Daten für Rerun-Entscheidungen.")
+    parser.add_argument("--input", required=True, help="Pfad zur Replay-Datendatei (CSV).")
+    parser.add_argument("--threshold", type=float, default=0.3, help="WARN-Schwelle (Standard: 0.3).")
+    parser.add_argument("--budget", type=int, default=1, help="Rerun-Budget (Standard: 1).")
+    return parser.parse_args()
+
+
+def load_runs_data(csv_path: Path) -> list[dict[str, Any]]:
+    if not csv_path.exists():
+        raise CLIError(f"Eingabedatei nicht gefunden: {csv_path}")
+    try:
+        df = pd.read_csv(csv_path)
+    except Exception as e:
+        raise CLIError(f"Fehler beim Lesen der CSV-Datei: {e}") from e
+
+    required_cols = {"run_id", "status", "unknown_rate", "rerun_helps", "rerun_shifts", "rerun_hurts"}
+    missing = required_cols - set(df.columns)
+    if missing:
+        raise CLIError(f"Fehlende Spalten in CSV: {missing}")
+
+    return df.to_dict(orient="records")
+
+
+def save_results(results: dict[str, Any], output_path: Path) -> None:
+    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, ensure_ascii=False)
+
+
+def main() -> None:
+    configure_logging()
+    args = parse_args()
+
+    csv_path = Path(args.input)
+    threshold = args.threshold
+    rerun_budget = args.budget
+
+    logging.info("Lade Replay-Daten aus %s", csv_path)
+    runs_data = load_runs_data(csv_path)
+
+    logging.info(
+        "Starte Analyse mit threshold=%.2f, rerun_budget=%d", threshold, rerun_budget
+    )
+    try:
+        results = core.analyze_runs(runs_data, threshold, rerun_budget)
+    except Exception as e:
+        logging.exception("Analysefehler: %s", e)
+        raise CLIError(f"Analysefehlgeschlagen: {e}") from e
+
+    output_path = Path("output/rerun_summary.json")
+    save_results(results, output_path)
+    logging.info("Analyse erfolgreich. Ergebnisse gespeichert unter %s", output_path)
+
+
+if __name__ == "__main__":
+    main()