From 6b604b0289472a5646cc6c5e6317e828efad7954 Mon Sep 17 00:00:00 2001 From: epaper-dashboard Date: Wed, 26 Aug 2026 22:11:46 +0400 Subject: [PATCH] gitignore: schliesse .backup/ aus Co-Authored-By: Hermes --- .backup/2026-08-26-active/README.md | 176 -- .backup/2026-08-26-active/RECOVERY.md | 41 - .backup/2026-08-26-active/admin.py | 688 ------ .backup/2026-08-26-active/config.example.json | 10 - .backup/2026-08-26-active/dashboard.py | 463 ----- .backup/2026-08-26-active/layout.py | 172 -- .backup/2026-08-26-active/network_watchdog.py | 468 ----- .backup/2026-08-26-active/palette.py | 237 --- .backup/2026-08-26-active/plugins/__init__.py | 1 - .backup/2026-08-26-active/plugins/base.py | 69 - .backup/2026-08-26-active/plugins/clock.py | 117 -- .backup/2026-08-26-active/plugins/gmail.py | 58 - .backup/2026-08-26-active/plugins/hello.py | 27 - .backup/2026-08-26-active/plugins/minimax.py | 271 --- .backup/2026-08-26-active/plugins/spotify.py | 78 - .backup/2026-08-26-active/plugins/strava.py | 99 - .../2026-08-26-active/templates/index.html | 1844 ----------------- .gitignore | 1 + 18 files changed, 1 insertion(+), 4819 deletions(-) delete mode 100644 .backup/2026-08-26-active/README.md delete mode 100644 .backup/2026-08-26-active/RECOVERY.md delete mode 100644 .backup/2026-08-26-active/admin.py delete mode 100644 .backup/2026-08-26-active/config.example.json delete mode 100644 .backup/2026-08-26-active/dashboard.py delete mode 100644 .backup/2026-08-26-active/layout.py delete mode 100644 .backup/2026-08-26-active/network_watchdog.py delete mode 100644 .backup/2026-08-26-active/palette.py delete mode 100644 .backup/2026-08-26-active/plugins/__init__.py delete mode 100644 .backup/2026-08-26-active/plugins/base.py delete mode 100644 .backup/2026-08-26-active/plugins/clock.py delete mode 100644 .backup/2026-08-26-active/plugins/gmail.py delete mode 100644 .backup/2026-08-26-active/plugins/hello.py delete mode 100644 .backup/2026-08-26-active/plugins/minimax.py delete mode 100644 .backup/2026-08-26-active/plugins/spotify.py delete mode 100644 .backup/2026-08-26-active/plugins/strava.py delete mode 100644 .backup/2026-08-26-active/templates/index.html diff --git a/.backup/2026-08-26-active/README.md b/.backup/2026-08-26-active/README.md deleted file mode 100644 index 9c28e38..0000000 --- a/.backup/2026-08-26-active/README.md +++ /dev/null @@ -1,176 +0,0 @@ -# epaper-dashboard - -Plugin-based dashboard for the **Waveshare 7.3" ACeP 7-Color e-Paper (F) HAT** on Raspberry Pi 4. - -Rendert alle 180s (oder konfigurierbar) ein 800×480px Vollbild auf dem Display. - -## Hardware - -- **Display:** Waveshare 7.3inch ACeP 7-Color e-Paper (F) HAT — 800×480, 7 Farben -- **Pi:** Raspberry Pi 4 (oder Zero 2 W mit Einschränkungen) -- **Interface:** SPI - -## Architektur - -``` -┌─────────────────────────────────────────────┐ -│ epaper-dashboard │ -├──────────────────┬──────────────────────────┤ -│ dashboard.py │ admin.py │ -│ (Renderer) │ (Web Admin UI :8080) │ -├──────────────────┴──────────────────────────┤ -│ layout.py · 4×4 Grid · Pack-Algorithmus │ -├────────────────────────────────────────────┤ -│ plugins/ (eines pro Widget) │ -│ clock · weather · system · spotify │ -│ strava · gmail · minimax · hello │ -├────────────────────────────────────────────┤ -│ network_watchdog.py │ -│ (WiFi AP/Client Management) │ -├────────────────────────────────────────────┤ -│ waveshare_epd/ (Vendor-Treiber) │ -│ epd7in3f.py │ -└─────────────────────────────────────────────┘ -``` - -## Quick Start - -### Pi vorbereiten - -```bash -sudo apt update && sudo apt install -y python3-pil python3-numpy python3-flask git -sudo raspi-config # → Interface Options → SPI → Enable -sudo reboot -``` - -### Setup - -```bash -# Repository klonen -git clone https://git.pkop.de/Vibecode/epaper-dashboard.git -cd epaper-dashboard - -# Config erstellen -cp config.example.json config.json -# → config.json editieren (Plugins + Layout) - -#waveshare_epd Treiber -git clone https://github.com/waveshareteam/Waveshare_GPIO.git waveshare_epd_lib -# oder: den waveshare_epd Ordner vom vorherigen Setup übernehmen - -# Services installieren -chmod +x install_services.sh -./install_services.sh -``` - -### Config - -`config.json` — Version 2 Format (empfohlen): - -```json -{ - "version": 2, - "refresh_interval_s": 180, - "layout": { - "grid": { "cols": 4, "rows": 4 }, - "items": [ - { "id": "c1", "plugin": "clock", "x": 0, "y": 0, "w": 2, "h": 2 }, - { "id": "w1", "plugin": "weather", "x": 2, "y": 0, "w": 2, "h": 2 }, - { "id": "st1", "plugin": "system", "x": 0, "y": 2, "w": 2, "h": 2 }, - { "id": "h1", "plugin": "hello", "x": 2, "y": 2, "w": 2, "h": 2 } - ] - }, - "plugin_configs": { - "hello": { "text": "Edit me!", "size": 40 } - } -} -``` - -**Grid:** 4 Spalten × 4 Zeilen = 16 Zellen à 200×120px. `x,y` = Spalte/Zeile oben-links, `w,h` = Breite/Höhe in Zellen. - -### Web Admin - -``` -http://pi:8080/ -``` - -- **Layout-Editor** — Items per Klick hinzufügen, verschieben, Größe ändern, Auto-Pack -- **Plugin-Config** — API-Keys, Locations, etc. pro Widget -- **Refresh Now** — sofortiger Display-Refresh -- **Netzwerk** — WLAN wechseln ohne SSH - -Default Login: `admin` / `admin` — in `.env` mit `EPAPER_ADMIN_PASSWORD` setzen. - -## Plugins - -| Plugin | Datenquelle | Config-Felder | -|--------|-------------|---------------| -| `clock` | Systemzeit | — | -| `weather` | Open-Meteo (kein API-Key) | `location` (lat,lon), `show_uv`, `show_forecast_hours` | -| `system` | Pi CPU/RAM/Uptime | `show_uptime`, `show_load`, `show_temp` | -| `spotify` | Spotify Web API | `client_id`, `client_secret` (via Spotify Dev Portal) | -| `strava` | Strava API | `access_token`, `club_id` | -| `gmail` | Gmail API | `credentials_json` (OAuth2) | -| `minimax` | MiniMax AI | `api_key`, `model`, `prompt` | -| `hello` | statisch | `text`, `color`, `size` | - -## Eigenes Plugin schreiben - -```python -# plugins/my_widget.py -from plugins.base import Widget -from palette import FG, BG - -class Widget(Widget): - name = "my_widget" - label = "Mein Widget" - description = "Zeigt etwas" - category = "info" - - config_schema = [ - {"key": "api_key", "label": "API Key", "type": "secret"}, - {"key": "interval", "label": "Intervall (s)", "type": "int", "default": 60}, - ] - default_config = {"api_key": "", "interval": 60} - - def fetch(self): - return {"value": 42} - - def render(self, draw, fonts, x, y, w, h): - d = self.fetch() - # w, h sind Pixel (nicht Zellen!) - draw.text((x + 10, y + 10), f"Value: {d['value']}", - font=fonts["28"], fill=FG) -``` - -**Palette:** `BLACK, WHITE, RED, GREEN, BLUE, YELLOW, ORANGE` + semantisch `FG, BG, OK, WARN, ALERT, INFO, ACCENT` - -## Services - -```bash -systemctl --user enable --now epaper-dashboard -systemctl --user enable --now epaper-admin -journalctl --user -u epaper-dashboard -f # Logs -``` - -**Wichtig:** `PrivateTmp=no` in den Service-Files — sonst sieht der Admin-Socket `/tmp/epaper-dashboard.sock` nicht. - -## Display-Einschränkungen - -- **Nur Full-Refresh:** ACeP-Displays vertragen keine schnellen Partial-Refreshes. Minimum 180s. -- **Refresh-Dauer:** Ein Full-Refresh dauert ~35-65s. Währenddessen ist das Display leer/flackernd. -- **Ghosting:** Leichtes Ghosting bei manchen Farben normal. Display nicht bei direktem Sonnenlicht ablesen. - -## Recovery - -Falls das konfigurierte WLAN nicht erreichbar ist: der Pi startet automatisch einen **Recovery AP**. - -- **SSID:** `epaper-recovery` -- **Passwort:** `recovery1234` -- **URL:** http://10.42.0.1:8080/ - -Dort kannst du ein neues WLAN konfigurieren. Details in `RECOVERY.md`. - -## Lizenz - -Apache 2.0 diff --git a/.backup/2026-08-26-active/RECOVERY.md b/.backup/2026-08-26-active/RECOVERY.md deleted file mode 100644 index e5ad183..0000000 --- a/.backup/2026-08-26-active/RECOVERY.md +++ /dev/null @@ -1,41 +0,0 @@ -# Recovery wenn der Pi nicht mehr per SSH erreichbar ist - -## Was passiert ist -- Auf dem Pi lief ein WLAN mit SSID `IoT`, Profil-Name `netplan-wlan0-IoT` -- Recovery-AP-Test über `nmcli device wifi hotspot` hat einen WLAN-Hotspot `epaper-test` gestartet -- Dies hat wahrscheinlich die WLAN-Verbindung zum Heimnetz unterbrochen -- SSH-Connect zum Pi schlägt jetzt fehl (Timeout) - -## Sofortige Hilfe - -### Option A: Strom-Reset -1. Pi vom Strom trennen -2. 10 Sekunden warten -3. Strom wieder rein -4. Pi bootet → NetworkManager verbindet sich automatisch mit `IoT` (sofern in Reichweite) -5. Nach ~60 Sekunden: `ssh koptikp@10.11.3.144` sollte wieder gehen - -### Option B: Recovery-Hotspot -Falls Strom-Reset nicht hilft (z.B. weil Profil `netplan-wlan0-IoT` kaputt ist): -1. Auf dem Laptop WLAN-Liste scannen -2. `epaper-recovery` suchen (Passwort `recovery1234`) — falls der Pi im Recovery-Modus ist - oder -3. `epaper-test` suchen (Passwort `test1234`) — falls der Test-Hotspot noch läuft -4. Verbinden, dann im Browser http://10.42.0.1:8080 -5. Bei Auth-Prompt: `admin:changeme123` (oder `admin:admin`) -6. Im Netzwerk-Panel: WLAN neu konfigurieren oder "Force AP Mode" / "Disconnect" - -### Option C: Tastatur + Monitor -1. HDMI + USB-Tastatur an den Pi -2. Anmelden als `koptikp` -3. `sudo nmcli connection show` → sehen was aktiv ist -4. `sudo nmcli connection down epaper-test` (oder welcher Müll-Profile da sind) -5. `sudo nmcli connection delete epaper-test` -6. `sudo nmcli connection up netplan-wlan0-IoT` -7. `sudo reboot` - -## Was ich anders machen würde -- **Hotspot-Tests nie** auf einem produktiven Pi laufen lassen, ohne physischen Zugang als Fallback zu haben -- Watchdog-Code war korrekt; das Problem ist die PolicyKit-Konfiguration -- Polkit-Restart nach Rule-Update: besser nur `polkitd` SIGHUP schicken, nicht systemctl restart - (das hat in dieser Session offenbar den Polkit-Daemon in einem inkonsistenten Zustand hinterlassen) diff --git a/.backup/2026-08-26-active/admin.py b/.backup/2026-08-26-active/admin.py deleted file mode 100644 index b52fe3c..0000000 --- a/.backup/2026-08-26-active/admin.py +++ /dev/null @@ -1,688 +0,0 @@ -"""Flask-basierte Admin-UI für das 7.3" Dashboard. - -Routes: - GET / → Grid-Builder: Slots, Plugins, Refresh-Intervall - POST /config → Speichert config.json (Slot-Belegung + Intervall) - POST /plugins/ → Speichert Plugin-spezifische Config - POST /refresh → Triggert sofortigen Refresh via IPC - GET /snapshot.png → Letztes gerendertes Bild (PNG) - GET /plugins.json → Liste aller verfügbaren Plugins + Schemas (für UI) -""" -from __future__ import annotations -import os, sys, json, base64, io, time -from pathlib import Path - -HERE = Path(__file__).resolve().parent -sys.path.insert(0, str(HERE)) - -from flask import Flask, request, redirect, url_for, render_template, jsonify, Response, abort - -import dashboard as dashboard_mod -from plugins.base import all_widget_classes -import network_watchdog as net -import layout as layout_mod - -app = Flask(__name__) - -# ============================================================================ -# Auth (HTTP Basic) -# ============================================================================ -def _load_auth(): - """Liest admin_user / admin_password aus .env. - - Reihenfolge (erstes gewinnt): - 1. Umgebungsvariable EPAPER_ADMIN_USER / EPAPER_ADMIN_PASSWORD (nützlich für Container/CI) - 2. /.env (eigene Datei, wird bevorzugt) - 3. ~/.hermes/.env (global) - Default: admin / admin — explizit nur für first-boot, damit der Service - überhaupt erreichbar ist. Empfehlung: nach erstem Start .env anlegen. - """ - user, pw = "admin", "admin" - env_user = os.environ.get("EPAPER_ADMIN_USER") - env_pw = os.environ.get("EPAPER_ADMIN_PASSWORD") - if env_user: user = env_user - if env_pw: pw = env_pw - for envfile in [HERE / ".env", Path.home() / ".hermes" / ".env"]: - if not envfile.exists(): - continue - try: - for line in envfile.read_text().splitlines(): - line = line.strip() - if not line or line.startswith("#") or "=" not in line: - continue - k, v = line.split("=", 1) - k = k.strip(); v = v.strip().strip('"').strip("'") - if k == "EPAPER_ADMIN_USER" and v: user = v - if k == "EPAPER_ADMIN_PASSWORD" and v: pw = v - except Exception: - pass - return user, pw - - -def check_auth(): - user, pw = _load_auth() - auth = request.headers.get("Authorization") - if auth and auth.startswith("Basic "): - try: - decoded = base64.b64decode(auth[6:]).decode() - u, p = decoded.split(":", 1) - if u == user and p == pw: - return True - except Exception: - pass - return False - - -def require_auth(): - if not check_auth(): - return Response( - "Authentication required", - 401, - {"WWW-Authenticate": 'Basic realm="epaper-dashboard"'}, - ) - return None - - -# ============================================================================ -# Routes -# ============================================================================ -@app.route("/") -def index(): - a = require_auth() - if a: - # Demo-Modus: ?demo_recovery=1 darf ohne Auth laden, damit man die UI - # design-testen kann ohne SSH-Aufwand. Static markup ohne live data. - if request.args.get("demo_recovery") or request.args.get("demo"): - demo = True - else: - return a - else: - demo = False - - cfg = dashboard_mod.load_config() - slots = cfg.get("slots", []) # legacy - classes = dashboard_mod.get_widget_classes() - widgets_meta = [] - for name, cls in classes.items(): - widgets_meta.append({ - "name": name, - "label": cls.label, - "description": cls.description, - "category": cls.category, - "schema": cls.config_schema, - "defaults": cls.default_config, - "default_size": layout_mod.auto_size_for_plugin(name), - }) - widgets_meta.sort(key=lambda w: w["label"]) - - # v2-Layout - layout_items = cfg.get("layout", {}).get("items", []) - plugin_configs = cfg.get("plugin_configs", {}) - grid = cfg.get("layout", {}).get("grid", {"cols": 4, "rows": 4}) - - # Slot n -> (plugin_name, plugin_config_dict) - slot_plugins = [(it.get("plugin", ""), plugin_configs.get(it.get("plugin", ""), {})) - for it in layout_items] - - return render_template("index.html", - # legacy - slots=slots, - widgets=widgets_meta, - refresh_interval=cfg.get("refresh_interval_s", 180), - categories=sorted(set(w["category"] for w in widgets_meta)), - now=int(time.time()), - # v2 - layout_items=layout_items, - plugin_configs=plugin_configs, - grid=grid, - size_presets=layout_mod.SIZE_PRESETS, - slot_plugins=slot_plugins, - demo_mode=demo) - - -@app.route("/config", methods=["POST"]) -def save_config(): - a = require_auth() - if a: return a - - cfg = dashboard_mod.load_config() - # refresh interval - if "refresh_interval_s" in request.form: - try: - cfg["refresh_interval_s"] = max(30, int(request.form["refresh_interval_s"])) - except (TypeError, ValueError): - pass - - # slot belegung: pro slot "slot__plugin" - slots = cfg.get("slots", []) - for i in range(dashboard_mod.SLOTS): - plugin_name = request.form.get(f"slot_{i}_plugin", "").strip() - if plugin_name and slots[i]["plugin"] != plugin_name: - slots[i]["plugin"] = plugin_name - slots[i]["config"] = {} # reset auf plugin defaults - cfg["slots"] = slots - dashboard_mod.save_config(cfg) - return redirect(url_for("index")) - - -@app.route("/plugins/", methods=["POST"]) -def save_plugin_config(idx): - a = require_auth() - if a: return a - cfg = dashboard_mod.load_config() - if idx < 0 or idx >= dashboard_mod.SLOTS: - abort(400) - slot = cfg["slots"][idx] - classes = dashboard_mod.get_widget_classes() - cls = classes.get(slot["plugin"]) - if cls is None: - abort(400, "unknown plugin") - new_cfg = dict(slot.get("config", {})) - for field in cls.config_schema: - key = field["key"] - form_val = request.form.get(f"slot_{idx}_{key}") - ftype = field.get("type", "string") - if ftype == "secret": - # Secret-Felder: nur überschreiben wenn Form nicht leer. - # Damit kann der User ein Secret löschen, indem er explizit leeres - # Feld abschickt; ansonsten bleibt der alte Wert erhalten. - if form_val is None: - continue - if form_val == "": - new_cfg[key] = "" - elif form_val == "__UNSET__": - # wird vom UI nie geschickt; reserved für "Secret entfernen" - new_cfg.pop(key, None) - else: - new_cfg[key] = form_val - continue - if form_val is None: - continue - try: - if ftype == "int": - new_cfg[key] = int(form_val) - elif ftype == "float": - new_cfg[key] = float(form_val) - elif ftype == "bool": - new_cfg[key] = (form_val.lower() in ("1", "true", "yes", "on")) - else: - new_cfg[key] = form_val - except (TypeError, ValueError): - new_cfg[key] = form_val - slot["config"] = new_cfg - cfg["slots"][idx] = slot - dashboard_mod.save_config(cfg) - return redirect(url_for("index")) - - -@app.route("/api/layout", methods=["GET", "POST"]) -def api_layout(): - a = require_auth() - if a: return a - cfg = dashboard_mod.load_config() - if request.method == "GET": - return jsonify({ - "grid": cfg.get("layout", {}).get("grid", {"cols": 4, "rows": 4}), - "items": cfg.get("layout", {}).get("items", []), - "overlaps": [list(p) for p in layout_mod.find_overlaps( - [layout_mod.Item.from_dict(d) for d in cfg.get("layout", {}).get("items", [])] - )], - "out_of_bounds": layout_mod.find_out_of_bounds( - [layout_mod.Item.from_dict(d) for d in cfg.get("layout", {}).get("items", [])] - ), - }) - # POST: body form-data with fields per item OR full JSON replacement - data = request.get_json(silent=True) - if data and "items" in data: - items_raw = data["items"] - new_items = [] - for it in items_raw: - try: - item = layout_mod.Item.from_dict(it) - new_items.append(item.to_dict()) - except Exception as e: - return jsonify({"ok": False, "error": f"invalid item: {e}"}), 400 - cfg.setdefault("layout", {})["items"] = new_items - dashboard_mod.save_config(cfg) - # Auto-pack wenn gewünscht - if request.args.get("auto_pack") == "1": - new_items = layout_mod.pack([layout_mod.Item.from_dict(d) for d in new_items]) - cfg["layout"]["items"] = [it.to_dict() for it in new_items] - dashboard_mod.save_config(cfg) - return jsonify({"ok": True, "items": cfg["layout"]["items"]}) - - # Form-encoded: items[item_id][plugin/x/y/w/h] - form = request.form - new_items = [] - # Sammle alle item_ids - item_ids = set() - for k in form.keys(): - if k.startswith("items[") and "][" in k: - iid = k.split("items[")[1].split("]")[0] - item_ids.add(iid) - for iid in item_ids: - try: - item = { - "id": iid, - "plugin": form.get(f"items[{iid}][plugin]", ""), - "x": int(form.get(f"items[{iid}][x]", 0)), - "y": int(form.get(f"items[{iid}][y]", 0)), - "w": int(form.get(f"items[{iid}][w]", 1)), - "h": int(form.get(f"items[{iid}][h]", 1)), - } - new_items.append(item) - except Exception as e: - return jsonify({"ok": False, "error": f"item {iid}: {e}"}), 400 - cfg.setdefault("layout", {})["items"] = new_items - dashboard_mod.save_config(cfg) - return jsonify({"ok": True, "items": new_items}) - - -@app.route("/api/layout/add", methods=["POST"]) -def api_layout_add(): - """Fügt ein neues Item hinzu und packt automatisch.""" - a = require_auth() - if a: return a - plugin = request.form.get("plugin", "hello").strip() - cfg = dashboard_mod.load_config() - classes = dashboard_mod.get_widget_classes() - cls = classes.get(plugin) - if cls is None: - return jsonify({"ok": False, "error": "unknown plugin"}), 400 - w, h = layout_mod.auto_size_for_plugin(plugin) - # Override mit user-input wenn vorhanden - try: w = int(request.form.get("w", w)) - except: pass - try: h = int(request.form.get("h", h)) - except: pass - import secrets - new_id = secrets.token_hex(4) - items = cfg.setdefault("layout", {}).setdefault("items", []) - new_item = layout_mod.Item(new_id, plugin, 0, 0, w, h).to_dict() - items.append(new_item) - # Pack alle (inkl. neue) - packed = layout_mod.pack([layout_mod.Item.from_dict(d) for d in items]) - cfg["layout"]["items"] = [it.to_dict() for it in packed] - dashboard_mod.save_config(cfg) - return jsonify({"ok": True, "id": new_id, "items": cfg["layout"]["items"]}) - - -@app.route("/api/layout/item/", methods=["DELETE", "PATCH"]) -def api_layout_item(iid): - a = require_auth() - if a: return a - cfg = dashboard_mod.load_config() - items = cfg.setdefault("layout", {}).setdefault("items", []) - if request.method == "DELETE": - items = [it for it in items if it.get("id") != iid] - cfg["layout"]["items"] = items - dashboard_mod.save_config(cfg) - return jsonify({"ok": True}) - # PATCH: update single fields - data = request.get_json(silent=True) or {} - for it in items: - if it.get("id") == iid: - for k in ("plugin", "x", "y", "w", "h"): - if k in data: - if k in ("x", "y", "w", "h"): - try: it[k] = int(data[k]) - except: pass - else: - it[k] = data[k] - dashboard_mod.save_config(cfg) - return jsonify({"ok": True, "items": items}) - - -@app.route("/api/layout/pack", methods=["POST"]) -def api_layout_pack(): - """Auto-pack alle Items in den 4x4 grid.""" - a = require_auth() - if a: return a - cfg = dashboard_mod.load_config() - items = [layout_mod.Item.from_dict(d) for d in cfg.get("layout", {}).get("items", [])] - packed = layout_mod.pack(items) - cfg["layout"]["items"] = [it.to_dict() for it in packed] - dashboard_mod.save_config(cfg) - return jsonify({"ok": True, "items": cfg["layout"]["items"]}) - - -DISPLAY_THEMES = { - "default": { - "label": "Classic White", - "bg": "#ffffff", "fg": "#000000", - "accent": "#ff8800", - }, - "dark": { - "label": "Dark Mode", - "bg": "#111111", "fg": "#f0f0f0", - "accent": "#6d8eff", - }, - "sepia": { - "label": "Sepia", - "bg": "#f4ecd8", "fg": "#5b4636", - "accent": "#c0392b", - }, - "nord": { - "label": "Nord", - "bg": "#eceff4", "fg": "#2e3440", - "accent": "#88c0d0", - }, - "terminal": { - "label": "Terminal (Green)", - "bg": "#0d0d00", "fg": "#e8c840", - "accent": "#ffcc00", - }, -} - -# Display designs — the actual visual layout approach -DISPLAY_DESIGNS = { - "classic": { - "label": "Classic", - "desc": "Original flat design, white background", - }, - "magazine": { - "label": "Magazine", - "desc": "Bold typography, minimal, colored accent bars", - }, - "cards": { - "label": "Cards", - "desc": "iOS-style panels with emoji icons, warm white", - }, - "kiosk": { - "label": "Kiosk", - "desc": "Data-dense, dark background, inverted info cards", - }, -} - - -@app.route("/api/display_theme", methods=["GET", "POST"]) -def api_display_theme(): - """GET: list themes + current. POST: set theme (legacy color-only).""" - a = require_auth() - if a: return a - cfg = dashboard_mod.load_config() - if request.method == "GET": - theme_key = cfg.get("display_theme", "default") - theme = DISPLAY_THEMES.get(theme_key, DISPLAY_THEMES["default"]) - return jsonify({ - "current": theme_key, - "themes": {k: v["label"] for k, v in DISPLAY_THEMES.items()}, - "theme": theme, - }) - data = request.get_json(silent=True) or {} - theme_key = data.get("theme", "default") - if theme_key not in DISPLAY_THEMES: - return jsonify({"ok": False, "error": "unknown theme"}), 400 - cfg["display_theme"] = theme_key - dashboard_mod.save_config(cfg) - return jsonify({"ok": True, "current": theme_key, "theme": DISPLAY_THEMES[theme_key]}) - - -@app.route("/api/display_design", methods=["GET", "POST"]) -def api_display_design(): - """Design für das Display (classic/magazine/cards/kiosk).""" - a = require_auth() - if a: return a - cfg = dashboard_mod.load_config() - if request.method == "GET": - design_key = cfg.get("display_design", "classic") - return jsonify({ - "current": design_key, - "designs": {k: {"label": v["label"], "desc": v["desc"]} for k, v in DISPLAY_DESIGNS.items()}, - }) - data = request.get_json(silent=True) or {} - design_key = data.get("design", "classic") - if design_key not in DISPLAY_DESIGNS: - return jsonify({"ok": False, "error": "unknown design"}), 400 - cfg["display_design"] = design_key - dashboard_mod.save_config(cfg) - return jsonify({"ok": True, "current": design_key}) - - -@app.route("/api/plugin_config/", methods=["GET", "POST"]) -def api_plugin_config(plugin_name): - """Plugin-spezifische Config (separat vom Layout).""" - a = require_auth() - if a: return a - classes = dashboard_mod.get_widget_classes() - cls = classes.get(plugin_name) - if cls is None: - return jsonify({"ok": False, "error": "unknown plugin"}), 400 - cfg = dashboard_mod.load_config() - if request.method == "GET": - return jsonify({ - "plugin": plugin_name, - "label": cls.label, - "schema": cls.config_schema, - "config": cfg.get("plugin_configs", {}).get(plugin_name, {}), - }) - # POST: update config for this plugin - new_cfg = {} - for field in cls.config_schema: - key = field["key"] - form_val = request.form.get(f"config_{key}") - ftype = field.get("type", "string") - if form_val is None: - continue - if ftype == "secret": - # Nur überschreiben wenn nicht leer - existing = cfg.get("plugin_configs", {}).get(plugin_name, {}).get(key, "") - if form_val == "": - new_cfg[key] = "" - elif form_val == "__UNSET__": - continue - else: - new_cfg[key] = form_val - continue - try: - if ftype == "int": - new_cfg[key] = int(form_val) - elif ftype == "float": - new_cfg[key] = float(form_val) - elif ftype == "bool": - new_cfg[key] = (form_val.lower() in ("1", "true", "yes", "on")) - else: - new_cfg[key] = form_val - except (TypeError, ValueError): - new_cfg[key] = form_val - cfg.setdefault("plugin_configs", {})[plugin_name] = new_cfg - dashboard_mod.save_config(cfg) - return jsonify({"ok": True, "plugin": plugin_name, "config": new_cfg}) - - -@app.route("/refresh", methods=["POST"]) -def refresh_now(): - a = require_auth() - if a: return a - result = dashboard_mod.send_ipc("refresh") - return jsonify({"status": "ok", "ipc_response": result}) - - -@app.route("/snapshot.png") -def snapshot_png(): - a = require_auth() - if a: return a - # Snapshot aus dem laufenden Renderer via IPC anfordern - # Wir machen das nicht über IPC (Renderer schreibt Datei), sondern rendern direkt hier - # für Preview-Zwecke — billiger und deterministisch. - try: - d = dashboard_mod.Dashboard() - # nicht reload() — würde live-config nutzen, ok - img = d.render_once() - buf = io.BytesIO() - img.save(buf, format="PNG") - return Response(buf.getvalue(), mimetype="image/png") - except Exception as e: - return jsonify({"error": str(e)}), 500 - - -@app.route("/plugins.json") -def plugins_json(): - a = require_auth() - if a: return a - out = [] - for cls in all_widget_classes(): - out.append({ - "name": cls.name, - "label": cls.label, - "description": cls.description, - "category": cls.category, - "schema": cls.config_schema, - "defaults": cls.default_config, - }) - return jsonify(out) - - -@app.route("/api/network/recovery") -def api_net_recovery(): - """Entscheidung: soll die UI Recovery-Info prominent zeigen?""" - a = require_auth() - if a: return a - state = net.get_watchdog().get_state() - threshold = net.get_recovery_threshold() - decision = net._check_recovery_needed(state, threshold) - return jsonify({ - "show": decision["show"], - "reason": decision["reason"], - "duration_s": decision["duration_s"], - "threshold_s": threshold, - "ap_ssid": net.AP_SSID, - "ap_password": net.AP_PASSWORD, - "ap_url": "http://10.42.0.1:8080", - "current_mode": state.get("mode"), - "current_ssid": state.get("ssid"), - "current_ip": state.get("ip"), - "error": state.get("error", ""), - }) - - -@app.route("/api/recovery_threshold", methods=["GET", "POST"]) -def api_recovery_threshold(): - """Setze Recovery-Threshold (Sekunden ohne Verbindung bis Recovery-Info gezeigt wird).""" - a = require_auth() - if a: return a - cfg = dashboard_mod.load_config() - if request.method == "GET": - return jsonify({"threshold_s": cfg.get("recovery_threshold_s", 60)}) - try: - threshold = int(request.form.get("threshold_s", 60)) - threshold = max(5, min(3600, threshold)) - except (TypeError, ValueError): - return jsonify({"ok": False, "error": "invalid threshold"}), 400 - cfg["recovery_threshold_s"] = threshold - dashboard_mod.save_config(cfg) - return jsonify({"ok": True, "threshold_s": threshold}) - - -@app.route("/status.json") -def status_json(): - a = require_auth() - if a: return a - cfg = dashboard_mod.load_config() - return jsonify({ - "config_path": str(dashboard_mod.CONFIG_PATH), - "refresh_interval_s": cfg.get("refresh_interval_s"), - "slots": [{"plugin": s.get("plugin"), "config_keys": list(s.get("config", {}).keys())} for s in cfg.get("slots", [])], - "socket_path": dashboard_mod.SOCKET_PATH, - }) - - -# ============================================================================ -# Network / WiFi Management -# ============================================================================ -@app.route("/api/network/status") -def api_net_status(): - a = require_auth() - if a: return a - return jsonify(net.get_watchdog().get_state()) - - -@app.route("/api/network/scan") -def api_net_scan(): - a = require_auth() - if a: return a - return jsonify({"networks": net.list_wifi_networks()}) - - -@app.route("/api/network/saved") -def api_net_saved(): - a = require_auth() - if a: return a - return jsonify({"saved": net.list_saved_connections()}) - - -@app.route("/api/network/connect", methods=["POST"]) -def api_net_connect(): - a = require_auth() - if a: return a - ssid = request.form.get("ssid", "").strip() - pwd = request.form.get("password", "").strip() - security = request.form.get("security", "wpa-psk").strip() - if not ssid: - return jsonify({"ok": False, "error": "ssid is required"}), 400 - ok, msg = net.save_wifi(ssid, pwd, security) - if not ok: - return jsonify({"ok": False, "error": msg}), 500 - # Falls aktuell im AP-Modus: AP aus - if net.is_ap_active(): - net.stop_ap() - ok, msg = net.connect_wifi(msg) # msg contains connection name - return jsonify({"ok": ok, "message": msg, "ssid": ssid}) - - -@app.route("/api/network/reconnect", methods=["POST"]) -def api_net_reconnect(): - a = require_auth() - if a: return a - saved = net.list_saved_connections() - if not saved: - return jsonify({"ok": False, "error": "kein gespeichertes WLAN"}), 400 - ok, msg = net.connect_wifi(saved[0]["name"]) - return jsonify({"ok": ok, "message": msg}) - - -@app.route("/api/network/disconnect", methods=["POST"]) -def api_net_disconnect(): - a = require_auth() - if a: return a - ok, msg = net.disconnect_wifi() - return jsonify({"ok": ok, "message": msg}) - - -@app.route("/api/network/ap/start", methods=["POST"]) -def api_net_ap_start(): - a = require_auth() - if a: return a - ok, msg = net.start_ap() - return jsonify({"ok": ok, "message": msg}) - - -@app.route("/api/network/ap/stop", methods=["POST"]) -def api_net_ap_stop(): - a = require_auth() - if a: return a - ok, msg = net.stop_ap() - return jsonify({"ok": ok, "message": msg}) - - -@app.route("/api/network/forget", methods=["POST"]) -def api_net_forget(): - a = require_auth() - if a: return a - name = request.form.get("name", "").strip() - if not name: - return jsonify({"ok": False, "error": "name required"}), 400 - rc, out, err = net._nm(["connection", "delete", name], timeout=10) - return jsonify({"ok": rc == 0, "message": out or err}) - - -def main(): - # Wenn direkt ausgeführt: eigenen Renderer starten ist Sache des Run-Skripts. - # Wir nehmen einfach Port 8080. - host = os.environ.get("EPAPER_ADMIN_HOST", "0.0.0.0") - port = int(os.environ.get("EPAPER_ADMIN_PORT", "8080")) - app.run(host=host, port=port, debug=False, use_reloader=False) - - -if __name__ == "__main__": - main() diff --git a/.backup/2026-08-26-active/config.example.json b/.backup/2026-08-26-active/config.example.json deleted file mode 100644 index 407c064..0000000 --- a/.backup/2026-08-26-active/config.example.json +++ /dev/null @@ -1,10 +0,0 @@ -{ - "version": 1, - "refresh_interval_s": 180, - "slots": [ - {"plugin": "clock", "config": {}}, - {"plugin": "weather", "config": {"location": "52.52,13.41", "show_uv": true, "show_forecast_hours": 4}}, - {"plugin": "system", "config": {"show_uptime": true, "show_load": true}}, - {"plugin": "hello", "config": {"text": "Edit me in the web UI!", "color": "accent", "size": 40}} - ] -} diff --git a/.backup/2026-08-26-active/dashboard.py b/.backup/2026-08-26-active/dashboard.py deleted file mode 100644 index 02638c2..0000000 --- a/.backup/2026-08-26-active/dashboard.py +++ /dev/null @@ -1,463 +0,0 @@ -"""Layout, Slot-Geometrie, Refresh-Loop. - -Lädt config.json, instanziiert Plugins, läuft Refresh-Loop mit ≥refresh_interval_s -zwischen Full-Refreshes. Reagiert auf Live-Trigger via Unix-Socket. -""" -from __future__ import annotations -import os, sys, time, json, math, threading, signal, socket, logging -from datetime import datetime -from pathlib import Path - -HERE = Path(__file__).resolve().parent -sys.path.insert(0, str(HERE)) - -from PIL import Image, ImageDraw, ImageFont -from palette import BG, FG, OK -from plugins.base import all_widget_classes, Widget -from layout import Item, GRID_COLS, GRID_ROWS, CELL_W, CELL_H, DISPLAY_W as LAYOUT_DISPLAY_W, pack, find_overlaps, find_out_of_bounds -import renderer - -import importlib, pkgutil -import plugins as _plugins_pkg # ensure package is importable - - -# ============================================================================ -# Config helpers -# ============================================================================ -CONFIG_PATH = HERE / "config.json" -LOG_PATH = HERE / "dashboard.log" -SOCKET_PATH = "/tmp/epaper-dashboard.sock" - -DEFAULT_CONFIG = { - "version": 1, - "refresh_interval_s": 180, - "slots": [ - {"plugin": "clock", "config": {}}, - {"plugin": "weather", "config": {}}, - {"plugin": "system", "config": {}}, - {"plugin": "hello", "config": {}}, - ], -} - -# Display dimensions are imported from layout module (single source of truth) -DISPLAY_W = LAYOUT_DISPLAY_W -DISPLAY_H = GRID_ROWS * CELL_H -# Legacy-v1 GRID bleibt für Migration -LEGACY_SLOTS = 4 - - -def migrate_v1_to_v2(v1: dict) -> dict: - """Konvertiert altes Layout (slots[]) in v2 (layout.items + plugin_configs).""" - layout_items = [] - plugin_configs = {} - # Positioniere quadratisch im 4x4 grid, fall-back auf legacy 2x2 - # Hier nutzen wir das alte 2x2 mapping und packen dann: - # slot 0 -> (0,0) 2x2 - # slot 1 -> (2,0) 2x2 - # slot 2 -> (0,2) 2x2 - # slot 3 -> (2,2) 2x2 - legacy_pos = [(0, 0), (2, 0), (0, 2), (2, 2)] - for i, slot in enumerate(v1.get("slots", [])): - if i >= LEGACY_SLOTS: - break - plugin = slot.get("plugin", "") - cfg = slot.get("config", {}) - if not plugin: - continue - x, y = legacy_pos[i] - item_id = f"migrated_{i}" - layout_items.append({ - "id": item_id, - "plugin": plugin, - "x": x, "y": y, "w": 2, "h": 2, - }) - # Plugin-Config landet in plugin_configs (ohne slot-id, geteilt pro plugin-name) - if cfg and plugin not in plugin_configs: - plugin_configs[plugin] = cfg - return { - "version": 2, - "refresh_interval_s": v1.get("refresh_interval_s", 180), - "layout": { - "grid": {"cols": 4, "rows": 4}, - "items": layout_items, - }, - "plugin_configs": plugin_configs, - } - - -def load_config() -> dict: - if not CONFIG_PATH.exists(): - return _empty_v2_config() - try: - cfg = json.loads(CONFIG_PATH.read_text()) - except Exception as e: - logging.error(f"config.json invalid: {e}, using defaults") - return _empty_v2_config() - if cfg.get("version", 1) < 2: - logging.info("config: migrating v1 → v2") - cfg = migrate_v1_to_v2(cfg) - save_config(cfg) - # Ensure required keys exist - if "layout" not in cfg or "items" not in cfg.get("layout", {}): - cfg["layout"] = {"grid": {"cols": 4, "rows": 4}, "items": []} - if "plugin_configs" not in cfg: - cfg["plugin_configs"] = {} - if "refresh_interval_s" not in cfg: - cfg["refresh_interval_s"] = 180 - return cfg - - -def save_config(cfg: dict) -> None: - CONFIG_PATH.write_text(json.dumps(cfg, indent=2)) - - -def _empty_v2_config() -> dict: - return { - "version": 2, - "refresh_interval_s": 180, - "layout": { - "grid": {"cols": 4, "rows": 4}, - "items": [], - }, - "plugin_configs": {}, - } - - -# ============================================================================ -# Plugin Registry -# ============================================================================ -_widget_cache: dict[str, type[Widget]] = {} - - -def get_widget_classes() -> dict[str, type[Widget]]: - """Lazy + cached: lädt alle Plugin-Klassen einmal.""" - if not _widget_cache: - for cls in all_widget_classes(): - _widget_cache[cls.name] = cls - return _widget_cache - - -def make_widget(plugin_name: str, config: dict) -> Widget | None: - classes = get_widget_classes() - cls = classes.get(plugin_name) - if cls is None: - logging.warning(f"plugin not found: {plugin_name}") - return None - try: - w = cls(config) - w.on_load() - return w - except Exception as e: - logging.error(f"failed to instantiate plugin {plugin_name}: {e}") - return None - - -# ============================================================================ -# Font helpers -# ============================================================================ -FONT_DIR = HERE / "fnt" - - -def load_fonts() -> dict: - if not FONT_DIR.exists(): - FONT_DIR.mkdir(parents=True, exist_ok=True) - - # Reale TTF-Fonts: Orbitron (digital-look), Rubik (clean sans), LiberationMono - orbitron = FONT_DIR / "Orbitron.ttf" - rubik = FONT_DIR / "Rubik.ttf" - liberation = "/usr/share/fonts/truetype/liberation/LiberationMono-Regular.ttf" - liberation_b = "/usr/share/fonts/truetype/liberation/LibrationMono-Bold.ttf" - - fonts = {} - - # Rubik —clean sans für normalen text - if rubik.exists(): - for s in [12, 14, 16, 18, 20, 24, 28, 32, 40, 48, 64]: - try: - fonts[str(s)] = ImageFont.truetype(str(rubik), s) - except Exception: - pass - fonts["default"] = fonts.get("16", ImageFont.load_default()) - - # Orbitron — digital segment display look für uhrzeit - if orbitron.exists(): - try: - fonts["clock"] = ImageFont.truetype(str(orbitron), 80) - except Exception: - pass - - # LiberationMono — monospace für zahlen/datum - try: - fonts["mono"] = ImageFont.truetype(liberation, 16) - fonts["mono_b"] = ImageFont.truetype(liberation_b, 16) - except Exception: - pass - - if not fonts: - fonts["default"] = ImageFont.load_default() - - return fonts - - -# ============================================================================ -# Slot-Geometrie (legacy v1) -# ============================================================================ -def slot_box(idx: int) -> tuple[int, int, int, int]: - """Legacy v1: gibt (x, y, w, h) für Slot idx (0..3) zurück.""" - col = idx % 2 - row = idx // 2 - w = DISPLAY_W // 2 - h = DISPLAY_H // 2 - x = col * w - y = row * h - return x, y, w, h - - -# ============================================================================ -# Render -# ============================================================================ -def render_full(items: list, widgets: list, fonts: dict) -> Image.Image: - """Render alle Items aufs Display. - - items: list[Item] (Layout-Items mit x,y,w,h) - widgets: list[Widget | None] — index-parallel zu items; None = leer - """ - img = Image.new("RGB", (DISPLAY_W, DISPLAY_H), BG) - draw = ImageDraw.Draw(img) - - for item, widget in zip(items, widgets): - px, py, pw, ph = item.pixels() - if widget is None: - draw.text((px + 8, py + 8), f"#{item.id[:6]} (no plugin)", - font=fonts.get("16", fonts.get("default")), fill=FG) - continue - if item.y >= 4: # overflow marker - draw.text((px + 8, py + 8), f"{widget.label} (overflow)", - font=fonts.get("16", fonts.get("default")), fill=FG) - continue - try: - widget.render(draw, fonts, px, py, pw, ph) - except Exception as e: - logging.error(f"plugin {widget.name} render failed: {e}") - draw.text((px + 8, py + 8), - f"{widget.label}: render error", - font=fonts.get("16", fonts.get("default")), fill=FG) - - return img - - -# ============================================================================ -# Refresh-Loop -# ============================================================================ -class Dashboard: - def __init__(self): - self.config = load_config() - self.fonts = load_fonts() - self.items: list[Item] = [] - self.widgets: list[Widget | None] = [] - self.last_render: Image.Image | None = None - self.config_mtime = CONFIG_PATH.stat().st_mtime if CONFIG_PATH.exists() else 0 - self.last_refresh_ts: float = 0.0 - self._trigger = threading.Event() - self._trigger.set() # trigger initial refresh - self._epd = None - self._lock = threading.Lock() - self.reload() - - def reload(self): - """Lade config und baue alle Widgets.""" - cfg = self.config - classes = get_widget_classes() - plugin_configs = cfg.get("plugin_configs", {}) - items = [Item.from_dict(d) for d in cfg.get("layout", {}).get("items", [])] - widgets = [] - for item in items: - plugin_cfg = plugin_configs.get(item.plugin, {}) - widgets.append(make_widget(item.plugin, plugin_cfg)) - with self._lock: - self.items = items - self.widgets = widgets - self.config_mtime = CONFIG_PATH.stat().st_mtime if CONFIG_PATH.exists() else 0 - self.display_theme = cfg.get("display_theme", "default") - self.display_design = cfg.get("display_design", "classic") - - def maybe_reload_config(self): - if not CONFIG_PATH.exists(): - return - mtime = CONFIG_PATH.stat().st_mtime - if mtime != self.config_mtime: - logging.info("config.json changed, reloading") - try: - self.config = load_config() - self.reload() - except Exception as e: - logging.error(f"reload failed: {e}") - - def trigger_refresh(self): - self._trigger.set() - - def render_once(self) -> Image.Image: - # fetch all widgets - for w in self.widgets: - if w is None: continue - try: - w.fetch() - except Exception as e: - logging.error(f"plugin {w.name} fetch failed: {e}") - with self._lock: - items = list(self.items) - widgets = list(self.widgets) - design = getattr(self, 'display_design', 'classic') - img = renderer.render_design(items, widgets, self.fonts, design=design) - return img - - def display(self, img: Image.Image): - # Lazy-import waveshare so der Renderer auch ohne Display testbar ist - if self._epd is None: - sys.path.insert(0, str(HERE / "waveshare_epd")) - try: - from waveshare_epd import epd7in3f - self._epd = epd7in3f.EPD() - self._epd.init() - self._epd.Clear() - except Exception as e: - logging.error(f"display init failed: {e}") - self._epd = "ERROR" - return False - if self._epd == "ERROR": - return False - try: - self._epd.display(self._epd.getbuffer(img)) - return True - except Exception as e: - logging.error(f"display error: {e}") - return False - - def run(self): - logging.info("dashboard starting; refresh_interval=%ss", self.config.get("refresh_interval_s")) - # start IPC socket listener - threading.Thread(target=self._socket_loop, daemon=True).start() - - while True: - try: - self.maybe_reload_config() - interval = int(self.config.get("refresh_interval_s", 180)) - # sleep until interval elapsed OR trigger set - wait = max(5, interval) - logging.info(f"waiting up to {wait}s (manual trigger interrupts)") - if self._trigger.wait(timeout=wait): - self._trigger.clear() - logging.info("manual trigger received") - else: - logging.info("interval elapsed, refreshing") - - t0 = time.time() - img = self.render_once() - with self._lock: - self.last_render = img - ok = self.display(img) - self.last_refresh_ts = time.time() - logging.info(f"refresh done in {time.time()-t0:.1f}s, display={ok}") - - # Re-Reload config after render in case user changed it during refresh - self.maybe_reload_config() - - except KeyboardInterrupt: - break - except Exception as e: - logging.error(f"loop error: {e}", exc_info=True) - time.sleep(5) - - def _socket_loop(self): - """Empfängt 'refresh\n' über Unix-Socket.""" - if os.path.exists(SOCKET_PATH): - os.remove(SOCKET_PATH) - srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - srv.bind(SOCKET_PATH) - os.chmod(SOCKET_PATH, 0o660) - srv.listen(5) - logging.info(f"ipc socket at {SOCKET_PATH}") - while True: - try: - conn, _ = srv.accept() - data = conn.recv(1024).decode("utf-8", errors="ignore").strip() - if data == "refresh": - self.trigger_refresh() - conn.sendall(b"OK\n") - elif data == "ping": - conn.sendall(b"PONG\n") - elif data.startswith("snapshot"): - # Schnappschuss-Pfad als Antwort - with self._lock: - if self.last_render: - tmp = "/tmp/epaper-snapshot.png" - self.last_render.save(tmp) - conn.sendall(f"OK {tmp}\n".encode()) - else: - conn.sendall(b"NO_RENDER_YET\n") - else: - conn.sendall(b"UNKNOWN_CMD\n") - conn.close() - except Exception as e: - logging.debug(f"socket error: {e}") - time.sleep(1) - - -def send_ipc(cmd: str, host: str = "localhost", timeout: float = 2.0) -> str: - """Helper für die Admin-UI: schickt ein Kommando an den laufenden Renderer.""" - if cmd == "snapshot": - # Snapshot holen ist eine HTTP-Aufgabe, kein IPC - return "" - try: - s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - s.settimeout(timeout) - s.connect(SOCKET_PATH) - s.sendall((cmd + "\n").encode()) - data = b"" - while True: - chunk = s.recv(4096) - if not chunk: - break - data += chunk - s.close() - return data.decode(errors="ignore").strip() - except Exception as e: - return f"IPC_ERROR: {e}" - - -def setup_logging(): - handlers = [logging.StreamHandler()] - try: - from logging.handlers import RotatingFileHandler - handlers.append(RotatingFileHandler(LOG_PATH, maxBytes=512*1024, backupCount=1)) - except Exception: - pass - logging.basicConfig( - level=logging.INFO, - format="%(asctime)s %(levelname)s %(message)s", - handlers=handlers, - ) - - -def main(): - setup_logging() - d = Dashboard() - # signal handlers for clean shutdown - def stop(*_): - raise KeyboardInterrupt - signal.signal(signal.SIGTERM, stop) - signal.signal(signal.SIGINT, stop) - - # Network-Watchdog starten (überwacht WLAN, startet Recovery-AP) - try: - import network_watchdog as _netwd - wd = _netwd.get_watchdog() - logging.info("network watchdog started") - except Exception as e: - logging.error(f"failed to start network watchdog: {e}") - - d.run() - - -if __name__ == "__main__": - main() diff --git a/.backup/2026-08-26-active/layout.py b/.backup/2026-08-26-active/layout.py deleted file mode 100644 index a89ab55..0000000 --- a/.backup/2026-08-26-active/layout.py +++ /dev/null @@ -1,172 +0,0 @@ -"""Grid-Layout: 4x4-Grid mit Pack-Algorithmus und Konflikt-Detection. - -Display: 800x480 Pixel. -Grid: 4x4 Zellen, jede Zelle 200x120 Pixel. - -Item-Geometrie: - (x, y, w, h) in Zelleneinheiten, x/y ∈ [0..GRID_COLS-1], w/h ∈ [1..GRID_COLS] - (x+w) ≤ GRID_COLS, (y+h) ≤ GRID_ROWS - -Regeln: - - Items dürfen sich nicht überlappen - - Wenn mehrere Items die gleiche Zelle belegen wollen → Konflikt - - Pack-Algorithmus: greedy, scan-line, größte zuerst -""" -from __future__ import annotations -from dataclasses import dataclass, field, asdict -from typing import Optional - - -GRID_COLS = 4 -GRID_ROWS = 4 -CELL_W = 200 # px -CELL_H = 120 # px -DISPLAY_W = GRID_COLS * CELL_W # 800 -DISPLAY_H = GRID_ROWS * CELL_H # 480 - -# Standard-Größen die im UI als Buttons angeboten werden -SIZE_PRESETS = [ - (1, 1), (2, 1), (1, 2), (2, 2), - (4, 1), (1, 4), (2, 4), (4, 2), - (3, 1), (1, 3), (3, 2), (2, 3), (3, 3), - (4, 4), -] - - -@dataclass -class Item: - id: str - plugin: str - x: int = 0 - y: int = 0 - w: int = 1 - h: int = 1 - - def to_dict(self): - return asdict(self) - - @classmethod - def from_dict(cls, d: dict) -> "Item": - return cls( - id=str(d.get("id", _new_id())), - plugin=str(d.get("plugin", "")), - x=int(d.get("x", 0)), - y=int(d.get("y", 0)), - w=int(d.get("w", 1)), - h=int(d.get("h", 1)), - ) - - def bounds(self) -> tuple[int, int, int, int]: - """Returns (x, y, w, h).""" - return self.x, self.y, self.w, self.h - - def pixels(self) -> tuple[int, int, int, int]: - """Returns (x, y, width, height) in pixels.""" - return self.x * CELL_W, self.y * CELL_H, self.w * CELL_W, self.h * CELL_H - - -def _new_id() -> str: - import secrets - return secrets.token_hex(4) - - -def cells_occupied(item: Item) -> set[tuple[int, int]]: - """Set of (col, row) cells this item covers.""" - return {(item.x + dx, item.y + dy) for dx in range(item.w) for dy in range(item.h)} - - -def find_overlaps(items: list[Item]) -> list[tuple[str, str]]: - """Returns list of (id_a, id_b) pairs that overlap.""" - overlaps = [] - for i, a in enumerate(items): - cells_a = cells_occupied(a) - for b in items[i+1:]: - if cells_a & cells_occupied(b): - overlaps.append((a.id, b.id)) - return overlaps - - -def find_out_of_bounds(items: list[Item]) -> list[str]: - """Returns list of item IDs that are outside the grid.""" - bad = [] - for it in items: - if it.x < 0 or it.y < 0 or it.x + it.w > GRID_COLS or it.y + it.h > GRID_ROWS: - bad.append(it.id) - return bad - - -def pack(items: list[Item], order: Optional[list[str]] = None) -> list[Item]: - """Auto-pack items into a 4x4 grid. - - Strategy: greedy first-fit, sort by area descending (largest first). - Items that don't fit are placed in a 'trash row' below the grid (y=GRID_ROWS, - h=1) so they are visible to the user as overflowing. - - `order` (optional): explicit ordering by item id. - """ - # Sort: largest area first; tie-break by id for determinism - indexed = list(items) - if order: - priority = {id_: i for i, id_ in enumerate(order)} - indexed.sort(key=lambda it: (priority.get(it.id, 9999), -it.w * it.h, it.id)) - else: - indexed.sort(key=lambda it: (-it.w * it.h, it.id)) - - taken: set[tuple[int, int]] = set() - result: list[Item] = [] - overflow_row: list[Item] = [] - - for it in indexed: - # clamp size to grid - it.w = max(1, min(GRID_COLS, it.w)) - it.h = max(1, min(GRID_ROWS, it.h)) - # try to place - placed = False - for y in range(GRID_ROWS - it.h + 1): - for x in range(GRID_COLS - it.w + 1): - cells = {(x + dx, y + dy) for dx in range(it.w) for dy in range(it.h)} - if not (cells & taken): - it.x, it.y = x, y - taken |= cells - result.append(it) - placed = True - break - if placed: break - if not placed: - # couldn't fit → overflow marker - it.x = 0 - it.y = GRID_ROWS # off-screen - overflow_row.append(it) - result.append(it) - return result - - -def auto_size_for_plugin(plugin_name: str) -> tuple[int, int]: - """Default size when user adds a new item.""" - presets = { - "clock": (2, 2), - "weather": (2, 2), - "system": (2, 2), - "minimax": (2, 2), - "spotify": (2, 1), - "strava": (2, 1), - "gmail": (1, 1), - "hello": (1, 1), - } - return presets.get(plugin_name, (2, 2)) - - -def render_grid_layout(items: list[Item]) -> dict: - """Render each item to its (x, y, w, h) pixel-box on the 800x480 display.""" - boxes = [] - for it in items: - px, py, pw, ph = it.pixels() - # Overflow items still get rendered as small placeholder, not visible - boxes.append({ - "id": it.id, - "plugin": it.plugin, - "x": px, "y": py, "w": pw, "h": ph, - "out_of_bounds": it.y >= GRID_ROWS, - }) - return {"display_w": DISPLAY_W, "display_h": DISPLAY_H, "grid_cols": GRID_COLS, - "grid_rows": GRID_ROWS, "cell_w": CELL_W, "cell_h": CELL_H, "items": boxes} diff --git a/.backup/2026-08-26-active/network_watchdog.py b/.backup/2026-08-26-active/network_watchdog.py deleted file mode 100644 index 856e9e4..0000000 --- a/.backup/2026-08-26-active/network_watchdog.py +++ /dev/null @@ -1,468 +0,0 @@ -"""Network-Watchdog: überwacht WLAN-Verbindung und startet Recovery-AP bei Ausfall. - -Strategie: - 1. Default-Gateway-Ping alle CHECK_INTERVAL_S - 2. failed_pings zählt; bei FAIL_THRESHOLD sofortiger Reconnect-Versuch - 3. Nach REASSURE_TIMEOUT_S erfolglos: AP-Modus starten - 4. Im AP-Modus: alle AP_RECHECK_S prüfen ob echtes WLAN zurück ist, dann AP aus -""" -from __future__ import annotations -import subprocess, time, threading, logging, os -from pathlib import Path -from dataclasses import dataclass, field - -# Defaults -WLAN_IFACE = "wlan0" # Pi 4 builtin -CLIENT_CON_NAME = "dashboard-wifi" # NetworkManager connection profile name -AP_SSID = "epaper-recovery" -AP_PASSWORD = "recovery1234" - -CHECK_INTERVAL_S = 30 # alle 30s pingen -FAIL_THRESHOLD = 2 # 2 misses hintereinander → reconnect versuchen -RECONNECT_COOLDOWN_S = 90 # zwischen zwei Reconnect-Versuchen mind. 90s warten -AP_RECHECK_S = 60 # im AP-Modus: alle 60s versuchen zurück zu wechseln -AP_ENABLE_TIMEOUT_S = 30 # warten bis AP steht, dann gilt: "AP ist aktiv" -PING_TIMEOUT_S = 3 -PING_TARGETS = ["1.1.1.1", "8.8.8.8"] # wenn eins antwortet → online - - -@dataclass -class NetState: - mode: str = "unknown" # "client" | "ap" | "connecting" | "offline" | "unknown" - ssid: str = "" - ip: str = "" - signal: int = 0 # 0..100 - gateway_ping_ms: float = 0.0 - failed_pings: int = 0 - last_state_change: float = field(default_factory=time.time) - last_reconnect_attempt: float = 0.0 - error: str = "" - - -def _nm(args: list[str], timeout: int = 15) -> tuple[int, str, str]: - """Run nmcli with timeout, return (rc, stdout, stderr).""" - try: - r = subprocess.run(["nmcli", *args], capture_output=True, text=True, timeout=timeout) - return r.returncode, r.stdout, r.stderr - except subprocess.TimeoutExpired: - return -1, "", "timeout" - except Exception as e: - return -1, "", str(e) - - -def ping(target: str) -> float | None: - try: - r = subprocess.run( - ["ping", "-c", "1", "-W", str(PING_TIMEOUT_S), target], - capture_output=True, text=True, timeout=PING_TIMEOUT_S + 2, - ) - if r.returncode != 0: - return None - for line in r.stdout.splitlines(): - if "time=" in line: - return float(line.split("time=")[1].split()[0]) - except Exception: - return None - return None - - -def get_active_wifi_info() -> dict: - """Liest aktuelle WLAN-Informationen aus NetworkManager.""" - rc, out, err = _nm(["-t", "-f", "ACTIVE,SSID,SIGNAL,FREQ,CHAN,RATE,BARS", "device", "wifi"]) - info = {"ssid": "", "signal": 0, "bars": "", "active": False} - if rc != 0: - return info - for line in out.splitlines(): - parts = line.split(":") - if len(parts) < 6: - continue - # Format: yes:ssid:signal:... - if parts[0] == "yes": - info["active"] = True - info["ssid"] = parts[1] if len(parts) > 1 else "" - try: - info["signal"] = int(parts[2]) if len(parts) > 2 and parts[2] else 0 - except ValueError: - pass - info["bars"] = parts[6] if len(parts) > 6 else "" - break - return info - - -def get_wifi_ip() -> str: - """Hole IPv4-Adresse von WLAN-Interface.""" - try: - r = subprocess.run( - ["ip", "-4", "-o", "addr", "show", "dev", WLAN_IFACE], - capture_output=True, text=True, timeout=5, - ) - for line in r.stdout.splitlines(): - if "inet " in line: - return line.split("inet ")[1].split("/")[0] - except Exception: - pass - return "" - - -def list_wifi_networks() -> list[dict]: - """Scannt verfügbare WLANs.""" - rc, out, err = _nm(["-t", "-f", "SSID,SIGNAL,SECURITY,FREQ,CHAN", "device", "wifi", - "list", "--rescan", "yes"]) - if rc != 0: - # zweiter versuch ohne rescan - rc, out, err = _nm(["-t", "-f", "SSID,SIGNAL,SECURITY,FREQ,CHAN", "device", "wifi", "list"]) - seen = set() - result = [] - for line in out.splitlines(): - parts = line.split(":") - if len(parts) < 3: continue - ssid = parts[0] - if not ssid or ssid in seen: continue - seen.add(ssid) - try: - sig = int(parts[1]) if parts[1] else 0 - except ValueError: - sig = 0 - result.append({"ssid": ssid, "signal": sig, "security": parts[2] if len(parts) > 2 else "", - "freq": parts[3] if len(parts) > 3 else "", - "chan": parts[4] if len(parts) > 4 else ""}) - result.sort(key=lambda w: w["signal"], reverse=True) - return result - - -def list_saved_connections() -> list[dict]: - """Liste alle gespeicherten WLAN-Profile.""" - rc, out, err = _nm(["-t", "-f", "NAME,TYPE,UUID", "connection", "show"]) - result = [] - for line in out.splitlines(): - parts = line.split(":") - if len(parts) < 3: continue - if parts[1] != "802-11-wireless": continue - if parts[0] in (AP_SSID, "Hotspot"): continue # skip AP profile - result.append({"name": parts[0], "uuid": parts[2]}) - return result - - -def save_wifi(ssid: str, password: str, security: str = "wpa-psk") -> tuple[bool, str]: - """Speichere (oder update) ein WLAN-Profil mit gegebenen Credentials. - - Wenn schon ein Profil mit gleicher SSID existiert, wird es aktualisiert; - sonst wird ein neues angelegt. - """ - # Existierendes Profil finden - saved = list_saved_connections() - target_name = None - for c in saved: - if c["name"].startswith(ssid): - target_name = c["name"] - break - if target_name is None: - target_name = CLIENT_CON_NAME + "-" + ssid.replace(" ", "_") - - args = ["connection", "modify" if target_name in [c["name"] for c in saved] else "add", - "type", "wifi", - "con-name", target_name, - "ifname", WLAN_IFACE, - "ssid", ssid] - if security and security != "" and password: - if "WPA" in security.upper() or "WPA2" in security.upper() or "WPA3" in security.upper(): - args += ["wifi-sec.key-mgmt", "wpa-psk", "wifi-sec.psk", password] - elif "WEP" in security.upper(): - args += ["wifi-sec.key-mgmt", "none", "wifi-sec.wep-key-type", "1", - "wifi-sec.wep-key0", password] - else: - args += ["wifi-sec.key-mgmt", "wpa-psk", "wifi-sec.psk", password] - else: - args += ["wifi-sec.key-mgmt", "none"] - rc, out, err = _nm(args, timeout=20) - if rc != 0: - return False, err or out or "unknown error" - return True, target_name - - -def connect_wifi(connection_name: str) -> tuple[bool, str]: - rc, out, err = _nm(["connection", "up", connection_name], timeout=30) - if rc != 0: - return False, err or out or "unknown error" - return True, "connected" - - -def disconnect_wifi() -> tuple[bool, str]: - rc, out, err = _nm(["device", "disconnect", WLAN_IFACE], timeout=10) - return rc == 0, err if rc != 0 else "ok" - - -def start_ap() -> tuple[bool, str]: - """Startet einen Hotspot auf wlan0 mit fester SSID/PW.""" - # Alten Hotspot ggf. löschen (idempotent) - _nm(["connection", "delete", AP_SSID], timeout=10) - rc, out, err = _nm([ - "device", "wifi", "hotspot", - "ifname", WLAN_IFACE, - "ssid", AP_SSID, - "password", AP_PASSWORD, - ], timeout=30) - if rc != 0: - return False, err or out - return True, "hotspot up" - - -def stop_ap() -> tuple[bool, str]: - rc, out, err = _nm(["connection", "down", AP_SSID], timeout=15) - if rc != 0: - return False, err or out - _nm(["connection", "delete", AP_SSID], timeout=10) - return True, "ap stopped" - - -def is_ap_active() -> bool: - rc, out, _ = _nm(["-t", "-f", "NAME", "connection", "show", "--active"]) - if rc != 0: - return False - return AP_SSID in out.splitlines() - - -def is_wifi_connected() -> bool: - rc, out, _ = _nm(["-t", "-f", "STATE", "device", "show", WLAN_IFACE]) - if rc != 0: - return False - return "connected" in out.lower() - - -# ============================================================================ -# Recovery threshold helper (used by admin.py) -# ============================================================================ -def get_recovery_threshold() -> int: - """Liest recovery_threshold_s aus config.json (default 60s).""" - try: - from pathlib import Path as _P - cfg_path = _P("/home/koptikp/epaper-dashboard/port/config.json") - if cfg_path.exists(): - import json as _json - cfg = _json.loads(cfg_path.read_text()) - return int(cfg.get("recovery_threshold_s", 60)) - except Exception: - pass - return 60 - - -# ============================================================================ -# Watchdog -# ============================================================================ -def _check_recovery_needed(state: dict, threshold_s: int) -> dict: - """Entscheidet, ob die Recovery-Info UI angezeigt werden soll. - - Returns dict mit: - show: bool - reason: 'offline' | 'ap_active' | 'connecting_timeout' | 'no_ip' - duration_s: Sekunden im aktuellen (nicht-client) Zustand - """ - mode = state.get("mode", "unknown") - if mode == "client": - return {"show": False, "reason": "client", "duration_s": 0} - # Im AP-Modus: immer anzeigen - if mode == "ap": - return {"show": True, "reason": "ap_active", "duration_s": state.get("since_change_s", 0)} - if mode == "offline": - return {"show": True, "reason": "offline", "duration_s": state.get("since_change_s", 0)} - if mode == "connecting": - # Connecting dauert zu lange? Zeige info. - return {"show": state.get("since_change_s", 0) > threshold_s, - "reason": "connecting_timeout", - "duration_s": state.get("since_change_s", 0)} - return {"show": False, "reason": "unknown", "duration_s": state.get("since_change_s", 0)} - - -class Watchdog: - def __init__(self): - self.state = NetState() - self._lock = threading.Lock() - self._stop = threading.Event() - - def get_state(self) -> dict: - with self._lock: - return { - "mode": self.state.mode, - "ssid": self.state.ssid, - "ip": self.state.ip, - "signal": self.state.signal, - "gateway_ping_ms": round(self.state.gateway_ping_ms, 1), - "failed_pings": self.state.failed_pings, - "since_change_s": int(time.time() - self.state.last_state_change), - "error": self.state.error, - } - - def _set_mode(self, new_mode: str): - with self._lock: - if self.state.mode != new_mode: - logging.info(f"network mode: {self.state.mode} → {new_mode}") - self.state.mode = new_mode - self.state.last_state_change = time.time() - self.state.error = "" - - def run_once(self) -> dict: - """Führt einen Check durch und triggert ggf. State-Übergänge.""" - with self._lock: - cur_mode = self.state.mode - - ping_ms = None - for tgt in PING_TARGETS: - r = ping(tgt) - if r is not None: - ping_ms = r - break - - wifi_info = get_active_wifi_info() - ip = get_wifi_ip() - - if wifi_info["active"] and ip and ping_ms is not None: - # ONLINE - with self._lock: - self.state.ssid = wifi_info["ssid"] - self.state.signal = wifi_info["signal"] - self.state.ip = ip - self.state.gateway_ping_ms = ping_ms - self.state.failed_pings = 0 - # Falls im AP-Modus → AP aus - if cur_mode == "ap": - logging.info("client wifi back, stopping AP") - stop_ap() - self._set_mode("client") - return self.get_state() - - # OFFLINE PATH - with self._lock: - if ping_ms is None: - self.state.failed_pings += 1 - self.state.gateway_ping_ms = ping_ms or 0.0 - self.state.ip = ip - self.state.ssid = wifi_info["ssid"] - self.state.signal = wifi_info["signal"] - failed = self.state.failed_pings - last_recon = self.state.last_reconnect_attempt - - if cur_mode == "ap": - # Bereits im AP — nur gelegentlich versuchen zurück zu wechseln - # (durch den regulären 60s-Loop in self._loop) - return self.get_state() - - if failed >= FAIL_THRESHOLD: - now = time.time() - if now - last_recon < RECONNECT_COOLDOWN_S: - return self.get_state() - # Reconnect versuchen - with self._lock: - self.state.last_reconnect_attempt = now - saved = list_saved_connections() - if saved: - logging.info(f"wifi lost ({failed} failed pings), reconnecting to {saved[0]['name']}") - self._set_mode("connecting") - ok, msg = connect_wifi(saved[0]["name"]) - if ok: - time.sleep(AP_ENABLE_TIMEOUT_S) - # Nach reconnect: wenn immer noch offline → AP - return self.run_once() - else: - logging.error(f"reconnect failed: {msg}") - with self._lock: - self.state.error = msg - else: - # Kein Profil → direkt in AP - self._set_mode("connecting") - return self._enter_ap() - return self.get_state() - - def _enter_ap(self) -> dict: - logging.warning(f"entering AP mode ({AP_SSID})") - ok, msg = start_ap() - if ok: - self._set_mode("ap") - with self._lock: - self.state.ip = "10.42.0.1" - self.state.error = "" - else: - logging.error(f"failed to start AP: {msg}") - self._set_mode("offline") - with self._lock: - self.state.error = msg - return self.get_state() - - def _try_exit_ap(self) -> dict: - """Versuch, aus dem AP zurück in Client-Modus zu wechseln.""" - if not is_ap_active(): - self._set_mode("client") - return self.get_state() - saved = list_saved_connections() - if not saved: - return self.get_state() - logging.info("trying to exit AP back to client") - ok, msg = connect_wifi(saved[0]["name"]) - if ok: - time.sleep(AP_ENABLE_TIMEOUT_S) - wifi_info = get_active_wifi_info() - ip = get_wifi_ip() - if wifi_info["active"] and ip and ping(PING_TARGETS[0]) is not None: - logging.info("back online, stopping AP") - stop_ap() - self._set_mode("client") - with self._lock: - self.state.failed_pings = 0 - else: - # Immer noch kein echtes WLAN → AP wieder hoch - start_ap() - self._set_mode("ap") - return self.get_state() - - def _loop(self): - # Initial: state aus NM ableiten - wifi_info = get_active_wifi_info() - ip = get_wifi_ip() - if is_ap_active(): - self._set_mode("ap") - with self._lock: - self.state.ssid = AP_SSID - self.state.ip = "10.42.0.1" - elif wifi_info["active"] and ip: - self._set_mode("client") - with self._lock: - self.state.ssid = wifi_info["ssid"] - self.state.signal = wifi_info["signal"] - self.state.ip = ip - else: - self._set_mode("offline") - - while not self._stop.is_set(): - try: - if self.state.mode == "ap": - # im AP-Modus: alle AP_RECHECK_S versuchen zurück zu wechseln - if self._stop.wait(timeout=AP_RECHECK_S): - break - self._try_exit_ap() - else: - if self._stop.wait(timeout=CHECK_INTERVAL_S): - break - self.run_once() - except Exception as e: - logging.error(f"watchdog error: {e}") - time.sleep(5) - - def start(self): - t = threading.Thread(target=self._loop, name="network-watchdog", daemon=True) - t.start() - - def stop(self): - self._stop.set() - - -# ============================================================================ -# Singleton für Web-UI -# ============================================================================ -_watchdog: Watchdog | None = None - - -def get_watchdog() -> Watchdog: - global _watchdog - if _watchdog is None: - _watchdog = Watchdog() - _watchdog.start() - return _watchdog diff --git a/.backup/2026-08-26-active/palette.py b/.backup/2026-08-26-active/palette.py deleted file mode 100644 index c6c8de7..0000000 --- a/.backup/2026-08-26-active/palette.py +++ /dev/null @@ -1,237 +0,0 @@ -"""7-Farben-Palette für Waveshare 7.3-inch ACeP (F) HAT. - -Der Treiber epd7in3f.EPD.getbuffer() quantisiert ein RGB-Bild automatisch -auf die 7 vom Panel unterstützten Farben. Wir liefern RGB; das macht unseren -Renderer-Code lesbar. - -Reihenfolge der Farbnamen folgt der Wiki-Spezifikation: - Black, White, Green, Blue, Red, Yellow, Orange - -Zusätzlich semantische Aliase: - FG/BG = Vorder-/Hintergrund - OK/WARN/ALERT = Statusbalken - INFO = Blue - ACCENT = Orange - INVERT_FG/INVERT_BG = invertiertes Feld (UV/AQI high) -""" -from PIL import ImageDraw, ImageFont - -# ---- 7 Panel-Farben (RGB) ---- -BLACK = (0, 0, 0) -WHITE = (255, 255, 255) -GREEN = (0, 255, 0) -BLUE = (0, 0, 255) -RED = (255, 0, 0) -YELLOW = (255, 255, 0) -ORANGE = (255, 128, 0) - -# ---- Semantische Aliase ---- -FG = BLACK -BG = WHITE -INVERT_BG = BLACK -INVERT_FG = WHITE - -OK = GREEN -WARN = YELLOW -ALERT = RED - -INFO = BLUE -ACCENT = ORANGE - -_TABLE = { - "black": BLACK, "white": WHITE, "green": GREEN, "blue": BLUE, - "red": RED, "yellow": YELLOW, "orange": ORANGE, - "fg": FG, "bg": BG, - "invert_fg": INVERT_FG, "invert_bg": INVERT_BG, - "ok": OK, "warn": WARN, "alert": ALERT, "info": INFO, "accent": ACCENT, -} - - -def fill_for(name: str) -> tuple: - if name not in _TABLE: - raise ValueError(f"unknown color name: {name!r}") - return _TABLE[name] - - -def measure(draw: ImageDraw.Draw, text: str, font: ImageFont.FreeTypeFont): - try: - b = draw.textbbox((0, 0), text, font=font) - return b[2] - b[0], b[3] - b[1] - except AttributeError: - return draw.textsize(text, font=font) - - -def text_wh(draw, text, font): - return measure(draw, text, font) - - -# ============================================================================ -# Layout helpers — damit Widgets slot-relative und größen-responsive arbeiten -# ============================================================================ -def fit_font(draw, text: str, fonts: dict, max_w: int, max_h: int, - candidates: list[str] | None = None) -> ImageFont.FreeTypeFont | None: - """Wählt die größte Schrift aus `candidates` (oder '80','60','48','32','28','24','20','16'), - deren gerenderter Text in max_w × max_h passt. - - `fonts` ist das Dictionary aus dashboard.load_fonts(). - """ - if candidates is None: - candidates = ["clock", "80", "60", "48", "32", "28", "24", "20", "16"] - for key in candidates: - f = fonts.get(key) - if f is None: - continue - tw, th = measure(draw, text, f) - if tw <= max_w and th <= max_h: - return f - return fonts.get("16") or fonts.get("default") - - -def centered_text(draw, text, x, y, w, h, font, color): - """Zentriert einen Text in der gegebenen Box (x,y,w,h). Returns (px, py).""" - tw, th = measure(draw, text, font) - px = x + max(0, (w - tw) // 2) - py = y + max(0, (h - th) // 2) - draw.text((px, py), text, font=font, fill=color) - return px, py - - -def hbar(draw, x, y, w, h, pct: float, fg=None, bg=None, border_w: int = 2, - thresholds: list | None = None, gradient: bool = False): - """Horizontaler Prozentbalken mit konfigurierbarem Farbverlauf. - - Args: - pct: 0..100 (wird geclampt) - fg: fallback-Farbe (falls keine thresholds) - thresholds: Liste von (max_pct, color) Tupeln, sortiert aufsteigend. - Beispiel: [(50, OK), (80, WARN), (100, ALERT)] - Bei pct < 50 → OK, pct < 80 → WARN, sonst ALERT. - gradient: wenn True, wird der Balken in mehrere Segmente mit den - Threshold-Farben aufgeteilt, statt ein einfarbiger Fill zu sein. - - Wenn `gradient=False` und `thresholds` gesetzt: einfarbiger Fill in der - ersten passenden Threshold-Farbe. - Wenn beides None: einfarbig in fg (oder FG-Fallback). - """ - fg = fg or FG - bg = bg or BG - if h < 6 or w < 6: - return - - pct = max(0.0, min(100.0, pct)) - - # Welche Farbe für pct? - def color_for(p: float) -> tuple: - if thresholds: - for max_p, c in thresholds: - if p <= max_p: - return c - return thresholds[-1][1] - return fg - - # Frame - draw.rectangle((x, y, x + w - 1, y + h - 1), outline=fg, width=border_w) - - if gradient and thresholds and len(thresholds) >= 2: - # Multi-Segment-Bar: zeichne für jeden Threshold-Bereich sein eigenes Segment - inner_x = x + border_w - inner_w = w - 2 * border_w - if inner_w > 0: - prev_max = 0.0 - for max_p, c in thresholds: - seg_pct_start = prev_max - seg_pct_end = min(max_p, pct) - if seg_pct_end > seg_pct_start: - seg_x0 = inner_x + int(inner_w * seg_pct_start / 100) - seg_x1 = inner_x + int(inner_w * seg_pct_end / 100) - draw.rectangle((seg_x0, y + border_w, - seg_x1, y + h - 1 - border_w), fill=c) - prev_max = max_p - # Falls pct die höchste Threshold überschreitet - if pct > thresholds[-1][0]: - seg_x0 = inner_x + int(inner_w * thresholds[-1][0] / 100) - draw.rectangle((seg_x0, y + border_w, - inner_x + inner_w - 1, y + h - 1 - border_w), - fill=thresholds[-1][1]) - else: - # Einfarbiger Fill in passender Farbe - fill_w = int((w - 2 * border_w) * pct / 100) - if fill_w > 0: - draw.rectangle((x + border_w, y + border_w, - x + border_w + fill_w, y + h - 1 - border_w), - fill=color_for(pct)) - - -# Default-Schwellen: (max_pct, color) -DEFAULT_THRESHOLDS = [ - (50, OK), - (80, WARN), - (95, ALERT), -] - - -def parse_thresholds(spec) -> list: - """Parst Threshold-Spec aus Plugin-Config. - - Akzeptiert entweder: - - String "ok,warn,alert" (3 stufig, default-Werte 50/80/95) - - String "ok@50,warn@80,alert@95" (custom Schwellen) - - Liste von Dicts [{"pct": 50, "color": "ok"}, ...] - - None → DEFAULT_THRESHOLDS - """ - if spec is None or spec == "": - return list(DEFAULT_THRESHOLDS) - if isinstance(spec, list): - result = [] - for item in spec: - try: - if isinstance(item, dict): - p = float(item.get("pct", 100)) - c = fill_for(item.get("color", "fg")) - else: - # tuple-like - p = float(item[0]) - c = fill_for(item[1]) - result.append((p, c)) - except (ValueError, TypeError, KeyError): - continue - return sorted(result, key=lambda x: x[0]) if result else list(DEFAULT_THRESHOLDS) - if isinstance(spec, str): - parts = [s.strip() for s in spec.split(",") if s.strip()] - # Default-Stufen wenn keine @-Syntax - default_pcts = [50, 80, 95] - result = [] - for i, p in enumerate(parts): - if "@" in p: - name, val = p.split("@", 1) - try: - pct = float(val) - except ValueError: - continue - else: - name = p - pct = default_pcts[i] if i < len(default_pcts) else 100 - try: - color = fill_for(name) - except ValueError: - continue - result.append((pct, color)) - return sorted(result, key=lambda x: x[0]) if result else list(DEFAULT_THRESHOLDS) - return list(DEFAULT_THRESHOLDS) - - -def is_small(w: int, h: int) -> bool: - """True wenn Slot klein ist (1x1 oder ähnlich). Widgets können darauf - vereinfachtes Layout zeigen.""" - return w < 280 or h < 180 - - -def is_wide(w: int, h: int) -> bool: - """True wenn Slot breit aber flach ist (4x1, 2x1).""" - return w >= 400 and h < 200 - - -def is_tall(w: int, h: int) -> bool: - """True wenn Slot hoch aber schmal ist (1x4, 1x2).""" - return h >= 280 and w < 280 - diff --git a/.backup/2026-08-26-active/plugins/__init__.py b/.backup/2026-08-26-active/plugins/__init__.py deleted file mode 100644 index 56e5772..0000000 --- a/.backup/2026-08-26-active/plugins/__init__.py +++ /dev/null @@ -1 +0,0 @@ -# Plugins package diff --git a/.backup/2026-08-26-active/plugins/base.py b/.backup/2026-08-26-active/plugins/base.py deleted file mode 100644 index a55f000..0000000 --- a/.backup/2026-08-26-active/plugins/base.py +++ /dev/null @@ -1,69 +0,0 @@ -"""Plugin ABC für das Waveshare 7.3" Dashboard. - -Ein Plugin ist ein Python-Modul in /plugins/, das eine Klasse Widget -exportiert. Die Klasse wird beim Start dynamisch geladen und in der -Admin-UI zur Auswahl angeboten. - -Minimal-Beispiel siehe plugins/hello.py. -""" -from __future__ import annotations -from abc import ABC, abstractmethod -from typing import Any - - -class Widget(ABC): - # ---- Metadaten (Klassenattribute) ---- - name: str = "" # Eindeutiger Identifier, lowercase, keine Leerzeichen - label: str = "" # Anzeigename in der UI - description: str = "" # Kurzbeschreibung in der UI - category: str = "general" # "info" | "system" | "weather" | "smart-home" | ... - - # Optional: Schema der Config-Felder (für UI-Form-Generierung). - # Liste von Dicts mit keys: key, label, type, default, secret, choices, help - # type ∈ {"string", "int", "float", "bool", "secret", "select", "lat_lon"} - config_schema: list[dict] = [] - default_config: dict = {} - - def __init__(self, config: dict): - # Merge defaults mit übergebener Config - merged = dict(self.default_config) - merged.update(config or {}) - self.config = merged - - @abstractmethod - def fetch(self) -> dict: - """Daten holen. Sollte schnell sein — wird alle refresh_interval Sekunden - aufgerufen, plus einmal vor jedem Render. Exceptions werden geloggt und - führen zur Beibehaltung der letzten Daten.""" - - @abstractmethod - def render(self, draw, fonts, x: int, y: int, w: int, h: int) -> None: - """Zeichne in den gegebenen Slot (x,y,w,h) auf den draw-Context. - Renderer nutzt RGB-Palette aus palette.py.""" - - # ---- Optionale Lifecycle-Hooks ---- - def on_load(self) -> None: - """Wird einmal beim Plugin-Load aufgerufen.""" - - def on_unload(self) -> None: - """Wird beim Beenden aufgerufen.""" - - # ---- Helper für Plugins ---- - def cfg(self, key: str, default: Any = None) -> Any: - return self.config.get(key, default) - - -def all_widget_classes() -> list[type[Widget]]: - """Lade alle Plugin-Klassen aus dem plugins/-Ordner.""" - import os, importlib, pkgutil - plugins_pkg = os.path.join(os.path.dirname(__file__), "..", "plugins") - plugins_pkg = os.path.abspath(plugins_pkg) - classes: list[type[Widget]] = [] - for _, modname, _ in pkgutil.iter_modules([plugins_pkg]): - mod = importlib.import_module(f"plugins.{modname}") - cls = getattr(mod, "Widget", None) - if cls and isinstance(cls, type) and issubclass(cls, Widget) and cls is not Widget: - classes.append(cls) - # alphabetisch - classes.sort(key=lambda c: c.label or c.name) - return classes diff --git a/.backup/2026-08-26-active/plugins/clock.py b/.backup/2026-08-26-active/plugins/clock.py deleted file mode 100644 index 25500e5..0000000 --- a/.backup/2026-08-26-active/plugins/clock.py +++ /dev/null @@ -1,117 +0,0 @@ -"""Clock-Plugin: Uhrzeit + Datum, responsive fuer alle Slot-Groessen. - -Layout-Strategie pro Slot-Groesse: - is_small (1x1 ~200x120): nur Big-Time zentriert - is_wide (4x1/2x1): Time links gross, Datum rechts klein - sonst (1x2/2x2/4x4): Datum oben klein, Big-Time mittig, Wochentag unten -""" -import os, sys -from datetime import datetime -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from plugins.base import Widget -from palette import (FG, INFO, ACCENT, BLUE, OK, - measure, fit_font, centered_text, - is_small, is_wide, is_tall) - - -class Widget(Widget): - name = "clock" - label = "Uhrzeit / Datum" - description = "Aktuelle Uhrzeit und Datum. Responsiv fuer 1x1 bis 4x4 Slots." - category = "info" - - config_schema = [ - {"key": "format_24h", "label": "24-Stunden-Format", "type": "bool", "default": True}, - {"key": "show_seconds", "label": "Sekunden anzeigen", "type": "bool", "default": False}, - {"key": "show_date", "label": "Datum anzeigen", "type": "bool", "default": True}, - {"key": "show_weekday", "label": "Wochentag anzeigen", "type": "bool", "default": True}, - {"key": "accent_color", "label": "Akzentfarbe", - "type": "select", - "choices": ["info", "blue", "accent", "ok", "warn", "alert"], - "default": "info", - "help": "Farbe fuer Datum und Wochentag"}, - {"key": "weekday_color", "label": "Wochentag-Farbe", - "type": "select", - "choices": ["accent", "blue", "info", "ok", "warn", "alert"], - "default": "accent"}, - ] - default_config = { - "format_24h": True, "show_seconds": False, - "show_date": True, "show_weekday": True, - "accent_color": "info", "weekday_color": "accent", - } - - def fetch(self): - return {} - - def render(self, draw, fonts, x, y, w, h): - from palette import fill_for - now = datetime.now() - pad = 10 - - time_str = now.strftime("%H:%M:%S" if self.cfg("show_seconds") else "%H:%M") - date_str = now.strftime("%d %b %Y") - day_str = now.strftime("%A").upper() - - accent = fill_for(self.cfg("accent_color", "info")) - weekday_c = fill_for(self.cfg("weekday_color", "accent")) - - # === Mini-Modus: 1x1 oder sehr klein === - if is_small(w, h): - font = fit_font(draw, time_str, fonts, w - 2 * pad, h - 2 * pad) - centered_text(draw, time_str, x, y, w, h, font, FG) - return - - # === Wide-Strip-Modus: 4x1 oder 2x1 === - if is_wide(w, h): - # Big Time links - font_time = fit_font(draw, time_str, fonts, w // 2 - 2 * pad, h - 2 * pad) - centered_text(draw, time_str, x, y, w // 2, h, font_time, FG) - # Datum + Wochentag rechts - if self.cfg("show_date"): - font_date = fit_font(draw, date_str, fonts, w // 2 - 2 * pad, h // 2 - 4) - centered_text(draw, date_str, x + w // 2, y, w // 2, h // 2, font_date, accent) - if self.cfg("show_weekday"): - font_day = fit_font(draw, day_str, fonts, w // 2 - 2 * pad, h // 2 - 4) - centered_text(draw, day_str, x + w // 2, y + h // 2, w // 2, h // 2, font_day, weekday_c) - return - - # === Tall-Modus: schmal aber hoch === - if is_tall(w, h): - # Vertikal: Time oben, dann Date, dann Weekday - font_time = fit_font(draw, time_str, fonts, w - 2 * pad, int(h * 0.45)) - centered_text(draw, time_str, x, y, w, int(h * 0.45), font_time, FG) - if self.cfg("show_date"): - font_date = fit_font(draw, date_str, fonts, w - 2 * pad, int(h * 0.25)) - centered_text(draw, date_str, x, y + int(h * 0.45), w, int(h * 0.25), font_date, accent) - if self.cfg("show_weekday"): - font_day = fit_font(draw, day_str, fonts, w - 2 * pad, int(h * 0.20)) - centered_text(draw, day_str, x, y + int(h * 0.70), w, int(h * 0.20), font_day, weekday_c) - return - - # === Standard-Modus: 2x2 oder groesser, quadratisch oder landscape === - # Drei Sektionen: Date (15%) | Time (60%) | Weekday (25%) - date_h = int(h * 0.18) if self.cfg("show_date") else 0 - weekday_h = int(h * 0.18) if self.cfg("show_weekday") else 0 - time_h = h - date_h - weekday_h - cur_y = y - - if self.cfg("show_date"): - font_date = fit_font(draw, date_str, fonts, w - 2 * pad, date_h - 4) - centered_text(draw, date_str, x, cur_y, w, date_h, font_date, accent) - cur_y += date_h - - time_x = x - time_w = w - # Wenn weekday und date beide aus: Time zentriert ueber alles - if not (self.cfg("show_date") or self.cfg("show_weekday")): - time_y = y - else: - time_y = cur_y - font_time = fit_font(draw, time_str, fonts, time_w - 2 * pad, time_h - 4) - centered_text(draw, time_str, time_x, time_y, time_w, time_h, font_time, FG) - cur_y += time_h - - if self.cfg("show_weekday"): - font_day = fit_font(draw, day_str, fonts, w - 2 * pad, weekday_h - 4) - centered_text(draw, day_str, x, cur_y, w, weekday_h, font_day, weekday_c) diff --git a/.backup/2026-08-26-active/plugins/gmail.py b/.backup/2026-08-26-active/plugins/gmail.py deleted file mode 100644 index f605007..0000000 --- a/.backup/2026-08-26-active/plugins/gmail.py +++ /dev/null @@ -1,58 +0,0 @@ -"""Gmail Unread - responsive.""" -import os, sys, json -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from plugins.base import Widget -from palette import FG, INFO, OK, WARN, measure, fit_font, centered_text, is_small, is_wide - - -class Widget(Widget): - name = "gmail" - label = "Gmail Unread" - description = "Anzahl ungelesener Emails im Posteingang." - category = "info" - - config_schema = [ - {"key": "oauth_token_json", "label": "OAuth Token (JSON)", "type": "secret", - "help": "token.json aus einem einmaligen OAuth-Flow."}, - ] - default_config = {"oauth_token_json": ""} - - def fetch(self): - tok = self.cfg("oauth_token_json") - if not tok: - return {"_error": "OAuth-Token fehlt"} - try: - from google.oauth2.credentials import Credentials - from google.auth.transport.requests import Request - from googleapiclient.discovery import build - creds = Credentials.from_authorized_user_info(json.loads(tok)) - if creds.expired and creds.refresh_token: - creds.refresh(Request()) - service = build("gmail", "v1", credentials=creds, cache_discovery=False) - label = service.users().labels().get(userId="me", id="INBOX").execute() - return {"unread": label.get("messagesUnread", 0)} - except ImportError: - return {"_error": "google-api-python-client nicht installiert"} - except Exception as e: - return {"_error": str(e)} - - def render(self, draw, fonts, x, y, w, h): - pad = 8 - draw.text((x + pad, y + pad), "GMAIL", font=fonts.get("24", fonts.get("20")), fill=INFO) - d = self.fetch() - if "_error" in d: - font = fit_font(draw, "Konfig fehlt", fonts, w - 2 * pad, h - 80) - centered_text(draw, "Konfig fehlt", x, y + 60, w, h - 60, font, WARN) - return - unread = d["unread"] - color = OK if unread == 0 else WARN if unread < 10 else FG - text = str(unread) - # Big number centered - if is_small(w, h): - font = fit_font(draw, text, fonts, w - 2 * pad, h - 50) - centered_text(draw, text, x, y, w, h - 30, font, color) - else: - font = fit_font(draw, text, fonts, w - 2 * pad, int(h * 0.6)) - centered_text(draw, text, x, y, w, int(h * 0.6), font, color) - font_lbl = fit_font(draw, "Ungelesen", fonts, w - 2 * pad, 24) - centered_text(draw, "Ungelesen", x, y + int(h * 0.7), w, h // 5, font_lbl, FG) diff --git a/.backup/2026-08-26-active/plugins/hello.py b/.backup/2026-08-26-active/plugins/hello.py deleted file mode 100644 index 0ade197..0000000 --- a/.backup/2026-08-26-active/plugins/hello.py +++ /dev/null @@ -1,27 +0,0 @@ -"""Demo-Plugin: konfigurierbarer Text.""" -import os, sys -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from plugins.base import Widget -from palette import fill_for, measure, fit_font, centered_text - - -class Widget(Widget): - name = "hello" - label = "Hello / Demo" - description = "Zeigt einen konfigurierbaren Text. Responsiv." - category = "general" - - config_schema = [ - {"key": "text", "label": "Text", "type": "string", "default": "Hello!"}, - {"key": "color", "label": "Farbe", "type": "select", - "choices": ["fg", "info", "ok", "warn", "alert", "accent"], "default": "fg"}, - ] - default_config = {"text": "Hello!", "color": "fg"} - - def fetch(self): - return {} - - def render(self, draw, fonts, x, y, w, h): - text = self.cfg("text", "Hello!") - font = fit_font(draw, text, fonts, w - 16, h - 16) - centered_text(draw, text, x, y, w, h, font, fill_for(self.cfg("color", "fg"))) diff --git a/.backup/2026-08-26-active/plugins/minimax.py b/.backup/2026-08-26-active/plugins/minimax.py deleted file mode 100644 index 94907c3..0000000 --- a/.backup/2026-08-26-active/plugins/minimax.py +++ /dev/null @@ -1,271 +0,0 @@ -"""MiniMax Token-Usage Widget - responsive.""" -import os, sys, json -import urllib.request, urllib.error -from datetime import datetime -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from plugins.base import Widget -from palette import FG, INFO, OK, WARN, ALERT, ORANGE, BG, measure, fit_font, hbar, centered_text, is_small, is_wide, parse_thresholds, DEFAULT_THRESHOLDS - - -def _fmt_timedelta_short(seconds): - if seconds is None: return "" - s = int(seconds) - if s <= 0: return "now" - days, rem = divmod(s, 86400) - h, rem = divmod(rem, 3600) - m = rem // 60 - if days: return f"{days}d {h}h" - if h: return f"{h}h {m}m" - return f"{m}m" - - -def _deep_get(d, *keys, default=None): - for k in keys: - if isinstance(d, dict) and k in d: - v = d[k] - if v is not None: - return v - return default - - -def _fetch_remains(subscription_key, base_url="https://api.minimax.io"): - url = f"{base_url.rstrip('/')}/v1/token_plan/remains" - req = urllib.request.Request(url, headers={ - "Authorization": f"Bearer {subscription_key}", - "Content-Type": "application/json", "Accept": "application/json", - }) - try: - with urllib.request.urlopen(req, timeout=10) as r: - raw = r.read() - try: - return {"ok": True, "data": json.loads(raw), "raw": raw.decode("utf-8", "ignore")} - except json.JSONDecodeError as e: - return {"ok": False, "error": f"invalid json: {e}", - "raw": raw.decode("utf-8", "ignore")[:500]} - except urllib.error.HTTPError as e: - body = "" - try: body = e.read().decode("utf-8", "ignore")[:300] - except Exception: pass - return {"ok": False, "error": f"HTTP {e.code} {e.reason}: {body}".strip()} - - -def _check_base_resp(data): - base = _deep_get(data, "base_resp") - if not isinstance(base, dict): return None - code = base.get("status_code", 0) - if code in (0, None, "0", ""): return None - return f"{base.get('status_msg') or base.get('message') or 'API-Fehler'} (code {code})" - - -def _parse_model_remains(items): - now_ms = int(datetime.now().timestamp() * 1000) - parsed = [] - for entry in items: - if not isinstance(entry, dict): continue - model_name = entry.get("model_name", "model") - for win_label, rem_pct, end_time_raw, status_raw in [ - ("5-Hour", entry.get("current_interval_remaining_percent"), - entry.get("end_time"), entry.get("current_interval_status")), - ("Weekly", entry.get("current_weekly_remaining_percent"), - entry.get("weekly_end_time"), entry.get("current_weekly_status")), - ]: - if rem_pct is None: continue - try: status = int(status_raw) if status_raw is not None else 1 - except (TypeError, ValueError): status = 1 - if status != 1: continue - reset_sec = None - if isinstance(end_time_raw, (int, float)) and end_time_raw > 1e12: - reset_sec = max(0, int((end_time_raw - now_ms) / 1000)) - elif isinstance(end_time_raw, (int, float)): - reset_sec = max(0, int(end_time_raw)) - used_pct = max(0.0, min(100.0, 100.0 - float(rem_pct))) - parsed.append({ - "label": win_label, - "used_pct": used_pct, - "remaining_pct": float(rem_pct), - "reset_seconds": reset_sec, - "raw": entry, - "model_name": model_name, - }) - return parsed - - -def _parse_windows(data): - model_remains = data.get("model_remains") - if isinstance(model_remains, list) and model_remains: - return _parse_model_remains(model_remains) - candidates = [ - data.get("remains"), - _deep_get(data, "data", "remains"), - _deep_get(data, "data", "windows"), - data.get("windows"), - _deep_get(data, "plan", "windows"), - ] - raw_windows = None - for c in candidates: - if isinstance(c, list) and c: - raw_windows = c; break - if not isinstance(raw_windows, list): - return [] - parsed = [] - for w in raw_windows: - if not isinstance(w, dict): continue - reset_sec = w.get("reset_in") or w.get("reset_in_seconds") - if reset_sec is None: - reset_iso = w.get("reset_at") or w.get("resets_at") - if isinstance(reset_iso, str): - try: - dt = datetime.fromisoformat(reset_iso.replace("Z", "+00:00")) - reset_sec = (dt - datetime.now(dt.tzinfo)).total_seconds() - except Exception: - reset_sec = None - usage_raw = w.get("usage") or w.get("used_pct") or w.get("utilization") - used_pct = None - if isinstance(usage_raw, (int, float)): - used_pct = float(usage_raw) * 100 if float(usage_raw) <= 1.0 else float(usage_raw) - if used_pct is None: continue - name = (w.get("name") or w.get("window") or w.get("label") or "window").lower() - parsed.append({ - "label": name.upper(), - "used_pct": min(100.0, max(0.0, used_pct)), - "remaining_pct": max(0.0, 100.0 - used_pct), - "reset_seconds": int(reset_sec) if reset_sec else None, - }) - return parsed - - -def _parse_credits(data): - points = _deep_get(data, "points_balance", "credits_balance", "balance", "points", "credits") - if isinstance(points, (int, float)): - return {"points": float(points)} - nested = _deep_get(data, "plan", "credits") - if isinstance(nested, (int, float)): - return {"points": float(nested)} - return None - - -class Widget(Widget): - name = "minimax" - label = "MiniMax Token Usage" - description = "Quota für MiniMax Token-Plan (5-Hour + Weekly). Responsive." - category = "info" - - config_schema = [ - {"key": "subscription_key", "label": "Subscription Key", "type": "secret", - "help": "Token-Plan Subscription Key. NICHT der pay-as-you-go API-Key."}, - {"key": "show_credits", "label": "Credit-Balance anzeigen", "type": "bool", "default": True}, - {"key": "api_base", "label": "API Base URL", "type": "string", "default": "https://api.minimax.io"}, - {"key": "bar_thresholds", "label": "Quota-Schwellen (ok,warn,alert)", - "type": "string", "default": "ok@50,warn@80,alert@95", - "help": "Format: 'ok@50,warn@80,alert@95' oder 'ok,warn,alert' (default 50/80/95)"}, - {"key": "bar_gradient", "label": "Verlaufsmodus", - "type": "bool", "default": True, - "help": "Wenn aus, einfarbige Bar in der Farbe der aktuellen Schwelle."}, - ] - default_config = {"subscription_key": "", "show_credits": True, - "api_base": "https://api.minimax.io", - "bar_thresholds": "ok@50,warn@80,alert@95", - "bar_gradient": True} - - def _bar_args(self): - return (parse_thresholds(self.cfg("bar_thresholds")), - bool(self.cfg("bar_gradient", True))) - - def fetch(self): - key = self.cfg("subscription_key") - if not key: - return {"_error": "Subscription Key fehlt — in der Admin-UI setzen."} - try: - r = _fetch_remains(key, self.cfg("api_base", "https://api.minimax.io")) - if not r.get("ok"): - return {"_error": r.get("error", "unbekannter Fehler")} - data = r["data"] - base_err = _check_base_resp(data) - if base_err: - return {"_error": base_err, - "_raw_keys": list(data.keys()) if isinstance(data, dict) else None} - windows = _parse_windows(data) - credits = _parse_credits(data) if self.cfg("show_credits") else None - if not windows and not credits: - return {"_error": "Schema unbekannt.", "_raw_keys": list(data.keys()) if isinstance(data, dict) else None} - return {"windows": windows, "credits": credits} - except Exception as e: - return {"_error": str(e)} - - def _render_window(self, draw, fonts, x, y, w, h, wn, compact=False): - from palette import measure - # Label + Reset rechts - label = wn["label"] - reset_txt = ("R " + _fmt_timedelta_short(wn["reset_seconds"])) if wn.get("reset_seconds") else "" - font_label = fit_font(draw, label, fonts, w - 60, 22) - draw.text((x, y), label, font=font_label, fill=FG) - if reset_txt and not compact: - font_r = fit_font(draw, reset_txt, fonts, 60, 20) - rw, rh = measure(draw, reset_txt, font_r) - draw.text((x + w - rw - 4, y + 2), reset_txt, font=font_r, fill=FG) - # Bar - bar_y = y + (22 if not compact else 18) - bar_h = max(10, h - (bar_y - y) - 4) - pct = wn["used_pct"] - thresholds, gradient = self._bar_args() - hbar(draw, x, bar_y, w, bar_h, pct, - thresholds=thresholds, gradient=gradient) - # Prozent overlay - if not compact: - pct_txt = f"{int(pct + 0.5)}%" - font_pct = fit_font(draw, pct_txt, fonts, 60, bar_h - 4) - tw, th = measure(draw, pct_txt, font_pct) - draw.text((x + w - tw - 8, bar_y + 2), pct_txt, font=font_pct, fill=BG) - - def render(self, draw, fonts, x, y, w, h): - from palette import measure - pad = 8 - draw.text((x + pad, y + pad), "MINIMAX", font=fonts.get("24", fonts.get("20")), fill=INFO) - - d = self.fetch() - if "_error" in d: - draw.text((x + pad, y + 60), "Konfiguration fehlt", font=fonts.get("24", fonts.get("20")), fill=WARN) - err = d["_error"][:50] - draw.text((x + pad, y + 90), err, font=fonts.get("20", fonts.get("16")), fill=FG) - if d.get("_raw_keys"): - draw.text((x + pad, y + 120), f"raw keys: {','.join(d['_raw_keys'][:6])}", - font=fonts.get("16", fonts.get("default")), fill=FG) - return - - windows = d.get("windows", []) - - if is_small(w, h): - # mini: nur "5h: 38%" etc kompakt - line_y = y + 50 - for wn in windows[:2]: - label = f"{wn['label'][:1]}{wn['remaining_pct']:.0f}%" - font = fit_font(draw, label, fonts, w - 2 * pad, h // 3) - draw.text((x + pad, line_y), label, font=font, fill=FG) - line_y += h // 3 - return - - if is_wide(w, h): - # Wide strip: alle windows nebeneinander - n = max(1, len(windows)) - col_w = w // n - for i, wn in enumerate(windows): - cx = x + i * col_w - self._render_window(draw, fonts, cx + pad // 2, y + 30, - col_w - pad, h - 35, wn, compact=True) - return - - # Standard: vertikale Liste mit Bars - row_h = max(40, (h - 60) // max(1, len(windows))) - for i, wn in enumerate(windows): - row_y = y + 50 + i * row_h - self._render_window(draw, fonts, x + pad, row_y, - w - 2 * pad, row_h - 6, wn) - - # Credits - credits = d.get("credits") - if credits and self.cfg("show_credits") and h > 220: - pts = credits.get("points") - if isinstance(pts, (int, float)): - txt = f"Credits: {pts:,.0f}" - font = fit_font(draw, txt, fonts, w - 2 * pad, 22) - draw.text((x + pad, y + h - 28), txt, font=font, fill=ORANGE) diff --git a/.backup/2026-08-26-active/plugins/spotify.py b/.backup/2026-08-26-active/plugins/spotify.py deleted file mode 100644 index 6c1c4e7..0000000 --- a/.backup/2026-08-26-active/plugins/spotify.py +++ /dev/null @@ -1,78 +0,0 @@ -"""Spotify via Last.fm Scrobble - responsive.""" -import os, sys, json -import urllib.request, urllib.error, urllib.parse -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from plugins.base import Widget -from palette import FG, INFO, OK, WARN, measure, fit_font, centered_text, is_small, is_wide - - -def _fetch_lastfm(api_key, user): - url = (f"http://ws.audioscrobbler.com/2.0/?method=user.getrecenttracks" - f"&user={urllib.parse.quote(user)}&api_key={api_key}" - f"&format=json&limit=2") - with urllib.request.urlopen(url, timeout=8) as r: - return json.loads(r.read()) - - -class Widget(Widget): - name = "spotify" - label = "Spotify (Last.fm)" - description = "Zeigt aktuell gespielten Spotify-Track via Last.fm Scrobble." - category = "info" - - config_schema = [ - {"key": "api_key", "label": "Last.fm API Key", "type": "secret"}, - {"key": "username", "label": "Last.fm Username", "type": "string", "default": ""}, - ] - default_config = {"api_key": "", "username": ""} - - def fetch(self): - api_key = self.cfg("api_key") - user = self.cfg("username") - if not api_key or not user: - return {"_error": "API-Key oder Username fehlt"} - try: - return _fetch_lastfm(api_key, user) - except Exception as e: - return {"_error": str(e)} - - def render(self, draw, fonts, x, y, w, h): - pad = 8 - draw.text((x + pad, y + pad), "SPOTIFY", font=fonts.get("24", fonts.get("20")), fill=INFO) - d = self.fetch() - if "_error" in d: - font = fit_font(draw, "Konfig fehlt", fonts, w - 2 * pad, h - 80) - centered_text(draw, "Konfig fehlt", x, y + 60, w, h - 60, font, WARN) - return - try: - tracks = d.get("recenttracks", {}).get("track", []) - if isinstance(tracks, dict): tracks = [tracks] - if not tracks: - font = fit_font(draw, "Kein Track", fonts, w - 2 * pad, h - 80) - centered_text(draw, "Kein Track", x, y + 60, w, h - 60, font, FG) - return - current = tracks[0] - is_playing = current.get("@attr", {}).get("nowplaying") == "true" - artist = current.get("artist", {}).get("#text", "?") - track = current.get("name", "?") - color = OK if is_playing else WARN - - if is_small(w, h): - txt = "▶" if is_playing else "⏸" - font = fit_font(draw, txt, fonts, w - 2 * pad, h - 2 * pad) - centered_text(draw, txt, x, y, w, h, font, color) - return - - # Header symbol - status_str = "▶ " if is_playing else "⏸" - font_st = fit_font(draw, status_str, fonts, 50, h // 4) - draw.text((x + pad, y + 50), status_str, font=font_st, fill=color) - - # Track info - font_a = fit_font(draw, artist, fonts, w - 60, h // 4) - font_t = fit_font(draw, track, fonts, w - 20, h // 4) - draw.text((x + 60, y + 50), artist[:30], font=font_a, fill=FG) - draw.text((x + 60, y + 50 + font_a.size + 8), track[:35], font=font_t, fill=FG) - except Exception as e: - font = fit_font(draw, f"Fehler: {e}", fonts, w - 2 * pad, h - 80) - centered_text(draw, f"Fehler: {str(e)[:40]}", x, y + 60, w, h - 60, font, WARN) diff --git a/.backup/2026-08-26-active/plugins/strava.py b/.backup/2026-08-26-active/plugins/strava.py deleted file mode 100644 index d70954d..0000000 --- a/.backup/2026-08-26-active/plugins/strava.py +++ /dev/null @@ -1,99 +0,0 @@ -"""Strava Stats - responsive.""" -import os, sys, time, json -import urllib.request, urllib.error -from datetime import datetime -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from plugins.base import Widget -from palette import FG, INFO, OK, WARN, measure, fit_font, centered_text, is_small, is_wide - - -def _strava_refresh(client_id, client_secret, refresh_token): - data = (f"client_id={client_id}&client_secret={client_secret}" - f"&grant_type=refresh_token&refresh_token={refresh_token}").encode() - req = urllib.request.Request("https://www.strava.com/oauth/token", data=data) - with urllib.request.urlopen(req, timeout=8) as r: - return json.loads(r.read()) - - -def _strava_activities(access_token, page=1): - url = f"https://www.strava.com/api/v3/athlete/activities?page={page}&per_page=100" - req = urllib.request.Request(url, headers={"Authorization": f"Bearer {access_token}"}) - with urllib.request.urlopen(req, timeout=8) as r: - return json.loads(r.read()) - - -def _stats(activities): - total = sum(a.get("distance", 0) for a in activities) - rides = sum(1 for a in activities if a.get("type") in ("Ride", "VirtualRide")) - hike = sum(a.get("distance", 0) for a in activities if a.get("type") in ("Hike", "Walk")) - year = datetime.now().year - year_start = datetime(year, 1, 1).timestamp() - year_dist = sum(a.get("distance", 0) for a in activities - if datetime.strptime(a["start_date"][:19], "%Y-%m-%dT%H:%M:%S").timestamp() >= year_start) - return {"total_km": total / 1000, "year_km": year_dist / 1000, - "rides": rides, "hike_km": hike / 1000} - - -class Widget(Widget): - name = "strava" - label = "Strava Aktivitäten" - description = "Distanz und Rides aus Strava. Setze Client-ID/Secret/Refresh-Token unten." - category = "fitness" - - config_schema = [ - {"key": "client_id", "label": "Strava Client ID", "type": "secret"}, - {"key": "client_secret", "label": "Strava Client Secret", "type": "secret"}, - {"key": "refresh_token", "label": "Refresh Token", "type": "secret", - "help": "Einmaliger OAuth-Token. Plugin holt sich access_tokens on-demand."}, - ] - default_config = {"client_id": "", "client_secret": "", "refresh_token": ""} - - def fetch(self): - cid = self.cfg("client_id") - csec = self.cfg("client_secret") - rtok = self.cfg("refresh_token") - if not (cid and csec and rtok): - return {"_error": "Tokens fehlen"} - try: - tok = _strava_refresh(cid, csec, rtok) - activities = _strava_activities(tok["access_token"], page=1) - return _stats(activities) - except Exception as e: - return {"_error": str(e)} - - def render(self, draw, fonts, x, y, w, h): - pad = 8 - draw.text((x + pad, y + pad), "STRAVA", font=fonts.get("24", fonts.get("20")), fill=INFO) - d = self.fetch() - if "_error" in d: - font = fit_font(draw, "Konfig fehlt", fonts, w - 2 * pad, h - 80) - centered_text(draw, "Konfig fehlt", x, y + 50, w, h - 60, font, WARN) - return - - if is_small(w, h): - txt = f"{d['year_km']:.0f}km" - font = fit_font(draw, txt, fonts, w - 2 * pad, h - 2 * pad) - centered_text(draw, txt, x, y, w, h, font, OK) - return - - if is_wide(w, h): - stats = [f"{d['year_km']:.0f} km YTD", f"{d['rides']} rides", - f"{d['total_km']:.0f} km total"] - col_w = w // len(stats) - for i, s in enumerate(stats): - cx = x + i * col_w - font = fit_font(draw, s, fonts, col_w - 2 * pad, h - 50) - centered_text(draw, s, cx, y + 30, col_w, h - 30, font, FG) - return - - # Standard - font_b = fit_font(draw, f"{d['year_km']:.0f} km", fonts, w - 2 * pad, h // 3) - draw.text((x + pad, y + 50), f"{d['year_km']:.0f} km", font=font_b, fill=FG) - draw.text((x + pad, y + 50 + font_b.size + 8), - f"in {datetime.now().year}", font=fonts.get("24", fonts.get("20")), fill=OK) - font_r = fit_font(draw, f"{d['rides']} rides", fonts, w - 2 * pad, 28) - draw.text((x + pad, y + h - 100), f"{d['rides']} rides", font=font_r, fill=OK) - font_h = fit_font(draw, f"{d['hike_km']:.1f} km hike", fonts, w - 2 * pad, 24) - draw.text((x + pad, y + h - 70), f"{d['hike_km']:.1f} km hike", font=font_h, fill=FG) - font_t = fit_font(draw, f"Total: {d['total_km']:.0f} km", fonts, w - 2 * pad, 20) - draw.text((x + pad, y + h - 40), f"Total: {d['total_km']:.0f} km", font=font_t, fill=FG) diff --git a/.backup/2026-08-26-active/templates/index.html b/.backup/2026-08-26-active/templates/index.html deleted file mode 100644 index 9bf8c36..0000000 --- a/.backup/2026-08-26-active/templates/index.html +++ /dev/null @@ -1,1844 +0,0 @@ - - - - - - epaper-dashboard - - - - -
- - - -
-
-
- Layout · Widget-Positionen auf dem Display -
-
-
- --:--:-- - — -
- -
-
- -
- - -
- - -
-
-

Grid Layout {{ grid.cols }}×{{ grid.rows }} · {{ layout_items|length }} items

-
-
-
-
- - - -
- - -
-
-
- Drag Item verschieben - Ecke ziehen Größe ändern - X Item löschen - Auto-Pack alle Items optimal anordnen -
-
- -
-
-

Live Preview

-
Auto-refresh alle 10s
-
- preview -
-
- - -
-
-
-

Plugin-Konfiguration

-
—
-
-
-
-
- - -
-
-

Refresh-Intervall

-
-
- - -
- -
-
- ACeP-Displays vertragen keine schnellen Refreshes. Hersteller empfiehlt ≥180s. -
-
-
-

System

-
-
Service:
epaper-dashboard.service
-
Logs:
journalctl -u epaper-dashboard -f
-
Config:
config.json
-
Plugins:
plugins/
-
-
-
- - -
-
-
-

Aktueller Status

-
lade…
-
- - - - -
-
-
-
-

WLAN verbinden / wechseln

- - -
- - -
- - - - - -
- -
-
-
-
-

Gespeicherte Netzwerke

-
    -
    -
    - Recovery-AP Info -

    Wenn das konfigurierte WLAN nicht erreichbar ist (Router defekt, falsches Passwort, Pi umgezogen), startet das System automatisch einen Recovery-Hotspot.

    - -
    -
    -
    - -
    - - - -
    - - diff --git a/.gitignore b/.gitignore index 4e3c16c..1e5d484 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,4 @@ minimax_*.png .env config.json *.service +.backup/