Add event_data_capture/src/event_data_capture/core.py

event_data_capture/src/event_data_capture/core.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import List
+
+
+class EventData:
+    """Repräsentiert Eventmetriken eines Stratums."""
+
+    def __init__(self, warn_rate: float, unknown_rate: float) -> None:
+        if not isinstance(warn_rate, (int, float)):
+            raise TypeError("warn_rate muss vom Typ float oder int sein.")
+        if not isinstance(unknown_rate, (int, float)):
+            raise TypeError("unknown_rate muss vom Typ float oder int sein.")
+        if not (0.0 <= warn_rate <= 1.0):
+            raise ValueError("warn_rate muss zwischen 0.0 und 1.0 liegen.")
+        if not (0.0 <= unknown_rate <= 1.0):
+            raise ValueError("unknown_rate muss zwischen 0.0 und 1.0 liegen.")
+        self.warn_rate: float = float(warn_rate)
+        self.unknown_rate: float = float(unknown_rate)
+
+    def to_dict(self) -> dict:
+        return {"warn_rate": self.warn_rate, "unknown_rate": self.unknown_rate}
+
+    def __repr__(self) -> str:
+        return f"EventData(warn_rate={self.warn_rate}, unknown_rate={self.unknown_rate})"
+
+
+def capture_event_data(event_stratum: str) -> List[EventData]:
+    """Erfasst und analysiert Events innerhalb eines gegebenen Stratums.
+
+    Args:
+        event_stratum: Bezeichnung des Event-Stratums ('pinned' oder 'unpinned').
+
+    Returns:
+        Liste von EventData-Objekten mit warn_rate und unknown_rate pro Stratum.
+    """
+    if not isinstance(event_stratum, str):
+        raise TypeError("event_stratum muss ein String sein.")
+
+    data_path = Path("data/run6_events.json")
+    if not data_path.exists():
+        raise FileNotFoundError(f"Eingabedatei {data_path} wurde nicht gefunden.")
+
+    with data_path.open("r", encoding="utf-8") as f:
+        try:
+            content = json.load(f)
+        except json.JSONDecodeError as e:
+            raise ValueError(f"Fehler beim Laden der JSON-Datei: {e}") from e
+
+    if event_stratum not in content:
+        raise KeyError(f"Stratum '{event_stratum}' nicht in den Eingabedaten gefunden.")
+
+    records = content[event_stratum]
+    if not isinstance(records, list):
+        raise ValueError("Erwartet eine Liste von Ereignisdaten im angegebenen Stratum.")
+
+    result: List[EventData] = []
+    for rec in records:
+        if not isinstance(rec, dict):
+            raise ValueError("Jedes Ereignis muss ein JSON-Objekt sein.")
+
+        warn_rate = rec.get("warn_rate")
+        unknown_rate = rec.get("unknown_rate")
+        result.append(EventData(warn_rate=warn_rate, unknown_rate=unknown_rate))
+
+    return result