ki
49de61bb26
Admin UI: Linear.app dark theme — near-black canvas, indigo-violet accent, Inter font Display: Minimal, Minimal Color, Bold, Analog Clock, Magazine, Cards, Kiosk, Classic
447 lines
15 KiB
Python
447 lines
15 KiB
Python
"""Layout, Slot-Geometrie, Refresh-Loop.
|
|
|
|
Lädt config.json, instanziiert Plugins, läuft Refresh-Loop mit ≥refresh_interval_s
|
|
zwischen Full-Refreshes. Reagiert auf Live-Trigger via Unix-Socket.
|
|
"""
|
|
from __future__ import annotations
|
|
import os, sys, time, json, math, threading, signal, socket, logging
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(HERE))
|
|
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
from palette import BG, FG, OK
|
|
from plugins.base import all_widget_classes, Widget
|
|
from layout import Item, GRID_COLS, GRID_ROWS, CELL_W, CELL_H, DISPLAY_W as LAYOUT_DISPLAY_W, pack, find_overlaps, find_out_of_bounds
|
|
import renderer
|
|
|
|
import importlib, pkgutil
|
|
import plugins as _plugins_pkg # ensure package is importable
|
|
|
|
|
|
# ============================================================================
|
|
# Config helpers
|
|
# ============================================================================
|
|
CONFIG_PATH = HERE / "config.json"
|
|
LOG_PATH = HERE / "dashboard.log"
|
|
SOCKET_PATH = "/tmp/epaper-dashboard.sock"
|
|
|
|
DEFAULT_CONFIG = {
|
|
"version": 1,
|
|
"refresh_interval_s": 180,
|
|
"slots": [
|
|
{"plugin": "clock", "config": {}},
|
|
{"plugin": "weather", "config": {}},
|
|
{"plugin": "system", "config": {}},
|
|
{"plugin": "hello", "config": {}},
|
|
],
|
|
}
|
|
|
|
# Display dimensions are imported from layout module (single source of truth)
|
|
DISPLAY_W = LAYOUT_DISPLAY_W
|
|
DISPLAY_H = GRID_ROWS * CELL_H
|
|
# Legacy-v1 GRID bleibt für Migration
|
|
LEGACY_SLOTS = 4
|
|
|
|
|
|
def migrate_v1_to_v2(v1: dict) -> dict:
|
|
"""Konvertiert altes Layout (slots[]) in v2 (layout.items + plugin_configs)."""
|
|
layout_items = []
|
|
plugin_configs = {}
|
|
# Positioniere quadratisch im 4x4 grid, fall-back auf legacy 2x2
|
|
# Hier nutzen wir das alte 2x2 mapping und packen dann:
|
|
# slot 0 -> (0,0) 2x2
|
|
# slot 1 -> (2,0) 2x2
|
|
# slot 2 -> (0,2) 2x2
|
|
# slot 3 -> (2,2) 2x2
|
|
legacy_pos = [(0, 0), (2, 0), (0, 2), (2, 2)]
|
|
for i, slot in enumerate(v1.get("slots", [])):
|
|
if i >= LEGACY_SLOTS:
|
|
break
|
|
plugin = slot.get("plugin", "")
|
|
cfg = slot.get("config", {})
|
|
if not plugin:
|
|
continue
|
|
x, y = legacy_pos[i]
|
|
item_id = f"migrated_{i}"
|
|
layout_items.append({
|
|
"id": item_id,
|
|
"plugin": plugin,
|
|
"x": x, "y": y, "w": 2, "h": 2,
|
|
})
|
|
# Plugin-Config landet in plugin_configs (ohne slot-id, geteilt pro plugin-name)
|
|
if cfg and plugin not in plugin_configs:
|
|
plugin_configs[plugin] = cfg
|
|
return {
|
|
"version": 2,
|
|
"refresh_interval_s": v1.get("refresh_interval_s", 180),
|
|
"layout": {
|
|
"grid": {"cols": 4, "rows": 4},
|
|
"items": layout_items,
|
|
},
|
|
"plugin_configs": plugin_configs,
|
|
}
|
|
|
|
|
|
def load_config() -> dict:
|
|
if not CONFIG_PATH.exists():
|
|
return _empty_v2_config()
|
|
try:
|
|
cfg = json.loads(CONFIG_PATH.read_text())
|
|
except Exception as e:
|
|
logging.error(f"config.json invalid: {e}, using defaults")
|
|
return _empty_v2_config()
|
|
if cfg.get("version", 1) < 2:
|
|
logging.info("config: migrating v1 → v2")
|
|
cfg = migrate_v1_to_v2(cfg)
|
|
save_config(cfg)
|
|
# Ensure required keys exist
|
|
if "layout" not in cfg or "items" not in cfg.get("layout", {}):
|
|
cfg["layout"] = {"grid": {"cols": 4, "rows": 4}, "items": []}
|
|
if "plugin_configs" not in cfg:
|
|
cfg["plugin_configs"] = {}
|
|
if "refresh_interval_s" not in cfg:
|
|
cfg["refresh_interval_s"] = 180
|
|
return cfg
|
|
|
|
|
|
def save_config(cfg: dict) -> None:
|
|
CONFIG_PATH.write_text(json.dumps(cfg, indent=2))
|
|
|
|
|
|
def _empty_v2_config() -> dict:
|
|
return {
|
|
"version": 2,
|
|
"refresh_interval_s": 180,
|
|
"layout": {
|
|
"grid": {"cols": 4, "rows": 4},
|
|
"items": [],
|
|
},
|
|
"plugin_configs": {},
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# Plugin Registry
|
|
# ============================================================================
|
|
_widget_cache: dict[str, type[Widget]] = {}
|
|
|
|
|
|
def get_widget_classes() -> dict[str, type[Widget]]:
|
|
"""Lazy + cached: lädt alle Plugin-Klassen einmal."""
|
|
if not _widget_cache:
|
|
for cls in all_widget_classes():
|
|
_widget_cache[cls.name] = cls
|
|
return _widget_cache
|
|
|
|
|
|
def make_widget(plugin_name: str, config: dict) -> Widget | None:
|
|
classes = get_widget_classes()
|
|
cls = classes.get(plugin_name)
|
|
if cls is None:
|
|
logging.warning(f"plugin not found: {plugin_name}")
|
|
return None
|
|
try:
|
|
w = cls(config)
|
|
w.on_load()
|
|
return w
|
|
except Exception as e:
|
|
logging.error(f"failed to instantiate plugin {plugin_name}: {e}")
|
|
return None
|
|
|
|
|
|
# ============================================================================
|
|
# Font helpers
|
|
# ============================================================================
|
|
FONT_DIR = HERE / "fnt"
|
|
|
|
|
|
def load_fonts() -> dict:
|
|
if not FONT_DIR.exists():
|
|
FONT_DIR.mkdir(parents=True, exist_ok=True)
|
|
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)
|
|
|
|
for item, widget in zip(items, widgets):
|
|
px, py, pw, ph = item.pixels()
|
|
if widget is None:
|
|
draw.text((px + 8, py + 8), f"#{item.id[:6]} (no plugin)",
|
|
font=fonts.get("16", fonts.get("default")), fill=FG)
|
|
continue
|
|
if item.y >= 4: # overflow marker
|
|
draw.text((px + 8, py + 8), f"{widget.label} (overflow)",
|
|
font=fonts.get("16", fonts.get("default")), fill=FG)
|
|
continue
|
|
try:
|
|
widget.render(draw, fonts, px, py, pw, ph)
|
|
except Exception as e:
|
|
logging.error(f"plugin {widget.name} render failed: {e}")
|
|
draw.text((px + 8, py + 8),
|
|
f"{widget.label}: render error",
|
|
font=fonts.get("16", fonts.get("default")), fill=FG)
|
|
|
|
return img
|
|
|
|
|
|
# ============================================================================
|
|
# Refresh-Loop
|
|
# ============================================================================
|
|
class Dashboard:
|
|
def __init__(self):
|
|
self.config = load_config()
|
|
self.fonts = load_fonts()
|
|
self.items: list[Item] = []
|
|
self.widgets: list[Widget | None] = []
|
|
self.last_render: Image.Image | None = None
|
|
self.config_mtime = CONFIG_PATH.stat().st_mtime if CONFIG_PATH.exists() else 0
|
|
self.last_refresh_ts: float = 0.0
|
|
self._trigger = threading.Event()
|
|
self._trigger.set() # trigger initial refresh
|
|
self._epd = None
|
|
self._lock = threading.Lock()
|
|
self.reload()
|
|
|
|
def reload(self):
|
|
"""Lade config und baue alle Widgets."""
|
|
cfg = self.config
|
|
classes = get_widget_classes()
|
|
plugin_configs = cfg.get("plugin_configs", {})
|
|
items = [Item.from_dict(d) for d in cfg.get("layout", {}).get("items", [])]
|
|
widgets = []
|
|
for item in items:
|
|
plugin_cfg = plugin_configs.get(item.plugin, {})
|
|
widgets.append(make_widget(item.plugin, plugin_cfg))
|
|
with self._lock:
|
|
self.items = items
|
|
self.widgets = widgets
|
|
self.config_mtime = CONFIG_PATH.stat().st_mtime if CONFIG_PATH.exists() else 0
|
|
self.display_theme = cfg.get("display_theme", "default")
|
|
self.display_design = cfg.get("display_design", "classic")
|
|
|
|
def maybe_reload_config(self):
|
|
if not CONFIG_PATH.exists():
|
|
return
|
|
mtime = CONFIG_PATH.stat().st_mtime
|
|
if mtime != self.config_mtime:
|
|
logging.info("config.json changed, reloading")
|
|
try:
|
|
self.config = load_config()
|
|
self.reload()
|
|
except Exception as e:
|
|
logging.error(f"reload failed: {e}")
|
|
|
|
def trigger_refresh(self):
|
|
self._trigger.set()
|
|
|
|
def render_once(self) -> Image.Image:
|
|
# fetch all widgets
|
|
for w in self.widgets:
|
|
if w is None: continue
|
|
try:
|
|
w.fetch()
|
|
except Exception as e:
|
|
logging.error(f"plugin {w.name} fetch failed: {e}")
|
|
with self._lock:
|
|
items = list(self.items)
|
|
widgets = list(self.widgets)
|
|
design = getattr(self, 'display_design', 'classic')
|
|
img = renderer.render_design(items, widgets, self.fonts, design=design)
|
|
return img
|
|
|
|
def display(self, img: Image.Image):
|
|
# Lazy-import waveshare so der Renderer auch ohne Display testbar ist
|
|
if self._epd is None:
|
|
sys.path.insert(0, str(HERE / "waveshare_epd"))
|
|
try:
|
|
from waveshare_epd import epd7in3f
|
|
self._epd = epd7in3f.EPD()
|
|
self._epd.init()
|
|
self._epd.Clear()
|
|
except Exception as e:
|
|
logging.error(f"display init failed: {e}")
|
|
self._epd = "ERROR"
|
|
return False
|
|
if self._epd == "ERROR":
|
|
return False
|
|
try:
|
|
self._epd.display(self._epd.getbuffer(img))
|
|
return True
|
|
except Exception as e:
|
|
logging.error(f"display error: {e}")
|
|
return False
|
|
|
|
def run(self):
|
|
logging.info("dashboard starting; refresh_interval=%ss", self.config.get("refresh_interval_s"))
|
|
# start IPC socket listener
|
|
threading.Thread(target=self._socket_loop, daemon=True).start()
|
|
|
|
while True:
|
|
try:
|
|
self.maybe_reload_config()
|
|
interval = int(self.config.get("refresh_interval_s", 180))
|
|
# sleep until interval elapsed OR trigger set
|
|
wait = max(5, interval)
|
|
logging.info(f"waiting up to {wait}s (manual trigger interrupts)")
|
|
if self._trigger.wait(timeout=wait):
|
|
self._trigger.clear()
|
|
logging.info("manual trigger received")
|
|
else:
|
|
logging.info("interval elapsed, refreshing")
|
|
|
|
t0 = time.time()
|
|
img = self.render_once()
|
|
with self._lock:
|
|
self.last_render = img
|
|
ok = self.display(img)
|
|
self.last_refresh_ts = time.time()
|
|
logging.info(f"refresh done in {time.time()-t0:.1f}s, display={ok}")
|
|
|
|
# Re-Reload config after render in case user changed it during refresh
|
|
self.maybe_reload_config()
|
|
|
|
except KeyboardInterrupt:
|
|
break
|
|
except Exception as e:
|
|
logging.error(f"loop error: {e}", exc_info=True)
|
|
time.sleep(5)
|
|
|
|
def _socket_loop(self):
|
|
"""Empfängt 'refresh\n' über Unix-Socket."""
|
|
if os.path.exists(SOCKET_PATH):
|
|
os.remove(SOCKET_PATH)
|
|
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
srv.bind(SOCKET_PATH)
|
|
os.chmod(SOCKET_PATH, 0o660)
|
|
srv.listen(5)
|
|
logging.info(f"ipc socket at {SOCKET_PATH}")
|
|
while True:
|
|
try:
|
|
conn, _ = srv.accept()
|
|
data = conn.recv(1024).decode("utf-8", errors="ignore").strip()
|
|
if data == "refresh":
|
|
self.trigger_refresh()
|
|
conn.sendall(b"OK\n")
|
|
elif data == "ping":
|
|
conn.sendall(b"PONG\n")
|
|
elif data.startswith("snapshot"):
|
|
# Schnappschuss-Pfad als Antwort
|
|
with self._lock:
|
|
if self.last_render:
|
|
tmp = "/tmp/epaper-snapshot.png"
|
|
self.last_render.save(tmp)
|
|
conn.sendall(f"OK {tmp}\n".encode())
|
|
else:
|
|
conn.sendall(b"NO_RENDER_YET\n")
|
|
else:
|
|
conn.sendall(b"UNKNOWN_CMD\n")
|
|
conn.close()
|
|
except Exception as e:
|
|
logging.debug(f"socket error: {e}")
|
|
time.sleep(1)
|
|
|
|
|
|
def send_ipc(cmd: str, host: str = "localhost", timeout: float = 2.0) -> str:
|
|
"""Helper für die Admin-UI: schickt ein Kommando an den laufenden Renderer."""
|
|
if cmd == "snapshot":
|
|
# Snapshot holen ist eine HTTP-Aufgabe, kein IPC
|
|
return ""
|
|
try:
|
|
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
s.settimeout(timeout)
|
|
s.connect(SOCKET_PATH)
|
|
s.sendall((cmd + "\n").encode())
|
|
data = b""
|
|
while True:
|
|
chunk = s.recv(4096)
|
|
if not chunk:
|
|
break
|
|
data += chunk
|
|
s.close()
|
|
return data.decode(errors="ignore").strip()
|
|
except Exception as e:
|
|
return f"IPC_ERROR: {e}"
|
|
|
|
|
|
def setup_logging():
|
|
handlers = [logging.StreamHandler()]
|
|
try:
|
|
from logging.handlers import RotatingFileHandler
|
|
handlers.append(RotatingFileHandler(LOG_PATH, maxBytes=512*1024, backupCount=1))
|
|
except Exception:
|
|
pass
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(message)s",
|
|
handlers=handlers,
|
|
)
|
|
|
|
|
|
def main():
|
|
setup_logging()
|
|
d = Dashboard()
|
|
# signal handlers for clean shutdown
|
|
def stop(*_):
|
|
raise KeyboardInterrupt
|
|
signal.signal(signal.SIGTERM, stop)
|
|
signal.signal(signal.SIGINT, stop)
|
|
|
|
# Network-Watchdog starten (überwacht WLAN, startet Recovery-AP)
|
|
try:
|
|
import network_watchdog as _netwd
|
|
wd = _netwd.get_watchdog()
|
|
logging.info("network watchdog started")
|
|
except Exception as e:
|
|
logging.error(f"failed to start network watchdog: {e}")
|
|
|
|
d.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|