Initial commit: epaper-dashboard for 7.3" ACeP 7-Color display

- dashboard.py: plugin-based renderer with 4x4 grid layout
- admin.py: web UI with layout editor + plugin configs
- layout.py: pack algorithm, item placement, grid system
- plugins/: clock, weather, system, spotify, strava, gmail, minimax, hello
- network_watchdog.py: WiFi AP/client mode management
- waveshare_epd_init.py: vendor driver stub
This commit is contained in:
ki
2026-08-26 14:12:43 +04:00
commit 28124c5617
38 changed files with 5408 additions and 0 deletions
+4
View File
@@ -0,0 +1,4 @@
# Admin-UI Credentials. Default im Code ist admin/admin — bitte ändern!
# Nach dem Ändern: admin.py neu starten.
EPAPER_ADMIN_USER=admin
EPAPER_ADMIN_PASSWORD=admin
+14
View File
@@ -0,0 +1,14 @@
__pycache__/
*.pyc
*.pyo
*.png
snap_*.png
render_*.png
sizes_*.png
admin_snapshot*.png
minimax_*.png
*.log
.DS_Store
.env
config.json
*.service
+9
View File
@@ -0,0 +1,9 @@
// Erlaubt koptikp (uid 1000) NetworkManager ohne Passwort zu steuern.
// Wir prüfen explizit auf uid, weil subject.local/active für SSH-Sessions
// oft nicht das tut, was man erwartet (logind behandelt SSH-Logins anders).
polkit.addRule(function(action, subject) {
if (action.id.indexOf("org.freedesktop.NetworkManager.") === 0 &&
subject.user === "koptikp") {
return polkit.Result.YES;
}
});
+76
View File
@@ -0,0 +1,76 @@
# 7.3" ePaper Dashboard
Plugin-basiertes Dashboard für Waveshare 7.3-inch ACeP 7-Color e-Paper (F) HAT auf einem Raspberry Pi 4.
## Architektur
- **Renderer (`dashboard.py`)**: Lädt `config.json`, instanziiert Plugins, refresht das Display alle `refresh_interval_s` Sekunden (≥180 empfohlen). Reagiert auf Live-Trigger via Unix-Socket.
- **Web-Admin (`admin.py`)**: HTTP Basic Auth, konfiguriert Slots, Plugins und Secrets im Browser. Triggert Live-Refresh.
- **Plugins (`plugins/`)**: Jedes Widget ist ein Python-Modul mit `Widget`-Klasse (siehe `plugins/base.py`). Eigenes Plugin = eine Datei in `plugins/`.
- **Templates (`templates/index.html`)**: Single-Page-UI, auto-reload-fähig.
## Setup
```bash
# Auf dem Pi:
sudo apt install python3-pil python3-numpy python3-flask git
git clone <dieses repo> # oder rsync von deinem Dev-Rechner
cd port/
# SPI in /boot/firmware/config.txt aktivieren (dtparam=spi=on)
# Reboot
python3 admin.py &
python3 dashboard.py &
```
Web-UI: <http://pi:8080/>
Default Login: `admin` / `admin` — bitte `EPAPER_ADMIN_PASSWORD` in `.env` setzen.
## Plugin schreiben
Eine Datei `plugins/my_widget.py`:
```python
from plugins.base import Widget
from palette import FG, INFO
class Widget(Widget):
name = "my_widget"
label = "Mein Widget"
description = "Zeigt etwas Nützliches"
category = "info"
config_schema = [
{"key": "api_key", "label": "API Key", "type": "secret"},
{"key": "interval", "label": "Intervall (Sekunden)", "type": "int", "default": 60},
{"key": "show_extra", "label": "Extra anzeigen", "type": "bool", "default": True},
]
default_config = {"api_key": "", "interval": 60, "show_extra": True}
def fetch(self):
# Daten holen (API call, etc.)
return {"value": 42}
def render(self, draw, fonts, x, y, w, h):
# In den Slot zeichnen
d = self.fetch()
draw.text((x + 10, y + 10), f"Value: {d['value']}", font=fonts["28"], fill=FG)
```
**Feld-Typen**: `string`, `int`, `float`, `bool`, `secret` (write-only), `select` (mit `choices`), `lat_lon`.
**Palette** (aus `palette.py`): `BLACK, WHITE, GREEN, BLUE, RED, YELLOW, ORANGE` + semantisch `FG, BG, OK, WARN, ALERT, INFO, ACCENT`.
## Mitgelieferte Plugins
- `clock` — Uhrzeit / Datum
- `weather` — Open-Meteo Wetter (kein API-Key)
- `system` — Pi CPU/RAM/Uptime
- `hello` — Demo
- `spotify` — Last.fm Scrobble
- `strava` — Strava Aktivitäten
- `gmail` — Gmail Unread
## Einschränkungen
- **Full Refresh only**: ACeP-Displays vertragen keine schnellen Partial-Refreshes (Verschleiß). Das Hersteller-empfohlene Minimum ist 180s.
- **Kein Live-Websocket**: Snapshot wird per Refresh-Now-Button ausgelöst. Bei Bedarf ergänzbar.
+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)
+545
View File
@@ -0,0 +1,545 @@
"""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: return a
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)
@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"]})
@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("/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()
+10
View File
@@ -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}}
]
}
+449
View File
@@ -0,0 +1,449 @@
"""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 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)
aldrich = FONT_DIR / "Aldrich-Regular.ttc"
clock = FONT_DIR / "advanced_led_board-7.ttc"
sizes = [16, 20, 24, 28, 32, 36, 48, 60, 80, 100]
fonts = {}
if aldrich.exists():
for s in sizes:
try:
fonts[str(s)] = ImageFont.truetype(str(aldrich), s)
except Exception:
pass
if clock.exists():
try:
fonts["clock"] = ImageFont.truetype(str(clock), 140)
except Exception:
pass
if not fonts:
fonts["default"] = ImageFont.load_default()
fonts["20"] = 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)
# Outer grid lines (subtle)
for c in range(1, 4):
draw.line((c * 200, 0, c * 200, 480), fill=FG, width=1)
draw.line((0, c * 120, 800, c * 120), fill=FG, 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)
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
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)
img = render_full(items, widgets, self.fonts)
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()
+1042
View File
File diff suppressed because one or more lines are too long
+31
View File
@@ -0,0 +1,31 @@
#!/bin/bash
# Erlaubt dem koptikp-User NetworkManager ohne Passwort zu steuern.
# Wird via polkit ausgewertet.
set -e
RULE='polkit.addRule(function(action, subject) {
if (action.id.indexOf("org.freedesktop.NetworkManager.") === 0 &&
subject.local === true && subject.active === true) {
return polkit.Result.YES;
}
});'
RULE_FILE="/etc/polkit-1/rules.d/50-allow-networkmanager.rules"
echo "Schreibe $RULE_FILE..."
sudo -n bash -c "cat > $RULE_FILE <<'RULEEOF'
$POLKIT_RULE
RULEEOF
chmod 644 $RULE_FILE"
echo "--- Inhalt: ---"
cat $RULE_FILE
echo
echo "--- Test: nmcli hotspot (kein pw mehr nötig) ---"
sudo -n nmcli device wifi hotspot ifname wlan0 ssid epaper-test password test1234
sleep 3
nmcli -t -f NAME,TYPE,DEVICE connection show --active
echo
echo "--- Cleanup ---"
sudo -n nmcli connection down epaper-test 2>&1
sudo -n nmcli connection delete epaper-test 2>&1
echo "Fertig."
+36
View File
@@ -0,0 +1,36 @@
#!/bin/bash
# Installiert die systemd-Services fuer das ePaper-Dashboard.
set -e
SERVICE_DIR=/etc/systemd/system
SERVICE_SRC=/home/koptikp/epaper-dashboard/port
echo "=== installiere service units ==="
sudo -n install -m 644 "$SERVICE_SRC/epaper-dashboard.service" "$SERVICE_DIR/epaper-dashboard.service"
sudo -n install -m 644 "$SERVICE_SRC/epaper-admin.service" "$SERVICE_DIR/epaper-admin.service"
echo
echo "=== stoppe alte Prozesse (manuell gestartet) ==="
pkill -f "python3.*admin.py" 2>/dev/null || true
pkill -f "python3.*dashboard.py" 2>/dev/null || true
sleep 2
echo "=== systemctl reload + enable + restart ==="
sudo -n systemctl daemon-reload
sudo -n systemctl enable epaper-dashboard.service
sudo -n systemctl enable epaper-admin.service
sudo -n systemctl restart epaper-dashboard.service
sudo -n systemctl restart epaper-admin.service
sleep 3
echo
echo "=== status dashboard ==="
sudo -n systemctl is-active epaper-dashboard.service
sudo -n systemctl is-active epaper-admin.service
echo
echo "=== ports ==="
ss -ltn | grep -E "8080" | head -3
echo
echo "Logs: journalctl -u epaper-dashboard -f"
echo " journalctl -u epaper-admin -f"
+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}
+427
View File
@@ -0,0 +1,427 @@
"""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()
# ============================================================================
# Watchdog
# ============================================================================
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
+1
View File
@@ -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
+70
View File
@@ -0,0 +1,70 @@
"""Clock: Uhrzeit + Datum. Responsive fuer alle Slot-Groessen."""
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
class Widget(Widget):
name = "clock"
label = "Uhrzeit / Datum"
description = "Aktuelle Uhrzeit und Datum. Responsives Layout fuer 1x1 bis 4x4."
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": ["fg", "info", "accent", "ok", "warn", "alert"], "default": "fg"},
]
default_config = {"format_24h": True, "show_seconds": False,
"show_date": True, "show_weekday": True, "accent_color": "fg"}
def fetch(self):
return {}
def render(self, draw, fonts, x, y, w, h):
from palette import fill_for, measure
now = datetime.now()
pad = 8
accent = fill_for(self.cfg("accent_color", "fg"))
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()
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
if is_wide(w, h):
# Wide strip: Uhrzeit links gross, Datum rechts klein
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)
if self.cfg("show_date"):
font_date = fit_font(draw, date_str, fonts, w // 2 - 2 * pad, h // 2)
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)
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)
# 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)
# 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)
+58
View File
@@ -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)
+27
View File
@@ -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")))
+271
View File
@@ -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)
+78
View File
@@ -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)
+99
View File
@@ -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)
+163
View File
@@ -0,0 +1,163 @@
"""System-Plugin: CPU-Last, RAM, Uptime. Responsive."""
import os, sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from plugins.base import Widget
from palette import FG, INFO, OK, WARN, ALERT, measure, fit_font, is_small, is_wide, hbar, parse_thresholds, DEFAULT_THRESHOLDS
def _read_loadavg():
try:
with open("/proc/loadavg") as f:
parts = f.read().split()
return float(parts[0])
except Exception:
return 0.0
def _read_mem():
try:
with open("/proc/meminfo") as f:
lines = f.readlines()
total = int([l for l in lines if l.startswith("MemTotal")][0].split()[1])
avail = int([l for l in lines if l.startswith("MemAvailable")][0].split()[1])
return total, total - avail, avail
except Exception:
return 1, 0, 1
def _read_uptime():
try:
with open("/proc/uptime") as f:
return float(f.read().split()[0])
except Exception:
return 0.0
def _fmt_uptime(s):
days, rem = divmod(int(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 _system_stats():
ncpu = os.cpu_count() or 1
la = _read_loadavg()
total, used, avail = _read_mem()
return {
"cpu_pct": min(int((la / ncpu) * 100), 100),
"ram_used_mb": used // 1024,
"ram_total_mb": total // 1024,
"ram_pct": int(used * 100 / total) if total else 0,
"uptime_s": _read_uptime(),
}
class Widget(Widget):
name = "system"
label = "System (CPU/RAM)"
description = "Live CPU-Last, RAM-Auslastung, Uptime. Responsive Layout fuer 1x1 bis 4x4."
category = "system"
config_schema = [
{"key": "show_uptime", "label": "Uptime anzeigen", "type": "bool", "default": True},
{"key": "show_load", "label": "Load Average anzeigen", "type": "bool", "default": True},
{"key": "compact", "label": "Kompaktmodus (nur Prozent)", "type": "bool", "default": False},
{"key": "bar_thresholds", "label": "CPU/RAM-Schwellen (Format: ok,warn,alert oder ok@50,warn@80,alert@95)",
"type": "string", "default": "ok@50,warn@80,alert@95",
"help": "Legt fest, ab wann eine Bar grün/gelb/rot wird. Mit '@P' setzt du die Schwelle in Prozent."},
{"key": "bar_gradient", "label": "Verlaufsmodus (Balken zeigt alle Stufen gleichzeitig)",
"type": "bool", "default": True,
"help": "Wenn aus, ist die Bar einfarbig in der Farbe der aktuellen Schwelle."},
]
default_config = {"show_uptime": True, "show_load": True, "compact": False,
"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):
return _system_stats()
def render(self, draw, fonts, x, y, w, h):
from palette import fill_for, measure
pad = 8
s = self.fetch()
cpu, ram = s["cpu_pct"], s["ram_pct"]
thresholds, gradient = self._bar_args()
# Im compact/small mode bleiben die Fallback-Farben (Text-Label reicht)
if self.cfg("compact") or is_small(w, h):
# mini/compact: nur CPU + RAM als 2 kleine bars
label_font = fit_font(draw, "CPU", fonts, w - 2 * pad, 20)
draw.text((x + pad, y + pad), "CPU", font=label_font, fill=FG)
bar_y = y + pad + 22
bar_w = w - 2 * pad
bar_h = max(8, (h - 30) // 4)
hbar(draw, x + pad, bar_y, bar_w, bar_h, cpu,
thresholds=thresholds, gradient=gradient)
ram_y = bar_y + bar_h + 6
draw.text((x + pad, ram_y), "RAM", font=label_font, fill=FG)
hbar(draw, x + pad, ram_y + 22, bar_w, bar_h, ram,
thresholds=thresholds, gradient=gradient)
return
if is_wide(w, h):
# Wide strip: CPU% | RAM% | Uptime (each third)
col_w = w // 3
for i, (label, pct) in enumerate([
("CPU", cpu), ("RAM", ram),
]):
cx = x + i * col_w
# Color = passende Schwelle
col_color = WARN
if thresholds:
for max_p, c in thresholds:
if pct <= max_p:
col_color = c; break
label_font = fit_font(draw, label, fonts, col_w - 2 * pad, 22)
draw.text((cx + pad, y + pad), f"{label} {pct}%",
font=label_font, fill=col_color)
if self.cfg("show_uptime"):
up_str = "up " + _fmt_uptime(s["uptime_s"])
font_up = fit_font(draw, up_str, fonts, col_w - 2 * pad, 22)
draw.text((x + 2 * col_w + pad, y + pad), up_str, font=font_up, fill=INFO)
return
# Standard: Header, CPU-Bar, RAM-Bar, Uptime
header_font = fonts.get("28") or fonts.get("24")
draw.text((x + pad, y + pad), "SYSTEM", font=header_font, fill=INFO)
# Big numbers + bars
cur_y = y + 40
# CPU row
text = f"CPU: {cpu}%"
draw.text((x + pad, cur_y), text, font=fonts.get("32", fonts.get("24")), fill=FG)
bar_y = cur_y + 36
bar_h = 28
hbar(draw, x + pad, bar_y, w - 2 * pad, bar_h, cpu,
thresholds=thresholds, gradient=gradient)
# RAM row
cur_y = bar_y + bar_h + 12
text = f"RAM: {s['ram_used_mb']}/{s['ram_total_mb']} MB ({ram}%)"
font_text = fit_font(draw, text, fonts, w - 2 * pad, 28)
draw.text((x + pad, cur_y), text, font=font_text, fill=FG)
bar_y2 = cur_y + 30
hbar(draw, x + pad, bar_y2, w - 2 * pad, bar_h, ram,
thresholds=thresholds, gradient=gradient)
# Bottom: load + uptime
bottom_y = bar_y2 + bar_h + 8
if self.cfg("show_load") or self.cfg("show_uptime"):
parts = []
if self.cfg("show_load"):
parts.append(f"load {s['cpu_pct']/100.0 * (os.cpu_count() or 1):.2f}")
if self.cfg("show_uptime"):
parts.append("up " + _fmt_uptime(s["uptime_s"]))
line = " · ".join(parts)
font_bot = fit_font(draw, line, fonts, w - 2 * pad, 22)
draw.text((x + pad, bottom_y), line, font=font_bot, fill=FG)
+179
View File
@@ -0,0 +1,179 @@
"""Wetter-Plugin (Open-Meteo, kein API-Key). Responsive."""
import os, sys, json, math
from datetime import datetime
import urllib.request, urllib.error
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, RED, BLUE, measure, fit_font, centered_text, is_small, is_wide
def _fetch_openmeteo(lat, lon):
url = (f"https://api.open-meteo.com/v1/forecast?"
f"latitude={lat}&longitude={lon}"
f"&current=temperature_2m,relative_humidity_2m,weather_code,"
f"wind_speed_10m,wind_direction_10m,surface_pressure,uv_index,is_day"
f"&hourly=temperature_2m,weather_code"
f"&forecast_days=2&timezone=auto")
try:
with urllib.request.urlopen(url, timeout=8) as r:
return json.loads(r.read())
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as e:
return {"_error": str(e)}
class Widget(Widget):
name = "weather"
label = "Wetter (Open-Meteo)"
description = "Aktuelles Wetter, Forecast und Windrose. Kein API-Key. Responsive."
category = "weather"
config_schema = [
{"key": "location", "label": "Standort (lat,lon)", "type": "lat_lon",
"default": "52.52,13.41", "help": "Beispiel: 52.52,13.41 für Berlin"},
{"key": "show_uv", "label": "UV-Index anzeigen", "type": "bool", "default": True},
{"key": "show_forecast_hours", "label": "Vorhersage-Stunden", "type": "int",
"default": 4, "help": "Wie viele Stunden voraus (1-8)"},
{"key": "show_wind_compass", "label": "Windrose anzeigen", "type": "bool", "default": True},
]
default_config = {"location": "52.52,13.41", "show_uv": True,
"show_forecast_hours": 4, "show_wind_compass": True}
def fetch(self):
loc = self.cfg("location", "52.52,13.41")
try:
parts = [s.strip() for s in loc.split(",")]
lat = float(parts[0]); lon = float(parts[1])
except Exception:
lat, lon = 52.52, 13.41
data = _fetch_openmeteo(lat, lon)
if "_error" not in data:
data["_lat"] = lat; data["_lon"] = lon
return data
def _draw_compass(self, draw, cx, cy, r, deg, fonts):
draw.ellipse((cx-r, cy-r, cx+r, cy+r), outline=FG, width=2)
for a in range(0, 360, 45):
rad = math.radians(a - 90)
inner = r - 8 if a % 90 == 0 else r - 4
x1 = cx + inner * math.cos(rad); y1 = cy + inner * math.sin(rad)
x2 = cx + r * math.cos(rad); y2 = cy + r * math.sin(rad)
draw.line((x1, y1, x2, y2), fill=FG, width=2)
f20 = fonts.get("20", fonts.get("16"))
for a, t in [(270,"N"), (0,"E"), (90,"S"), (180,"W")]:
rad = math.radians(a - 90)
tx = cx + (r+12) * math.cos(rad) - 6
ty = cy + (r+12) * math.sin(rad) - 9
draw.text((tx, ty), t, font=f20, fill=FG)
rad_arrow = math.radians(deg - 90)
tip_x = cx + (r - 14) * math.cos(rad_arrow)
tip_y = cy + (r - 14) * math.sin(rad_arrow)
base = math.radians(150)
lx = cx + 18 * math.cos(rad_arrow + base); ly = cy + 18 * math.sin(rad_arrow + base)
rx = cx + 18 * math.cos(rad_arrow - base); ry = cy + 18 * math.sin(rad_arrow - base)
draw.polygon([(tip_x, tip_y), (lx, ly), (rx, ry)], fill=RED)
draw.ellipse((cx-3, cy-3, cx+3, cy+3), fill=FG)
def render(self, draw, fonts, x, y, w, h):
from palette import fill_for
pad = 8
d = self.fetch()
if "_error" in d:
draw.text((x + pad, y + pad), "WETTER", font=fonts.get("24", fonts.get("20")), fill=INFO)
draw.text((x + pad, y + 60), f"Fehler: {d['_error'][:40]}", font=fonts.get("20", fonts.get("16")), fill=ALERT)
return
cur = d.get("current", {})
temp = cur.get("temperature_2m", 0)
temp_color = RED if temp >= 28 else ORANGE if temp >= 20 else BLUE if temp <= 5 else FG
if is_small(w, h):
# mini: nur Big Temp + ggf UV
temp_str = f"{int(temp + 0.5)}°"
font = fit_font(draw, temp_str, fonts, w - 2 * pad, int(h * 0.6))
centered_text(draw, temp_str, x, y, w, int(h * 0.6), font, temp_color)
if self.cfg("show_uv"):
uv_int = int(cur.get("uv_index", 0) + 0.5)
uv_str = f"UV {uv_int}"
font_uv = fit_font(draw, uv_str, fonts, w - 2 * pad, h // 4)
centered_text(draw, uv_str, x, y + int(h * 0.62), w, h // 4, font_uv, FG)
return
if is_wide(w, h):
# Wide strip: Temp + Humidity + Wind
temp_str = f"{int(temp + 0.5)}°"
font = fit_font(draw, temp_str, fonts, w // 3 - 2 * pad, h - 2 * pad)
centered_text(draw, temp_str, x, y, w // 3, h, font, temp_color)
hum = cur.get("relative_humidity_2m", 0)
ws = cur.get("wind_speed_10m", 0)
wd = cur.get("wind_direction_10m", 0)
stats = f"Hum {hum}%\nWind {ws} km/h\nDir {wd}°"
font_s = fit_font(draw, "Hum 100%", fonts, w - w // 3 - 2 * pad, h // 3)
y_off = y + pad
for line in stats.split("\n"):
centered_text(draw, line, x + w // 3, y_off, w - w // 3, h // 3, font_s, FG)
y_off += h // 3
return
# Standard: Big temp, info, forecast
# Header
draw.text((x + pad, y + pad), "WETTER", font=fonts.get("24", fonts.get("20")), fill=INFO)
# Big temp
temp_str = f"{int(temp + 0.5)}°"
font = fit_font(draw, temp_str, fonts, w // 2 - 2 * pad, int(h * 0.6))
draw.text((x + pad, y + 50), temp_str, font=font, fill=temp_color)
# Stats right
hum = cur.get("relative_humidity_2m", 0)
ws = cur.get("wind_speed_10m", 0)
wd = cur.get("wind_direction_10m", 0)
stats_y = y + 60
for line, col in [(f"Hum {hum}%", BLUE), (f"Wind {ws} km/h", INFO), (f"Dir {wd}°", FG)]:
font_l = fit_font(draw, line, fonts, w // 2 - 2 * pad, 24)
draw.text((x + w // 2 + pad, stats_y), line, font=font_l, fill=col)
stats_y += 30
# UV box top right
if self.cfg("show_uv"):
uv = cur.get("uv_index", 0)
uv_int = int(uv + 0.5)
box_w = min(80, w // 5)
box_x = x + w - box_w - pad
box_y = y + pad
box_h = 36
if uv_int >= 6:
draw.rectangle((box_x, box_y, box_x + box_w, box_y + box_h), fill=ALERT)
draw.text((box_x + 6, box_y + 4), f"UV {uv_int}", font=fonts.get("20", fonts.get("16")), fill=BG)
else:
draw.rectangle((box_x, box_y, box_x + box_w, box_y + box_h), outline=FG, width=2)
draw.text((box_x + 6, box_y + 4), f"UV {uv_int}", font=fonts.get("20", fonts.get("16")), fill=FG)
# Compass optional
if self.cfg("show_wind_compass") and w >= 280 and h >= 280:
cr = min(60, w // 8, h // 6)
cx = x + w - cr - pad
cy = y + h - cr - pad
self._draw_compass(draw, cx, cy, cr, wd, fonts)
# Forecast bottom
fc_h = max(1, min(int(self.cfg("show_forecast_hours", 4)), 8))
times = d.get("hourly", {}).get("time", [])
temps = d.get("hourly", {}).get("temperature_2m", [])
if times and h > 200:
cur_iso = datetime.now().strftime("%Y-%m-%dT%H:00")
try: idx0 = times.index(cur_iso)
except ValueError: idx0 = 0
forecast_y = y + h - 70
# Forecast-Strip nimmt 60% der Slot-Breite, von links
strip_w = int(w * 0.6)
cell_w = strip_w // fc_h
for i in range(fc_h):
j = idx0 + i + 1
if j >= len(times): break
hh = times[j].split("T")[1][:5]
tt = int(temps[j] + 0.5)
color = ORANGE if tt >= 25 else BLUE if tt <= 5 else FG
cx2 = x + pad + i * cell_w
draw.text((cx2, forecast_y), hh, font=fonts.get("20", fonts.get("16")), fill=FG)
font_t = fit_font(draw, f"{tt}°", fonts, cell_w - 4, 40)
draw.text((cx2, forecast_y + 22), f"{tt}°", font=font_t, fill=color)
+8
View File
@@ -0,0 +1,8 @@
import sys
sys.path.insert(0, ".")
import dashboard
d = dashboard.Dashboard()
img = d.render_once()
img.save("/tmp/render_test.png")
print("rendered:", img.size)
print("widgets:", [(w.name if w else None) for w in d.widgets])
+14
View File
@@ -0,0 +1,14 @@
import os, socket
p = "/tmp/test.sock"
if os.path.exists(p):
os.remove(p)
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
try:
srv.bind(p)
srv.listen(1)
print("bound, listening")
print("exists immediately after bind:", os.path.exists(p))
finally:
srv.close()
if os.path.exists(p):
os.remove(p)
+7
View File
@@ -0,0 +1,7 @@
#!/bin/bash
# startet admin.py im Hintergrund mit korrektem CWD
cd /home/koptikp/epaper-dashboard/port
pkill -f admin.py 2>/dev/null
sleep 1
nohup env EPAPER_ADMIN_PORT=8080 python3 admin.py > /tmp/admin.log 2>&1 &
echo "admin pid: $!"
+4
View File
@@ -0,0 +1,4 @@
#!/bin/bash
chmod +x /home/koptikp/epaper-dashboard/port/start_admin.sh 2>/dev/null
# Sende das Script selbst
cat /home/koptikp/epaper-dashboard/port/start_admin.sh
+9
View File
@@ -0,0 +1,9 @@
#!/bin/bash
cd /home/koptikp/epaper-dashboard/port
pkill -f dashboard.py 2>/dev/null
sleep 1
nohup env python3 dashboard.py > /tmp/dashboard.log 2>&1 &
echo "dashboard pid: $!"
sleep 2
echo "--- log preview ---"
head -20 /tmp/dashboard.log
+1002
View File
File diff suppressed because it is too large Load Diff
+20
View File
@@ -0,0 +1,20 @@
#!/bin/bash
cat > /home/koptikp/epaper-dashboard/port/.env <<'ENVEOF'
EPAPER_ADMIN_USER=admin
EPAPER_ADMIN_PASSWORD=changeme123
ENVEOF
chmod 600 /home/koptikp/epaper-dashboard/port/.env
echo "=== .env geschrieben ==="
ls -la /home/koptikp/epaper-dashboard/port/.env
echo
echo "=== restart admin ==="
bash /home/koptikp/epaper-dashboard/port/start_admin.sh
sleep 2
echo
echo "=== test mit altem passwort (sollte 401) ==="
curl -s -u admin:admin -o /dev/null -w "admin:admin -> HTTP %{http_code}\n" http://127.0.0.1:8080/status.json
echo "=== test mit neuem passwort (sollte 200) ==="
curl -s -u admin:changeme123 -o /dev/null -w "admin:changeme123 -> HTTP %{http_code}\n" http://127.0.0.1:8080/status.json
echo
echo "=== sicherheits-check: passwort in logs? ==="
grep -E "changeme123|admin:" /tmp/admin.log || echo "(kein Treffer in logs)"
+40
View File
@@ -0,0 +1,40 @@
"""Demo: Progress-Bars mit Gradient und konfigurierbaren Schwellen."""
import sys; sys.path.insert(0, ".")
from PIL import Image, ImageDraw, ImageFont
from palette import hbar, parse_thresholds, FG, OK, WARN, ALERT, ORANGE, BLUE, GREEN, RED, YELLOW
def font(s): return ImageFont.truetype("fnt/Aldrich-Regular.ttc", s)
fonts = {str(s): font(s) for s in [16, 20, 24]}
img = Image.new("RGB", (800, 600), (255,255,255))
draw = ImageDraw.Draw(img)
# Title
draw.text((20, 10), "Gradient Bars - 3-stufig default (50/80/95)", font=font(20), fill=FG)
# Row 1: gradient on (default thresholds)
draw.text((20, 50), "Gradient AN, Thresholds 50/80/95", font=font(16), fill=FG)
hbar(draw, 20, 80, 760, 30, 25, thresholds=parse_thresholds("ok@50,warn@80,alert@95"), gradient=True)
draw.text((20, 115), "25% (ok)", font=font(16), fill=FG)
hbar(draw, 20, 140, 760, 30, 65, thresholds=parse_thresholds("ok@50,warn@80,alert@95"), gradient=True)
draw.text((20, 175), "65% (warn)", font=font(16), fill=FG)
hbar(draw, 20, 200, 760, 30, 90, thresholds=parse_thresholds("ok@50,warn@80,alert@95"), gradient=True)
draw.text((20, 235), "90% (alert)", font=font(16), fill=FG)
# Row 2: gradient off
draw.text((20, 280), "Gradient AUS (einfarbig in aktueller Schwelle)", font=font(20), fill=FG)
hbar(draw, 20, 310, 760, 30, 25, thresholds=parse_thresholds("ok@50,warn@80,alert@95"), gradient=False)
draw.text((20, 345), "25% einfarbig green", font=font(16), fill=FG)
hbar(draw, 20, 370, 760, 30, 65, thresholds=parse_thresholds("ok@50,warn@80,alert@95"), gradient=False)
draw.text((20, 405), "65% einfarbig yellow", font=font(16), fill=FG)
hbar(draw, 20, 430, 760, 30, 90, thresholds=parse_thresholds("ok@50,warn@80,alert@95"), gradient=False)
draw.text((20, 465), "90% einfarbig red", font=font(16), fill=FG)
# Row 3: custom thresholds (mehr Stufen)
draw.text((20, 510), "Custom: 4 Stufen 30/60/80/95, custom colors", font=font(20), fill=FG)
spec = [(30, GREEN), (60, YELLOW), (80, ORANGE), (95, RED)]
hbar(draw, 20, 545, 760, 30, 50, thresholds=spec, gradient=True)
draw.text((20, 580), "50% → yellow segment", font=font(16), fill=FG)
img.save("/tmp/gradient_demo.png")
print("saved /tmp/gradient_demo.png")
+32
View File
@@ -0,0 +1,32 @@
import sys; sys.path.insert(0, ".")
import json
from layout import Item
from dashboard import render_full, load_fonts
cfg = json.load(open("config.json"))
items = [Item.from_dict(d) for d in cfg["layout"]["items"]]
plugins = cfg.get("plugin_configs", {})
from plugins.base import all_widget_classes
classes = {c.name: c for c in all_widget_classes()}
widgets = []
for it in items:
cls = classes.get(it.plugin)
if cls:
try:
w = cls(plugins.get(it.plugin, {}))
w.fetch()
widgets.append(w)
except Exception as e:
print(f"widget {it.plugin} failed: {e}")
widgets.append(None)
else:
widgets.append(None)
fonts = load_fonts()
print("items:", [(i.id, i.plugin, i.x, i.y, i.w, i.h) for i in items])
print("widgets:", [w.name if w else None for w in widgets])
img = render_full(items, widgets, fonts)
img.save("/tmp/render_grid_v2.png")
print(f"saved /tmp/render_grid_v2.png size={img.size}")
+26
View File
@@ -0,0 +1,26 @@
import sys; sys.path.insert(0, ".")
from PIL import Image, ImageDraw, ImageFont
from plugins.minimax import Widget
class FakeWidget(Widget):
def fetch(self):
return {
"windows": [
{"label": "5-Hour", "used_pct": 47.0, "remaining_pct": 53.0, "reset_seconds": 1820, "raw": {}},
{"label": "Weekly", "used_pct": 23.0, "remaining_pct": 77.0, "reset_seconds": 124800, "raw": {}},
],
"credits": {"points": 12345.0},
}
img = Image.new("RGB", (400, 240), (255, 255, 255))
draw = ImageDraw.Draw(img)
fonts = {
"16": ImageFont.truetype("fnt/Aldrich-Regular.ttc", 16),
"20": ImageFont.truetype("fnt/Aldrich-Regular.ttc", 20),
"24": ImageFont.truetype("fnt/Aldrich-Regular.ttc", 24),
"28": ImageFont.truetype("fnt/Aldrich-Regular.ttc", 28),
}
w = FakeWidget({"show_credits": True, "label_5h": "5-Hour", "label_weekly": "Weekly"})
w.render(draw, fonts, 0, 0, 400, 240)
img.save("/tmp/minimax_fake.png")
print("saved /tmp/minimax_fake.png")
+49
View File
@@ -0,0 +1,49 @@
"""Test: clock als 4x1-Strip (200x120), minimax als 2x2, weather als 2x2 (lower-right).
Zeigt ob Widgets auf verschiedene Groessen reagieren."""
import sys; sys.path.insert(0, ".")
from PIL import Image, ImageDraw, ImageFont
from layout import Item, GRID_COLS, GRID_ROWS, CELL_W, CELL_H, pack
from plugins.clock import Widget as ClockW
from plugins.minimax import Widget as MiniMaxW
from plugins.weather import Widget as WeatherW
from plugins.hello import Widget as HelloW
# Force-load fonts
def font(size):
return ImageFont.truetype("fnt/Aldrich-Regular.ttc", size)
fonts = {
"16": font(16), "20": font(20), "24": font(24),
"28": font(28), "32": font(32), "48": font(48),
"60": font(60), "80": font(80),
"clock": ImageFont.truetype("fnt/advanced_led_board-7.ttc", 100),
}
# Layout:
# clock (4x1) | y=0
# minimax (2x1) hello (2x1) | y=1
# weather (4x2) | y=2..3
items = [
Item("a", "clock", 0, 0, 4, 1),
Item("b", "clock", 0, 0, 1, 1), # tiny
Item("c", "clock", 0, 0, 2, 2), # large
Item("d", "clock", 0, 0, 4, 4), # full
]
print("Testing CLOCK widget at different sizes:")
print("="*60)
img = Image.new("RGB", (800, 480), (255,255,255))
draw = ImageDraw.Draw(img)
# 4x1 strip top
w = ClockW({})
w.fetch()
w.render(draw, fonts, 0, 0, 800, 120)
# 2x2 below
w = ClockW({"show_date": True})
w.fetch()
w.render(draw, fonts, 0, 120, 400, 240)
# mini 1x1 bottom-right
w = ClockW({})
w.fetch()
w.render(draw, fonts, 600, 360, 200, 120)
img.save("/tmp/sizes_clock.png")
print("saved /tmp/sizes_clock.png")
+57
View File
@@ -0,0 +1,57 @@
"""Test alle Widgets in verschiedenen Groessen, um Layout-Probleme zu finden."""
import sys; sys.path.insert(0, ".")
from PIL import Image, ImageDraw, ImageFont
from layout import Item
from plugins.clock import Widget as ClockW
from plugins.weather import Widget as WeatherW
from plugins.system import Widget as SystemW
from plugins.minimax import Widget as MiniMaxW
from plugins.hello import Widget as HelloW
import json
def font(size):
return ImageFont.truetype("fnt/Aldrich-Regular.ttc", size)
fonts = {
"16": font(16), "20": font(20), "24": font(24),
"28": font(28), "32": font(32), "48": font(48),
"60": font(60), "80": font(80),
"clock": ImageFont.truetype("fnt/advanced_led_board-7.ttc", 100),
}
plugins_cfg = json.load(open("config.json")).get("plugin_configs", {})
# Layout test grid: 4x4
# (0,0)-(3,0): system 4x1 wide
# (0,1)-(1,1): weather 2x1
# (2,1)-(3,1): minimax 2x1
# (0,2)-(1,3): hello 2x2
# (2,2)-(3,3): mix
img = Image.new("RGB", (800, 480), (255,255,255))
draw = ImageDraw.Draw(img)
# system 4x1 (full width strip)
w = SystemW(plugins_cfg.get("system", {}))
w.fetch()
w.render(draw, fonts, 0, 0, 800, 120)
# weather 2x1
w = WeatherW(plugins_cfg.get("weather", {"location": "52.52,13.41"}))
w.fetch()
w.render(draw, fonts, 0, 120, 400, 120)
# minimax 2x1
w = MiniMaxW(plugins_cfg.get("minimax", {}))
w.fetch()
w.render(draw, fonts, 400, 120, 400, 120)
# hello 2x2
w = HelloW({"text": "Layout Test", "color": "accent"})
w.render(draw, fonts, 0, 240, 400, 240)
# clock 2x2
w = ClockW({"show_date": True})
w.fetch()
w.render(draw, fonts, 400, 240, 400, 240)
img.save("/tmp/sizes_mix.png")
print("saved /tmp/sizes_mix.png")
+29
View File
@@ -0,0 +1,29 @@
"""Test: System Widget mit verschiedenen Threshold-Konfigurationen."""
import sys; sys.path.insert(0, ".")
from PIL import Image, ImageDraw, ImageFont
from plugins.system import Widget as SystemW
def font(s): return ImageFont.truetype("fnt/Aldrich-Regular.ttc", s)
fonts = {str(s): font(s) for s in [16, 20, 24, 28, 32]}
img = Image.new("RGB", (800, 480), (255,255,255))
draw = ImageDraw.Draw(img)
# Drei Konfigurationen nebeneinander
configs = [
{"name": "Default (50/80/95, gradient)",
"cfg": {"bar_thresholds": "ok@50,warn@80,alert@95", "bar_gradient": True}},
{"name": "Strenger (30/60/85, gradient)",
"cfg": {"bar_thresholds": "ok@30,warn@60,alert@85", "bar_gradient": True}},
{"name": "Einfarbig (40/70/95)",
"cfg": {"bar_thresholds": "ok@40,warn@70,alert@95", "bar_gradient": False}},
]
for i, c in enumerate(configs):
x = i * 267
draw.text((x + 8, 5), c["name"], font=font(14), fill=(0,0,0))
w = SystemW({**c["cfg"], "compact": False})
w.fetch()
w.render(draw, fonts, x, 30, 260, 200)
img.save("/tmp/system_thresholds.png")
print("saved /tmp/system_thresholds.png")
+3
View File
@@ -0,0 +1,3 @@
"""Waveshare 7.3-inch ACeP 7-Color e-Paper (F) — vendored driver."""
from .epd7in3f import EPD
from .epdconfig import module_exit