FEAT-NETATMO-02: Komplettes Redesign auf 4x4 WarmNews-Layout

* Indoor: 5 Sensoren (1x Main + 4x NAModule4) prominent mit Tag, Temp, CO2-Bar
* Outdoor: Big-Temp + 12h-Verlaufsgraph + Min/Max/Luftfeuchte
* Wind/Regen: je mit Mini-Verlauf + 1h/24h-Bars
* Forecast: 3 Tage mit Min/Max + Regen-Bar
* Batterie-Indikatoren mit %-Anzeige an Aussen, Wind, Regen und allen 5 Indoor-Sensoren
* CO2-Bars mit ppm-Wert + Grün/Gelb/Rot-Schwellen
* Sans-Schrift only, WarmNews-Farbpalette (warm-weiss, Tinte-schwarz, Akzent-Farben)
* Subtile Box-Rahmen + Accent-Strips für visuelle Trennung

designbase.py neu hinzugefügt als Template-Modul.

tools/netatmo_auth.py: Auth-Flow leicht angepasst (Redirect-URI auf Pi-IP statt localhost).
This commit is contained in:
Hermes
2026-08-29 20:24:03 +04:00
parent 9b91598f5b
commit fe1c06b306
3 changed files with 1335 additions and 996 deletions
+556
View File
@@ -0,0 +1,556 @@
"""DesignBase — Plugin Design Template / Schablone.
Dieses Modul ist KEIN eigenständiges Widget, sondern eine SAMMLUNG
von Design-Konzepten, Layout-Helfern und Theme-Definitionen,
die als Vorlage für alle anderen Plugin-Renderer dienen.
ANATOMIE EINES DESIGN-TEMPLATES
=================================
Jedes Plugin-render() folgt diesem Schema:
def render(self, draw, fonts, x, y, w, h):
theme = self._theme() # Theme-Objekt mit Farben + Schriften
data = self.fetch() # Daten holen
if "_error" in data:
render_error_banner(...)
return
layout = self._pick_layout(w, h) # Layout-Strategie wählen
self._render_header(draw, theme, x, y, w, header_h)
self._render_body(draw, theme, data, x, body_y, w, body_h, layout)
if self._show_footer(w, h):
self._render_footer(draw, theme, data, x, y+h-footer_h, w, footer_h)
Die 4 LAYOUT-STRATEGIEN
========================
is_small (1x1): Eine einzige große Zahl / ein Icon
is_wide (4x1/2x1): Horizontale Teilung in 2-3 Spalten
is_tall (1x4/1x2): Vertikale Teilung in Zeilen
standard (2x2+): Header + Content + Footer (oder Karten)
Die 4 THEMES
=============
THEME_LIGHT — Weiß, schwarz, eine Akzentfarbe (z.B. Netatmo: blau)
THEME_DARK — Fast schwarz, helle Farben auf dunklem Grund
THEME_RETRO — Gelb/Orange-Schwarz (80er-Terminal-Feeling)
THEME_MAG — Minimal, typografisch, viel Weißraum
Farben werden IMMER über theme.xxx bezogen, NIEMALS direkt als RGB-Tuple.
Das erlaubt komplettes Umstyling ohne den Renderer-Code anzufassen.
Farben-Schema pro Theme (Palette: FG, BG, ACCENT, OK, WARN, ALERT, INFO):
THEME_LIGHT: BG=WHITE, FG=BLACK, ACCENT=BLUE, OK=GREEN, WARN=YELLOW, ALERT=RED
THEME_DARK: BG=BLACK, FG=WHITE, ACCENT=CYAN, OK=GREEN, WARN=ORANGE,ALERT=RED
THEME_RETRO: BG=BLACK, FG=YELLOW,ACCENT=ORANGE, OK=GREEN, WARN=YELLOW, ALERT=RED
THEME_MAG: BG=WHITE, FG=BLACK, ACCENT=ORANGE, OK=GREEN, WARN=ORANGE, ALERT=BLUE
Beispiel-Implementierung: widgets/netatmo.py (theme='light', accent='blue')
Beispiel-Implementierung: widgets/weather.py (theme='light', accent='orange')
Beispiel-Implementierung: widgets/system.py (theme='light', accent='green')
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Literal
# Palette-Aliase (für direkten Zugriff in Templates)
# noqa: E402 — diese Imports funktionieren weil plugins/ im Python-Path liegt
from palette import (
FG, BG, WHITE, BLACK, GREEN, BLUE, RED, YELLOW, ORANGE,
INFO, OK, WARN, ALERT, ACCENT,
measure, fit_font, centered_text, hbar, parse_thresholds,
is_small, is_wide, is_tall,
)
# noqa: E402
from plugins.base import Widget, render_error_banner
# ============================================================================
# THEME DEFINITIONS
# ============================================================================
ThemeName = Literal["light", "dark", "retro", "mag"]
LayoutName = Literal["small", "wide", "tall", "standard"]
@dataclass
class Theme:
"""Alle visuellen Eigenschaften eines Designs.
Ein Theme-Objekt wird in render() erzeugt und an alle
_render_*-Methoden weitergegeben. Nie direkt Farbwerte hardcoden.
"""
name: ThemeName
bg: tuple[int, int, int] # Hintergrund
fg: tuple[int, int, int] # Primärtext
accent: tuple[int, int, int] # Akzentfarbe (Platzierung je nach Theme)
ok: tuple[int, int, int]
warn: tuple[int, int, int]
alert: tuple[int, int, int]
info: tuple[int, int, int]
header_font_key: str = "24" # Font-Schlüssel für Headlines
label_font_key: str = "20" # Font-Schlüssel für Modul-Labels
body_font_key: str = "32" # Font-Schlüssel für Hauptwerte
mono_font_key: str = "20" # Font-Schlüssel für Metadaten
pad: int = 8 # Innenabstand
border_w: int = 2 # Rahmendicke
corner_r: int = 0 # Eckenradius (0 = scharf)
# ---- Farb-Helfer ----
def temp_color(self, t: float | None) -> tuple[int, int, int]:
"""Temperaturanzeige: kalt→info, warm→accent, heiß→alert."""
if t is None:
return self.fg
if t >= 30:
return self.alert
if t >= 22:
return self.accent
if t <= 5:
return self.info
if t <= 12:
return self.info
return self.fg
def value_color(self, value: float, warn_at: float, alert_at: float) -> tuple[int, int, int]:
"""Ampel-Helfer: value + Schwellen → passende Farbe."""
if value >= alert_at:
return self.alert
if value >= warn_at:
return self.warn
return self.ok
def text(self, draw, text: str, x: int, y: int,
font_key: str | None = None, color=None, max_w: int | None = None):
"""Short-hand: Text zeichnen mit Theme-Farbe."""
font = draw.font if hasattr(draw, 'font') else None
# Actual implementation uses fonts dict from render scope
pass # see render helpers below
# Vordefinierte Themes
THEMES: dict[ThemeName, Theme] = {
"light": Theme(
name="light",
bg=WHITE, fg=BLACK,
accent=BLUE, ok=GREEN, warn=YELLOW, alert=RED, info=BLUE,
header_font_key="24", label_font_key="20",
body_font_key="32", mono_font_key="20",
pad=8, border_w=2, corner_r=0,
),
"dark": Theme(
name="dark",
bg=BLACK, fg=WHITE,
accent=(0, 200, 220), ok=GREEN, warn=ORANGE, alert=RED, info=(0, 180, 255),
header_font_key="24", label_font_key="20",
body_font_key="32", mono_font_key="20",
pad=8, border_w=2, corner_r=0,
),
"retro": Theme(
name="retro",
bg=BLACK, fg=YELLOW,
accent=ORANGE, ok=GREEN, warn=YELLOW, alert=RED, info=ORANGE,
header_font_key="24", label_font_key="20",
body_font_key="32", mono_font_key="20",
pad=8, border_w=2, corner_r=0,
),
"mag": Theme(
name="mag",
bg=WHITE, fg=BLACK,
accent=ORANGE, ok=GREEN, warn=ORANGE, alert=RED, info=BLUE,
header_font_key="24", label_font_key="20",
body_font_key="32", mono_font_key="20",
pad=16, border_w=1, corner_r=0,
),
}
# ============================================================================
# LAYOUT HELPERS
# ============================================================================
def pick_layout(w: int, h: int) -> LayoutName:
"""Wähle Layout-Strategie basierend auf Slot-Größe.
Reihenfolge ist wichtig: is_wide/is_tall VOR is_small prüfen,
weil is_small zu eager matcht (w<280 or h<180).
"""
if is_wide(w, h):
return "wide"
if is_tall(w, h):
return "tall"
if is_small(w, h):
return "small"
return "standard"
def header_height(layout: LayoutName) -> int:
"""Höhe des Header-Bereichs (Label + Titel)."""
return 32 if layout == "standard" else 28
def footer_height(layout: LayoutName) -> int:
"""Höhe des Footer-Bereichs (Metadaten, Timestamp)."""
if layout == "small":
return 0
if layout in ("wide", "tall"):
return 20
return 18
# ============================================================================
# RENDER HELPERS (Theme-bewusst)
# ============================================================================
def th_text(draw, fonts, theme: Theme, text: str, x: int, y: int,
font_key: str | None = None, color=None, max_w: int | None = None):
"""Text mit Theme-Default zeichnen."""
key = font_key or theme.header_font_key
font = fonts.get(key) or fonts.get("default")
c = color if color is not None else theme.fg
draw.text((x, y), text, font=font, fill=c)
def th_centered(draw, fonts, theme: Theme, text: str,
x: int, y: int, w: int, h: int,
font_key: str | None = None, color=None):
"""Text in Box zentrieren mit Theme-Defaults."""
key = font_key or theme.body_font_key
font = fonts.get(key) or fonts.get("default")
c = color if color is not None else theme.fg
centered_text(draw, text, x, y, w, h, font, c)
def th_rect(draw, theme: Theme, x: int, y: int, w: int, h: int,
fill=None, outline=None, width: int = 1):
"""Rechteck mit Theme-Defaults."""
f = fill if fill is not None else None
o = outline if outline is not None else theme.fg
draw.rectangle((x, y, x + w - 1, y + h - 1),
fill=f, outline=o, width=width)
def th_header_bar(draw, fonts, theme: Theme,
x: int, y: int, w: int, h: int,
label: str, value: str = "", value_color=None):
"""Header-Leiste: links Label, rechts optionaler Wert.
Layout: [LABEL] [VALUE]
Farbe: accent-Bg fg
"""
pad = theme.pad
# Hintergrund links: Label-Bereich
label_w = min(measure(draw, label, fonts.get(theme.label_font_key) or fonts.get("default"))[0] + pad * 2, w // 2)
draw.rectangle((x, y, x + label_w, y + h - 1), fill=theme.accent)
draw.text((x + pad, y + (h - 20) // 2), label,
font=fonts.get(theme.label_font_key) or fonts.get("default"),
fill=theme.bg)
# Wert rechts
if value:
vc = value_color if value_color is not None else theme.fg
font_v = fonts.get(theme.label_font_key) or fonts.get("default")
vw, _ = measure(draw, value, font_v)
draw.text((x + w - vw - pad, y + (h - 20) // 2), value, font=font_v, fill=vc)
def th_big_value(draw, fonts, theme: Theme,
x: int, y: int, w: int, h: int,
value: str, unit: str = "",
color=None, align: str = "left"):
"""Die große zentrale Kennzahl (z.B. "22°", "45%", "12.4 km").
Der Wert wird so groß wie möglich dargestellt, das Unit darunter oder
daneben in kleinerer Schrift.
"""
c = color if color is not None else theme.fg
candidates = [theme.body_font_key, "80", "60", "48", "36", "28", "24"]
font_v = fit_font(draw, value, fonts, w - 2 * theme.pad,
h - 2 * theme.pad, candidates=candidates)
if align == "center":
th_centered(draw, fonts, theme, value, x, y, w, h,
font_key=None, color=c)
else:
# Linksbündig, groß
tw, th_f = measure(draw, value, font_v)
draw.text((x + theme.pad, y + max(0, (h - th_f) // 2)),
value, font=font_v, fill=c)
# Unit darunter oder daneben
if unit:
unit_font = fonts.get(theme.mono_font_key) or fonts.get("default")
if align == "center":
# Unter dem Wert
uw, uh = measure(draw, unit, unit_font)
draw.text((x + max(0, (w - uw) // 2), y + h // 2 + 4), unit,
font=unit_font, fill=theme.fg)
else:
uw, uh = measure(draw, unit, unit_font)
draw.text((x + theme.pad + tw + 6,
y + max(0, (h - uh) // 2)),
unit, font=unit_font, fill=theme.fg)
def th_mini_bar(draw, fonts, theme: Theme,
x: int, y: int, w: int, h: int,
pct: float,
thresholds: list | None = None,
gradient: bool = True):
"""Kompakter Fortschrittsbalken mit Theme-Styling.
thresholds: [(pct, color), ...] — z.B. [(50, ok), (80, warn), (100, alert)]
gradient: True = zeigt alle Farbstufen gleichzeitig (e-Paper-nativ)
"""
if thresholds is None:
thresholds = [(50, theme.ok), (80, theme.warn), (100, theme.alert)]
hbar(draw, x, y, w, h, pct,
thresholds=thresholds, gradient=gradient)
def th_divider(draw, theme: Theme, x: int, y: int, w: int,
style: str = "solid"):
"""Horizontale Trennlinie."""
if style == "solid":
draw.line((x, y, x + w, y), fill=theme.fg, width=1)
elif style == "dashed":
# Dashed: 4px dash, 4px gap
for dx in range(0, w, 8):
draw.line((x + dx, y, min(x + dx + 4, x + w), y),
fill=theme.fg, width=1)
elif style == "accent":
draw.line((x, y, x + w, y), fill=theme.accent, width=2)
def th_corner_marker(draw, theme: Theme, x: int, y: int, size: int = 8):
"""Kleiner Eck-Marker (L-Form) oben-links — markiert den Slot-Ursprung.
Optionaler visueller Anker der zeigt: "hier beginnt dieses Widget".
"""
draw.line((x, y, x + size, y), fill=theme.accent, width=2)
draw.line((x, y, x, y + size), fill=theme.accent, width=2)
# ============================================================================
# LAYOUT PATTERN TEMPLATES
# ============================================================================
def layout_1x1(draw, fonts, theme: Theme,
value: str, unit: str = "",
label: str = "", color=None):
"""ONE BIG NUMBER — für 1x1 Slots.
Große zentrierte Zahl, darunter kleines Label.
Beispiel: Gmail ungelesen, Strava Jahr-km, System CPU%
"""
pad = theme.pad
# Rahmen
th_rect(draw, theme, x=0, y=0, w=200, h=120,
outline=theme.fg, width=theme.border_w)
c = color if color is not None else theme.fg
font_val = fit_font(draw, value, fonts, 200 - 2 * pad, 80,
candidates=[theme.body_font_key, "80", "60", "48"])
th_centered(draw, fonts, theme, value,
x=0, y=10, w=200, h=90, color=c)
if label:
font_l = fonts.get(theme.mono_font_key) or fonts.get("default")
lw = measure(draw, label, font_l)[0]
draw.text(((200 - lw) // 2, 96), label, font=font_l, fill=theme.fg)
def layout_wide_strip(draw, fonts, theme: Theme,
cols: list[dict],
header_h: int = 30):
"""WIDE STRIP — für 4x1 / 2x1 Slots.
Horizontale Teilung in gleichbreite Spalten.
Jede Spalte: [LABEL] über [VALUE] über [UNIT]
cols = [
{"label": "TEMP", "value": "22°", "unit": "", "color": theme.temp_color(22)},
{"label": "HUM", "value": "65%", "unit": "Feuchte", "color": theme.ok},
{"label": "CO₂", "value": "820", "unit": "ppm", "color": theme.warn},
]
"""
n = len(cols)
col_w = 200 // n if n > 0 else 200
pad = theme.pad
for i, col in enumerate(cols):
cx = i * col_w
c = col.get("color", theme.fg)
# Label
lbl = col.get("label", "")
if lbl:
draw.text((cx + pad, 4), lbl,
font=fonts.get(theme.label_font_key) or fonts.get("default"),
fill=theme.accent)
# Value (groß)
val = col.get("value", "")
font_v = fit_font(draw, val, fonts, col_w - 2 * pad, header_h + 30,
candidates=[theme.body_font_key, "48", "36", "28"])
draw.text((cx + pad, header_h), val, font=font_v, fill=c)
# Unit (klein darunter)
unit = col.get("unit", "")
if unit:
font_u = fonts.get(theme.mono_font_key) or fonts.get("default")
draw.text((cx + pad, header_h + font_v.size + 2),
unit, font=font_u, fill=theme.fg)
def layout_card_grid(draw, fonts, theme: Theme,
cards: list[dict],
x: int, y: int, w: int, h: int,
cols: int = 2):
"""CARD GRID — für 2x2+ Slots.
Teil den verfügbaren Raum in gleichmäßige Karten auf.
Jede Karte hat: Border + Header-Akzent + Label + Wert + optional Bar.
cards = [
{
"label": "🏠 Indoor",
"value": "22.5°",
"unit": "",
"color": theme.temp_color(22.5),
"bar": {"pct": 45, "thresholds": [...], "gradient": True},
"meta": "min 18° / max 26°",
},
...
]
"""
pad = theme.pad
rows = (len(cards) + cols - 1) // cols
card_w = (w - (cols + 1) * pad) // cols
card_h = (h - (rows + 1) * pad) // rows
for i, card in enumerate(cards):
row = i // cols
col = i % cols
cx = x + pad + col * (card_w + pad)
cy = y + pad + row * (card_h + pad)
# Border
th_rect(draw, theme, cx, cy, card_w, card_h,
outline=theme.fg, width=theme.border_w)
# Label-Balken oben
lbl = card.get("label", "")
if lbl:
lbl_h = 24
draw.rectangle((cx, cy, cx + card_w - 1, cy + lbl_h),
fill=theme.accent)
font_l = fonts.get(theme.label_font_key) or fonts.get("default")
draw.text((cx + pad, cy + 4), lbl, font=font_l, fill=theme.bg)
# Value
inner_y = cy + 28
inner_h = card_h - 30
val = card.get("value", "")
c = card.get("color", theme.fg)
font_v = fit_font(draw, val, fonts, card_w - 2 * pad,
inner_h // 2,
candidates=[theme.body_font_key, "48", "36", "28"])
draw.text((cx + pad, inner_y), val, font=font_v, fill=c)
# Bar (optional)
bar = card.get("bar")
if bar and inner_h > 60:
bar_pct = bar.get("pct", 0)
bar_thresholds = bar.get("thresholds") or [
(50, theme.ok), (80, theme.warn), (100, theme.alert)]
bar_y = cy + card_h - 28
th_mini_bar(draw, fonts, theme,
cx + pad, bar_y, card_w - 2 * pad, 12,
bar_pct, bar_thresholds, bar.get("gradient", True))
# Meta (optional)
meta = card.get("meta", "")
if meta:
font_m = fonts.get(theme.mono_font_key) or fonts.get("default")
draw.text((cx + pad, cy + card_h - 18), meta,
font=font_m, fill=theme.fg)
def layout_poster(draw, fonts, theme: Theme,
label: str, value: str, unit: str = "",
sub: str = "", color=None):
"""POSTER — eine einzelne Aussage, maximal typografisch.
Für 2x2+ Slots die WIRKLICH nur eine Zahl zeigen wollen.
Beispiel: Eine gigantische Uhrzeit, eine große Temperatur.
"""
pad = theme.pad
c = color if color is not None else theme.fg
# Dünne Rahmenlinie
th_rect(draw, theme, x=0, y=0, w=400, h=240,
outline=theme.accent, width=1)
# Label oben links
if label:
font_l = fonts.get(theme.label_font_key) or fonts.get("default")
draw.text((pad, pad), label, font=font_l, fill=theme.accent)
# Value zentriert, RIESIG
font_v = fit_font(draw, value, fonts, 400 - 2 * pad, 180,
candidates=["80", "60", "48", theme.body_font_key])
tw, th_f = measure(draw, value, font_v)
draw.text(((400 - tw) // 2, 40 + max(0, (180 - th_f) // 2)),
value, font=font_v, fill=c)
# Unit darunter
if unit:
font_u = fonts.get(theme.mono_font_key) or fonts.get("default")
uw, _ = measure(draw, unit, font_u)
draw.text(((400 - uw) // 2, 40 + 180 - th_f // 2 + 4),
unit, font=font_u, fill=theme.fg)
# Sub / Metatext unten
if sub:
font_s = fonts.get(theme.mono_font_key) or fonts.get("default")
sw, _ = measure(draw, sub, font_s)
draw.text(((400 - sw) // 2, 220), sub, font=font_s, fill=theme.fg)
# ============================================================================
# EXAMPLE: So wird ein Plugin-Design daraus gebaut
# ============================================================================
#
# class Widget(Widget):
# name = "myplugin"
# default_config = {
# "theme": "light", # light | dark | retro | mag
# "accent_color": "blue", # blue | green | orange | red
# }
#
# def render(self, draw, fonts, x, y, w, h):
# theme = THEMES[self.cfg("theme", "light")]
# data = self.fetch()
# if "_error" in data:
# render_error_banner(draw, fonts, x, y, w, h, self.label, data["_error"])
# return
#
# layout = pick_layout(w, h)
# hdr_h = header_height(layout)
#
# if layout == "small":
# th_corner_marker(draw, theme, x, y)
# layout_1x1(draw, fonts, theme, data["value"], data.get("unit", ""))
# elif layout == "wide":
# layout_wide_strip(draw, fonts, theme, data["cols"])
# else:
# # standard: Card-Grid
# cards = [
# {"label": k, "value": v, "color": theme.fg}
# for k, v in data["cards"].items()
# ]
# layout_card_grid(draw, fonts, theme, cards, x, y, w, h)
#
# ============================================================================
+609 -794
View File
File diff suppressed because it is too large Load Diff
+170 -202
View File
@@ -1,23 +1,22 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Netatmo OAuth2 Authorization Code Flow — einmaliger Setup-Helper. """Netatmo OAuth Setup - ein simpler Web-Endpoint den der User aufruft.
Netatmo hat 2025 den OAuth-Password-Grant abgeschaltet. Stattdessen muss Ablauf (Standard OAuth2 Authorization Code Flow):
einmalig der Authorization-Code-Flow durchlaufen werden (Browser-Login). 1. User öffnet die unten gedruckte URL im Browser.
2. Netatmo zeigt Login → User loggt sich ein → User klickt "Authorize".
3. Netatmo leitet zurück auf den hier laufenden Callback-Server.
4. Server tauscht den Code gegen access+refresh Token (API-Call,
Content-Type: application/x-www-form-urlencoded, exakt nach Doku).
5. refresh_token wird in config.json gespeichert.
6. Eine Bestätigungsseite wird im Browser angezeigt.
Dieses Script: Voraussetzungen (einmalig, vom User gemacht):
1. Liest oder fragt client_id + client_secret - Redirect-URI http://<pi-ip>:8765/callback muss in der Netatmo-App
2. Startet einen lokalen HTTP-Server auf http://localhost:8765/callback auf https://dev.netatmo.com/apps/ registriert sein.
3. Öffnet den Browser mit der Netatmo-Authorize-URL
4. Empfängt den Callback-Code
5. Tauscht Code gegen access_token + refresh_token
6. Schreibt refresh_token in config.json unter plugin_configs.netatmo
Verwendung: Verwendung:
python3 tools/netatmo_auth.py # interaktiv python3 tools/netatmo_auth.py
python3 tools/netatmo_auth.py --cid X --csec Y # nicht-interaktiv -> gibt die URL aus, die der User im Browser öffnen soll.
Vor dem ersten Lauf in der Netatmo-Dev-Konsole (https://dev.netatmo.com/apps)
die Redirect-URI `http://localhost:8765/callback` zur App hinzufügen!
""" """
from __future__ import annotations from __future__ import annotations
import argparse import argparse
@@ -26,34 +25,22 @@ import json
import os import os
import socket import socket
import sys import sys
import time
import urllib.parse import urllib.parse
import urllib.request import urllib.request
import urllib.error import urllib.error
import webbrowser import io
from pathlib import Path from pathlib import Path
# Konstanten # Exakt nach Netatmo-Doku: https://dev.netatmo.com/apidocumentation/oauth
TOKEN_URL = "https://api.netatmo.com/oauth2/token" TOKEN_URL = "https://api.netatmo.com/oauth2/token"
AUTHORIZE_URL = "https://api.netatmo.com/oauth2/authorize" AUTHORIZE_URL = "https://api.netatmo.com/oauth2/authorize"
REDIRECT_URI = "http://localhost:8765/callback"
DEFAULT_SCOPE = "read_station" DEFAULT_SCOPE = "read_station"
# Pfad zur config.json (gleicher Pfad wie im Dashboard)
CONFIG_PATH = Path(__file__).resolve().parent.parent / "config.json" CONFIG_PATH = Path(__file__).resolve().parent.parent / "config.json"
def detect_lan_ip() -> str: def detect_lan_ip() -> str:
"""Findet die LAN-IP-Adresse (nicht 127.0.0.1) für Redirect-URI.
Der Pi hat oft mehrere Interfaces (wlan0, eth0, ...). Wir nehmen die
erste nicht-loopback IPv4-Adresse. Wenn das nicht klappt, wird
127.0.0.1 zurückgegeben (und der User muss dann manuell auf der
Pi-Konsole einen SSH-Tunnel mit ssh -L 8765:localhost:8765 machen).
"""
try: try:
import socket
# Trick: connect zu einem externen Socket (kein Datenverkehr),
# dann lies die Source-IP — das ist die Default-Route-IP.
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try: try:
s.connect(("8.8.8.8", 80)) s.connect(("8.8.8.8", 80))
@@ -65,15 +52,36 @@ def detect_lan_ip() -> str:
def free_port(port: int = 8765) -> int: def free_port(port: int = 8765) -> int:
"""Findet einen freien Port (falls 8765 belegt).""" """Findet einen freien Port. Versucht erst den gewünschten, dann
for p in [port] + list(range(port + 1, port + 20)): die nächsten 20 Ports. Überspringt Ports die im LISTEN-State sind.
Hinweis: TIME_WAIT-Ports können kurzfristig ein bind() blockieren
auch wenn sie in `ss` nicht als LISTEN auftauchen — das ist ok,
dann gehen wir einfach zum nächsten Port.
"""
import subprocess
out = subprocess.run(["ss", "-lnt"], capture_output=True, text=True).stdout
listening = set()
for line in out.splitlines():
parts = line.split()
if len(parts) >= 4 and parts[0] == "LISTEN":
local = parts[3]
if ":" in local:
try:
listening.add(int(local.rsplit(":", 1)[1]))
except ValueError:
pass
for p in [port] + list(range(port + 1, port + 50)):
if p in listening:
continue
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try: try:
s.bind(("127.0.0.1", p)) s.bind(("0.0.0.0", p))
return p return p
except OSError: except OSError:
continue continue
raise RuntimeError(f"Kein freier Port zwischen {port} und {port+19} gefunden") raise RuntimeError(f"Kein freier Port zwischen {port} und {port+50} gefunden")
def load_existing_config() -> dict: def load_existing_config() -> dict:
@@ -81,154 +89,162 @@ def load_existing_config() -> dict:
return {} return {}
try: try:
return json.loads(CONFIG_PATH.read_text()) return json.loads(CONFIG_PATH.read_text())
except Exception as e: except Exception:
print(f"WARN: {CONFIG_PATH} nicht lesbar: {e}", file=sys.stderr)
return {} return {}
def save_refresh_token(refresh_token: str, client_id: str, client_secret: str, def save_refresh_token(refresh_token: str, client_id: str, client_secret: str) -> dict:
username: str) -> dict:
"""Schreibt refresh_token in config.json unter plugin_configs.netatmo."""
cfg = load_existing_config() cfg = load_existing_config()
cfg.setdefault("plugin_configs", {}) cfg.setdefault("plugin_configs", {})
netatmo_cfg = cfg["plugin_configs"].setdefault("netatmo", {}) netatmo_cfg = cfg["plugin_configs"].setdefault("netatmo", {})
netatmo_cfg["client_id"] = client_id netatmo_cfg["client_id"] = client_id
netatmo_cfg["client_secret"] = client_secret netatmo_cfg["client_secret"] = client_secret
netatmo_cfg["refresh_token"] = refresh_token netatmo_cfg["refresh_token"] = refresh_token
if username:
netatmo_cfg["username"] = username
CONFIG_PATH.write_text(json.dumps(cfg, indent=2)) CONFIG_PATH.write_text(json.dumps(cfg, indent=2))
return cfg return cfg
def exchange_code_for_tokens(code: str, client_id: str, client_secret: str) -> dict: def exchange_code_for_tokens(code: str, client_id: str, client_secret: str,
"""Authorization-Code → Access + Refresh Token.""" redirect_uri: str) -> dict:
data = json.dumps({ """Authorization Code → Access+Refresh Token via API-Call.
Exakt nach Netatmo-Doku: POST /oauth2/token mit
Content-Type: application/x-www-form-urlencoded;charset=UTF-8
WICHTIG: redirect_uri muss EXAKT der Wert sein der beim /authorize-Aufruf
benutzt wurde. Wenn der eine 10.11.3.144 war, muss der hier auch
10.11.3.144 sein, sonst 'invalid_grant'.
"""
data = urllib.parse.urlencode({
"grant_type": "authorization_code", "grant_type": "authorization_code",
"client_id": client_id, "client_id": client_id,
"client_secret": client_secret, "client_secret": client_secret,
"code": code, "code": code,
"redirect_uri": REDIRECT_URI, "redirect_uri": redirect_uri,
"scope": DEFAULT_SCOPE, "scope": DEFAULT_SCOPE,
}).encode("utf-8") }).encode("utf-8")
req = urllib.request.Request(TOKEN_URL, data=data, req = urllib.request.Request(
headers={"Content-Type": "application/json", "Accept": "application/json"}) TOKEN_URL, data=data,
headers={"Content-Type": "application/x-www-form-urlencoded;charset=UTF-8",
"Accept": "application/json"},
)
try: try:
with urllib.request.urlopen(req, timeout=15) as r: with urllib.request.urlopen(req, timeout=15) as r:
return json.loads(r.read()) return json.loads(r.read())
except urllib.error.HTTPError as e: except urllib.error.HTTPError as e:
body = e.read().decode("utf-8", "ignore") body = e.read().decode("utf-8", errors="ignore")
raise SystemExit(f"Token-Exchange fehlgeschlagen: HTTP {e.code}\n{body}") raise SystemExit(f"Token-Exchange fehlgeschlagen: HTTP {e.code}\n{body}")
def run_callback_server(port: int, expected_state: str) -> str: def run_callback_server(port: int, expected_state: str, client_id: str,
"""Startet einen HTTP-Server, der auf den Callback wartet und den Code extrahiert.""" client_secret: str, redirect_uri: str) -> None:
captured = {} """Startet einen HTTP-Server, der den Callback empfängt und den Token speichert.
Läuft bis ein Token gespeichert wurde (max 5 Minuten)."""
saved = {}
def make_response(status: int, body: bytes, content_type: str = "text/html; charset=utf-8"):
return (status, [("Content-Type", content_type)], body)
class Handler(http.server.BaseHTTPRequestHandler): class Handler(http.server.BaseHTTPRequestHandler):
def log_message(self, *args, **kwargs): def log_message(self, format, *args):
pass # quiet # Logge jeden Request damit wir sehen was passiert
print(f" >> {self.command} {self.path} from {self.client_address[0]}")
def do_GET(self): def do_GET(self):
parsed = urllib.parse.urlparse(self.path) parsed = urllib.parse.urlparse(self.path)
qs = urllib.parse.parse_qs(parsed.query) qs = urllib.parse.parse_qs(parsed.query)
print(f" qs: {dict(qs)}")
if "error" in qs: if "error" in qs:
captured["error"] = qs.get("error", ["unknown"])[0] err = qs.get("error", ["unknown"])[0]
self.send_response(400) status, hdrs, body = make_response(400, (
self.send_header("Content-Type", "text/html; charset=utf-8") f"<h1>Fehler bei Netatmo-Authentifizierung</h1>"
f"<p>Grund: {err}</p>"
f"<p>Du kannst dieses Fenster schliessen und es nochmal "
f"probieren.</p>").encode())
self.send_response(status)
for k, v in hdrs: self.send_header(k, v)
self.end_headers() self.end_headers()
self.wfile.write( self.wfile.write(body)
b"<h1>Fehler bei Netatmo-Authentifizierung</h1>" saved["error"] = err
b"<p>Bitte zurueck zum Terminal gehen.</p>")
return return
if "code" not in qs: if "code" not in qs:
self.send_response(400) # Health-Check oder 404
self.send_response(204)
self.end_headers() self.end_headers()
return return
if qs.get("state", [None])[0] != expected_state: if qs.get("state", [None])[0] != expected_state:
self.send_response(400) status, hdrs, body = make_response(400, b"<h1>State mismatch (CSRF-Schutz)</h1>")
self.send_header("Content-Type", "text/html; charset=utf-8") self.send_response(status)
for k, v in hdrs: self.send_header(k, v)
self.end_headers() self.end_headers()
self.wfile.write(b"<h1>State mismatch (CSRF-Schutz)</h1>") self.wfile.write(body)
saved["error"] = "state_mismatch"
return return
captured["code"] = qs["code"][0] code = qs["code"][0]
self.send_response(200) print(f" code erhalten, tausche gegen Token...")
self.send_header("Content-Type", "text/html; charset=utf-8") try:
tok = exchange_code_for_tokens(code, client_id, client_secret, redirect_uri)
refresh_token = tok["refresh_token"]
save_refresh_token(refresh_token, client_id, client_secret)
saved["refresh_token"] = refresh_token
print(f" refresh_token gespeichert!")
html = (
"<h1 style='color:green'>OK!</h1>"
"<p>Refresh-Token wurde in config.json gespeichert.</p>"
"<p>Das Netatmo-Plugin ist jetzt aktiv. Du kannst dieses "
"Fenster schliessen.</p>"
f"<p style='font-family:monospace;font-size:0.8em;color:#666'>"
f"refresh_token: {refresh_token[:24]}...</p>"
).encode()
status, hdrs, body = make_response(200, html)
except SystemExit as e:
saved["error"] = str(e)
status, hdrs, body = make_response(500, (
f"<h1 style='color:red'>Token-Exchange fehlgeschlagen</h1>"
f"<pre>{e}</pre>").encode())
self.send_response(status)
for k, v in hdrs: self.send_header(k, v)
self.end_headers() self.end_headers()
self.wfile.write( self.wfile.write(body)
b"<h1>OK!</h1>"
b"<p>Du kannst dieses Fenster jetzt schliessen. "
b"Zurueck zum Terminal fuer den Refresh-Token.</p>")
server = http.server.HTTPServer(("0.0.0.0", port), Handler) server = http.server.HTTPServer(("0.0.0.0", port), Handler)
server.timeout = 180 # 3 min timeout server.timeout = 300 # 5 min
server.handle_request() print(f"Server listening on 0.0.0.0:{port}, warte auf Callback...",
server.handle_request() # 2nd for /favicon.ico falls Browser fragt flush=True)
if "error" in captured: # Läuft bis refresh_token gespeichert oder 5 Minuten Timeout
raise SystemExit(f"Netatmo-Authentifizierung fehlgeschlagen: {captured['error']}") end_time = time.time() + 300
if "code" not in captured: while time.time() < end_time and "refresh_token" not in saved:
raise SystemExit("Timeout oder kein Code empfangen.") server.handle_request()
return captured["code"] return saved
def main(): def main():
parser = argparse.ArgumentParser(description="Netatmo OAuth2-Setup") parser = argparse.ArgumentParser(description="Netatmo OAuth Setup")
parser.add_argument("--cid", help="Client-ID") parser.add_argument("--cid", help="Client-ID")
parser.add_argument("--csec", help="Client-Secret") parser.add_argument("--csec", help="Client-Secret")
parser.add_argument("--port", type=int, default=8765, parser.add_argument("--redirect", default=None,
help="Lokaler Port für den Callback-Server") help="Redirect-URI (default: http://<lan-ip>:8765/callback)")
parser.add_argument("--scope", default=DEFAULT_SCOPE, parser.add_argument("--port", type=int, default=8765)
help=f"OAuth-Scopes (default: {DEFAULT_SCOPE})") parser.add_argument("--scope", default=DEFAULT_SCOPE)
args = parser.parse_args() args = parser.parse_args()
# 1) Client-Credentials laden existing = load_existing_config().get("plugin_configs", {}).get("netatmo", {})
client_id = args.cid client_id = args.cid or existing.get("client_id") or input("Netatmo Client-ID: ").strip()
if not client_id: client_secret = args.csec or existing.get("client_secret") or input("Netatmo Client-Secret: ").strip()
existing = load_existing_config().get("plugin_configs", {}).get("netatmo", {})
client_id = existing.get("client_id") or input("Netatmo Client-ID: ").strip()
client_secret = args.csec
if not client_secret:
existing = load_existing_config().get("plugin_configs", {}).get("netatmo", {})
client_secret = existing.get("client_secret") or input("Netatmo Client-Secret: ").strip()
if not (client_id and client_secret): if not (client_id and client_secret):
raise SystemExit("Client-ID und Client-Secret werden benötigt.") raise SystemExit("Client-ID und Client-Secret erforderlich.")
# 2) Port + State + Redirect-URI
port = free_port(args.port) port = free_port(args.port)
lan_ip = detect_lan_ip()
if args.redirect:
redirect = args.redirect
else:
redirect = f"http://{lan_ip}:{port}/callback"
import secrets import secrets
state = secrets.token_urlsafe(16) state = secrets.token_urlsafe(16)
# Wenn Chromium auf dem Pi vorhanden ist UND ein X-Server läuft,
# öffnen wir den Browser auf dem Pi selbst. Dann ist 'localhost' die
# korrekte Redirect-URI. Das ist der einfachste Fall — User muss
# nichts weiter konfigurieren.
lan_ip = detect_lan_ip()
import os
if os.path.exists("/tmp/.X11-unix/X0") and not os.environ.get("DISPLAY"):
os.environ["DISPLAY"] = ":0"
chromium_local = any(os.path.exists(p) for p in
("/usr/bin/chromium", "/usr/bin/chromium-browser",
"/usr/bin/google-chrome"))
if chromium_local and os.environ.get("DISPLAY"):
# Chromium läuft auf dem Pi selbst → localhost funktioniert.
redirect = f"http://localhost:{port}/callback"
target_host = "localhost"
print()
print("Info: Chromium auf Display :0 erkannt. Verwende localhost.")
print(" (Öffne ein Terminal-Fenster auf dem Pi und schau zu.)")
else:
# Kein lokaler Browser → Redirect muss LAN-IP haben, und der
# User öffnet die URL in seinem Laptop/Phone-Browser.
redirect = f"http://{lan_ip}:{port}/callback"
target_host = lan_ip
if redirect.startswith("http://localhost"):
print()
print("WARN: Konnte keine LAN-IP ermitteln. Verwende localhost.")
print(" Falls du per SSH verbunden bist: exit und dann")
print(" 'ssh -L 8765:localhost:8765 pi@...' um den Tunnel zu öffnen.")
# 3) Authorize-URL
auth_url = ( auth_url = (
f"{AUTHORIZE_URL}" f"{AUTHORIZE_URL}"
f"?client_id={urllib.parse.quote(client_id)}" f"?client_id={urllib.parse.quote(client_id)}"
@@ -238,87 +254,39 @@ def main():
f"&response_type=code" f"&response_type=code"
) )
print(f"\n=== Netatmo OAuth Setup ===\n")
print(f"Local callback server listening on {redirect}")
print() print()
print(f"WICHTIG: Diese Redirect-URI muss EXAKT in deiner Netatmo-App") print("=" * 70)
print(f"auf https://dev.netatmo.com/apps/ registriert sein!") print(" NETATMO OAUTH SETUP")
print("=" * 70)
print() print()
print(f" Aktuelle Redirect-URI:") print("Voraussetzung: Die folgende Redirect-URI muss EXAKT in deiner")
print(f" {redirect}") print("Netatmo-App auf https://dev.netatmo.com/apps/ registriert sein:")
print() print()
if redirect.startswith("http://localhost"): print(f" {redirect}")
print(f" Falls deine App auf https://dev.netatmo.com/apps/ mit der") print()
print(f" URI 'http://localhost:8765/callback' registriert ist, dann") print("=" * 70)
print(f" funktioniert der Helper NUR wenn du direkt auf dem Pi ein") print(" BITTE IM BROWSER OEFFNEN:")
print(f" Terminal hast (Browser auf dem Pi). Falls du per SSH") print("=" * 70)
print(f" verbunden bist: 'exit' und mit 'ssh -L 8765:localhost:8765'") print()
print(f" neu verbinden, dann Browser auf localhost:8765 öffnen.") print(auth_url)
print()
print("=" * 70)
print(" Nach dem Authorize-Klick landest du auf der Login-Seite,")
print(" loggst dich ein, klickst 'Authorize', und wirst zurueckgeleitet.")
print(" Der Server holt dann den Token und speichert ihn in config.json.")
print("=" * 70)
print()
result = run_callback_server(port, state, client_id, client_secret, redirect)
if "refresh_token" in result:
print() print()
print(f" Alternative: du kannst die App-URI auf") print("ERFOLG! Refresh-Token gespeichert in", CONFIG_PATH)
print(f" '{redirect}' ändern, dann funktioniert der Helper vom") print()
print(f" Laptop/Phone aus über diese URL.") sys.exit(0)
print() elif "error" in result:
print(f"Oeffne in deinem Browser:\n {auth_url}\n") print()
try: print("FEHLER:", result["error"], file=sys.stderr)
# Wenn ein X-Server läuft (Pi mit Bildschirm, Desktop-Session), sys.exit(1)
# explizit DISPLAY=:0 setzen und Chromium direkt öffnen. Das
# umgeht xdg-open-Fallbacks die oft auf headless-Setups scheitern.
import os
if os.path.exists("/tmp/.X11-unix/X0") and not os.environ.get("DISPLAY"):
os.environ["DISPLAY"] = ":0"
# Chromium-Binary auf dem Pi finden
chromium = None
for cand in ("chromium", "chromium-browser", "google-chrome"):
from shutil import which
p = which(cand)
if p:
chromium = p
break
if chromium and os.environ.get("DISPLAY"):
# Chromium im App-Mode öffnen — kein Toolbar, sieht wie eine
# native App aus. Fenster schließt nach Redirect automatisch
# durch unser Callback-Handler.
print(f"Oeffne Chromium auf Display {os.environ['DISPLAY']}...")
import subprocess
subprocess.Popen(
[chromium, f"--app={auth_url}",
"--no-default-browser-check",
"--no-first-run"],
env=dict(os.environ),
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
)
elif webbrowser.open(auth_url):
print("(Browser wurde geoeffnet.)")
else:
print("(Browser konnte nicht automatisch geoeffnet werden — bitte URL manuell öffnen.)")
except Exception as e:
print(f"(Browser-Fehler: {e} — bitte URL manuell öffnen.)")
# 4) Auf Callback warten
print("\nWarte auf Callback...")
code = run_callback_server(port, state)
# 5) Code → Tokens
print("\nTausche Code gegen Tokens...")
tok = exchange_code_for_tokens(code, client_id, client_secret)
refresh_token = tok.get("refresh_token")
access_token = tok.get("access_token")
expires_in = tok.get("expires_in", 0)
if not refresh_token:
raise SystemExit(f"Antwort enthält keinen refresh_token: {tok}")
print(f" access_token: {access_token[:24]}...")
print(f" refresh_token: {refresh_token[:24]}...")
print(f" expires_in: {expires_in}s ({expires_in // 60} min)")
# 6) In config.json speichern
save_refresh_token(refresh_token, client_id, client_secret,
username="")
print(f"\nGespeichert in {CONFIG_PATH}")
print("\nFertig! Das Plugin holt sich ab jetzt selbst neue Access-Tokens.")
print("Tipp: dashboard.py / epaper-admin restarten damit das Plugin den "
"neuen refresh_token liest.\n")
if __name__ == "__main__": if __name__ == "__main__":