Add exit_metrics_logging/src/exit_metrics_logging/core.py

exit_metrics_logging/src/exit_metrics_logging/core.py

@@ -0,0 +1,59 @@
+import json
+import os
+from pathlib import Path
+from datetime import datetime
+from typing import Any, Dict
+
+
+class RunMetrics:
+    """Repräsentiert eine Sammlung von Metriken für einen Run."""
+
+    def __init__(self, run_id: str, warn_rate: float, unknown_rate: float, delta_t: float) -> None:
+        if not isinstance(run_id, str) or not run_id:
+            raise ValueError("run_id muss ein nicht-leerer String sein.")
+        for name, value in ("warn_rate", warn_rate), ("unknown_rate", unknown_rate), ("delta_t", delta_t):
+            if not isinstance(value, (int, float)):
+                raise TypeError(f"{name} muss numerisch (float oder int) sein.")
+            if value < 0:
+                raise ValueError(f"{name} darf nicht negativ sein.")
+
+        self.run_id: str = run_id
+        self.warn_rate: float = float(warn_rate)
+        self.unknown_rate: float = float(unknown_rate)
+        self.delta_t: float = float(delta_t)
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Konvertiert die RunMetrics-Instanz in ein Dictionary zur JSON-Speicherung."""
+        return {
+            "run_id": self.run_id,
+            "warn_rate": self.warn_rate,
+            "unknown_rate": self.unknown_rate,
+            "delta_t": self.delta_t,
+            "timestamp": datetime.utcnow().isoformat() + "Z",
+        }
+
+
+def log_metrics(run_id: str, warn_rate: float, unknown_rate: float, delta_t: float) -> None:
+    """Protokolliert Exit-Metriken eines einzelnen Runs in eine JSON-Datei."""
+
+    metrics = RunMetrics(run_id, warn_rate, unknown_rate, delta_t)
+    data = metrics.to_dict()
+
+    output_path = Path("output/exit_metrics.json")
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    existing_data = []
+    if output_path.exists():
+        try:
+            with output_path.open("r", encoding="utf-8") as f:
+                existing_data = json.load(f)
+            if not isinstance(existing_data, list):
+                existing_data = [existing_data]
+        except (json.JSONDecodeError, FileNotFoundError):
+            existing_data = []
+
+    existing_data.append(data)
+    with output_path.open("w", encoding="utf-8") as f:
+        json.dump(existing_data, f, indent=2, ensure_ascii=False)
+
+    assert output_path.exists(), "Fehler: Datei wurde nicht erstellt."