Vibecode/epaper-dashboard
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2073039f43048f0e3effe1dd7e05782cc3ee11cc
epaper-dashboard/dashboard.py
T
Patrik 5de086e1b1 FEAT-DEEPSLEEP: Display-Refresh-Timer + CPU-Throttle + Admin Power-Tab 
DeepSleep-Modus fuer das ePaper-Display:
- dashboard.py: _is_deepsleep(), _next_wake(), _seconds_until_wake(),
                _apply_cpu_throttle() (Pi auf 600MHz im Sleep)
- 60s sleep-chunks in main loop statt festes Intervall
- render_full() mit deepsleep_active + next_wake params
- ZZZ-Indikator + CPU-Throttle-Badge wenn schlafend
- load_config() auto-filled 'power' defaults aus POWER_DEFAULTS

Admin UI (templates/power.html + admin.py):
- /power (GET+POST): DeepSleep-Toggle, Wake-Times, CPU-Throttle-Settings
- /api/power (JSON): live status polling fuer Sidebar-Badge
- /api/wake (POST): manuelles Display-Aufwecken aus dem Admin
- templates/index.html: Energie-Sidebar-Link + Live-Status-Badge
  (Zz / wach / an / aus) via refreshPowerStatus()
2026-08-31 00:47:46 +02:00

588 lines
22 KiB
Python
Raw Blame History

"""Layout, Slot-Geometrie, Refresh-Loop mit DeepSleep-Support.
Lädt config.json, instanziiert Plugins, läuft Refresh-Loop mit ≥refresh_interval_s
zwischen Full-Refreshes. Reagiert auf Live-Trigger via Unix-Socket.
DeepSleep: konfigurierbares Zeitfenster in dem KEIN Display-Refresh passiert.
Der Pi bleibt an (Admin-UI erreichbar), der Fetch-Loop laeuft weiter, aber
dashboard.display() wird uebersprungen. ePaper haelt das letzte Bild ohne Strom.
"""
from __future__ import annotations
import os, sys, time, json, math, threading, signal, socket, logging
from datetime import datetime, timedelta
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, POWER_DEFAULTS
from plugins.base import all_widget_classes, Widget, draw_deepsleep_indicator
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
# Power-Defaults (DeepSleep) sicherstellen
if "power" not in cfg or not isinstance(cfg["power"], dict):
cfg["power"] = dict(POWER_DEFAULTS)
else:
for k, v in POWER_DEFAULTS.items():
cfg["power"].setdefault(k, v)
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": {},
"power": dict(POWER_DEFAULTS),
}
# ============================================================================
# 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,
deepsleep_active: bool = False, next_wake: str = "") -> 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)
# Grid-Trennlinien NUR im freien Hintergrund (zwischen leeren Zellen).
# Niemals durch Widgets durchgehen — Widgets sind klar abgegrenzt.
GRID_LINE = (215, 215, 215)
item_cells = set()
for item in items:
for dx in range(item.w):
for dy in range(item.h):
item_cells.add((item.x + dx, item.y + dy))
# Vertikale Linien: an jeder Spaltengrenze nur dort zeichnen, wo
# die LINKE und RECHTE Zelle beide LEER sind (kein Item).
for c in range(1, 4):
for r in range(4):
if (c - 1, r) not in item_cells and (c, r) not in item_cells:
y0 = r * 120
draw.line((c * 200, y0, c * 200, y0 + 120), fill=GRID_LINE, width=1)
# Horizontale Linien: analog.
for r in range(1, 4):
for c in range(4):
if (c, r - 1) not in item_cells and (c, r) not in item_cells:
x0 = c * 200
draw.line((x0, r * 120, x0 + 200, r * 120), fill=GRID_LINE, width=1)
for item, widget in zip(items, widgets):
px, py, pw, ph = item.pixels()
# Cell border
draw.rectangle((px, py, px + pw - 1, py + ph - 1), outline=FG, width=2)
if widget is None:
draw.text((px + 8, py + 8), f"#{item.id[:6]} (no plugin)",
font=fonts.get("16", fonts.get("default")), fill=FG)
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
# Fonts an Widget durchreichen VOR render()
if hasattr(widget, "set_fonts"):
widget.set_fonts(fonts)
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)
# DeepSleep-Indikator im oberen rechten Bereich
if deepsleep_active:
ind_w = min(280, DISPLAY_W // 3)
draw_deepsleep_indicator(draw, DISPLAY_W - ind_w - 6, 4, ind_w, 18, next_wake)
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()
# ------------------------------------------------------------------
# DeepSleep: Zeitschaltung fuer Display-Refresh
# ------------------------------------------------------------------
def _parse_hhmm(self, s: str) -> int:
"""'HH:MM' → Minuten seit Mitternacht. Fallback: 0."""
try:
parts = s.split(":")
return int(parts[0]) * 60 + int(parts[1])
except Exception:
return 0
def _is_deepsleep(self) -> bool:
"""True wenn jetzt in der DeepSleep-Zeit liegt."""
cfg = self.config.get("power", {})
if not cfg.get("deepsleep_enabled", False):
return False
now = datetime.now()
now_min = now.hour * 60 + now.minute
start = self._parse_hhmm(cfg.get("deepsleep_start", "23:00"))
end = self._parse_hhmm(cfg.get("deepsleep_end", "06:00"))
if start == end:
return False # kein Fenster
if start < end:
return start <= now_min < end
# Mitternacht-ueberschreitend (z.B. 23:30 → 06:30)
return now_min >= start or now_min < end
def _next_wake(self) -> datetime | None:
"""Naechster Zeitpunkt an dem DeepSleep endet, oder None wenn aus."""
if not self._is_deepsleep():
return None
cfg = self.config.get("power", {})
end = self._parse_hhmm(cfg.get("deepsleep_end", "06:00"))
now = datetime.now()
end_dt = now.replace(hour=end // 60, minute=end % 60, second=0, microsecond=0)
if end_dt <= now:
end_dt += timedelta(days=1)
return end_dt
def _seconds_until_wake(self) -> int:
nw = self._next_wake()
if nw is None:
return 0
return max(5, int((nw - datetime.now()).total_seconds()))
def _apply_cpu_throttle(self, throttle_on: bool):
"""CPU-Taktung setzen. Nur auf Linux/Pi wirksam; sonst noop."""
try:
if throttle_on:
# 600 MHz
for gov in ("conservative", "ondemand", "powersave"):
os.system(f"echo {gov} > /sys/devices/system/cpu/cpu0/cpufreq/scaling_governor 2>/dev/null")
os.system("echo 600000 > /sys/devices/system/cpu/cpu0/cpufreq/scaling_max_freq 2>/dev/null")
else:
# Default: ondemand, max 1.5 GHz
for gov in ("ondemand", "performance"):
os.system(f"echo {gov} > /sys/devices/system/cpu/cpu0/cpufreq/scaling_governor 2>/dev/null")
os.system("echo 1500000 > /sys/devices/system/cpu/cpu0/cpufreq/scaling_max_freq 2>/dev/null")
except Exception:
pass
def render_once(self, with_deepsleep: bool = True) -> Image.Image:
# fetch all widgets (auch im DeepSleep — damit Daten frisch sind wenn wir aufwachen)
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)
# DeepSleep-Status + Indikator
sleep_active = with_deepsleep and self._is_deepsleep()
next_wake = ""
if sleep_active:
nw = self._next_wake()
if nw:
next_wake = nw.strftime("%H:%M")
img = render_full(items, widgets, self.fonts,
deepsleep_active=sleep_active, next_wake=next_wake)
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()
_throttle_applied = False
while True:
try:
self.maybe_reload_config()
interval = int(self.config.get("refresh_interval_s", 180))
# ----- DeepSleep Check -----
if self._is_deepsleep():
secs = self._seconds_until_wake()
nw = self._next_wake()
nw_str = nw.strftime("%H:%M") if nw else "?"
logging.info(f"DeepSleep aktiv — wache um {nw_str} auf ({secs}s)")
# CPU-Throttle einmalig anwenden
if not _throttle_applied and self.config.get("power", {}).get("cpu_throttle", True):
self._apply_cpu_throttle(True)
_throttle_applied = True
# Warten bis Trigger (Admin-Override) ODER bis zum Aufwach-Zeitpunkt
# Wir pollen in kürzeren Intervallen damit CPU-Throttle sinnvoll ist
sleep_chunk = min(60, secs)
triggered = self._trigger.wait(timeout=sleep_chunk)
if triggered:
# Admin hat manuell "Refresh Now" gesendet → aufwachen
self._trigger.clear()
logging.info("DeepSleep durch Admin-Trigger beendet")
if _throttle_applied:
self._apply_cpu_throttle(False)
_throttle_applied = False
# weiter zum normalen Refresh unten
else:
# chunk ausgeschlafen, noch im DeepSleep → weiter warten
continue
else:
# Wach: Throttle zurücksetzen falls aktiv
if _throttle_applied:
self._apply_cpu_throttle(False)
_throttle_applied = False
# normaler Refresh
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()
Reference in New Issue View Git Blame Copy Permalink