Add data_analysis/src/data_analysis/io_utils.py

data_analysis/src/data_analysis/io_utils.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any
+
+import pandas as pd
+import numpy as np
+
+__all__ = ["LogEntry"]
+
+logger = logging.getLogger(__name__)
+
+
+class LogEntryError(ValueError):
+    """Custom exception for LogEntry validation errors."""
+
+
+@dataclass(slots=True)
+class LogEntry:
+    """Repräsentiert einen einzelnen Datensatz einer Rover-Messung.
+
+    Attribute:
+        timestamp (datetime): Zeitstempel der Messung.
+        luminosity (int): Lichtintensität in Lux.
+        sound_level (float): Geräuschpegel in Dezibel A.
+        temperature (float): Temperatur in Grad Celsius.
+        inference (float): Inferenzergebnis (Wahrscheinlichkeit eines Ereignisses).
+    """
+
+    timestamp: datetime
+    luminosity: int
+    sound_level: float
+    temperature: float
+    inference: float
+    raw: dict[str, Any] = field(default_factory=dict, repr=False)
+
+    def __init__(
+        self,
+        timestamp: str | datetime,
+        luminosity: int | float,
+        sound_level: float,
+        temperature: float,
+        inference: float,
+    ) -> None:
+        try:
+            if isinstance(timestamp, str):
+                self.timestamp = datetime.fromisoformat(timestamp)
+            elif isinstance(timestamp, datetime):
+                self.timestamp = timestamp
+            else:
+                raise TypeError("timestamp must be str or datetime")
+
+            self.luminosity = int(luminosity)
+            self.sound_level = float(sound_level)
+            self.temperature = float(temperature)
+            self.inference = float(inference)
+            self.raw = {
+                "t": self.timestamp.isoformat(),
+                "Lx": self.luminosity,
+                "dB": self.sound_level,
+                "Temp": self.temperature,
+                "Inference": self.inference,
+            }
+        except Exception as e:
+            logger.error("Error initializing LogEntry: %s", e)
+            raise LogEntryError(str(e)) from e
+
+    @classmethod
+    def from_json_record(cls, record: dict[str, Any]) -> LogEntry:
+        """Erzeugt eine LogEntry-Instanz aus einem JSON-Datensatz und validiert Felder."""
+        required = {"t", "Lx", "dB", "Temp", "Inference"}
+
+        missing = required - set(record.keys())
+        if missing:
+            raise LogEntryError(f"Missing fields: {', '.join(sorted(missing))}")
+
+        return cls(
+            timestamp=record["t"],
+            luminosity=record["Lx"],
+            sound_level=record["dB"],
+            temperature=record["Temp"],
+            inference=record["Inference"],
+        )
+
+    def to_dict(self) -> dict[str, Any]:
+        """Konvertiert den LogEntry zurück in ein Dictionary."""
+        return self.raw.copy()
+
+    def __post_init__(self) -> None:  # pragma: no cover
+        # Redundant safeguard if dataclass init is auto-used
+        assert isinstance(self.timestamp, datetime)
+        assert isinstance(self.luminosity, int)
+        assert isinstance(self.sound_level, float)
+        assert isinstance(self.temperature, float)
+        assert isinstance(self.inference, float)
+
+    def __repr__(self) -> str:
+        return (
+            f"LogEntry(time={self.timestamp.isoformat()}, Lx={self.luminosity}, "
+            f"dB={self.sound_level:.1f}, Temp={self.temperature:.1f}, Inference={self.inference:.2f})"
+        )