Add hotspot_logging/src/hotspot_logging/core.py

hotspot_logging/src/hotspot_logging/core.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from pathlib import Path
+from dataclasses import dataclass, asdict
+from typing import Any
+
+
+@dataclass
+class LogEntry:
+    """Repräsentiert einen einzelnen Hotspot-Logeintrag."""
+    timestamp: str
+    hotspot_ratio: float
+
+
+def log_hotspot_ratio(time_window: float, near_expiry_jobs: int, total_jobs: int, output_path: str | Path = "output/hotspot_log.json") -> None:
+    """Berechnet den Anteil von near-expiry-unpinned-Jobs zu Gesamtjobs und schreibt ihn als JSON-Logeintrag."""
+
+    # Input-Validierung
+    assert isinstance(time_window, (float, int)), "time_window muss float oder int sein"
+    assert isinstance(near_expiry_jobs, int) and near_expiry_jobs >= 0, "near_expiry_jobs muss >= 0 sein"
+    assert isinstance(total_jobs, int) and total_jobs >= 0, "total_jobs muss >= 0 sein"
+
+    # Edge Case: Kein Job im Zeitfenster
+    ratio = 0.0 if total_jobs == 0 else near_expiry_jobs / total_jobs
+
+    entry = LogEntry(timestamp=datetime.utcnow().isoformat() + "Z", hotspot_ratio=ratio)
+
+    output_path = Path(output_path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    if output_path.exists():
+        try:
+            with output_path.open("r", encoding="utf-8") as f:
+                data = json.load(f)
+                if not isinstance(data, list):
+                    data = []
+        except (json.JSONDecodeError, OSError):
+            data = []
+    else:
+        data = []
+
+    data.append(asdict(entry))
+
+    with output_path.open("w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2)
+
+    # Kein Rückgabewert — nur Seiteneffekt (Logging)
+    return None