Sync to Pi: alle Features die live deployed sind

Aus dem Backup und Live-Pull vom Pi (10.11.3.144):
- dashboard.py: Grid-Linien nur im freien Hintergrund (nicht durch Widgets)
- templates/index.html: komplett redesigned mit Sidebar + Topbar + Toast + Modal
- plugins/clock.py: responsive Layout (1x1 bis 4x4)
- plugins/system.py, weather.py, minimax.py: mit Threshold-Bars und Color-Variants
- plugins/base.py: NEU — fetch_with_retry Helper (3x retry mit backoff)
  + render_error_banner für fehlgeschlagene API-Plugins
  (grosses rotes "!" Icon mit Plugin-Name und Fehler statt Crash)

Cleanup: Helfer-Chaos (renderer.py/2/3, design_a/b/c.html, clock_classic.py,
23x clock_*.png, alte test_*.py) wurde bereits im vorherigen Commit entfernt.

Co-Authored-By: Hermes <noreply@hermes.local>
This commit is contained in:
epaper-dashboard
2026-08-26 22:11:36 +04:00
co-authored by Hermes
parent af9a99a379
commit 1f142f5245
23 changed files with 6250 additions and 950 deletions
+176
View File
@@ -0,0 +1,176 @@
# 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
+41
View File
@@ -0,0 +1,41 @@
# 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)
+688
View File
@@ -0,0 +1,688 @@
"""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/<idx> → 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. <port>/.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_<idx>_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/<int:idx>", 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/<iid>", 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/<plugin_name>", 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()
@@ -0,0 +1,10 @@
{
"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}}
]
}
+463
View File
@@ -0,0 +1,463 @@
"""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()
+172
View File
@@ -0,0 +1,172 @@
"""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}
@@ -0,0 +1,468 @@
"""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
+237
View File
@@ -0,0 +1,237 @@
"""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
@@ -0,0 +1 @@
# Plugins package
+69
View File
@@ -0,0 +1,69 @@
"""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
+117
View File
@@ -0,0 +1,117 @@
"""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)
@@ -0,0 +1,58 @@
"""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)
@@ -0,0 +1,27 @@
"""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")))
@@ -0,0 +1,271 @@
"""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)
@@ -0,0 +1,78 @@
"""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)
@@ -0,0 +1,99 @@
"""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)
File diff suppressed because it is too large Load Diff
+52 -2
View File
@@ -89,7 +89,15 @@ def require_auth():
@app.route("/")
def index():
a = require_auth()
if a: return a
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
@@ -128,7 +136,8 @@ def index():
plugin_configs=plugin_configs,
grid=grid,
size_presets=layout_mod.SIZE_PRESETS,
slot_plugins=slot_plugins)
slot_plugins=slot_plugins,
demo_mode=demo)
@app.route("/config", methods=["POST"])
@@ -524,6 +533,47 @@ def plugins_json():
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()
+25 -5
View File
@@ -15,7 +15,6 @@ 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
@@ -208,8 +207,32 @@ def render_full(items: list, widgets: list, fonts: dict) -> Image.Image:
img = Image.new("RGB", (DISPLAY_W, DISPLAY_H), BG)
draw = ImageDraw.Draw(img)
# Grid-Trennlinien NUR im freien Hintergrund (zwischen leeren Zellen).
# Niemals durch Widgets durchgehen — Widgets sind klar abgegrenzt.
GRID_LINE = (215, 215, 215)
item_cells = set()
for item in items:
for dx in range(item.w):
for dy in range(item.h):
item_cells.add((item.x + dx, item.y + dy))
# Vertikale Linien: an jeder Spaltengrenze nur dort zeichnen, wo
# die LINKE und RECHTE Zelle beide LEER sind (kein Item).
for c in range(1, 4):
for r in range(4):
if (c - 1, r) not in item_cells and (c, r) not in item_cells:
y0 = r * 120
draw.line((c * 200, y0, c * 200, y0 + 120), fill=GRID_LINE, width=1)
# Horizontale Linien: analog.
for r in range(1, 4):
for c in range(4):
if (c, r - 1) not in item_cells and (c, r) not in item_cells:
x0 = c * 200
draw.line((x0, r * 120, x0 + 200, r * 120), fill=GRID_LINE, width=1)
for item, widget in zip(items, widgets):
px, py, pw, ph = item.pixels()
# Cell border
draw.rectangle((px, py, px + pw - 1, py + ph - 1), outline=FG, width=2)
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)
@@ -261,8 +284,6 @@ class Dashboard:
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():
@@ -290,8 +311,7 @@ class Dashboard:
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)
img = render_full(items, widgets, self.fonts)
return img
def display(self, img: Image.Image):
+41
View File
@@ -229,9 +229,50 @@ def is_wifi_connected() -> bool:
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()
+141 -19
View File
@@ -10,45 +10,168 @@ from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any
import time
import urllib.request
import urllib.error
import socket as _socket
from PIL import ImageDraw
# ============================================================================
# Fetch-Helper mit Retry-Logik
# ============================================================================
def fetch_with_retry(fn, retries: int = 3, delay_s: float = 0.5):
"""Ruft fn() bis zu retries Mal auf. Bei 3x fail gibt es None + error_string."""
last_err = None
for attempt in range(1, retries + 1):
try:
return fn(), None
except Exception as e:
last_err = e
if attempt < retries:
time.sleep(delay_s * attempt)
return None, _format_error(last_err)
def fetch_url(url: str, headers=None, timeout: int = 10, retries: int = 3):
"""HTTP-GET mit Retry. Returns (data, error)."""
def _do():
req = urllib.request.Request(url, headers=headers or {})
with urllib.request.urlopen(req, timeout=timeout) as r:
return r.read()
data, err = fetch_with_retry(_do, retries=retries, delay_s=0.5)
if err:
return None, err
return data, None
def _format_error(exc):
"""Kurze, menschenlesbare Fehlermeldung."""
if isinstance(exc, urllib.error.HTTPError):
return f"HTTP {exc.code} {exc.reason or ''}".strip()
if isinstance(exc, urllib.error.URLError):
return f"URL-Fehler: {exc.reason}"
if isinstance(exc, _socket.timeout):
return "Timeout (keine Antwort)"
if isinstance(exc, _socket.gaierror):
return f"DNS-Fehler: {exc}"
if isinstance(exc, ConnectionRefusedError):
return "Connection refused"
if isinstance(exc, ConnectionResetError):
return "Connection reset"
if isinstance(exc, TimeoutError):
return "Timeout"
if isinstance(exc, (KeyError, ValueError, TypeError)):
return f"Datenfehler: {str(exc)[:60]}"
return f"{type(exc).__name__}: {str(exc)[:60]}"
# ============================================================================
# Error-Banner für fehlgeschlagene API-Plugins
# ============================================================================
ERROR_COLORS = {
"icon": (200, 0, 0),
"title": (180, 0, 0),
"subtitle": (60, 60, 60),
"muted": (120, 120, 120),
"bg": (245, 244, 240),
}
def render_error_banner(draw, fonts, x, y, w, h, plugin_name, error_msg,
last_success=None):
"""Zeichnet ein auffaelliges Fehler-Schild in den Slot.
Layout: Grosses rotes "!" Icon links, Plugin-Name + Fehler rechts.
"""
from palette import measure
pad = 12
# Border rot
draw.rectangle((x, y, x + w - 1, y + h - 1),
outline=ERROR_COLORS["icon"], width=3)
icon_size = min(h - 2 * pad, 100)
if icon_size < 30:
icon_size = 30
cx = x + pad
cy = y + pad
draw.rectangle((cx, cy, cx + icon_size - 1, cy + icon_size - 1),
fill=ERROR_COLORS["icon"])
font_icon = fonts.get(str(min(icon_size, 80))) or fonts.get("60") or fonts.get("default")
tw, th = measure(draw, "!", font_icon)
draw.text((cx + (icon_size - tw) // 2 - 2, cy + (icon_size - th) // 2 - 4),
"!", font=font_icon, fill=ERROR_COLORS["bg"])
tx = cx + icon_size + 12
tw_avail = w - (tx - x) - pad
title = f"Plugin: {plugin_name}"
font_title = fonts.get("24") or fonts.get("20") or fonts.get("default")
for try_font in [font_title, fonts.get("20"), fonts.get("16")]:
tw, _ = measure(draw, title, try_font)
if tw <= tw_avail or try_font is fonts.get("16"):
font_title = try_font
break
draw.text((tx, cy + 2), title, font=font_title, fill=ERROR_COLORS["title"])
sub_y = cy + 30
sub = f"API nicht erreichbar: {error_msg}"
font_sub = fonts.get("20") or fonts.get("16")
if measure(draw, sub, font_sub)[0] > tw_avail:
words = sub.split()
lines, cur = [], ""
for w in words:
cand = (cur + " " + w).strip()
if measure(draw, cand, font_sub)[0] <= tw_avail:
cur = cand
else:
if cur: lines.append(cur)
cur = w
if cur: lines.append(cur)
for i, ln in enumerate(lines[:3]):
draw.text((tx, sub_y + i * 22), ln, font=font_sub, fill=ERROR_COLORS["subtitle"])
else:
draw.text((tx, sub_y), sub, font=font_sub, fill=ERROR_COLORS["subtitle"])
if last_success:
font_meta = fonts.get("16") or fonts.get("default")
meta_y = y + h - 22
draw.text((tx, meta_y), f"Letzte Aktualisierung: {last_success}",
font=font_meta, fill=ERROR_COLORS["muted"])
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" | ...
# ---- Metadaten ----
name: str = ""
label: str = ""
description: str = ""
category: str = "general"
# 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."""
"""Daten holen.
Empfohlen: nutze `fetch_with_retry(self._fetch_internal)` für 3x retry.
Bei 3x fail: gib {"_error": "..."} zurück statt zu crashen.
"""
@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."""
"""Zeichne in den gegebenen Slot."""
# ---- Optionale Lifecycle-Hooks ----
def on_load(self) -> None:
"""Wird einmal beim Plugin-Load aufgerufen."""
pass
def on_unload(self) -> None:
"""Wird beim Beenden aufgerufen."""
pass
# ---- Helper für Plugins ----
def cfg(self, key: str, default: Any = None) -> Any:
return self.config.get(key, default)
@@ -64,6 +187,5 @@ def all_widget_classes() -> list[type[Widget]]:
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
+72 -25
View File
@@ -1,15 +1,23 @@
"""Clock: Uhrzeit + Datum. Responsive fuer alle Slot-Groessen."""
"""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, measure, fit_font, centered_text, is_small, is_wide
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. Responsives Layout fuer 1x1 bis 4x4."
description = "Aktuelle Uhrzeit und Datum. Responsiv fuer 1x1 bis 4x4 Slots."
category = "info"
config_schema = [
@@ -17,54 +25,93 @@ class Widget(Widget):
{"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": ["fg", "info", "accent", "ok", "warn", "alert"], "default": "fg"},
{"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": "fg"}
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, measure
from palette import fill_for
now = datetime.now()
pad = 8
accent = fill_for(self.cfg("accent_color", "fg"))
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):
# mini: nur Uhr, zentriert
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):
# Wide strip: Uhrzeit links gross, Datum rechts klein
# 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)
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)
centered_text(draw, day_str, x + w // 2, y + h // 2, w // 2, h // 2, font_day, INFO)
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
# Standard: Header (Datum), Big Time, Day unten
# Big Time zentriert
font_time = fit_font(draw, time_str, fonts, w - 2 * pad, int(h * 0.55))
centered_text(draw, time_str, x, y + pad, w, int(h * 0.55), font_time, FG)
# === 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
# Datum darunter
if self.cfg("show_date"):
font_date = fit_font(draw, date_str, fonts, w - 2 * pad, h // 5)
centered_text(draw, date_str, x, y + int(h * 0.55), w, h // 5, font_date, accent)
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
# Wochentag
if self.cfg("show_weekday"):
font_day = fit_font(draw, day_str, fonts, w - 2 * pad, h // 6)
centered_text(draw, day_str, x, y + int(h * 0.78), w, h // 6, font_day, INFO)
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)
+1100 -899
View File
File diff suppressed because it is too large Load Diff