Add debug_artifact/src/debug_artifact/core.py

debug_artifact/src/debug_artifact/core.py

@@ -0,0 +1,80 @@
+import json
+import argparse
+import os
+from pathlib import Path
+from dataclasses import dataclass, asdict
+from typing import List, Dict, Any
+
+
+@dataclass
+class DebugInfo:
+    """Repräsentiert die Debug-Information über das schlimmste Mischfenster."""
+    worst_mischfenster: float
+    corr_id: str
+
+
+def create_debug_artifact(raw_events: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """Erstellt ein Debug-JSON-Artefakt basierend auf Roh-Events unter Berücksichtigung von p95-basierten Worst-Case-Mischfenstern.
+
+    Args:
+        raw_events: Liste von Roh-Event-Daten, die Metriken und Mischfensterinformationen enthalten.
+
+    Returns:
+        dict: Ein Dictionary mit Debug-Informationen, das Felder zu 'worst_mischfenster' und 'corr_id' enthält.
+    """
+    if not isinstance(raw_events, list):
+        raise TypeError("raw_events muss eine Liste sein")
+
+    if not raw_events:
+        return asdict(DebugInfo(worst_mischfenster=0.0, corr_id="none"))
+
+    # Validierung und Suche nach dem schlimmsten Mischfenster
+    valid_events = []
+    for event in raw_events:
+        if not isinstance(event, dict):
+            continue
+        value = event.get("mischfenster") or event.get("p95")
+        corr_id = event.get("corr_id") or event.get("id")
+        if isinstance(value, (int, float)) and isinstance(corr_id, str):
+            valid_events.append((float(value), corr_id))
+
+    if not valid_events:
+        return asdict(DebugInfo(worst_mischfenster=0.0, corr_id="none"))
+
+    worst = max(valid_events, key=lambda x: x[0])
+    debug_info = DebugInfo(worst_mischfenster=worst[0], corr_id=worst[1])
+
+    result = asdict(debug_info)
+    assert "worst_mischfenster" in result and "corr_id" in result, "Fehlende Debug-Info-Felder"
+
+    return result
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Erzeugt ein Debug-Artefakt aus Roh-Event-Daten.")
+    parser.add_argument("--input", required=True, help="Pfad zur Eingabedatei mit Roh-Events im JSON-Format.")
+    parser.add_argument("--output", required=True, help="Zielpfad für das erzeugte Debug-Artefakt.")
+    return parser.parse_args()
+
+
+def _main() -> None:
+    args = _parse_args()
+
+    input_path = Path(args.input)
+    output_path = Path(args.output)
+
+    if not input_path.exists():
+        raise FileNotFoundError(f"Eingabedatei nicht gefunden: {input_path}")
+
+    with input_path.open("r", encoding="utf-8") as f:
+        raw_events = json.load(f)
+
+    debug_info = create_debug_artifact(raw_events)
+
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    with output_path.open("w", encoding="utf-8") as f:
+        json.dump(debug_info, f, indent=2, sort_keys=True)
+
+
+if __name__ == "__main__":
+    _main()