78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
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", "sum_rain_1": "sum_rain_1"})
|
|
|
|
# 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"]),
|
|
{
|
|
"temperature": _number(data.get("Temperature")),
|
|
"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)
|