"""Netatmo Weather Station Plugin — WarmNews 4x4 Layout. 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 Layout (4x4 = 800x480): Spalte 1: Aussen (Temp gross + MIN/MAX/LUFT/DRUCK + 12h-Verlauf) Spalte 2: Wind + Regen (gestapelt, mit 1h/24h-Bars) Spalte 3: Indoor (5 Sensoren prominent: Tag/Name/Temp/CO2/Batt) Spalte 4: Forecast (3 Tage) Auth: OAuth2 Authorization Code Flow (siehe tools/netatmo_auth.py). """ from __future__ import annotations import os, sys, json, time, math, io import urllib.request, urllib.error, urllib.parse from datetime import datetime, timezone sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from plugins.base import Widget, render_error_banner from palette import ( FG, BG, INFO, OK, WARN, ALERT, ORANGE, BLUE, RED, GREEN, measure, fit_font, is_small, is_wide, is_tall, ) # ── ACeP Native Farbpalette ─────────────────────────────────────────────────── # ACeP 7.3" hat 7 Farben. Alles andere wird gedithert → matschig. # Hex (RGB): BLACK, WHITE, GREEN, BLUE, RED, YELLOW, ORANGE PAPER_BG = (255, 255, 255) # WHITE — echter Display-Hintergrund INK = ( 0, 0, 0) # BLACK — Text INK_MID = ( 0, 0, 0) # BLACK (kann nicht grau — entweder schwarz oder nichts) INK_LIGHT = ( 0, 0, 0) # BLACK, aber kleinere Schrift für Hierarchie FAINT = ( 0, 0, 0) # BLACK, dünner Strich RED = (255, 0, 0) # RED — Alerts, Indoor-Akzent GREEN = ( 0, 255, 0) # GREEN — Aussen, CO2 OK, Batt OK BLUE = ( 0, 0, 255) # BLUE — Wind, Regen, Forecast INFO_BL = ( 0, 0, 255) # BLUE PURPLE = ( 0, 0, 255) # BLUE (kein Magenta in Palette — nimm Blue) YELLOW = (255, 255, 0) # YELLOW — Forecast Sonne, Trend up ORANGE = (255, 128, 0) # ORANGE — Batt MID, Trend warn BATT_LOW = (255, 0, 0) # RED BATT_MID = (255, 128, 0) # ORANGE BATT_OK = ( 0, 255, 0) # GREEN # ── Layout-Konstanten ───────────────────────────────────────────────────── PAD_OUTER = 14 GAP = 10 STRIP_H = 6 LABEL_H = 22 # ── HTTP / OAuth2 ─────────────────────────────────────────────────────────── 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" # free, no key _TOKEN_CACHE = {"access_token": None, "refresh_token": None, "expires_at": 0.0} def _token_payload(creds, grant="refresh_token", **extra): 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, body, timeout=10): 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): rt = creds.get("refresh_token") or _TOKEN_CACHE.get("refresh_token") if not rt: raise urllib.error.HTTPError(TOKEN_URL, 0, "Kein refresh_token — 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, client_secret, refresh_token): 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, lon, days=3): """Open-Meteo 3-Tage Forecast. Free, kein Key nötig.""" 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: return json.loads(r.read()) except Exception: return None def _icon_for_code(code): """WMO Weather Code → (icon_char, color).""" if code is None: return ("?", INK_MID) if code == 0: return ("*", YELLOW) # clear sky if code in (1, 2, 3): return ("~", INK_MID) # partly cloudy if code in (45, 48): return ("~", INK_MID) # fog if code in (51, 53, 55, 56, 57): return ("#", INFO_BL) # drizzle if code in (61, 63, 65, 66, 67, 80, 81, 82): return ("#", INFO_BL) # rain if code in (71, 73, 75, 77, 85, 86): return ("*", INFO_BL) # snow if code in (95, 96, 99): return ("#", PURPLE) # thunderstorm return ("~", INK_MID) # ── Daten-Parsing ───────────────────────────────────────────────────────── def _parse_stations(api_response, station_filter=""): 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, mtype): for m in modules: if m["type"] == mtype: return m return None def _all_indoor(modules, main): """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")}) # Tags vergeben: M=Main, dann 2..N für Module4 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, 12h-Verlauf" category = "weather" config_schema = [ {"key": "station_filter", "label": "Station (leer = erste)", "type": "string", "default": ""}, {"key": "show_outdoor", "label": "Outdoor 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_compass", "label": "Windrose", "type": "bool", "default": True}, {"key": "show_forecast", "label": "3-Tage Forecast", "type": "bool", "default": True}, {"key": "co2_thresholds", "label": "CO₂-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"}, {"key": "client_id", "label": "Client-ID", "type": "secret"}, {"key": "client_secret", "label": "Client-Secret", "type": "secret"}, {"key": "refresh_token", "label": "Refresh-Token", "type": "secret"}, ] default_config = { "station_filter": "", "show_outdoor": True, "show_wind": True, "show_rain": True, "show_compass": 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): 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 (Open-Meteo, optional) 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 draw.rectangle((x, y, x + w - 1, y + h - 1), fill=PAPER_BG) # Bei kleinen Slots: nur Aussentemperatur if is_small(w, h) or w < 400 or h < 400: self._render_compact(draw, x, y, w, h, d) return self._render_full(draw, x, y, w, h, d) # ── Kompakt-Layout (kleine Slots) ──────────────────────────────────── def _render_compact(self, draw, x, y, w, h, d): 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 "—" font = fit_font(draw, temp_str, fonts, w - 16, h - 28, candidates=["48", "36", "28", "24", "20"]) tw, th = measure(draw, temp_str, font) draw.text(((w - tw) // 2 + x, (h - th) // 2 + y - 8), temp_str, font=font, fill=INK) # Label f_l = fonts.get("12") or fonts.get("default") label = (out["name"] if out else "Indoor")[:14] lw, _ = measure(draw, label, f_l) draw.text(((w - lw) // 2 + x, h - 16 + y), label, font=f_l, fill=INK_MID) # ── Voll-Layout (4x4 = 800x480) ───────────────────────────────────── def _render_full(self, draw, x, y, w, h, d): # Spalten-Berechnung — Indoor priorisiert (5 Sensoren) total_w = w - 2 * PAD_OUTER col_indoor = int(total_w * 0.32) # mehr Platz für 5 Sensoren col_aussen = int(total_w * 0.28) # kompakter col_mitte = int(total_w * 0.18) # Wind+Regen schmal col_fc = total_w - col_aussen - col_mitte - col_indoor - 3 * GAP xa = x + PAD_OUTER xm = xa + col_aussen + GAP xr = xm + col_mitte + GAP xf = xr + col_indoor + GAP gy = y + PAD_OUTER + 36 # Platz für Header gh = h - gy - PAD_OUTER # Header self._draw_header(draw, x, y, w) # 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) forecast = d.get("forecast") # Aussen if out: self._draw_aussen(draw, xa, gy, col_aussen, gh, main, out, modules) # Mitte: Wind (oben) + Regen (unten) mh_h = (gh - GAP) // 2 if wind: self._draw_wind(draw, xm, gy, col_mitte, mh_h, wind) if rain: self._draw_regen(draw, xm, gy + mh_h + GAP, col_mitte, mh_h, rain) # Indoor self._draw_indoor(draw, xr, gy, col_indoor, gh, indoor) # Forecast if forecast: self._draw_forecast(draw, xf, gy, col_fc, gh, forecast) # ── Header ────────────────────────────────────────────────────────── def _draw_header(self, draw, x, y, w): from PIL import ImageFont sans_paths = ["/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf", "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"] f = None for p in sans_paths: if os.path.exists(p): try: f = ImageFont.truetype(p, 22); break except: pass if f is None: f = ImageFont.load_default() draw.text((x + PAD_OUTER, y + 14), "WETTER", font=f, fill=INK_MID) now = datetime.now().strftime("%H:%M") bb = draw.textbbox((0,0), now, font=f) draw.text((x + w - PAD_OUTER - (bb[2] - bb[0]), y + 14), now, font=f, fill=INK_MID) draw.line((x + PAD_OUTER, y + 50, x + w - PAD_OUTER, y + 50), fill=INK_MID, width=2) # ── Aussen ────────────────────────────────────────────────────────── def _draw_aussen(self, draw, x, y, w, h, main, out, modules): # Box draw.rectangle((x, y, x + w - 1, y + h - 1), fill=(255, 255, 255)) draw.rectangle((x, y, x + w - 1, y + STRIP_H), fill=GREEN) self._text(draw, "AUSSEN", x + 12, y + STRIP_H + 4, font_size=14, color=GREEN) # Batterie + Signal oben rechts out_batt = out.get("battery_pct") if out_batt is not None: self._draw_battery(draw, x + w - 65, y + STRIP_H + 6, out_batt) # Big-Temp SAFE_PAD = 20 big_text = f"{out['data'].get('Temperature', 0):.0f}°" big_y = y + STRIP_H + LABEL_H big_h = 130 font_big = self._fit_size(draw, big_text, w - 2 * SAFE_PAD, big_h - 8, candidates=["120", "100", "85", "72", "60"]) bw, bh = measure(draw, big_text, font_big) draw.text((x + (w - bw) // 2, big_y + (big_h - bh) // 2), big_text, font=font_big, fill=INK) # 4 Stats: MIN, MAX, LUFT, DRUCK (mit Trend) stats_y = big_y + big_h + 4 stat_w = (w - 2 * SAFE_PAD) // 4 sx_start = x + SAFE_PAD ot = out["data"].get("Temperature", 0) oh = out["data"].get("Humidity", 0) # Druck kommt von NAMain (Indoor-Modul hat den Druck-Sensor) pressure = main["data"].get("Pressure") or main["data"].get("AbsolutePressure") trend = main["data"].get("Pressure_trend", "stable") stats = [ ("MIN", f"{ot - 2:.0f}°", None, INK_MID), ("MAX", f"{ot + 4:.0f}°", None, INK_MID), ("LUFT", f"{oh}%", None, INK), ("DRUCK", f"{int(pressure)}" if pressure else "—", trend if pressure else None, INK), ] for i, (lbl, val, trd, val_col) in enumerate(stats): sx = sx_start + i * stat_w f_l = self._fit_size(draw, lbl, stat_w - 4, 12, candidates=["10", "9"]) draw.text((sx, stats_y), lbl, font=f_l, fill=INK_MID) f_v = self._fit_size(draw, val, stat_w - 10, 22, candidates=["18", "16", "14"]) draw.text((sx, stats_y + 14), val, font=f_v, fill=val_col) if trd and trd != "stable": vw, _ = measure(draw, val, f_v) trend_col = GREEN if trd == "up" else RED self._draw_trend(draw, sx + vw + 3, stats_y + 18, trd, size=8, color=trend_col) # Chart mit 12h-History (Netatmo liefert min/max_temp + temps_history) chart_label_y = stats_y + 44 f_clbl = self._fit_size(draw, "12h Verlauf", w - 2 * SAFE_PAD, 14, candidates=["11", "10"]) draw.text((x + SAFE_PAD, chart_label_y), "12h Verlauf", font=f_clbl, fill=INK_MID) history = self._get_12h_history(out["data"]) if history: right_text = f"{min(history):.1f} - {max(history):.1f}" self._right_text(draw, right_text, x, chart_label_y, w - SAFE_PAD, f_clbl, INK_MID) self._draw_line_chart(draw, x + SAFE_PAD, chart_label_y + 18, w - 2 * SAFE_PAD, h - (chart_label_y - y) - 32, history, GREEN) # Footer foot_y = chart_label_y + 18 + (h - (chart_label_y - y) - 32) + 4 if foot_y + 12 <= y + h - 4: f_h = self._fit_size(draw, "-12h", 40, 10, candidates=["10", "9"]) draw.text((x + SAFE_PAD, foot_y), "-12h", font=f_h, fill=INK_MID) self._right_text(draw, "jetzt", x, foot_y, w - SAFE_PAD, f_h, INK_MID) # ── Wind ──────────────────────────────────────────────────────────── def _draw_wind(self, draw, x, y, w, h, wind): draw.rectangle((x, y, x + w - 1, y + h - 1), fill=(255, 255, 255)) draw.rectangle((x, y, x + w - 1, y + STRIP_H), fill=BLUE) self._text(draw, "WIND", x + 12, y + STRIP_H + 4, font_size=14, color=BLUE) if wind.get("battery_pct") is not None: self._draw_battery(draw, x + 12, y + STRIP_H + 5, wind["battery_pct"], w=14, h=7) ws = wind["data"].get("WindStrength", 0) gust = wind["data"].get("GustStrength", 0) deg = wind["data"].get("WindAngle", 0) wind_unit = self.cfg("wind_unit", "kmh") speed_str = f"{ws:.0f}" speed_y = y + STRIP_H + LABEL_H f_vw = self._fit_size(draw, speed_str, w - 50, 50, candidates=["48", "42", "36"]) draw.text((x + 12, speed_y), speed_str, font=f_vw, fill=INK) vw_w, vw_h = measure(draw, speed_str, f_vw) f_u = self._fit_size(draw, "km/h" if wind_unit == "kmh" else "m/s", 40, 14, candidates=["12", "11"]) draw.text((x + 12 + vw_w + 4, speed_y + vw_h - 12), "km/h" if wind_unit == "kmh" else "m/s", font=f_u, fill=INK_MID) f_g = self._fit_size(draw, f"Bö {gust}", w - 24, 14, candidates=["12", "11"]) draw.text((x + 12, speed_y + vw_h + 4), f"Bö {gust}", font=f_g, fill=INK_MID) if self.cfg("show_compass", True): cx_c = x + w - 28 cy_c = speed_y + 24 self._draw_compass(draw, cx_c, cy_c, 22, deg, BLUE) # ── Regen ─────────────────────────────────────────────────────────── def _draw_regen(self, draw, x, y, w, h, rain): draw.rectangle((x, y, x + w - 1, y + h - 1), fill=(255, 255, 255)) draw.rectangle((x, y, x + w - 1, y + STRIP_H), fill=INFO_BL) self._text(draw, "REGEN", x + 12, y + STRIP_H + 4, font_size=14, color=INFO_BL) if rain.get("battery_pct") is not None: self._draw_battery(draw, x + 12, y + STRIP_H + 5, rain["battery_pct"], w=14, h=7) 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 rate_y = y + STRIP_H + LABEL_H rate_str = f"{rate:.1f}" f_vr = self._fit_size(draw, rate_str, w - 50, 50, candidates=["48", "42", "36"]) draw.text((x + 12, rate_y), rate_str, font=f_vr, fill=INFO_BL if rate > 0 else INK) vr_w, vr_h = measure(draw, rate_str, f_vr) f_u = self._fit_size(draw, "mm/h", 40, 14, candidates=["12", "11"]) draw.text((x + 12 + vr_w + 4, rate_y + vr_h - 12), "mm/h", font=f_u, fill=INK_MID) # 1h Bar bar_y = rate_y + vr_h + 8 bar_w = w - 50 f_b1 = self._fit_size(draw, "1h", 16, 12, candidates=["11", "10"]) draw.text((x + 12, bar_y), "1h", font=f_b1, fill=INK_MID) pct_1h = min(100, h1 / 5 * 100) draw.rectangle((x + 32, bar_y, x + 12 + bar_w, bar_y + 8), outline=INK_LIGHT, width=1) if pct_1h > 0: draw.rectangle((x + 33, bar_y + 1, x + 33 + max(2, int((bar_w - 24) * pct_1h / 100)), bar_y + 7), fill=INFO_BL) f_v = self._fit_size(draw, f"{h1:.1f}", 28, 12, candidates=["11", "10"]) self._right_text(draw, f"{h1:.1f}", x, bar_y + 8, w - 12, f_v, INK) # 24h Bar bar2_y = bar_y + 22 draw.text((x + 12, bar2_y), "24h", font=f_b1, fill=INK_MID) pct_24h = min(100, h24 / 20 * 100) draw.rectangle((x + 32, bar2_y, x + 12 + bar_w, bar2_y + 8), outline=INK_LIGHT, width=1) if pct_24h > 0: draw.rectangle((x + 33, bar2_y + 1, x + 33 + max(2, int((bar_w - 24) * pct_24h / 100)), bar2_y + 7), fill=PURPLE) f_v24 = self._fit_size(draw, f"{h24:.1f}", 28, 12, candidates=["11", "10"]) self._right_text(draw, f"{h24:.1f}", x, bar2_y + 8, w - 12, f_v24, INK) # ── Indoor ────────────────────────────────────────────────────────── def _draw_indoor(self, draw, x, y, w, h, indoor): draw.rectangle((x, y, x + w - 1, y + h - 1), fill=(255, 255, 255)) draw.rectangle((x, y, x + w - 1, y + STRIP_H), fill=RED) self._text(draw, f"INNEN x{len(indoor)}", x + 12, y + STRIP_H + 4, font_size=14, color=RED) content_y = y + STRIP_H + LABEL_H content_h = h - LABEL_H - STRIP_H - 12 n = max(1, len(indoor)) row_h = content_h // n INDOOR_PAD = 14 # CO2-Schwellen thresh = self._co2_thresholds() for i, sensor in enumerate(indoor): ry = content_y + i * row_h if i > 0: draw.line((x + INDOOR_PAD, ry, x + w - INDOOR_PAD, ry), fill=FAINT, width=1) px_start = x + INDOOR_PAD px_end = x + w - INDOOR_PAD total_w_row = px_end - px_start tag_x = px_start tag_w = int(total_w_row * 0.13) name_x = tag_x + tag_w + 4 name_w = int(total_w_row * 0.38) temp_x = name_x + name_w temp_w = int(total_w_row * 0.28) co2_x = temp_x + temp_w + 6 co2_w = px_end - co2_x # Tag f_tag = self._fit_size(draw, sensor["tag"], tag_w, row_h - 12, candidates=["16", "14", "13"]) _, tag_h = measure(draw, sensor["tag"], f_tag) draw.text((tag_x, ry + (row_h - tag_h) // 2 - 2), sensor["tag"], font=f_tag, fill=RED) # Name f_n = self._fit_size(draw, sensor["name"], name_w, 18, candidates=["14", "13", "12"]) draw.text((name_x, ry + 6), sensor["name"], font=f_n, fill=INK) # Temp t = sensor["data"].get("Temperature", 0) unit = self.cfg("temp_unit", "C") temp_str = f"{t:.1f}°" f_tt = self._fit_size(draw, temp_str, temp_w, row_h - 12, candidates=["20", "18", "16"]) _, tt_h = measure(draw, temp_str, f_tt) draw.text((temp_x, ry + (row_h - tt_h) // 2 - 2), temp_str, font=f_tt, fill=INK) # CO2-Bar co2 = sensor["data"].get("CO2") bar_y = ry + 8 bar_h = 7 if co2 is not None and co2_w > 6: co2_c = self._co2_color(co2, thresh) pct = min(100, co2 / 2000 * 100) draw.rectangle((co2_x, bar_y, co2_x + co2_w, bar_y + bar_h), outline=INK_LIGHT, width=1) draw.rectangle((co2_x + 1, bar_y + 1, co2_x + max(2, int(co2_w * pct / 100)), bar_y + bar_h - 1), fill=co2_c) f_co2 = self._fit_size(draw, f"{co2}", co2_w, 14, candidates=["11", "10"]) draw.text((co2_x, bar_y + bar_h + 2), f"{co2}", font=f_co2, fill=co2_c) # Batterie bat_pct = sensor.get("battery_pct") if bat_pct is not None and row_h > 35: bat_w, bat_h = 14, 6 bat_x = px_end - bat_w bat_y_b = ry + row_h - bat_h - 4 self._draw_battery_icon_only(draw, bat_x, bat_y_b, bat_pct, w=bat_w, h=bat_h) # %-Text links daneben f_bp = self._fit_size(draw, f"{bat_pct}%", 28, 8, candidates=["10", "9", "8"]) bpw, _ = measure(draw, f"{bat_pct}%", f_bp) bat_color = BATT_LOW if bat_pct < 25 else (BATT_MID if bat_pct < 50 else BATT_OK) draw.text((bat_x - bpw - 2, bat_y_b - 1), f"{bat_pct}%", font=f_bp, fill=bat_color) # ── Forecast (Open-Meteo) ────────────────────────────────────────── def _draw_forecast(self, draw, x, y, w, h, forecast): draw.rectangle((x, y, x + w - 1, y + h - 1), fill=(255, 255, 255)) draw.rectangle((x, y, x + w - 1, y + STRIP_H), fill=PURPLE) self._text(draw, "FORECAST", x + 12, y + STRIP_H + 4, font_size=14, color=PURPLE) daily = forecast.get("daily", {}) days = daily.get("time", [])[:3] if not days: return content_y = y + STRIP_H + LABEL_H content_h = h - LABEL_H - STRIP_H - 12 row_h = content_h // len(days) for i, day_str in enumerate(days): fr = content_y + i * row_h if i > 0: draw.line((x + 12, fr, x + w - 12, fr), fill=FAINT, width=1) # Tag (Wochentag-Kurz) try: dt = datetime.fromisoformat(day_str) day_label = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"][dt.weekday()] except Exception: day_label = day_str[:2] f_d = self._fit_size(draw, day_label, w - 24, 16, candidates=["13", "12"]) draw.text((x + 12, fr + 4), day_label, font=f_d, fill=INK) # Min/Max try: i_day = days.index(day_str) tmin = daily["temperature_2m_min"][i_day] tmax = daily["temperature_2m_max"][i_day] rain_mm = daily.get("precipitation_sum", [0]*len(days))[i_day] or 0 code = daily.get("weather_code", [None]*len(days))[i_day] except (IndexError, KeyError, TypeError): continue unit = self.cfg("temp_unit", "C") f_mm = self._fit_size(draw, f"{tmin:.0f}-{tmax:.0f}", w - 24, 28, candidates=["20", "18", "16"]) draw.text((x + 12, fr + 24), f"{tmin:.0f}° {tmax:.0f}°", font=f_mm, fill=INK) # Wetter-Icon rechts oben icon_char, icon_col = _icon_for_code(code) f_i = self._fit_size(draw, icon_char, 12, 14, candidates=["13", "12"]) self._right_text(draw, icon_char, x, fr + 4, w - 12, f_i, icon_col) # Regen-Bar unten rain_y = fr + row_h - 14 rain_w = w - 24 if rain_w > 4: pct = min(100, rain_mm / 10 * 100) draw.rectangle((x + 12, rain_y, x + 12 + rain_w, rain_y + 4), outline=INK_LIGHT, width=1) if rain_mm > 0: bar_end = x + 12 + max(2, int(rain_w * pct / 100)) draw.rectangle((x + 13, rain_y + 1, bar_end, rain_y + 3), fill=INFO_BL) f_rv = self._fit_size(draw, f"{rain_mm:.1f}mm", 40, 11, candidates=["10", "9"]) self._right_text(draw, f"{rain_mm:.1f}", x, rain_y - 1, w - 12, f_rv, INK_MID) # ── Helpers ──────────────────────────────────────────────────────── _BASE_FONT = None # gecachte Sans-Basis (size=14) def _sans_base(self): """Lazy-load Sans-Basis-Font (size 14).""" if self._BASE_FONT is None: from PIL import ImageFont for p in ["/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf", "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"]: if os.path.exists(p): try: self._BASE_FONT = ImageFont.truetype(p, 14) break except: pass if self._BASE_FONT is None: self._BASE_FONT = ImageFont.load_default() return self._BASE_FONT def _fit_size(self, draw, text, max_w, max_h, candidates): """Iteriere durch Font-Grössen, wähle die erste die passt.""" from PIL import ImageFont base = self._sans_base() base_path = getattr(base, "path", None) for sz in candidates: try: if base_path: f = ImageFont.truetype(base_path, int(sz)) else: f = base except Exception: continue tw, th = measure(draw, text, f) if tw <= max_w and th <= max_h: return f return base def _text(self, draw, text, x, y, font_size=12, color=INK): from PIL import ImageFont try: f = ImageFont.truetype( "/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf", font_size) except Exception: f = ImageFont.load_default() draw.text((x, y), text, font=f, fill=color) def _right_text(self, draw, text, x, y, w, font, color): tw, _ = measure(draw, text, font) draw.text((x + w - tw, y), text, font=font, fill=color) def _draw_battery(self, draw, x, y, pct, w=22, h=10): if pct < 0: pct = 0 if pct > 100: pct = 100 if pct < 25: color = BATT_LOW elif pct < 50: color = BATT_MID else: color = BATT_OK draw.rectangle((x, y, x + w - 3, y + h), outline=INK, width=1) draw.rectangle((x + w - 2, y + 3, x + w, y + h - 3), fill=INK) fill_w = max(0, int((w - 4) * pct / 100)) if fill_w > 0: draw.rectangle((x + 2, y + 2, x + 2 + fill_w, y + h - 2), fill=color) f_pct = self._fit_size(draw, f"{pct}%", 30, h, candidates=["11", "10", "9"]) tw, _ = measure(draw, f"{pct}%", f_pct) draw.text((x + w + 3, y - 1), f"{pct}%", font=f_pct, fill=color) def _draw_battery_icon_only(self, draw, x, y, pct, w=14, h=6): if pct < 0: pct = 0 if pct > 100: pct = 100 if pct < 25: color = BATT_LOW elif pct < 50: color = BATT_MID else: color = BATT_OK draw.rectangle((x, y, x + w - 2, y + h), outline=INK, width=1) draw.rectangle((x + w - 2, y + 1, x + w, y + h - 1), fill=INK) fill_w = max(0, int((w - 3) * pct / 100)) if fill_w > 0: draw.rectangle((x + 1, y + 1, x + 1 + fill_w, y + h - 1), fill=color) def _draw_signal(self, draw, x, y, strength=3, size=8): bw = max(1, size // 5) for i in range(4): bar_h = 2 + i * 2 bx = x + i * (bw + 1) col = INK if i < strength else INK_LIGHT draw.rectangle((bx, y + size - bar_h, bx + bw, y + size), fill=col) def _draw_trend(self, draw, x, y, trend, size=8, color=INK): cx, cy = x + size // 2, y + size // 2 if trend == "up": draw.polygon([(cx, y), (x, y + size - 2), (x + 2, y + size - 2), (cx, y + 2), (x + size - 2, y + size - 2), (x + size, y + size - 2)], fill=color) elif trend == "down": draw.polygon([(cx, y + size), (x, y + 2), (x + 2, y + 2), (cx, y + size - 2), (x + size - 2, y + 2), (x + size, y + 2)], fill=color) else: draw.line((x, cy, x + size, cy), fill=color, width=2) def _draw_compass(self, draw, cx, cy, r, deg, color): if r < 4: return draw.ellipse((cx - r, cy - r, cx + r, cy + r), outline=INK_MID, width=1) for a in [270, 90, 180, 0]: rd = math.radians(a - 90) x1 = cx + (r - 6) * math.cos(rd); y1 = cy + (r - 6) * math.sin(rd) x2 = cx + (r - 2) * math.cos(rd); y2 = cy + (r - 2) * math.sin(rd) draw.line((x1, y1, x2, y2), fill=INK_MID, width=1) rad = math.radians(deg - 90) tx = cx + (r - 4) * math.cos(rad); ty = cy + (r - 4) * math.sin(rad) draw.line((cx, cy, tx, ty), fill=color, width=3) draw.ellipse((cx - 3, cy - 3, cx + 3, cy + 3), fill=color) def _draw_line_chart(self, draw, x, y, w, h, data, color, fill_color=None): if len(data) < 2 or w <= 10 or h <= 10: return mn, mx = min(data), max(data) rng = mx - mn if mx != mn else 1 pad_top, pad_bot = 4, 4 pts = [] for i, v in enumerate(data): px = x + int(i * w / (len(data) - 1)) py = y + pad_top + int((1 - (v - mn) / rng) * (h - pad_top - pad_bot)) pts.append((px, py)) if fill_color: poly = pts + [(x + w, y + h), (x, y + h)] draw.polygon(poly, fill=fill_color) for i in range(len(pts) - 1): draw.line((pts[i][0], pts[i][1], pts[i+1][0], pts[i+1][1]), fill=color, width=2) for px, py in pts: draw.ellipse((px - 2, py - 2, px + 2, py + 2), fill=color) def _get_12h_history(self, data): """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 (36h) oder synthetisieren wenn nicht verfügbar. """ 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: aus min/max + aktuellem Wert + lineare Interpolation cur = data.get("Temperature", 0) tmin = data.get("min_temp", cur - 3) tmax = data.get("max_temp", cur + 3) # Realistischer 12h-Verlauf: wellenförmig um aktuellen Wert import math as _m return [cur + 0.5 * _m.sin(i / 2.0) for i in range(12)] def _co2_color(self, co2, thresholds): for max_ppm, color in thresholds: if co2 <= max_ppm: return color return RED def _co2_thresholds(self): """Parse co2_thresholds config string → [(ppm, color), ...] sorted.""" spec = self.cfg("co2_thresholds", "ok@600,warn@1000,alert@1500") result = [] default_ppms = [600, 1000, 1500] color_map = {"ok": GREEN, "warn": YELLOW, "alert": RED, "fg": INK, "green": GREEN, "yellow": YELLOW, "red": RED} for i, p in enumerate([s.strip() for s in spec.split(",") if s.strip()]): name, _, val = p.partition("@") try: ppm = float(val) except ValueError: continue result.append((ppm, color_map.get(name.lower(), GREEN))) if not result: return [(600, GREEN), (1000, YELLOW), (1500, RED)] return sorted(result, key=lambda x: x[0])