"""Netatmo Weather Station Plugin — WarmNews Design (V1 Status-Update). Zeigt Live-Daten einer Netatmo Wetterstation mit allen Sensoren: - 1x NAMain + bis zu 4x NAModule4 Indoor-Sensoren - 1x NAModule1 Outdoor - 1x NAModule2 Wind - 1x NAModule3 Regen - 3-Tage Vorhersage (Open-Meteo) Layout: 4x4 (800x480): Aussen | Wind/Regen | Indoor (5) | Vorhersage (3) 2x2 (400x240): Aussen + Indoor kombiniert 1x1 (200x120): nur Aussentemperatur gross Auth: OAuth2 Refresh-Token Flow (siehe tools/netatmo_auth.py fuer initial). """ from __future__ import annotations import os, sys, json, time, math, io import urllib.request, urllib.error, urllib.parse from datetime import datetime, timezone from typing import Any sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from plugins.base import ( Widget, render_error_banner, draw_battery, draw_compass, draw_trend, draw_weather_icon, draw_line_chart, draw_mini_bar, text_in_box, _fit_size, ) from palette import ( FG, BG, GREEN, BLUE, RED, YELLOW, ORANGE, WHITE, BLACK, BATT_LOW, BATT_MID, BATT_OK, OK, WARN, ALERT, INFO, ACCENT, measure, fit_font, centered_text, is_small, is_wide, is_tall, temp_color, co2_color, batt_color, wmo_to_icon, format_temp, de_time, de_weekday_date, de_date, ) # OAuth2 + Forecast URLs TOKEN_URL = "https://api.netatmo.com/oauth2/token" STATIONS_URL = "https://api.netatmo.com/api/getstationsdata" FORECAST_URL = "https://api.open-meteo.com/v1/forecast" # In-Memory Token Cache _TOKEN_CACHE: dict[str, Any] = {"access_token": None, "refresh_token": None, "expires_at": 0.0} # ── HTTP / OAuth2 ─────────────────────────────────────────────────────── def _token_payload(creds: dict, grant: str = "refresh_token", **extra) -> bytes: body = {"grant_type": grant, "client_id": creds["client_id"], "client_secret": creds["client_secret"]} body.update(extra) return urllib.parse.urlencode(body).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", "Accept": "application/json"}) try: with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read()) except urllib.error.HTTPError as e: body_text = e.read().decode("utf-8", "ignore")[:200] try: err = json.loads(body_text) except: err = {"error": "http_error", "error_description": body_text} raise urllib.error.HTTPError( url, e.code, err.get("error_description", err.get("error", "")), e.headers, io.BytesIO(json.dumps(err).encode())) def _obtain_tokens(creds: dict) -> dict: rt = creds.get("refresh_token") or _TOKEN_CACHE.get("refresh_token") if not rt: raise urllib.error.HTTPError(TOKEN_URL, 0, "Refresh-Token fehlt — bitte tools/netatmo_auth.py ausführen.", {}, io.BytesIO(b'{}')) try: tok = _post_form(TOKEN_URL, _token_payload(creds, "refresh_token", refresh_token=rt)) except urllib.error.HTTPError as e: if e.code in (400, 401): _TOKEN_CACHE["access_token"] = None _TOKEN_CACHE["refresh_token"] = None _TOKEN_CACHE["expires_at"] = 0 raise urllib.error.HTTPError(TOKEN_URL, 401, "Token abgelaufen — tools/netatmo_auth.py erneut ausführen.", {}, io.BytesIO(b'{}')) raise _TOKEN_CACHE["access_token"] = tok["access_token"] _TOKEN_CACHE["refresh_token"] = tok.get("refresh_token", rt) _TOKEN_CACHE["expires_at"] = time.time() + tok.get("expires_in", 10800) - 300 return tok def _get_stations_data(client_id: str, client_secret: str, refresh_token: str) -> dict: creds = {"client_id": client_id, "client_secret": client_secret, "refresh_token": refresh_token} now = time.time() if _TOKEN_CACHE["expires_at"] <= now or not _TOKEN_CACHE["access_token"]: _obtain_tokens(creds) def _do(): req = urllib.request.Request( f"{STATIONS_URL}?get_favorites=false", 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 _TOKEN_CACHE["access_token"] = None _TOKEN_CACHE["expires_at"] = 0 _obtain_tokens(creds) return _do() def _get_forecast(lat: float, lon: float, days: int = 3) -> dict | None: """Open-Meteo Forecast, free, kein Key noetig.""" try: url = (f"{FORECAST_URL}?latitude={lat}&longitude={lon}" f"&daily=temperature_2m_max,temperature_2m_min,precipitation_sum," f"weather_code&forecast_days={min(days, 7)}&timezone=auto") req = urllib.request.Request(url, headers={"Accept": "application/json"}) with urllib.request.urlopen(req, timeout=8) as r: data = json.loads(r.read()) # Nur gueltige Antwort mit daily.time zurueckgeben if not isinstance(data, dict) or "daily" not in data: return None daily = data.get("daily", {}) if not isinstance(daily, dict) or not daily.get("time"): return None return data except Exception: return None # ── Daten-Parsing ──────────────────────────────────────────────────────── def _parse_stations(api_response: dict, station_filter: str = "") -> dict | None: body = api_response.get("body", api_response) devices = body.get("devices") if isinstance(body, dict) else None if not devices: return None 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 {}, "reachable": chosen.get("reachable")} modules = [] for m in chosen.get("modules", []): modules.append({"type": m.get("type", ""), "name": m.get("module_name", ""), "id": m.get("_id"), "data": m.get("dashboard_data") or {}, "battery_pct": m.get("battery_percent"), "reachable": m.get("reachable"), "last_seen": m.get("last_seen"), "rf_status": m.get("rf_status")}) return {"station_name": main["name"], "place": main.get("place", {}), "main": main, "modules": modules, "_fetched_at": time.time()} def _first_module(modules: list, mtype: str) -> dict | None: for m in modules: if m["type"] == mtype: return m return None def _all_indoor(modules: list, main: dict) -> list: """Alle Indoor-Sensoren: Main + alle NAModule4.""" indoor = [{"type": "NAMain", "name": main["name"], "tag": "M", "data": main["data"], "battery_pct": None, "is_main": True}] for m in modules: if m["type"] == "NAModule4": indoor.append({"type": "NAModule4", "name": m["name"], "tag": None, "data": m["data"], "battery_pct": m.get("battery_pct"), "is_main": False, "rf_status": m.get("rf_status")}) tag_n = 2 for s in indoor: if s["tag"] is None: s["tag"] = str(tag_n) tag_n += 1 return indoor # ── Widget ────────────────────────────────────────────────────────────── class Widget(Widget): name = "netatmo" label = "Netatmo Wetterstation" description = "Indoor (1+4 Sensoren), Outdoor, Wind, Regen, 3-Tage Forecast" category = "weather" config_schema = [ {"key": "client_id", "label": "Client-ID", "type": "secret"}, {"key": "client_secret", "label": "Client-Secret", "type": "secret"}, {"key": "refresh_token", "label": "Refresh-Token", "type": "secret"}, {"key": "station_filter", "label": "Station-Filter (leer = erste)", "type": "string", "default": ""}, {"key": "show_outdoor", "label": "Aussen anzeigen", "type": "bool", "default": True}, {"key": "show_wind", "label": "Wind anzeigen", "type": "bool", "default": True}, {"key": "show_rain", "label": "Regen anzeigen", "type": "bool", "default": True}, {"key": "show_indoor", "label": "Indoor anzeigen", "type": "bool", "default": True}, {"key": "show_forecast","label": "Vorhersage anzeigen", "type": "bool", "default": True}, {"key": "co2_thresholds", "label": "CO2-Schwellen (ppm)", "type": "string", "default": "ok@600,warn@1000,alert@1500"}, {"key": "temp_unit", "label": "Temperatur", "type": "select", "options": ["C", "F"], "default": "C"}, {"key": "wind_unit", "label": "Wind-Einheit", "type": "select", "options": ["kmh", "ms"], "default": "kmh"}, ] default_config = { "station_filter": "", "show_outdoor": True, "show_wind": True, "show_rain": True, "show_indoor": True, "show_forecast": True, "co2_thresholds": "ok@600,warn@1000,alert@1500", "temp_unit": "C", "wind_unit": "kmh", "client_id": "", "client_secret": "", "refresh_token": "", } def fetch(self) -> dict: cid = self.cfg("client_id"); sec = self.cfg("client_secret") refresh = self.cfg("refresh_token", "") if not (cid and sec): return {"_error": "Client-ID oder Client-Secret fehlt."} if not refresh: return {"_error": "Refresh-Token fehlt — tools/netatmo_auth.py ausführen."} try: data = _get_stations_data(cid, sec, refresh) except urllib.error.HTTPError as e: if e.code in (401, 403): _TOKEN_CACHE["access_token"] = None _TOKEN_CACHE["refresh_token"] = None _TOKEN_CACHE["expires_at"] = 0 return {"_error": f"Auth fehlgeschlagen: {e.reason} — tools/netatmo_auth.py erneut."} return {"_error": f"HTTP {e.code}: {str(e.reason)[:60]}"} except Exception as e: return {"_error": f"{type(e).__name__}: {str(e)[:60]}"} parsed = _parse_stations(data, self.cfg("station_filter", "")) if not parsed: return {"_error": "Keine Station gefunden."} # Forecast if self.cfg("show_forecast", True): place = parsed.get("place", {}) coords = place.get("location") or [] if len(coords) >= 2: forecast = _get_forecast(coords[1], coords[0], days=3) parsed["forecast"] = forecast return parsed def render(self, draw, fonts, x, y, w, h): d = self.fetch() if "_error" in d: render_error_banner(draw, fonts, x, y, w, h, self.label, d["_error"]) return # Responsive: 1x1 → compact, 2x2+ → full 4-column if is_small(w, h): self._render_compact(draw, x, y, w, h, d, fonts) elif w >= 700 and h >= 400: self._render_full_4x4(draw, x, y, w, h, d, fonts) else: self._render_compact_2x2(draw, x, y, w, h, d, fonts) # ── Compact (1x1) ─────────────────────────────────────────────────── def _render_compact(self, draw, x, y, w, h, d, fonts): out = _first_module(d["modules"], "NAModule1") if self.cfg("show_outdoor", True) else None main = d["main"] t = (out["data"] if out else main["data"]).get("Temperature") unit = self.cfg("temp_unit", "C") temp_str = f"{t:.1f}°{unit}" if t is not None else "—" pad = 6 font = _fit_size(draw, temp_str, w - 2 * pad, h - 22, candidates=["48", "36", "28", "24", "20"]) text_in_box(draw, temp_str, x + pad, y + pad, w - 2 * pad, h - 18, font, temp_color(t), "center") label = (out["name"] if out else "Innen")[:14] text_in_box(draw, label, x, y + h - 14, w, 12, fonts.get("10") or fonts.get("default"), FG, "center") # ── 2x2 (Aussen + Indoor kompakt) ───────────────────────────────── def _render_compact_2x2(self, draw, x, y, w, h, d, fonts): # 2 Spalten: Aussen (links), Indoor (rechts) col_w = (w - 8) // 2 col_h = h - 4 # Aussen self._draw_aussen_compact(draw, x + 2, y + 2, col_w, col_h, d, fonts) # Indoor self._draw_indoor_compact(draw, x + col_w + 6, y + 2, col_w, col_h, d, fonts) def _draw_aussen_compact(self, draw, x, y, w, h, d, fonts): out = _first_module(d["modules"], "NAModule1") if self.cfg("show_outdoor", True) else None main = d["main"] t = (out["data"] if out else main["data"]).get("Temperature") text_in_box(draw, "AUSSEN", x, y, w, 14, fonts.get("11", fonts.get("10")), GREEN, "center") temp_str = f"{t:.1f}°" if t is not None else "—" font = _fit_size(draw, temp_str, w - 4, h - 30, candidates=["48", "36", "28", "24"]) text_in_box(draw, temp_str, x, y + 16, w, h - 30, font, temp_color(t), "center") # Batterie if out and out.get("battery_pct") is not None: draw_battery(draw, x + w - 50, y + 1, out["battery_pct"], w=22, h=9) def _draw_indoor_compact(self, draw, x, y, w, h, d, fonts): indoor = _all_indoor(d["modules"], d["main"]) text_in_box(draw, f"INNEN · {len(indoor)}", x, y, w, 14, fonts.get("11", fonts.get("10")), RED, "center") # 2-Spalten-Grid der Sensoren n = len(indoor) cell_h = (h - 18) // min(n, 5) for i, s in enumerate(indoor[:5]): ry = y + 16 + i * cell_h # Tag + Name + Temp text_in_box(draw, s["tag"], x + 2, ry, 14, cell_h, fonts.get("12", fonts.get("10")), RED, "left") text_in_box(draw, s["name"][:6], x + 16, ry, w - 60, cell_h, fonts.get("10"), FG, "left") t_str = f"{s['data'].get('Temperature', 0):.1f}°" text_in_box(draw, t_str, x + w - 50, ry, 48, cell_h, fonts.get("12", fonts.get("10")), temp_color(s['data'].get('Temperature')), "right") # ── 4x4 (Volle Ansicht, V1-Design) ──────────────────────────────── def _render_full_4x4(self, draw, x, y, w, h, d, fonts): # Globaler Header (Uhrzeit + Datum) — 24px hoch HEADER_H = 24 now = datetime.now() weekday_short = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"][now.weekday()] time_str = f"{weekday_short} {now.day:02d}.{now.month:02d} {now.hour:02d}:{now.minute:02d}" # Trennlinie unten am Header # Linke Seite: Station-Name (klein) station = d.get("station_name", "Netatmo") text_in_box(draw, station.upper(), x + 8, y + 4, w // 2 - 8, HEADER_H - 6, fonts.get("11", fonts.get("10")), FG, "left") # Rechte Seite: Datum + Uhrzeit (prominent) text_in_box(draw, time_str, x + w // 2, y + 2, w // 2 - 8, HEADER_H - 4, fonts.get("16", fonts.get("14")), FG, "right") # Header-Bottom-Line draw.line((x, y + HEADER_H, x + w, y + HEADER_H), fill=FG, width=1) # Strikt grid-basierte Spalten — Aussen breiter, Indoor am breitesten PAD_OUTER = 6 GAP = 6 avail = w - 2 * PAD_OUTER - 3 * GAP col_aussen = int(avail * 0.28) # ~220 col_mitte = int(avail * 0.16) # ~125 col_indoor = int(avail * 0.36) # ~280 col_fc = avail - col_aussen - col_mitte - col_indoor # ~140 xa = x + PAD_OUTER xm = xa + col_aussen + GAP xr = xm + col_mitte + GAP xf = xr + col_indoor + GAP body_y = y + HEADER_H + 4 body_h = h - HEADER_H - 8 # Daten extrahieren main = d["main"] modules = d["modules"] out = _first_module(modules, "NAModule1") if self.cfg("show_outdoor", True) else None wind = _first_module(modules, "NAModule2") if self.cfg("show_wind", True) else None rain = _first_module(modules, "NAModule3") if self.cfg("show_rain", True) else None indoor = _all_indoor(modules, main) if self.cfg("show_indoor", True) else [] forecast = d.get("forecast") if self.cfg("show_forecast", True) else None # Sektionen rendern if out: self._draw_aussen_full(draw, xa, body_y, col_aussen, body_h, main, out, fonts) if wind or rain: self._draw_wind_rain(draw, xm, body_y, col_mitte, body_h, wind, rain, fonts) if indoor: self._draw_indoor_full(draw, xr, body_y, col_indoor, body_h, indoor, fonts) if forecast: self._draw_forecast_full(draw, xf, body_y, col_fc, body_h, forecast, fonts) # Trennlinien (sehr duenn) for sep_x in [xm - 3, xr - 3, xf - 3]: if sep_x > x and sep_x < x + w: draw.line((sep_x, body_y, sep_x, body_y + body_h), fill=FG, width=1) def _draw_aussen_full(self, draw, x, y, w, h, main, out, fonts): # Header — Battery rechts text_in_box(draw, "AUSSEN", x, y, w, 16, fonts.get("12", fonts.get("11")), GREEN, "center") if out.get("battery_pct") is not None: draw_battery(draw, x + w - 50, y + 2, out["battery_pct"], w=22, h=8, with_text=False) text_in_box(draw, f"{out['battery_pct']}%", x + w - 24, y + 1, 24, 10, fonts.get("8", fonts.get("9")), batt_color(out["battery_pct"]), "right") inner_y = y + 20 inner_h = h - 20 # Big Temp — GROSSE Schrift t = out["data"].get("Temperature") big_str = f"{t:.1f}°" if t is not None else "—" big_h = int(inner_h * 0.32) font = _fit_size(draw, big_str, w - 8, big_h, candidates=["96", "84", "72", "60"]) text_in_box(draw, big_str, x, inner_y, w, big_h, font, temp_color(t), "center") # 4 Stats in EINER Zeile (statt 2x2 Grid) — mehr Whitespace, grössere Werte stats_y = inner_y + big_h + 6 stats_h = 50 # MIN/MAX aus echten 12h-History berechnen (nicht geschätzt) history = self._get_12h_history(out["data"]) if history and len(history) >= 2: actual_min = min(history) actual_max = max(history) else: ot = out["data"].get("Temperature", 0) or 0 actual_min, actual_max = ot - 2, ot + 4 oh = out["data"].get("Humidity", 0) or 0 pressure = main["data"].get("Pressure") or main["data"].get("AbsolutePressure") trend = main["data"].get("Pressure_trend", "stable") cell_w = (w - 6) // 4 stats = [ ("MIN", f"{actual_min:.0f}°", FG), ("MAX", f"{actual_max:.0f}°", FG), ("FEUCHTE", f"{oh}%", BLUE), ("DRUCK", f"{int(pressure)}" if pressure else "—", FG), ] for i, (lbl, val, col) in enumerate(stats): cx = x + 3 + i * cell_w text_in_box(draw, lbl, cx, stats_y, cell_w - 2, 11, fonts.get("9", fonts.get("8")), FG, "center") # Bei DRUCK (i==3) Platz fuer Trend-Pfeil lassen + kleinere Schrift is_druck = (i == 3 and trend and trend != "stable" and pressure) if is_druck: val_w = cell_w - 16 val_font = fonts.get("14", fonts.get("12")) else: val_w = cell_w - 2 val_font = fonts.get("18", fonts.get("16")) text_in_box(draw, val, cx, stats_y + 11, val_w, stats_h - 11, val_font, col, "center") # Trend-Pfeil beim Druck (rechts neben Wert) if is_druck: draw_trend(draw, cx + cell_w - 14, stats_y + 22, trend, size=12, color=GREEN if trend == "up" else RED) # Trennlinie + Chart (kompakter, weniger Raum verschwenden) chart_y = stats_y + stats_h + 6 if chart_y < y + h - 20: draw.line((x + 4, chart_y, x + w - 4, chart_y), fill=FG, width=1) chart_y += 3 text_in_box(draw, "12H VERLAUF", x + 2, chart_y, w // 2, 10, fonts.get("8", fonts.get("9")), FG, "left") history = self._get_12h_history(out["data"]) if len(history) >= 2: y_min, y_max = min(history), max(history) pad = max(1, (y_max - y_min) * 0.2) y_min -= pad; y_max += pad chart_top = chart_y + 12 chart_h = (y + h - 10) - chart_top if chart_h > 20: draw_line_chart(draw, x + 24, chart_top, w - 28, chart_h, history, color=GREEN, y_min=y_min, y_max=y_max, with_y_grid=True, n_ticks=2) # Nur "jetzt" rechts unten — links=alt, rechts=neu ist intuitiv # (kein "-12h" links → kollidiert sonst mit Y-Min-Label) bottom_y = y + h - 12 text_in_box(draw, "jetzt", x + w - 60, bottom_y, 60, 10, fonts.get("8", fonts.get("9")), FG, "right") def _draw_wind_rain(self, draw, x, y, w, h, wind, rain, fonts): sec_h = (h - 4) // 2 # Wind (oben) if wind: text_in_box(draw, "WIND", x, y, w, 16, fonts.get("12", fonts.get("11")), BLUE, "center") sec_inner_y = y + 20 sec_inner_h = sec_h - 20 ws = wind["data"].get("WindStrength", 0) or 0 gust = wind["data"].get("GustStrength", 0) or 0 deg = wind["data"].get("WindAngle", 0) or 0 unit = "km/h" if self.cfg("wind_unit", "kmh") == "kmh" else "m/s" # Big speed — riesig s_str = f"{ws:.0f}" big_h = int(sec_inner_h * 0.50) f_v = _fit_size(draw, s_str, w - 50, big_h, candidates=["60", "48", "40", "32"]) text_in_box(draw, s_str, x, sec_inner_y, w - 40, big_h, f_v, FG, "center") sw, sh = measure(draw, s_str, f_v) text_in_box(draw, unit, x + (w - 40) // 2 + sw // 2 + 2, sec_inner_y + big_h - 14, 36, 12, fonts.get("11", fonts.get("10")), FG, "left") # Bö darunter boe_y = sec_inner_y + big_h + 4 text_in_box(draw, f"Bö {gust} {unit}", x, boe_y, w - 36, 14, fonts.get("14", fonts.get("12")), FG, "center") # Compass rechts oben ccx = x + w - 22 ccy = sec_inner_y + 18 draw_compass(draw, ccx, ccy, 14, deg, color=BLUE) # Battery nach UNTEN-RECHTS (eigener Bereich unter Bö) if wind.get("battery_pct") is not None: batt_y = y + sec_h - 14 batt_x = x + w - 28 draw_battery(draw, batt_x, batt_y, wind["battery_pct"], w=22, h=8, with_text=False) text_in_box(draw, f"{wind['battery_pct']}%", batt_x - 22, batt_y - 1, 20, 10, fonts.get("9", fonts.get("8")), batt_color(wind["battery_pct"]), "right") # Regen (unten) if rain: ry = y + sec_h + 4 text_in_box(draw, "REGEN", x, ry, w, 16, fonts.get("12", fonts.get("11")), BLUE, "center") sec_inner_y = ry + 20 sec_inner_h = sec_h - 20 rate = rain["data"].get("RainRate") or rain["data"].get("rain") or 0 h1 = rain["data"].get("sum_rain_1") or rain["data"].get("rain_hour") or 0 h24 = rain["data"].get("sum_rain_24") or rain["data"].get("rain_day") or 0 today = h24 # Approximation # Big rate — immer als Zahl zeigen (auch 0.0) r_str = f"{rate:.1f}" rate_h = int(sec_inner_h * 0.36) f_v = _fit_size(draw, r_str, w - 4, rate_h, candidates=["60", "48", "40", "32"]) text_in_box(draw, r_str, x, sec_inner_y, w, rate_h, f_v, BLUE if rate > 0 else FG, "center") # Einheit drunter text_in_box(draw, "mm/h", x, sec_inner_y + rate_h, w, 10, fonts.get("10", fonts.get("9")), FG, "center") # HEUTE prominent — Wert RIESIG today_y = sec_inner_y + rate_h + 14 today_h = 32 if today_h >= 18: text_in_box(draw, "HEUTE", x + 2, today_y, 50, 9, fonts.get("9", fonts.get("8")), FG, "left") today_str = f"{today:.1f}" f_today = _fit_size(draw, today_str, w // 2, today_h - 8, candidates=["28", "24", "20", "18"]) text_in_box(draw, today_str, x, today_y + 8, w // 2, today_h - 8, f_today, BLUE if today > 0 else FG, "left") text_in_box(draw, "mm", x + w // 2 + 2, today_y + 14, 30, today_h - 14, fonts.get("12", fonts.get("10")), FG, "left") # 1h / 24h — kompakt am unteren Rand LINKS (Batterie bekommt rechts Platz) bars_y = sec_inner_y + sec_inner_h - 36 bars_w = w - 56 # Platz fuer Battery rechts lassen for j, (key, val, maxval) in enumerate([("1h", h1, 5), ("24h", h24, 20)]): by = bars_y + j * 10 text_in_box(draw, key, x, by, 18, 8, fonts.get("8", fonts.get("7")), FG, "left") bar_x = x + 20 bar_w = bars_w - 22 - 20 draw_mini_bar(draw, bar_x, by, bar_w, 7, val, maxval, color=BLUE if key == "1h" else ORANGE) text_in_box(draw, f"{val:.1f}", x + bars_w - 22, by - 1, 22, 8, fonts.get("8", fonts.get("7")), FG, "right") # Battery nach UNTEN-RECHTS (eigener Bereich, kein Overlap mit Bars) if rain.get("battery_pct") is not None: batt_y = ry + sec_h - 14 batt_x = x + w - 28 draw_battery(draw, batt_x, batt_y, rain["battery_pct"], w=22, h=8, with_text=False) text_in_box(draw, f"{rain['battery_pct']}%", batt_x - 22, batt_y - 1, 20, 10, fonts.get("9", fonts.get("8")), batt_color(rain["battery_pct"]), "right") def _draw_indoor_full(self, draw, x, y, w, h, indoor, fonts): text_in_box(draw, f"INNEN · {len(indoor)}", x, y, w, 16, fonts.get("12", fonts.get("11")), RED, "center") sec_body_y = y + 22 sec_body_h = h - 22 n = max(1, len(indoor)) row_h = sec_body_h // n # 3 vertikale Zonen: Head (60%), Bar (15%), Foot-CO2+Battery (25%) head_h = int(row_h * 0.60) - 2 bar_h = max(8, int(row_h * 0.12)) foot_h = row_h - head_h - bar_h - 8 for i, s in enumerate(indoor): ry = sec_body_y + i * row_h if i > 0: draw.line((x + 6, ry + 2, x + w - 6, ry + 2), fill=FG, width=1) inner_x = x + 6 inner_w = w - 12 # HEAD: Tag | Name (GROSS) | Temp (RIESIG) tag_w = 30 draw.rectangle((inner_x, ry + 2, inner_x + tag_w, ry + 2 + head_h - 4), fill=RED) f_tag = _fit_size(draw, s["tag"], tag_w - 4, head_h - 8, candidates=["28", "24", "22", "20"]) tw_, th_ = measure(draw, s["tag"], f_tag) draw.text((inner_x + (tag_w - tw_) // 2, ry + 2 + (head_h - 4 - th_) // 2), s["tag"], font=f_tag, fill=WHITE) # Name gross, fett name_x = inner_x + tag_w + 8 name_w = inner_w - tag_w - 8 - 130 f_name = _fit_size(draw, s["name"], name_w, head_h - 4, candidates=["24", "20", "18", "16"]) text_in_box(draw, s["name"], name_x, ry + 2, name_w, head_h - 4, f_name, FG, "left") # Temp RIESIG, in Temperatur-Farbe t = s["data"].get("Temperature") t_str = f"{t:.1f}°" if t is not None else "—" f_t = _fit_size(draw, t_str, 130, head_h - 4, candidates=["48", "40", "36", "32"]) text_in_box(draw, t_str, inner_x + inner_w - 130, ry + 2, 130, head_h - 4, f_t, temp_color(t), "right") # BAR: CO2-Bar (volle Breite) co2 = s["data"].get("CO2") if co2 is not None and inner_w > 6: bar_y = ry + head_h + 4 draw_mini_bar(draw, inner_x, bar_y, inner_w, bar_h, co2, 2000, color=co2_color(co2)) # FOOT: CO2 Wert (links) + Battery (rechts) — sauber vertikal ausgerichtet foot_y = ry + head_h + 4 + bar_h + 2 if foot_h >= 10 and co2 is not None: co2_str = f"CO₂ {co2} ppm" text_in_box(draw, co2_str, inner_x, foot_y, inner_w // 2, foot_h, fonts.get("10", fonts.get("9")), co2_color(co2), "left") if s.get("battery_pct") is not None: # Battery rechtsbuendig: Icon + Prozent, beide auf gleicher y batt_x = inner_x + inner_w - 30 # Icon-Endposition draw_battery(draw, batt_x, foot_y, s["battery_pct"], w=22, h=8, with_text=False) pct_str = f"{s['battery_pct']}%" text_in_box(draw, pct_str, batt_x - 26, foot_y, 24, foot_h, fonts.get("10", fonts.get("9")), batt_color(s["battery_pct"]), "right") def _draw_forecast_full(self, draw, x, y, w, h, forecast, fonts): text_in_box(draw, "VORHERSAGE", x, y, w, 16, fonts.get("12", fonts.get("11")), ORANGE, "center") daily = forecast.get("daily", {}) days = daily.get("time", [])[:3] if not days: return sec_body_y = y + 22 sec_body_h = h - 22 row_h = sec_body_h // len(days) # Layout: Day+Icon+Rain oben (32%), Min/Max RIESIG (68%) l1_h = max(30, int(row_h * 0.32)) l2_h = row_h - l1_h - 2 for i, day_str in enumerate(days): ry = sec_body_y + i * row_h if i > 0: draw.line((x + 6, ry + 2, x + w - 6, ry + 2), fill=FG, width=1) # Tag (links) try: dt = datetime.fromisoformat(day_str) day_label = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"][dt.weekday()] except Exception: day_label = day_str[:2] text_in_box(draw, day_label.upper(), x + 2, ry, 32, l1_h, fonts.get("18", fonts.get("16")), FG, "left") # Wetter-Icon (rechts) try: i_day = days.index(day_str) code = daily.get("weather_code", [None] * len(days))[i_day] except (IndexError, KeyError, TypeError): code = None icon_size = min(26, l1_h - 4) draw_weather_icon(draw, x + w - icon_size - 2, ry + (l1_h - icon_size) // 2, icon_size, code) # REGEN prominent in der Day-Row: groesseres Drop-Icon + groessere Schrift try: rain_mm = (daily.get("precipitation_sum", [0] * len(days)) or [0] * len(days))[i_day] or 0 except (IndexError, KeyError, TypeError): rain_mm = 0 # Drop-Icon (groesser) drop_w = 12 drop_h = 16 drop_x = x + 36 drop_y = ry + (l1_h - drop_h) // 2 if rain_mm > 0: # Tropfen mit Fuellung in Blau draw.polygon([(drop_x + drop_w // 2, drop_y), (drop_x, drop_y + drop_h - 2), (drop_x + drop_w, drop_y + drop_h - 2)], fill=BLUE) # Highlight (1-2 weisse Pixel oben fuer 3D-Effekt) draw.point((drop_x + drop_w // 2 + 1, drop_y + 4), fill=WHITE) else: # Leerer Tropfen in FG draw.polygon([(drop_x + drop_w // 2, drop_y), (drop_x, drop_y + drop_h - 2), (drop_x + drop_w, drop_y + drop_h - 2)], outline=FG) # Regen-Wert GROSS daneben rain_str = f"{rain_mm:.1f}" f_rain = _fit_size(draw, rain_str, 50, l1_h - 2, candidates=["20", "18", "16", "14"]) text_in_box(draw, rain_str, drop_x + drop_w + 2, ry, 50, l1_h, f_rain, BLUE if rain_mm > 0 else FG, "left") # "mm" Einheit klein dahinter text_in_box(draw, "mm", drop_x + drop_w + 2 + 32, ry, 18, l1_h, fonts.get("10", fonts.get("9")), FG, "left") # Min/Max RIESIG (volle Breite, beide sichtbar) try: tmin = daily["temperature_2m_min"][i_day] tmax = daily["temperature_2m_max"][i_day] except (IndexError, KeyError, TypeError): continue min_str = f"{tmin}°" max_str = f"{tmax}°" f_min = _fit_size(draw, min_str, w - 2, l2_h, candidates=["40", "36", "32", "28"]) f_max = _fit_size(draw, max_str, w - 2, l2_h, candidates=["40", "36", "32", "28"]) text_in_box(draw, min_str, x + 2, ry + l1_h, w // 2 - 4, l2_h, f_min, BLUE, "center") text_in_box(draw, max_str, x + w // 2, ry + l1_h, w // 2 - 4, l2_h, f_max, ORANGE, "center") # ── History (12h Temperatur-Verlauf) ────────────────────────────── def _get_12h_history(self, data: dict) -> list: """Versuche 12h-Temp-Verlauf aus Netatmo-Daten zu extrahieren. Netatmo liefert 'Temp_history' als String '14.0;13.5;...' (3h-Intervalle). Wir nehmen die letzten 12 Werte. """ hist_str = data.get("Temp_history") or data.get("temp_history") if hist_str: try: vals = [float(v) for v in hist_str.split(";") if v] if len(vals) >= 4: return vals[-12:] if len(vals) >= 12 else vals except (ValueError, AttributeError): pass # Fallback: Sinus-basierter Demo-Verlauf cur = data.get("Temperature", 0) or 0 return [cur + 0.5 * math.sin(i / 2.0) for i in range(12)] # ── CO2 Threshold Parser ──────────────────────────────────────── def _co2_thresholds(self) -> list: spec = self.cfg("co2_thresholds", "ok@600,warn@1000,alert@1500") result = [] color_map = {"ok": GREEN, "warn": YELLOW, "alert": RED, "fg": FG, "green": GREEN, "yellow": YELLOW, "red": RED} for p in [s.strip() for s in spec.split(",") if s.strip()]: name, _, val = p.partition("@") try: ppm = float(val) result.append((ppm, color_map.get(name.lower(), GREEN))) except ValueError: continue if not result: return [(600, GREEN), (1000, YELLOW), (1500, RED)] return sorted(result, key=lambda x: x[0])