hermes
25b0432757
OAuth2 Password-Grant + Refresh-Token on-the-fly. Holt sich automatisch einen neuen Access-Token wenn der alte abläuft (3h Gültigkeit). Module: NAMain (Indoor), NAModule1 (Outdoor), NAModule2 (Wind), NAModule3 (Regen), NAModule4 (Extra Indoor). Slot-responsive: 1x1 — Outdoor-Temp prominent + Mini-Status 4x1 — Indoor | Outdoor | Wind/Regen kompakt (3 Spalten) 1x4 — Vertikale Liste aller Module 2x2 — Indoor-Card (mit CO2-Bar) + Outdoor-Card + Wind/Regen-Bereich 4x4 — Volle Ansicht mit Min/Max, Windrose, Timestamps Config-Optionen: client_id/client_secret/username/password (secrets via Admin-UI) station_filter (substring-match für Multi-Setup) show_indoor/outdoor/wind/rain/compass/secondary (bools) co2_thresholds (ppm, default ok@600,warn@1000,alert@1500) temp_unit (C/F), wind_unit (kmh/ms) Plus: - plugins/NETATMO.md: vollständige Doku (Setup, Optionen, Diagnose) - config.netatmo.example.json: copy-paste Beispiel-Config - layout.py: netatmo default-size = 4x4 (4 Sub-Cards brauchen Platz) - README.md: Plugin-Tabelle erweitert Verifiziert: /api/plugins.json listet netatmo, Auto-Loader erkennt es, Admin-UI zeigt das Config-Form.
883 lines
34 KiB
Python
883 lines
34 KiB
Python
"""Netatmo Weather Station Plugin - responsive.
|
||
|
||
Zeigt Live-Daten der heimischen Netatmo Wetterstation (Hauptmodul NAMain +
|
||
beliebige Zusatzmodule: NAModule1=Outdoor, NAModule2=Wind, NAModule3=Regen,
|
||
NAModule4=Indoor).
|
||
|
||
Auth: OAuth2 Password-Grant für Erst-Token, dann Refresh-Token-Rotation
|
||
on-the-fly. Access-Token wird 3h gültig sein, in-memory gecached und
|
||
automatisch erneuert (kein Disk-IO pro Render).
|
||
|
||
Konfiguration (in der Admin-UI bzw. config.json unter `plugin_configs.netatmo`):
|
||
client_id — App-Client-ID (https://dev.netatmo.com/apps)
|
||
client_secret — App-Client-Secret
|
||
username — Netatmo-Login (E-Mail)
|
||
password — Netatmo-Passwort
|
||
station_filter — Name der Station falls mehrere vorhanden (Default: erste)
|
||
show_indoor — Indoor-Modul anzeigen (bool, default True)
|
||
show_outdoor — Outdoor-Modul anzeigen (bool, default True)
|
||
show_wind — Wind-Modul anzeigen (bool, default True)
|
||
show_rain — Regen-Modul anzeigen (bool, default True)
|
||
show_compass — Windrose zeichnen (bool, default True)
|
||
co2_thresholds — CO2-Schwellen (Format "ok@600,warn@1000,alert@1500")
|
||
temp_unit — "C" oder "F" (Default "C", Netatmo liefert Celsius)
|
||
wind_unit — "kmh" oder "ms" (Default "kmh", Netatmo liefert km/h)
|
||
"""
|
||
from __future__ import annotations
|
||
import os
|
||
import sys
|
||
import json
|
||
import time
|
||
import math
|
||
import urllib.request
|
||
import urllib.error
|
||
import urllib.parse
|
||
from datetime import datetime, timezone
|
||
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||
from plugins.base import Widget, fetch_url, render_error_banner # noqa: E402
|
||
from palette import ( # noqa: E402
|
||
FG, BG, INFO, OK, WARN, ALERT, ORANGE, BLUE, RED, GREEN,
|
||
measure, fit_font, centered_text, hbar, parse_thresholds,
|
||
is_small, is_wide, is_tall,
|
||
)
|
||
|
||
|
||
# ============================================================================
|
||
# OAuth2 + API Client
|
||
# ============================================================================
|
||
TOKEN_URL = "https://api.netatmo.net/oauth2/token"
|
||
STATIONS_URL = "https://api.netatmo.net/api/getstationsdata"
|
||
|
||
# Module-Type → Friendly-Name
|
||
MODULE_TYPES = {
|
||
"NAMain": {"label": "Indoor", "icon": "🏠", "short": "in"},
|
||
"NAModule1": {"label": "Outdoor", "icon": "🌳", "short": "out"},
|
||
"NAModule2": {"label": "Wind", "icon": "💨", "short": "wind"},
|
||
"NAModule3": {"label": "Regen", "icon": "🌧", "short": "rain"},
|
||
"NAModule4": {"label": "Extra", "icon": "📍", "short": "extra"},
|
||
}
|
||
|
||
|
||
def _token_payload(creds: dict, grant: str = "password", **extra) -> bytes:
|
||
"""Body für /oauth2/token. grant: 'password' oder 'refresh_token'."""
|
||
base = {
|
||
"grant_type": grant,
|
||
"client_id": creds["client_id"],
|
||
"client_secret": creds["client_secret"],
|
||
"scope": "read_station",
|
||
}
|
||
base.update(extra)
|
||
return urllib.parse.urlencode(base).encode("utf-8")
|
||
|
||
|
||
def _post_form(url: str, body: bytes, timeout: int = 10) -> dict:
|
||
req = urllib.request.Request(
|
||
url, data=body,
|
||
headers={"Content-Type": "application/x-www-form-urlencoded;charset=UTF-8"},
|
||
)
|
||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||
return json.loads(r.read())
|
||
|
||
|
||
# In-Memory Token-Cache: pro Prozess ein Access-Token.
|
||
# Kein Disk-IO bei jedem Render. Access-Token ist 3h gültig (10800s),
|
||
# wir holen einen frischen, wenn weniger als 5 Min Restlaufzeit.
|
||
_TOKEN_CACHE: dict = {
|
||
"access_token": None,
|
||
"refresh_token": None,
|
||
"expires_at": 0.0,
|
||
}
|
||
|
||
|
||
def _obtain_tokens(creds: dict) -> dict:
|
||
"""Holt einen frischen Access-Token via Password- oder Refresh-Grant.
|
||
|
||
Beim ersten Aufruf wird Password-Grant verwendet (initial token issuance).
|
||
Wenn ein refresh_token gecached ist, wird der Refresh-Grant bevorzugt
|
||
(saves rate limit).
|
||
"""
|
||
cached_refresh = _TOKEN_CACHE.get("refresh_token")
|
||
if cached_refresh:
|
||
try:
|
||
tok = _post_form(TOKEN_URL,
|
||
_token_payload(creds, "refresh_token",
|
||
refresh_token=cached_refresh))
|
||
except urllib.error.HTTPError as e:
|
||
# Refresh-Token evtl. revoked → fallback auf password
|
||
if e.code != 400:
|
||
raise
|
||
tok = None
|
||
if tok:
|
||
_TOKEN_CACHE["access_token"] = tok["access_token"]
|
||
_TOKEN_CACHE["refresh_token"] = tok.get("refresh_token", cached_refresh)
|
||
_TOKEN_CACHE["expires_at"] = time.time() + tok.get("expires_in", 10800) - 300
|
||
return tok
|
||
|
||
# Password-Grant (initial)
|
||
tok = _post_form(TOKEN_URL,
|
||
_token_payload(creds, "password",
|
||
username=creds["username"],
|
||
password=creds["password"]))
|
||
_TOKEN_CACHE["access_token"] = tok["access_token"]
|
||
_TOKEN_CACHE["refresh_token"] = tok.get("refresh_token")
|
||
_TOKEN_CACHE["expires_at"] = time.time() + tok.get("expires_in", 10800) - 300
|
||
return tok
|
||
|
||
|
||
def _get_stations_data(client_id: str, client_secret: str,
|
||
username: str, password: str) -> dict:
|
||
"""Holt die /getstationsdata Response. Returns parsed dict."""
|
||
creds = {"client_id": client_id, "client_secret": client_secret,
|
||
"username": username, "password": password}
|
||
|
||
now = time.time()
|
||
if _TOKEN_CACHE["expires_at"] <= now or not _TOKEN_CACHE["access_token"]:
|
||
_obtain_tokens(creds)
|
||
|
||
def _do():
|
||
url = f"{STATIONS_URL}?get_favorites=false"
|
||
req = urllib.request.Request(url, headers={
|
||
"Authorization": f"Bearer {_TOKEN_CACHE['access_token']}",
|
||
"Accept": "application/json",
|
||
})
|
||
with urllib.request.urlopen(req, timeout=10) as r:
|
||
return json.loads(r.read())
|
||
|
||
try:
|
||
return _do()
|
||
except urllib.error.HTTPError as e:
|
||
if e.code != 401:
|
||
raise
|
||
# 401 → Token abgelaufen / widerrufen, einmal neu holen
|
||
_TOKEN_CACHE["access_token"] = None
|
||
_TOKEN_CACHE["expires_at"] = 0
|
||
_obtain_tokens(creds)
|
||
return _do()
|
||
|
||
|
||
# ============================================================================
|
||
# Parsing Helpers
|
||
# ============================================================================
|
||
def _fmt_temp(t: float | None, unit: str = "C") -> str:
|
||
if t is None:
|
||
return "—"
|
||
return f"{t:.1f}°{unit}"
|
||
|
||
|
||
def _fmt_pct(v: float | None) -> str:
|
||
if v is None:
|
||
return "—"
|
||
return f"{int(v)}%"
|
||
|
||
|
||
def _fmt_wind(w: float | None, unit: str = "kmh") -> str:
|
||
if w is None:
|
||
return "—"
|
||
if unit == "ms":
|
||
return f"{w / 3.6:.1f} m/s"
|
||
return f"{w:.1f} km/h"
|
||
|
||
|
||
def _fmt_rain(mm: float | None) -> str:
|
||
if mm is None:
|
||
return "—"
|
||
return f"{mm:.1f} mm"
|
||
|
||
|
||
def _time_short(epoch_s: int | None) -> str:
|
||
"""Unix timestamp → 'HH:MM' oder '' wenn None."""
|
||
if not epoch_s:
|
||
return ""
|
||
try:
|
||
dt = datetime.fromtimestamp(epoch_s, tz=timezone.utc).astimezone()
|
||
return dt.strftime("%H:%M")
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def _temp_color(t: float | None) -> tuple:
|
||
if t is None:
|
||
return FG
|
||
if t >= 30:
|
||
return RED
|
||
if t >= 22:
|
||
return ORANGE
|
||
if t <= 0:
|
||
return BLUE
|
||
if t <= 8:
|
||
return INFO
|
||
return FG
|
||
|
||
|
||
def _co2_pct(co2: float | None) -> float:
|
||
"""CO2 in ppm → 0..100 % der Skala bis 2000 ppm."""
|
||
if co2 is None:
|
||
return 0
|
||
return min(100.0, max(0.0, co2 / 2000.0 * 100.0))
|
||
|
||
|
||
def _co2_color(co2: float | None) -> tuple:
|
||
if co2 is None:
|
||
return FG
|
||
if co2 >= 1500:
|
||
return ALERT
|
||
if co2 >= 1000:
|
||
return WARN
|
||
return OK
|
||
|
||
|
||
def _humidity_color(h: float | None) -> tuple:
|
||
if h is None:
|
||
return FG
|
||
if h < 30 or h > 65:
|
||
return WARN
|
||
return OK
|
||
|
||
|
||
def _parse_stations(api_response: dict, station_filter: str = "") -> dict | None:
|
||
"""Extrahiert die passende Station und alle Module.
|
||
|
||
Returns dict mit:
|
||
- station_name, place
|
||
- main: dict (NAMain dashboard_data + type)
|
||
- modules: list[dict] (alle Zusatzmodule mit type + dashboard_data)
|
||
Oder None wenn keine Station gefunden.
|
||
"""
|
||
body = api_response.get("body") if "body" in api_response else api_response
|
||
devices = body.get("devices") if isinstance(body, dict) else None
|
||
if not devices:
|
||
return None
|
||
|
||
# Falls ein Filter gesetzt ist, matche auf station_name oder module_name
|
||
chosen = None
|
||
if station_filter:
|
||
sf = station_filter.strip().lower()
|
||
for dev in devices:
|
||
if sf in (dev.get("station_name") or "").lower():
|
||
chosen = dev
|
||
break
|
||
for m in dev.get("modules", []):
|
||
if sf in (m.get("module_name") or "").lower():
|
||
chosen = dev
|
||
break
|
||
if chosen:
|
||
break
|
||
if chosen is None:
|
||
chosen = devices[0]
|
||
|
||
main = {
|
||
"type": "NAMain",
|
||
"name": chosen.get("station_name", "Station"),
|
||
"data": chosen.get("dashboard_data") or {},
|
||
"place": chosen.get("place") or {},
|
||
"wifi_status": chosen.get("wifi_status"),
|
||
"reachable": chosen.get("reachable"),
|
||
}
|
||
modules = []
|
||
for m in chosen.get("modules", []):
|
||
mtype = m.get("type", "")
|
||
# Module ohne aktuelle Daten (rf_strength, battery_level, last_seen)
|
||
modules.append({
|
||
"type": mtype,
|
||
"name": m.get("module_name") or MODULE_TYPES.get(mtype, {}).get("label", mtype),
|
||
"id": m.get("_id"),
|
||
"data": m.get("dashboard_data") or {},
|
||
"battery_pct": m.get("battery_percent"),
|
||
"rf_status": m.get("rf_status"),
|
||
"reachable": m.get("reachable"),
|
||
"last_seen": m.get("last_seen"),
|
||
"last_message": m.get("last_message"),
|
||
})
|
||
return {
|
||
"station_name": main["name"],
|
||
"place": main.get("place", {}),
|
||
"main": main,
|
||
"modules": modules,
|
||
"_fetched_at": time.time(),
|
||
}
|
||
|
||
|
||
# ============================================================================
|
||
# Plugin-Klasse
|
||
# ============================================================================
|
||
class Widget(Widget):
|
||
name = "netatmo"
|
||
label = "Netatmo Wetterstation"
|
||
description = ("Live-Daten der heimischen Netatmo Station: Indoor, Outdoor, "
|
||
"Wind, Regen, CO₂. Responsive 1×1..4×4.")
|
||
category = "weather"
|
||
|
||
config_schema = [
|
||
{"key": "client_id", "label": "Netatmo Client-ID", "type": "secret",
|
||
"help": "App-Client-ID von https://dev.netatmo.com/apps/"},
|
||
{"key": "client_secret", "label": "Netatmo Client-Secret", "type": "secret"},
|
||
{"key": "username", "label": "Netatmo Login (E-Mail)", "type": "secret"},
|
||
{"key": "password", "label": "Netatmo Passwort", "type": "secret",
|
||
"help": "Wird nur lokal für den ersten OAuth-Token verwendet."},
|
||
{"key": "station_filter", "label": "Station (leer = erste)",
|
||
"type": "string", "default": ""},
|
||
{"key": "show_indoor", "label": "Indoor-Modul anzeigen", "type": "bool",
|
||
"default": True},
|
||
{"key": "show_outdoor", "label": "Outdoor-Modul anzeigen", "type": "bool",
|
||
"default": True},
|
||
{"key": "show_wind", "label": "Wind-Modul anzeigen", "type": "bool",
|
||
"default": True},
|
||
{"key": "show_rain", "label": "Regen-Modul anzeigen", "type": "bool",
|
||
"default": True},
|
||
{"key": "show_compass", "label": "Windrose", "type": "bool",
|
||
"default": True},
|
||
{"key": "show_secondary", "label": "Min/Max + Letzte Aktualisierung",
|
||
"type": "bool", "default": True,
|
||
"help": "Zeigt Tages-Min/Max und Timestamp. Bei großen Slots."},
|
||
{"key": "co2_thresholds",
|
||
"label": "CO₂-Schwellen (ppm)",
|
||
"type": "string", "default": "ok@600,warn@1000,alert@1500",
|
||
"help": "ppm-Schwellen für die CO₂-Bar (grün/gelb/rot)."},
|
||
{"key": "bar_gradient",
|
||
"label": "CO₂-Bar Verlaufsmodus", "type": "bool", "default": True},
|
||
{"key": "temp_unit", "label": "Temperatur-Einheit", "type": "select",
|
||
"options": ["C", "F"], "default": "C"},
|
||
{"key": "wind_unit", "label": "Wind-Einheit", "type": "select",
|
||
"options": ["kmh", "ms"], "default": "kmh"},
|
||
]
|
||
default_config = {
|
||
"client_id": "",
|
||
"client_secret": "",
|
||
"username": "",
|
||
"password": "",
|
||
"station_filter": "",
|
||
"show_indoor": True,
|
||
"show_outdoor": True,
|
||
"show_wind": True,
|
||
"show_rain": True,
|
||
"show_compass": True,
|
||
"show_secondary": True,
|
||
"co2_thresholds": "ok@600,warn@1000,alert@1500",
|
||
"bar_gradient": True,
|
||
"temp_unit": "C",
|
||
"wind_unit": "kmh",
|
||
}
|
||
|
||
# ---- Helper ----
|
||
def _co2_thresholds(self) -> list:
|
||
# Wenn user ok@600 geschrieben hat, parst parse_thresholds das richtig.
|
||
# Aber parse_thresholds erwartet Prozent. Hier rechnen wir auf %
|
||
# um, damit wir die gleiche hbar-Routine nutzen können.
|
||
spec = self.cfg("co2_thresholds", "ok@600,warn@1000,alert@1500")
|
||
if isinstance(spec, str):
|
||
result = []
|
||
default_pcts = [600, 1000, 1500]
|
||
for i, p in enumerate([s.strip() for s in spec.split(",") if s.strip()]):
|
||
if "@" in p:
|
||
name, val = p.split("@", 1)
|
||
try:
|
||
ppm = float(val)
|
||
except ValueError:
|
||
continue
|
||
else:
|
||
name = p
|
||
ppm = default_pcts[i] if i < len(default_pcts) else 2000
|
||
try:
|
||
color = {
|
||
"ok": OK, "warn": WARN, "alert": ALERT,
|
||
"fg": FG, "green": GREEN, "yellow": WARN,
|
||
"red": RED, "orange": ORANGE, "blue": BLUE,
|
||
}[name]
|
||
except KeyError:
|
||
continue
|
||
# Convert ppm → % of 2000 scale
|
||
result.append((min(100.0, ppm / 2000.0 * 100.0), color))
|
||
return sorted(result, key=lambda x: x[0]) if result else [
|
||
(30.0, OK), (50.0, WARN), (75.0, ALERT),
|
||
]
|
||
return [(30.0, OK), (50.0, WARN), (75.0, ALERT)]
|
||
|
||
def _bar_args(self):
|
||
return (self._co2_thresholds(),
|
||
bool(self.cfg("bar_gradient", True)))
|
||
|
||
# ---- Fetch ----
|
||
def fetch(self) -> dict:
|
||
cid = self.cfg("client_id")
|
||
sec = self.cfg("client_secret")
|
||
user = self.cfg("username")
|
||
pw = self.cfg("password")
|
||
if not (cid and sec and user and pw):
|
||
return {"_error": "Client-ID / Secret / Login / Passwort fehlen."}
|
||
|
||
try:
|
||
data = _get_stations_data(cid, sec, user, pw)
|
||
except urllib.error.HTTPError as e:
|
||
body = ""
|
||
try:
|
||
body = e.read().decode("utf-8", "ignore")[:120]
|
||
except Exception:
|
||
pass
|
||
if e.code in (401, 403):
|
||
# Auth-Fehler: Token-Cache zurücksetzen für nächsten Versuch
|
||
_TOKEN_CACHE["access_token"] = None
|
||
_TOKEN_CACHE["expires_at"] = 0
|
||
return {"_error": f"Auth fehlgeschlagen (HTTP {e.code}). "
|
||
f"Credentials prüfen."}
|
||
return {"_error": f"HTTP {e.code} {e.reason}: {body}".strip()}
|
||
except Exception as e:
|
||
return {"_error": f"{type(e).__name__}: {str(e)[:80]}"}
|
||
|
||
parsed = _parse_stations(data, self.cfg("station_filter", ""))
|
||
if not parsed:
|
||
return {"_error": "Keine Station gefunden."}
|
||
|
||
parsed["_raw_count"] = len(data.get("body", data).get("devices", []))
|
||
return parsed
|
||
|
||
# ---- Render ----
|
||
def render(self, draw, fonts, x, y, w, h):
|
||
pad = 8
|
||
d = self.fetch()
|
||
|
||
if "_error" in d:
|
||
render_error_banner(draw, fonts, x, y, w, h,
|
||
self.label, d["_error"])
|
||
return
|
||
|
||
# Header
|
||
header_font = fonts.get("24") or fonts.get("20") or fonts.get("default")
|
||
draw.text((x + pad, y + pad), "NETATMO", font=header_font, fill=INFO)
|
||
sub = d["station_name"]
|
||
if len(sub) > 24:
|
||
sub = sub[:23] + "…"
|
||
font_sub = fit_font(draw, sub, fonts, w - 2 * pad - 4, 18)
|
||
draw.text((x + w - measure(draw, sub, font_sub)[0] - pad, y + pad + 4),
|
||
sub, font=font_sub, fill=FG)
|
||
|
||
# Module nach Typ filtern
|
||
main = d["main"]
|
||
modules = [m for m in d["modules"]
|
||
if self._module_visible(m["type"])]
|
||
|
||
# Dispatches je Slot-Größe. Reihenfolge wichtig:
|
||
# wide/tall zuerst prüfen, weil `is_small(w,h)` sehr eager ist
|
||
# (alles mit w<280 oder h<180) und sonst 4x1/1x4 Slots überschattet.
|
||
if is_wide(w, h):
|
||
self._render_wide(draw, fonts, x, y, w, h, main, modules)
|
||
return
|
||
if is_tall(w, h):
|
||
self._render_tall(draw, fonts, x, y, w, h, main, modules)
|
||
return
|
||
if is_small(w, h):
|
||
self._render_small(draw, fonts, x, y, w, h, main, modules)
|
||
return
|
||
# Standard (2x2 oder größer quadratisch)
|
||
self._render_standard(draw, fonts, x, y, w, h, main, modules, d)
|
||
|
||
def _module_visible(self, mtype: str) -> bool:
|
||
return {
|
||
"NAMain": self.cfg("show_indoor", True),
|
||
"NAModule1": self.cfg("show_outdoor", True),
|
||
"NAModule2": self.cfg("show_wind", True),
|
||
"NAModule3": self.cfg("show_rain", True),
|
||
"NAModule4": self.cfg("show_indoor", True), # Extra-Indoor
|
||
}.get(mtype, True)
|
||
|
||
# ---- Layouts ----
|
||
def _render_small(self, draw, fonts, x, y, w, h, main, modules):
|
||
"""1x1 — Outdoor-Temp prominent + Mini-Status."""
|
||
out = self._first_module(modules, "NAModule1")
|
||
if out:
|
||
t = out["data"].get("Temperature")
|
||
temp = _fmt_temp(t, self.cfg("temp_unit", "C"))
|
||
color = _temp_color(t)
|
||
font = fit_font(draw, temp, fonts, w - 2 * 8, int(h * 0.55))
|
||
centered_text(draw, temp, x, y + 26, w, int(h * 0.55), font, color)
|
||
else:
|
||
in_t = main["data"].get("Temperature")
|
||
temp = _fmt_temp(in_t, self.cfg("temp_unit", "C"))
|
||
color = _temp_color(in_t)
|
||
font = fit_font(draw, temp, fonts, w - 2 * 8, int(h * 0.55))
|
||
centered_text(draw, temp, x, y + 26, w, int(h * 0.55), font, color)
|
||
|
||
# Mini-Status-Zeile unten: Hum + CO2
|
||
sub_y = y + h - 22
|
||
parts = []
|
||
if out:
|
||
h_val = out["data"].get("Humidity")
|
||
if h_val is not None:
|
||
parts.append(f"💧{int(h_val)}%")
|
||
co2 = main["data"].get("CO2")
|
||
if co2 is not None:
|
||
parts.append(f"CO₂ {int(co2)}")
|
||
line = " ".join(parts) or "—"
|
||
font = fit_font(draw, line, fonts, w - 2 * 8, 18)
|
||
centered_text(draw, line, x, sub_y, w, 22, font, FG)
|
||
|
||
def _render_wide(self, draw, fonts, x, y, w, h, main, modules):
|
||
"""4x1 — Indoor | Outdoor | Wind/Regen kompakt."""
|
||
col_w = w // 3
|
||
# Col 1: Indoor Temp + Hum + CO2-bar
|
||
self._draw_module_cell(
|
||
draw, fonts,
|
||
x + 4, y + 28, col_w - 8, h - 32,
|
||
main, label="🏠 IN", show_co2_bar=True,
|
||
)
|
||
# Col 2: Outdoor
|
||
out = self._first_module(modules, "NAModule1")
|
||
if out:
|
||
self._draw_module_cell(
|
||
draw, fonts,
|
||
x + col_w + 4, y + 28, col_w - 8, h - 32,
|
||
out, label="🌳 OUT", show_co2_bar=False,
|
||
)
|
||
# Col 3: Wind + Rain stacked
|
||
cx3 = x + 2 * col_w + 4
|
||
cw3 = col_w - 8
|
||
row_h = (h - 32) // 2
|
||
wind = self._first_module(modules, "NAModule2")
|
||
if wind:
|
||
self._draw_module_cell(
|
||
draw, fonts, cx3, y + 28, cw3, row_h,
|
||
wind, label="💨", show_co2_bar=False, compact=True,
|
||
)
|
||
rain = self._first_module(modules, "NAModule3")
|
||
if rain:
|
||
self._draw_module_cell(
|
||
draw, fonts, cx3, y + 28 + row_h, cw3, row_h,
|
||
rain, label="🌧", show_co2_bar=False, compact=True,
|
||
)
|
||
|
||
def _render_tall(self, draw, fonts, x, y, w, h, main, modules):
|
||
"""1x4 / 1x2 — vertikale Liste aller Module."""
|
||
items = []
|
||
if self.cfg("show_indoor", True):
|
||
items.append(("🏠", "IN", main))
|
||
for m in modules:
|
||
label = MODULE_TYPES.get(m["type"], {}).get("short", m["type"][-2:]).upper()
|
||
items.append((MODULE_TYPES.get(m["type"], {}).get("icon", "•"), label, m))
|
||
|
||
row_h = max(40, (h - 32) // max(1, len(items)))
|
||
for i, (icon, label, mod) in enumerate(items):
|
||
ry = y + 30 + i * row_h
|
||
self._draw_module_cell(
|
||
draw, fonts, x + 4, ry, w - 8, row_h - 4,
|
||
mod, label=f"{icon} {label}", show_co2_bar=False,
|
||
compact=(row_h < 70),
|
||
)
|
||
|
||
def _render_standard(self, draw, fonts, x, y, w, h, main, modules, d):
|
||
"""Standard 2x2+ Layout."""
|
||
pad = 8
|
||
# Top-Header-Zone (NETATMO + station name) wird in render() gemacht.
|
||
body_y = y + 36
|
||
body_h = h - 36
|
||
|
||
# Indoor Card (links)
|
||
card_w = (w - 3 * pad) // 2
|
||
card_h = (body_h - pad) // 2
|
||
if self.cfg("show_indoor", True):
|
||
self._draw_card(
|
||
draw, fonts, x + pad, body_y,
|
||
card_w, card_h,
|
||
"🏠 Indoor", main,
|
||
show_co2=True, show_minmax=self.cfg("show_secondary", True),
|
||
)
|
||
|
||
# Outdoor Card (rechts)
|
||
out = self._first_module(modules, "NAModule1")
|
||
if out and self.cfg("show_outdoor", True):
|
||
self._draw_card(
|
||
draw, fonts, x + pad + card_w + pad, body_y,
|
||
card_w, card_h,
|
||
"🌳 Outdoor", out,
|
||
show_co2=False, show_minmax=self.cfg("show_secondary", True),
|
||
)
|
||
|
||
# Bottom: Wind-Modul + Rain-Modul (zusammen in einer Card)
|
||
bottom_y = body_y + card_h + pad
|
||
wind = self._first_module(modules, "NAModule2")
|
||
rain = self._first_module(modules, "NAModule3")
|
||
self._draw_wind_rain_card(
|
||
draw, fonts, x + pad, bottom_y, w - 2 * pad, card_h,
|
||
wind, rain,
|
||
show_compass=self.cfg("show_compass", True),
|
||
show_secondary=self.cfg("show_secondary", True),
|
||
)
|
||
|
||
# Footer: Letzte Aktualisierung + Modul-Status (nur wenn Platz)
|
||
if self.cfg("show_secondary", True) and h >= 280:
|
||
self._draw_footer(draw, fonts, x, y + h - 16, w, main, modules)
|
||
|
||
# ---- Card-Drawing Helpers ----
|
||
def _draw_card(self, draw, fonts, x, y, w, h, title, mod,
|
||
show_co2=True, show_minmax=False):
|
||
"""Eine Modul-Card mit Header, Temperatur, Humidity, CO2-Bar."""
|
||
pad = 6
|
||
# Border
|
||
draw.rectangle((x, y, x + w - 1, y + h - 1), outline=FG, width=2)
|
||
# Title bar
|
||
title_font = fit_font(draw, title, fonts, w - 2 * pad, 22)
|
||
draw.rectangle((x, y, x + w - 1, y + 24), fill=FG)
|
||
draw.text((x + pad, y + 4), title,
|
||
font=title_font, fill=BG)
|
||
# Name des Moduls rechts (klein)
|
||
name = mod.get("name", "")
|
||
if name and name.lower() != title.split(" ", 1)[-1].lower():
|
||
short = name[:18] + ("…" if len(name) > 18 else "")
|
||
font_n = fit_font(draw, short, fonts, w // 3, 16)
|
||
tw, _ = measure(draw, short, font_n)
|
||
draw.text((x + w - tw - pad - 2, y + 6),
|
||
short, font=font_n, fill=BG)
|
||
|
||
# Body
|
||
inner_y = y + 28
|
||
inner_h = h - 30
|
||
data = mod.get("data", {}) or {}
|
||
|
||
# Big Temp
|
||
t = data.get("Temperature")
|
||
temp_str = _fmt_temp(t, self.cfg("temp_unit", "C"))
|
||
temp_color = _temp_color(t)
|
||
font_t = fit_font(draw, temp_str, fonts,
|
||
w - 2 * pad - 4, int(inner_h * 0.6))
|
||
draw.text((x + pad, inner_y + 4), temp_str,
|
||
font=font_t, fill=temp_color)
|
||
|
||
# Min/Max als kleine Labels darunter
|
||
if show_minmax and inner_h > 80:
|
||
mn = data.get("min_temp")
|
||
mx = data.get("max_temp")
|
||
if mn is not None or mx is not None:
|
||
tmin = (data.get("date_min_temp") or 0)
|
||
tmax = (data.get("date_max_temp") or 0)
|
||
line = ""
|
||
if mn is not None:
|
||
line += f"↓{mn:.1f}° {_time_short(tmin) or '—'}"
|
||
if mx is not None:
|
||
line += f" ↑{mx:.1f}° {_time_short(tmax) or '—'}"
|
||
font_m = fit_font(draw, line, fonts, w - 2 * pad, 16)
|
||
draw.text((x + pad, inner_y + 4 + font_t.size + 4),
|
||
line, font=font_m, fill=FG)
|
||
|
||
# Humidity badge rechts oben
|
||
h_val = data.get("Humidity")
|
||
if h_val is not None and w > 180:
|
||
h_str = f"💧 {int(h_val)}%"
|
||
font_h = fit_font(draw, h_str, fonts, w // 3, 22)
|
||
tw, th = measure(draw, h_str, font_h)
|
||
draw.text((x + w - tw - pad - 2, inner_y + 6),
|
||
h_str, font=font_h, fill=_humidity_color(h_val))
|
||
|
||
# CO2-Bar (nur für Indoor)
|
||
if show_co2:
|
||
co2 = data.get("CO2")
|
||
if co2 is not None:
|
||
# Label
|
||
label = f"CO₂ {int(co2)} ppm"
|
||
font_l = fit_font(draw, label, fonts, w - 2 * pad, 18)
|
||
draw.text((x + pad, y + h - 38),
|
||
label, font=font_l, fill=_co2_color(co2))
|
||
# Bar
|
||
thresholds, gradient = self._bar_args()
|
||
hbar(draw, x + pad, y + h - 18, w - 2 * pad, 12,
|
||
_co2_pct(co2), thresholds=thresholds, gradient=gradient)
|
||
|
||
# Pressure (nur wenn da, z.B. NAMain)
|
||
if w > 280 and "Pressure" in data:
|
||
p = data.get("Pressure")
|
||
if p is not None:
|
||
p_str = f"{p:.0f} mbar"
|
||
font_p = fit_font(draw, p_str, fonts, w // 4, 18)
|
||
tw, _ = measure(draw, p_str, font_p)
|
||
draw.text((x + w - tw - pad - 2, inner_y + 30),
|
||
p_str, font=font_p, fill=INFO)
|
||
|
||
def _draw_wind_rain_card(self, draw, fonts, x, y, w, h, wind, rain,
|
||
show_compass=True, show_secondary=True):
|
||
"""Untere Card: Wind-Modul links, Regen-Modul rechts."""
|
||
pad = 6
|
||
draw.rectangle((x, y, x + w - 1, y + h - 1), outline=FG, width=2)
|
||
|
||
# 2 Spalten
|
||
col_w = (w - 3 * pad) // 2
|
||
# ----- Wind -----
|
||
wx = x + pad
|
||
wy = y + 4
|
||
wh = h - 8
|
||
if wind:
|
||
wd = wind["data"]
|
||
title_font = fit_font(draw, "💨 Wind", fonts, col_w - 2 * pad, 20)
|
||
draw.text((wx + pad, wy), "💨 Wind", font=title_font, fill=FG)
|
||
|
||
speed = wd.get("WindStrength")
|
||
gust = wd.get("GustStrength")
|
||
direction = wd.get("WindAngle")
|
||
unit = self.cfg("wind_unit", "kmh")
|
||
s_str = _fmt_wind(speed, unit)
|
||
font_s = fit_font(draw, s_str, fonts, col_w - 2 * pad, int(wh * 0.55))
|
||
draw.text((wx + pad, wy + 24), s_str, font=font_s, fill=ORANGE)
|
||
|
||
if gust is not None:
|
||
g_str = f"♨ {_fmt_wind(gust, unit)}"
|
||
font_g = fit_font(draw, g_str, fonts, col_w - 2 * pad, 18)
|
||
draw.text((wx + pad, wy + 24 + font_s.size + 4),
|
||
g_str, font=font_g, fill=FG)
|
||
|
||
if direction is not None and show_compass:
|
||
cr = min(28, col_w // 4, (wh - 24) // 3)
|
||
cx = wx + col_w - cr - pad - 4
|
||
cy = wy + wh - cr - 4
|
||
self._draw_compass(draw, cx, cy, cr, direction, fonts)
|
||
|
||
if show_secondary and wh > 110:
|
||
# Min/Max
|
||
mn = wd.get("min_wind_str")
|
||
mx = wd.get("max_wind_str")
|
||
line = ""
|
||
if mn is not None:
|
||
line += f"↓{mn:.0f} "
|
||
if mx is not None:
|
||
line += f"↑{mx:.0f}"
|
||
if line:
|
||
font_l = fit_font(draw, line, fonts, col_w - 2 * pad - 30, 16)
|
||
draw.text((wx + pad, wy + wh - 20),
|
||
line, font=font_l, fill=FG)
|
||
else:
|
||
font_e = fit_font(draw, "kein Wind-Sensor", fonts, col_w - 2 * pad, 18)
|
||
draw.text((wx + pad, wy + 12), "kein Wind-Sensor",
|
||
font=font_e, fill=FG)
|
||
|
||
# Trennlinie zwischen Wind und Rain
|
||
sep_x = x + col_w + 2 * pad
|
||
draw.line((sep_x, y + 6, sep_x, y + h - 6), fill=FG, width=1)
|
||
|
||
# ----- Regen -----
|
||
rx = x + col_w + 3 * pad
|
||
ry = y + 4
|
||
if rain:
|
||
rd = rain["data"]
|
||
title_font = fit_font(draw, "🌧 Regen", fonts, col_w - 2 * pad, 20)
|
||
draw.text((rx + pad, ry), "🌧 Regen", font=title_font, fill=FG)
|
||
|
||
# Aktuelle Regenrate
|
||
rate = rd.get("RainRate") or rd.get("rain_rate") or 0
|
||
# Verschiedene Felder je nach Netatmo-Variante
|
||
cur = (rd.get("rain") if rate is None else rate)
|
||
cur_str = f"{cur:.1f} mm/h" if cur else "trocken"
|
||
font_r = fit_font(draw, cur_str, fonts, col_w - 2 * pad, int(wh * 0.4))
|
||
draw.text((rx + pad, ry + 22),
|
||
cur_str, font=font_r,
|
||
fill=BLUE if (cur and cur > 0) else FG)
|
||
|
||
# Summen
|
||
if show_secondary and wh > 110:
|
||
h1 = rd.get("sum_rain_1") or rd.get("rain_hour") or 0
|
||
h24 = rd.get("sum_rain_24") or rd.get("rain_day") or 0
|
||
font_s = fit_font(draw, "1h: 1.0 mm", fonts, col_w - 2 * pad, 18)
|
||
yl = ry + 22 + font_r.size + 6
|
||
draw.text((rx + pad, yl),
|
||
f"1h: {_fmt_rain(h1)}",
|
||
font=font_s, fill=FG)
|
||
draw.text((rx + pad, yl + font_s.size + 2),
|
||
f"24h: {_fmt_rain(h24)}",
|
||
font=font_s, fill=FG)
|
||
else:
|
||
font_e = fit_font(draw, "kein Regen-Sensor", fonts, col_w - 2 * pad, 18)
|
||
draw.text((rx + pad, ry + 12),
|
||
"kein Regen-Sensor", font=font_e, fill=FG)
|
||
|
||
def _draw_module_cell(self, draw, fonts, x, y, w, h, mod, label="",
|
||
show_co2_bar=False, compact=False):
|
||
"""Kleine Modul-Zelle (z.B. für wide/tall Layout)."""
|
||
pad = 4
|
||
data = mod.get("data", {}) or {}
|
||
t = data.get("Temperature")
|
||
temp = _fmt_temp(t, self.cfg("temp_unit", "C"))
|
||
color = _temp_color(t)
|
||
|
||
# Top label
|
||
font_l = fit_font(draw, label, fonts, w - 2 * pad, 18)
|
||
draw.text((x + pad, y + 2), label, font=font_l, fill=INFO)
|
||
|
||
if compact:
|
||
font_t = fit_font(draw, temp, fonts, w - 2 * pad, h - 20)
|
||
draw.text((x + pad, y + 22), temp, font=font_t, fill=color)
|
||
else:
|
||
font_t = fit_font(draw, temp, fonts, w - 2 * pad, int(h * 0.5))
|
||
draw.text((x + pad, y + 22), temp, font=font_t, fill=color)
|
||
# Humidity
|
||
h_val = data.get("Humidity")
|
||
if h_val is not None:
|
||
h_str = f"💧 {int(h_val)}%"
|
||
font_h = fit_font(draw, h_str, fonts, w - 2 * pad, 18)
|
||
draw.text((x + pad, y + 22 + font_t.size + 6),
|
||
h_str, font=font_h, fill=_humidity_color(h_val))
|
||
|
||
# CO2-Bar wenn Indoor
|
||
if show_co2_bar:
|
||
co2 = data.get("CO2")
|
||
if co2 is not None and h > 60:
|
||
thresholds, gradient = self._bar_args()
|
||
bar_y = y + h - 14
|
||
hbar(draw, x + pad, bar_y, w - 2 * pad, 8,
|
||
_co2_pct(co2), thresholds=thresholds, gradient=gradient)
|
||
# Label
|
||
lbl = f"CO₂ {int(co2)}"
|
||
font_c = fit_font(draw, lbl, fonts, w // 2, 14)
|
||
draw.text((x + pad, bar_y - 14),
|
||
lbl, font=font_c, fill=_co2_color(co2))
|
||
|
||
def _draw_compass(self, draw, cx, cy, r, deg, fonts):
|
||
"""Windrose: Kreis + N/E/S/W + Pfeil auf deg (0=N)."""
|
||
draw.ellipse((cx - r, cy - r, cx + r, cy + r), outline=FG, width=1)
|
||
for a in range(0, 360, 90):
|
||
rad = math.radians(a - 90)
|
||
x1 = cx + (r - 5) * math.cos(rad)
|
||
y1 = cy + (r - 5) * math.sin(rad)
|
||
x2 = cx + r * math.cos(rad)
|
||
y2 = cy + r * math.sin(rad)
|
||
draw.line((x1, y1, x2, y2), fill=FG, width=1)
|
||
# N/E/S/W Labels
|
||
for a, t in [(270, "N"), (0, "E"), (90, "S"), (180, "W")]:
|
||
rad = math.radians(a - 90)
|
||
tx = cx + (r + 2) * math.cos(rad) - 4
|
||
ty = cy + (r + 2) * math.sin(rad) - 5
|
||
font_x = fonts.get("16") or fonts.get("default")
|
||
draw.text((tx, ty), t, font=font_x, fill=FG)
|
||
# Pfeil
|
||
rad_arrow = math.radians(deg - 90)
|
||
tip_x = cx + (r - 4) * math.cos(rad_arrow)
|
||
tip_y = cy + (r - 4) * math.sin(rad_arrow)
|
||
base = math.radians(150)
|
||
lx = cx + 7 * math.cos(rad_arrow + base)
|
||
ly = cy + 7 * math.sin(rad_arrow + base)
|
||
rx = cx + 7 * math.cos(rad_arrow - base)
|
||
ry = cy + 7 * math.sin(rad_arrow - base)
|
||
draw.polygon([(tip_x, tip_y), (lx, ly), (rx, ry)], fill=RED)
|
||
draw.ellipse((cx - 2, cy - 2, cx + 2, cy + 2), fill=FG)
|
||
|
||
def _draw_footer(self, draw, fonts, x, y, w, main, modules):
|
||
"""Footer: letzte Aktualisierung + Reichweite."""
|
||
# 'last_message' der Hauptstation ist meist die letzte Datenaktualisierung
|
||
last_msg = main.get("data", {}).get("time_utc") or 0
|
||
# Manche API-Versionen liefern 'last_message' als Unix-Sekunden
|
||
if not last_msg:
|
||
last_msg = main.get("last_message", 0)
|
||
ts = _time_short(last_msg)
|
||
reach = main.get("reachable", True)
|
||
text = f"last update: {ts or '—'}"
|
||
if not reach:
|
||
text += " ⚠ offline"
|
||
# Reichweite der Module
|
||
offline = sum(1 for m in modules if not m.get("reachable", True))
|
||
if offline:
|
||
text += f" · {offline} offline"
|
||
|
||
font_f = fit_font(draw, text, fonts, w - 16, 14)
|
||
draw.text((x + 8, y), text, font=font_f, fill=FG)
|
||
|
||
# ---- Utility ----
|
||
@staticmethod
|
||
def _first_module(modules: list, mtype: str) -> dict | None:
|
||
for m in modules:
|
||
if m["type"] == mtype:
|
||
return m
|
||
return None |