1st POC

netatmo_service/data.py

@@ -0,0 +1,65 @@
+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 five 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"})
+
+    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)
+