service_netatmo/netatmo_service/data.py

from __future__ import annotations

from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True)
class Sample:
    timestamp: int
    values: dict[str, float | None]


def _dashboard(item: dict[str, Any]) -> dict[str, Any]:
    return item.get("dashboard_data") or {}


def extract_samples(payload: dict, names: dict[str, str]) -> dict[str, Sample]:
    """Translate a getstationsdata response into the service's RRD samples."""
    devices = payload.get("body", {}).get("devices", [])
    if not devices:
        raise ValueError("Netatmo response contains no weather station")

    device = devices[0]
    items = [device, *device.get("modules", [])]
    by_name = {item.get("module_name"): item for item in items}
    result: dict[str, Sample] = {}

    def sample(key: str, fields: dict[str, str]) -> None:
        item = by_name.get(names[key])
        if not item or not _dashboard(item):
            return
        data = _dashboard(item)
        result[key] = Sample(
            int(data["time_utc"]),
            {target: _number(data.get(source)) for target, source in fields.items()},
        )

    sample("bedroom", {"temperature": "Temperature", "co2": "CO2", "humidity": "Humidity"})
    sample("study", {"temperature": "Temperature", "co2": "CO2", "humidity": "Humidity"})
    sample("living", {"temperature": "Temperature", "co2": "CO2", "humidity": "Humidity"})
    sample("wind", {"gust": "GustStrength", "average": "WindStrength", "angle": "WindAngle"})
    # Netatmo resets this cumulative millimetre total at local midnight. Store
    # it unchanged so historical graphs show the total accumulated each day.
    sample("rain", {"sum_rain_24": "sum_rain_24"})

    # Pressure belongs to the main indoor station rather than a separate
    # module, so give it its own sample and use that station's timestamp.
    station_data = _dashboard(device)
    if station_data:
        result["pressure"] = Sample(
            int(station_data["time_utc"]),
            {"pressure": _number(station_data.get("Pressure"))},
        )

    outdoor = by_name.get(names["outdoor"])
    if outdoor and _dashboard(outdoor):
        data = _dashboard(outdoor)
        # Pressure is measured by the main indoor station; associate it with the
        # outdoor/weather graph while preserving the outdoor module timestamp.
        pressure = _dashboard(device).get("Pressure")
        result["outdoor"] = Sample(
            int(data["time_utc"]),
            {
                "min_temp": _number(data.get("min_temp", data.get("Temperature"))),
                "max_temp": _number(data.get("max_temp", data.get("Temperature"))),
                "humidity": _number(data.get("Humidity")),
                "pressure": _number(pressure),
            },
        )
    return result


def _number(value: Any) -> float | None:
    if value is None:
        return None
    return float(value)