Add report_generator/src/report_generator/core.py

report_generator/src/report_generator/core.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+import json
+from dataclasses import dataclass, asdict
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+import os
+
+
+@dataclass
+class DriftMetrics:
+    warn_count: int
+    fail_count: int
+    total_runs: int
+    window_size: int
+
+    def validate(self) -> None:
+        assert isinstance(self.warn_count, int) and self.warn_count >= 0, "warn_count muss eine nicht-negative Ganzzahl sein."
+        assert isinstance(self.fail_count, int) and self.fail_count >= 0, "fail_count muss eine nicht-negative Ganzzahl sein."
+        assert isinstance(self.total_runs, int) and self.total_runs > 0, "total_runs muss eine positive Ganzzahl sein."
+        assert isinstance(self.window_size, int) and self.window_size > 0, "window_size muss eine positive Ganzzahl sein."
+        assert self.warn_count <= self.total_runs, "warn_count darf total_runs nicht überschreiten."
+
+
+@dataclass
+class DriftReport:
+    warn_rate: float
+    threshold: float
+    alert: bool
+    timestamp: str
+
+
+class InputValidationError(Exception):
+    """Wird ausgelöst, wenn Eingangsparameter ungültig sind."""
+    pass
+
+
+def generate_report(warn_count: int, total_runs: int, threshold: float) -> str:
+    """Berechnet die Warnrate und erstellt einen Bericht über WARN-/FAIL-Raten in Drift-Analysen.
+
+    Args:
+        warn_count (int): Anzahl der WARN-Ergebnisse im Rolling-Window.
+        total_runs (int): Gesamtanzahl der Runs im betrachteten Rolling-Window.
+        threshold (float): Schwellenwert, ab welcher WARN-Rate ein Alarm ausgelöst wird.
+
+    Returns:
+        str: Pfad zur generierten Berichtdatei im JSON-Format.
+    """
+
+    # Input Validation
+    if not isinstance(warn_count, int) or not isinstance(total_runs, int) or not isinstance(threshold, (int, float)):
+        raise InputValidationError("Ungültige Typen für Eingabeparameter.")
+    if total_runs <= 0:
+        raise InputValidationError("total_runs muss größer als 0 sein.")
+    if warn_count < 0 or warn_count > total_runs:
+        raise InputValidationError("warn_count darf nicht negativ sein und total_runs nicht überschreiten.")
+    if not (0 <= threshold <= 1):
+        raise InputValidationError("threshold muss zwischen 0 und 1 liegen.")
+
+    # Berechnung
+    warn_rate = warn_count / total_runs if total_runs > 0 else 0.0
+    alert = warn_rate > threshold
+
+    # Bericht erstellen
+    report = DriftReport(
+        warn_rate=float(round(warn_rate, 4)),
+        threshold=float(threshold),
+        alert=alert,
+        timestamp=datetime.now().isoformat(),
+    )
+
+    # Zielpfad
+    output_path = Path("output/drift_report.json")
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    # Schreiben im JSON-Format
+    with output_path.open("w", encoding="utf-8") as f:
+        json.dump(asdict(report), f, indent=2, ensure_ascii=False)
+
+    return str(output_path.resolve())