"""Flask-basierte Admin-UI für das 7.3" Dashboard. Routes: GET / → Grid-Builder: Slots, Plugins, Refresh-Intervall POST /config → Speichert config.json (Slot-Belegung + Intervall) POST /plugins/ → Speichert Plugin-spezifische Config POST /refresh → Triggert sofortigen Refresh via IPC GET /snapshot.png → Letztes gerendertes Bild (PNG) GET /plugins.json → Liste aller verfügbaren Plugins + Schemas (für UI) """ from __future__ import annotations import os, sys, json, base64, io, time from pathlib import Path HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) from flask import Flask, request, redirect, url_for, render_template, jsonify, Response, abort import dashboard as dashboard_mod from plugins.base import all_widget_classes import network_watchdog as net import layout as layout_mod app = Flask(__name__) # ============================================================================ # Auth (HTTP Basic) # ============================================================================ def _load_auth(): """Liest admin_user / admin_password aus .env. Reihenfolge (erstes gewinnt): 1. Umgebungsvariable EPAPER_ADMIN_USER / EPAPER_ADMIN_PASSWORD (nützlich für Container/CI) 2. /.env (eigene Datei, wird bevorzugt) 3. ~/.hermes/.env (global) Default: admin / admin — explizit nur für first-boot, damit der Service überhaupt erreichbar ist. Empfehlung: nach erstem Start .env anlegen. """ user, pw = "admin", "admin" env_user = os.environ.get("EPAPER_ADMIN_USER") env_pw = os.environ.get("EPAPER_ADMIN_PASSWORD") if env_user: user = env_user if env_pw: pw = env_pw for envfile in [HERE / ".env", Path.home() / ".hermes" / ".env"]: if not envfile.exists(): continue try: for line in envfile.read_text().splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) k = k.strip(); v = v.strip().strip('"').strip("'") if k == "EPAPER_ADMIN_USER" and v: user = v if k == "EPAPER_ADMIN_PASSWORD" and v: pw = v except Exception: pass return user, pw def check_auth(): user, pw = _load_auth() auth = request.headers.get("Authorization") if auth and auth.startswith("Basic "): try: decoded = base64.b64decode(auth[6:]).decode() u, p = decoded.split(":", 1) if u == user and p == pw: return True except Exception: pass return False def require_auth(): if not check_auth(): return Response( "Authentication required", 401, {"WWW-Authenticate": 'Basic realm="epaper-dashboard"'}, ) return None # ============================================================================ # Routes # ============================================================================ @app.route("/") def index(): a = require_auth() if a: # Demo-Modus: ?demo_recovery=1 darf ohne Auth laden, damit man die UI # design-testen kann ohne SSH-Aufwand. Static markup ohne live data. if request.args.get("demo_recovery") or request.args.get("demo"): demo = True else: return a else: demo = False cfg = dashboard_mod.load_config() slots = cfg.get("slots", []) # legacy classes = dashboard_mod.get_widget_classes() widgets_meta = [] for name, cls in classes.items(): widgets_meta.append({ "name": name, "label": cls.label, "description": cls.description, "category": cls.category, "schema": cls.config_schema, "defaults": cls.default_config, "default_size": layout_mod.auto_size_for_plugin(name), }) widgets_meta.sort(key=lambda w: w["label"]) # v2-Layout layout_items = cfg.get("layout", {}).get("items", []) plugin_configs = cfg.get("plugin_configs", {}) grid = cfg.get("layout", {}).get("grid", {"cols": 4, "rows": 4}) # Slot n -> (plugin_name, plugin_config_dict) slot_plugins = [(it.get("plugin", ""), plugin_configs.get(it.get("plugin", ""), {})) for it in layout_items] return render_template("index.html", # legacy slots=slots, widgets=widgets_meta, refresh_interval=cfg.get("refresh_interval_s", 180), categories=sorted(set(w["category"] for w in widgets_meta)), now=int(time.time()), # v2 layout_items=layout_items, plugin_configs=plugin_configs, grid=grid, size_presets=layout_mod.SIZE_PRESETS, slot_plugins=slot_plugins, demo_mode=demo) @app.route("/config", methods=["POST"]) def save_config(): a = require_auth() if a: return a cfg = dashboard_mod.load_config() # refresh interval if "refresh_interval_s" in request.form: try: cfg["refresh_interval_s"] = max(30, int(request.form["refresh_interval_s"])) except (TypeError, ValueError): pass # NOTE: legacy v1 slot-X-belegung wird nicht mehr unterstützt — die # Admin-UI nutzt seit v2 die Drag&Drop-API unter /api/layout für # Slot-Belegung. /config macht heute NUR noch den Refresh-Intervall. dashboard_mod.save_config(cfg) return redirect(url_for("index")) @app.route("/plugins/", methods=["POST"]) def save_plugin_config(idx): a = require_auth() if a: return a cfg = dashboard_mod.load_config() if idx < 0 or idx >= dashboard_mod.SLOTS: abort(400) slot = cfg["slots"][idx] classes = dashboard_mod.get_widget_classes() cls = classes.get(slot["plugin"]) if cls is None: abort(400, "unknown plugin") new_cfg = dict(slot.get("config", {})) for field in cls.config_schema: key = field["key"] form_val = request.form.get(f"slot_{idx}_{key}") ftype = field.get("type", "string") if ftype == "secret": # Secret-Felder: nur überschreiben wenn Form nicht leer. # Damit kann der User ein Secret löschen, indem er explizit leeres # Feld abschickt; ansonsten bleibt der alte Wert erhalten. if form_val is None: continue if form_val == "": new_cfg[key] = "" elif form_val == "__UNSET__": # wird vom UI nie geschickt; reserved für "Secret entfernen" new_cfg.pop(key, None) else: new_cfg[key] = form_val continue if form_val is None: if ftype == "bool": new_cfg[key] = False else: 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 OHNE bestehende Items zu verschieben (BUG-04). BUG-04: Vorher wurde `layout_mod.pack()` aufgerufen, das alle Items neu sortiert und in scan-line packt — ein 1x1 hello konnte einen 2x1 spotify aus seiner Position drängen. Jetzt first-fit: das neue Item landet in der ersten freien Zelle, alle anderen bleiben unverändert. Falls kein Platz: HTTP 409 mit Hinweis auf den Auto-Pack-Button. """ 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_list = cfg.setdefault("layout", {}).setdefault("items", []) existing = [layout_mod.Item.from_dict(d) for d in items_list] candidate = layout_mod.Item(new_id, plugin, 0, 0, w, h) placed = layout_mod.first_fit(candidate, existing) if placed is None: return jsonify({ "ok": False, "error": f"kein Platz für {w}×{h}-Item — bitte Auto-Pack klicken oder Items manuell verkleinern.", "hint": "use_auto_pack", }), 409 items_list.append(placed.to_dict()) dashboard_mod.save_config(cfg) return jsonify({"ok": True, "id": new_id, "items": items_list}) @app.route("/api/layout/item/", methods=["DELETE", "PATCH"]) def api_layout_item(iid): a = require_auth() if a: return a cfg = dashboard_mod.load_config() items = cfg.setdefault("layout", {}).setdefault("items", []) if request.method == "DELETE": items = [it for it in items if it.get("id") != iid] cfg["layout"]["items"] = items dashboard_mod.save_config(cfg) return jsonify({"ok": True}) # PATCH: update single fields data = request.get_json(silent=True) or {} for it in items: if it.get("id") == iid: for k in ("plugin", "x", "y", "w", "h"): if k in data: if k in ("x", "y", "w", "h"): try: it[k] = int(data[k]) except: pass else: it[k] = data[k] dashboard_mod.save_config(cfg) return jsonify({"ok": True, "items": items}) @app.route("/api/layout/pack", methods=["POST"]) def api_layout_pack(): """Auto-pack alle Items in den 4x4 grid.""" a = require_auth() if a: return a cfg = dashboard_mod.load_config() items = [layout_mod.Item.from_dict(d) for d in cfg.get("layout", {}).get("items", [])] packed = layout_mod.pack(items) cfg["layout"]["items"] = [it.to_dict() for it in packed] dashboard_mod.save_config(cfg) return jsonify({"ok": True, "items": cfg["layout"]["items"]}) DISPLAY_THEMES = { "default": { "label": "Classic White", "bg": "#ffffff", "fg": "#000000", "accent": "#ff8800", }, "dark": { "label": "Dark Mode", "bg": "#111111", "fg": "#f0f0f0", "accent": "#6d8eff", }, "sepia": { "label": "Sepia", "bg": "#f4ecd8", "fg": "#5b4636", "accent": "#c0392b", }, "nord": { "label": "Nord", "bg": "#eceff4", "fg": "#2e3440", "accent": "#88c0d0", }, "terminal": { "label": "Terminal (Green)", "bg": "#0d0d00", "fg": "#e8c840", "accent": "#ffcc00", }, } # Display designs — the actual visual layout approach DISPLAY_DESIGNS = { "classic": { "label": "Classic", "desc": "Original flat design, white background", }, "magazine": { "label": "Magazine", "desc": "Bold typography, minimal, colored accent bars", }, "cards": { "label": "Cards", "desc": "iOS-style panels with emoji icons, warm white", }, "kiosk": { "label": "Kiosk", "desc": "Data-dense, dark background, inverted info cards", }, } @app.route("/api/display_theme", methods=["GET", "POST"]) def api_display_theme(): """GET: list themes + current. POST: set theme (legacy color-only).""" a = require_auth() if a: return a cfg = dashboard_mod.load_config() if request.method == "GET": theme_key = cfg.get("display_theme", "default") theme = DISPLAY_THEMES.get(theme_key, DISPLAY_THEMES["default"]) return jsonify({ "current": theme_key, "themes": {k: v["label"] for k, v in DISPLAY_THEMES.items()}, "theme": theme, }) data = request.get_json(silent=True) or {} theme_key = data.get("theme", "default") if theme_key not in DISPLAY_THEMES: return jsonify({"ok": False, "error": "unknown theme"}), 400 cfg["display_theme"] = theme_key dashboard_mod.save_config(cfg) return jsonify({"ok": True, "current": theme_key, "theme": DISPLAY_THEMES[theme_key]}) @app.route("/api/display_design", methods=["GET", "POST"]) def api_display_design(): """Design für das Display (classic/magazine/cards/kiosk).""" a = require_auth() if a: return a cfg = dashboard_mod.load_config() if request.method == "GET": design_key = cfg.get("display_design", "classic") return jsonify({ "current": design_key, "designs": {k: {"label": v["label"], "desc": v["desc"]} for k, v in DISPLAY_DESIGNS.items()}, }) data = request.get_json(silent=True) or {} design_key = data.get("design", "classic") if design_key not in DISPLAY_DESIGNS: return jsonify({"ok": False, "error": "unknown design"}), 400 cfg["display_design"] = design_key dashboard_mod.save_config(cfg) return jsonify({"ok": True, "current": design_key}) @app.route("/api/plugin_config/", methods=["GET", "POST"]) def api_plugin_config(plugin_name): """Plugin-spezifische Config (separat vom Layout).""" a = require_auth() if a: return a classes = dashboard_mod.get_widget_classes() cls = classes.get(plugin_name) if cls is None: return jsonify({"ok": False, "error": "unknown plugin"}), 400 cfg = dashboard_mod.load_config() if request.method == "GET": return jsonify({ "plugin": plugin_name, "label": cls.label, "schema": cls.config_schema, "config": cfg.get("plugin_configs", {}).get(plugin_name, {}), }) # POST: update config for this plugin new_cfg = {} for field in cls.config_schema: key = field["key"] form_val = request.form.get(f"config_{key}") ftype = field.get("type", "string") if form_val is None: if ftype == "bool": new_cfg[key] = False # unchecked checkbox → False else: continue if ftype == "secret": # Nur überschreiben wenn nicht leer existing = cfg.get("plugin_configs", {}).get(plugin_name, {}).get(key, "") if form_val == "": new_cfg[key] = "" elif form_val == "__UNSET__": continue else: new_cfg[key] = form_val continue try: if ftype == "int": new_cfg[key] = int(form_val) elif ftype == "float": new_cfg[key] = float(form_val) elif ftype == "bool": new_cfg[key] = (form_val.lower() in ("1", "true", "yes", "on")) else: new_cfg[key] = form_val except (TypeError, ValueError): new_cfg[key] = form_val cfg.setdefault("plugin_configs", {})[plugin_name] = new_cfg dashboard_mod.save_config(cfg) return jsonify({"ok": True, "plugin": plugin_name, "config": new_cfg}) @app.route("/refresh", methods=["POST"]) def refresh_now(): a = require_auth() if a: return a result = dashboard_mod.send_ipc("refresh") return jsonify({"status": "ok", "ipc_response": result}) @app.route("/snapshot.png") def snapshot_png(): a = require_auth() if a: return a # Snapshot aus dem laufenden Renderer via IPC anfordern # Wir machen das nicht über IPC (Renderer schreibt Datei), sondern rendern direkt hier # für Preview-Zwecke — billiger und deterministisch. try: d = dashboard_mod.Dashboard() # nicht reload() — würde live-config nutzen, ok img = d.render_once() buf = io.BytesIO() img.save(buf, format="PNG") return Response(buf.getvalue(), mimetype="image/png") except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/plugins.json") def plugins_json(): a = require_auth() if a: return a out = [] for cls in all_widget_classes(): out.append({ "name": cls.name, "label": cls.label, "description": cls.description, "category": cls.category, "schema": cls.config_schema, "defaults": cls.default_config, }) return jsonify(out) @app.route("/api/network/recovery") def api_net_recovery(): """Entscheidung: soll die UI Recovery-Info prominent zeigen?""" a = require_auth() if a: return a state = net.get_watchdog().get_state() threshold = net.get_recovery_threshold() decision = net._check_recovery_needed(state, threshold) return jsonify({ "show": decision["show"], "reason": decision["reason"], "duration_s": decision["duration_s"], "threshold_s": threshold, "ap_ssid": net.AP_SSID, "ap_password": net.AP_PASSWORD, "ap_url": "http://10.42.0.1:8080", "current_mode": state.get("mode"), "current_ssid": state.get("ssid"), "current_ip": state.get("ip"), "error": state.get("error", ""), }) @app.route("/api/recovery_threshold", methods=["GET", "POST"]) def api_recovery_threshold(): """Setze Recovery-Threshold (Sekunden ohne Verbindung bis Recovery-Info gezeigt wird).""" a = require_auth() if a: return a cfg = dashboard_mod.load_config() if request.method == "GET": return jsonify({"threshold_s": cfg.get("recovery_threshold_s", 60)}) try: threshold = int(request.form.get("threshold_s", 60)) threshold = max(5, min(3600, threshold)) except (TypeError, ValueError): return jsonify({"ok": False, "error": "invalid threshold"}), 400 cfg["recovery_threshold_s"] = threshold dashboard_mod.save_config(cfg) return jsonify({"ok": True, "threshold_s": threshold}) @app.route("/status.json") def status_json(): a = require_auth() if a: return a cfg = dashboard_mod.load_config() return jsonify({ "config_path": str(dashboard_mod.CONFIG_PATH), "refresh_interval_s": cfg.get("refresh_interval_s"), "slots": [{"plugin": s.get("plugin"), "config_keys": list(s.get("config", {}).keys())} for s in cfg.get("slots", [])], "socket_path": dashboard_mod.SOCKET_PATH, }) # ============================================================================ # Network / WiFi Management # ============================================================================ @app.route("/api/network/status") def api_net_status(): a = require_auth() if a: return a return jsonify(net.get_watchdog().get_state()) @app.route("/api/network/scan") def api_net_scan(): a = require_auth() if a: return a return jsonify({"networks": net.list_wifi_networks()}) @app.route("/api/network/saved") def api_net_saved(): a = require_auth() if a: return a return jsonify({"saved": net.list_saved_connections()}) @app.route("/api/network/connect", methods=["POST"]) def api_net_connect(): a = require_auth() if a: return a ssid = request.form.get("ssid", "").strip() pwd = request.form.get("password", "").strip() security = request.form.get("security", "wpa-psk").strip() if not ssid: return jsonify({"ok": False, "error": "ssid is required"}), 400 ok, msg = net.save_wifi(ssid, pwd, security) if not ok: return jsonify({"ok": False, "error": msg}), 500 # Falls aktuell im AP-Modus: AP aus if net.is_ap_active(): net.stop_ap() ok, msg = net.connect_wifi(msg) # msg contains connection name return jsonify({"ok": ok, "message": msg, "ssid": ssid}) @app.route("/api/network/reconnect", methods=["POST"]) def api_net_reconnect(): a = require_auth() if a: return a saved = net.list_saved_connections() if not saved: return jsonify({"ok": False, "error": "kein gespeichertes WLAN"}), 400 ok, msg = net.connect_wifi(saved[0]["name"]) return jsonify({"ok": ok, "message": msg}) @app.route("/api/network/disconnect", methods=["POST"]) def api_net_disconnect(): a = require_auth() if a: return a ok, msg = net.disconnect_wifi() return jsonify({"ok": ok, "message": msg}) @app.route("/api/network/ap/start", methods=["POST"]) def api_net_ap_start(): a = require_auth() if a: return a ok, msg = net.start_ap() return jsonify({"ok": ok, "message": msg}) @app.route("/api/network/ap/stop", methods=["POST"]) def api_net_ap_stop(): a = require_auth() if a: return a ok, msg = net.stop_ap() return jsonify({"ok": ok, "message": msg}) @app.route("/api/network/forget", methods=["POST"]) def api_net_forget(): a = require_auth() if a: return a name = request.form.get("name", "").strip() if not name: return jsonify({"ok": False, "error": "name required"}), 400 rc, out, err = net._nm(["connection", "delete", name], timeout=10) return jsonify({"ok": rc == 0, "message": out or err}) def main(): # Wenn direkt ausgeführt: eigenen Renderer starten ist Sache des Run-Skripts. # Wir nehmen einfach Port 8080. host = os.environ.get("EPAPER_ADMIN_HOST", "0.0.0.0") port = int(os.environ.get("EPAPER_ADMIN_PORT", "8080")) app.run(host=host, port=port, debug=False, use_reloader=False) if __name__ == "__main__": main()