Add data_logging/src/data_logging/core.py

data_logging/src/data_logging/core.py

@@ -0,0 +1,69 @@
+import json
+from dataclasses import dataclass, asdict
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+
+@dataclass
+class LogEntry:
+    """Datenmodell für eine Umweltdatenerfassung."""
+    timestamp: datetime
+    temperature: float
+    wind_speed: float
+
+    def to_serializable(self) -> dict[str, Any]:
+        return {
+            "timestamp": self.timestamp.isoformat(),
+            "temperature": self.temperature,
+            "wind_speed": self.wind_speed,
+        }
+
+
+def _validate_input(timestamp: datetime, temperature: float, wind_speed: float) -> None:
+    """Validiert Eingabedaten zur Sicherstellung der Datenintegrität."""
+    if not isinstance(timestamp, datetime):
+        raise TypeError("timestamp muss ein datetime-Objekt sein.")
+    if not isinstance(temperature, (float, int)):
+        raise TypeError("temperature muss vom Typ float sein.")
+    if not isinstance(wind_speed, (float, int)):
+        raise TypeError("wind_speed muss vom Typ float sein.")
+    # Einfache physikalische Plausibilitätsprüfung
+    assert -100.0 <= temperature <= 100.0, "Ungültiger Temperaturbereich"
+    assert 0.0 <= wind_speed <= 400.0, "Ungültiger Windgeschwindigkeitsbereich"
+
+
+def log_data(timestamp: datetime, temperature: float, wind_speed: float, output_path: str | Path = "output/environment_log.json") -> None:
+    """Protokolliert Zeitstempel, Temperatur und Windgeschwindigkeit in einer JSON-Datei.
+
+    Parameter:
+        timestamp (datetime): Zeitpunkt der Messung im ISO-Format
+        temperature (float): Gemessene Temperatur in Grad Celsius
+        wind_speed (float): Gemessene Windgeschwindigkeit in km/h
+        output_path (str | Path): Pfad zur Zieldatei
+    """
+    _validate_input(timestamp, temperature, wind_speed)
+    entry = LogEntry(timestamp=timestamp, temperature=float(temperature), wind_speed=float(wind_speed))
+
+    path = Path(output_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+
+    data: list[dict[str, Any]] = []
+
+    if path.exists():
+        try:
+            with path.open("r", encoding="utf-8") as f:
+                existing = json.load(f)
+                if isinstance(existing, list):
+                    data = existing
+        except (json.JSONDecodeError, OSError):
+            # Datei war leer oder korrupt → Überschreiben
+            data = []
+
+    data.append(entry.to_serializable())
+
+    with path.open("w", encoding="utf-8") as f:
+        json.dump(data, f, ensure_ascii=False, indent=2)
+
+    # CI-Check Assertion (integritätsprüfung)
+    assert path.exists() and path.stat().st_size > 0, "Logdatei wurde nicht erstellt oder ist leer"