diff --git a/data_analysis/src/data_analysis/io_utils.py b/data_analysis/src/data_analysis/io_utils.py new file mode 100644 index 0000000..941fa4b --- /dev/null +++ b/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})" + ) \ No newline at end of file