feat(station): E-Paper-Anzeige fuer Modus, Lampe und Auslieferungen
render.lines() zerlegt den zusammengefuehrten Zustand (status.json + control.json + SSID aus der Konfigurationsdatei) in Textzeilen, ohne Pillow und ohne Hardware zu brauchen - render.compose() importiert Pillow erst innerhalb der Funktion. epdconfig.py ist eine eigene GPIO/SPI-Schicht auf gpiozero/spidev, weil RPi.GPIO auf Debian 13 mit Python 3.13 nicht mehr traegt; epd2in13_V2.py/epd2in13_V3.py sind unveraendert aus dem Waveshare-Repository uebernommen (siehe HERKUNFT.md), nur der epdconfig-Import zeigt auf unser Modul. display.py laeuft als eigener systemd-Dienst (Wants=, Restart=on-failure), fehlendes Panel/SPI/Bibliothek beendet ihn sauber mit sys.exit(0) statt einer Neustartschleife - eine Migration darf davon nie abhaengen. Verdrahtung in setup_pi.sh/deploy.sh/ninlight-station folgt in einem eigenen Durchgang.
This commit is contained in:
Michael Sauer
@@ -0,0 +1,286 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Daemon fuer die E-Paper-Anzeige der ninlight-Station: alle 5 s status.json,
|
||||
control.json und die Konfigurationsdatei lesen, daraus einen Text zeichnen und - nur bei
|
||||
Aenderung - aufs Panel bringen.
|
||||
|
||||
Panel: Waveshare 2,13" (250x122 px), SPI. Ob das verbaute Panel Revision V2 oder V3 ist,
|
||||
laesst sich am Geraet selbst nicht auslesen - die Revision kommt deshalb aus der
|
||||
Umgebungsvariable EPD_REVISION (Vorgabe "V3", von setup_pi.sh in
|
||||
/etc/ninlight-station.conf geschrieben und von der systemd-Unit als Environment=
|
||||
weitergereicht).
|
||||
|
||||
So bestimmst du die richtige Revision empirisch: erst
|
||||
EPD_REVISION=V3 python3 display.py --test
|
||||
laufen lassen und das Panel ansehen. Zeigt es ein sauberes, seitenrichtiges Testbild ohne
|
||||
Geisterbilder/Streifen, ist V3 richtig. Zeigt es nichts, ein invertiertes/verschobenes Bild
|
||||
oder haengt beim SWRESET (ReadBusy() kehrt nie zurueck), mit
|
||||
EPD_REVISION=V2 python3 display.py --test
|
||||
wiederholen. Zwischen den beiden Versuchen kurz die Stromversorgung des Panels trennen
|
||||
(--clear reicht nicht immer, falsche Revision kann das Panel in einem Zwischenzustand
|
||||
lassen).
|
||||
|
||||
Die Anzeige ist nirgends Voraussetzung fuer die Migration: fehlt das Panel, SPI oder die
|
||||
gpiozero/spidev-Bibliothek, faengt main() das als DisplayUnavailable ab, protokolliert
|
||||
GENAU EINE Zeile und beendet sich mit sys.exit(0) - nicht mit einem Fehler, sonst wuerde
|
||||
systemd (Restart=on-failure) den Dienst endlos neu starten und das Journal fluten. Ein
|
||||
Display-Fehler darf niemals eine Migration verhindern.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
# Flacher Import wie bei server.py/products.py: dieses Verzeichnis ist kein Package,
|
||||
# display.py kann aber von ueberall aus gestartet werden (die systemd-Unit ruft es mit
|
||||
# vollem Pfad auf).
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
|
||||
import render # noqa: E402
|
||||
from epdconfig import DisplayUnavailable # noqa: E402
|
||||
|
||||
log = logging.getLogger("ninlight-display")
|
||||
|
||||
# Dieselben Umgebungsvariablen wie tools/station/ninlight-station verwendet, damit Tests
|
||||
# und Handbetrieb ohne echtes /run bzw. /etc auskommen.
|
||||
RUN_DIR = Path(os.environ.get("NINLIGHT_RUN_DIR", "/run/ninlight-station"))
|
||||
CONF_PATH = Path(os.environ.get("NINLIGHT_STATION_CONF", "/etc/ninlight-station.conf"))
|
||||
SERVICE_NAME = os.environ.get("NINLIGHT_STATION_SERVICE", "ninlight-station")
|
||||
|
||||
POLL_INTERVAL_S = 5
|
||||
# E-Paper altert mit jedem Bildaufbau - ein Vollaufbau nur bei Moduswechsel und
|
||||
# mindestens alle 30 Minuten, sonst Teilaktualisierung (siehe Modul-Docstring der
|
||||
# Panel-Treiber: haeufige Teilaktualisierungen ohne gelegentlichen Vollaufbau lassen
|
||||
# Geisterbilder stehen).
|
||||
FULL_REFRESH_INTERVAL_S = 30 * 60
|
||||
|
||||
|
||||
def _read_json(path: Path) -> dict:
|
||||
"""Geparster Inhalt von `path`, oder {} wenn er fehlt/unlesbar/kaputt/kein Objekt
|
||||
ist - status.json und control.json koennen jederzeit in genau diesem Zustand sein
|
||||
(frisch aufgesetzter Pi, Schreibfehler mitten im Schreiben)."""
|
||||
try:
|
||||
data = json.loads(path.read_text())
|
||||
except (OSError, ValueError):
|
||||
return {}
|
||||
return data if isinstance(data, dict) else {}
|
||||
|
||||
|
||||
def _read_conf(path: Path) -> dict[str, str]:
|
||||
"""Minimaler Parser fuer /etc/ninlight-station.conf: KEY='VALUE'-Zeilen, wie
|
||||
setup_pi.sh sie schreibt. Kein echtes Shell-Sourcing (das CLI braucht das fuer $-
|
||||
und `-Escaping in Passwoertern, siehe dessen Kommentar - display.py liest hier nur
|
||||
SSID/MIGRATE_SSID, reine Bezeichner ohne Sonderzeichen)."""
|
||||
values: dict[str, str] = {}
|
||||
try:
|
||||
text = path.read_text()
|
||||
except OSError:
|
||||
return values
|
||||
for raw_line in text.splitlines():
|
||||
line = raw_line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
key, _, value = line.partition("=")
|
||||
key = key.strip()
|
||||
value = value.strip()
|
||||
if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
|
||||
value = value[1:-1]
|
||||
if key:
|
||||
values[key] = value
|
||||
return values
|
||||
|
||||
|
||||
def _read_mode_file(run_dir: Path) -> str:
|
||||
"""Modus, in dem der Dienst laut RUN_DIR/mode tatsaechlich gestartet wurde. Fehlt die
|
||||
Datei oder ist ihr Inhalt kein bekannter Modus, gilt "recovery" - derselbe Vorgabewert
|
||||
wie in server.py:resolve_mode()."""
|
||||
try:
|
||||
value = (run_dir / "mode").read_text().strip()
|
||||
except OSError:
|
||||
return "recovery"
|
||||
return value if value in ("recovery", "migration") else "recovery"
|
||||
|
||||
|
||||
def _service_active(service_name: str) -> bool:
|
||||
"""Laeuft der Hauptdienst (server.py) gerade? "ninlight-station off" stoppt ihn und
|
||||
faehrt den Zugangspunkt herunter, laesst status.json dabei aber unangetastet liegen -
|
||||
ein reines Lesen von status.json wuerde nach "off" also faelschlich weiter den letzten
|
||||
Modus vor dem Abschalten zeigen. Deshalb ist die Dienstaktivitaet selbst (nicht das
|
||||
Vorhandensein von status.json) die Quelle fuer den Anzeigemodus "off"."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["systemctl", "is-active", "--quiet", service_name],
|
||||
timeout=5, check=False)
|
||||
return result.returncode == 0
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
# systemctl selbst nicht erreichbar (z.B. Handlauf ohne systemd) - dann anhand
|
||||
# der Statusdatei entscheiden statt "off" zu behaupten, ohne es zu wissen.
|
||||
return True
|
||||
|
||||
|
||||
def build_state(run_dir: Path = RUN_DIR, conf_path: Path = CONF_PATH,
|
||||
service_name: str = SERVICE_NAME) -> dict:
|
||||
"""Baut den zusammengefuehrten Zustand, den render.lines()/compose() erwarten -
|
||||
status.json (served/recent/last_seen), control.json (phase/armed_target, nur im
|
||||
Migrationsmodus massgeblich), die SSID aus der Konfigurationsdatei und "off", wenn
|
||||
der Hauptdienst gar nicht laeuft."""
|
||||
if not _service_active(service_name):
|
||||
return {"mode": "off"}
|
||||
|
||||
status = _read_json(run_dir / "status.json")
|
||||
mode = status.get("mode")
|
||||
if mode not in ("recovery", "migration"):
|
||||
# status.json fehlt/ist kaputt/stammt (noch) nicht vom laufenden Prozess -
|
||||
# RUN_DIR/mode ist das, was "ninlight-station on"/"migrate on" zuletzt
|
||||
# geschrieben hat und ist unabhaengig von status.json verfuegbar.
|
||||
mode = _read_mode_file(run_dir)
|
||||
|
||||
conf = _read_conf(conf_path)
|
||||
state: dict = {"mode": mode, "served": status.get("served"), "recent": status.get("recent")}
|
||||
|
||||
if mode == "recovery":
|
||||
state["ssid"] = conf.get("SSID")
|
||||
else: # migration
|
||||
state["ssid"] = conf.get("MIGRATE_SSID")
|
||||
state["last_seen"] = status.get("last_seen")
|
||||
# Die Phase kommt bewusst aus control.json, nicht aus status.json: server.py
|
||||
# schreibt status.json nur beim Start und nach einer Auslieferung, waehrend
|
||||
# "ninlight-station arm"/"listen" nur die Steuerdatei anfasst (Spec/CLI
|
||||
# status_cmd folgt demselben Muster). Und bewusst nur HIER gelesen - im
|
||||
# Rettungsmodus ignoriert server.py control.json ebenfalls, ein
|
||||
# liegengebliebener Inhalt aus einer frueheren Migration darf dort nicht
|
||||
# faelschlich "Phase: FREIGEGEBEN" zeigen.
|
||||
control = _read_json(run_dir / "control.json")
|
||||
phase = control.get("phase")
|
||||
state["phase"] = phase if phase in ("auto", "listen", "armed") else "auto"
|
||||
state["armed_target"] = control.get("armed_target")
|
||||
|
||||
return state
|
||||
|
||||
|
||||
def _load_epd_module(revision: str):
|
||||
"""Laedt den passenden Panel-Treiber. Nicht ueber ein Dict aus Modulnamen, damit ein
|
||||
statischer Import-Checker (und ein Mensch, der `grep -n epd2in13` tippt) beide
|
||||
moeglichen Importe findet."""
|
||||
if revision == "V2":
|
||||
import epd2in13_V2 as epd_module
|
||||
elif revision == "V3":
|
||||
import epd2in13_V3 as epd_module
|
||||
else:
|
||||
raise DisplayUnavailable(
|
||||
f"unbekannte EPD_REVISION '{revision}' - erwartet 'V2' oder 'V3'")
|
||||
return epd_module
|
||||
|
||||
|
||||
def _draw(epd, image, full: bool) -> None:
|
||||
buf = epd.getbuffer(image)
|
||||
if full:
|
||||
# epd.init() laedt die volle LUT neu - noetig, weil displayPartial() unten immer
|
||||
# die Teil-LUT setzt und die naechste display()-Auslieferung sonst mit der
|
||||
# falschen Wellenform liefe.
|
||||
epd.init()
|
||||
epd.display(buf)
|
||||
else:
|
||||
epd.displayPartial(buf)
|
||||
|
||||
|
||||
def run_test(revision: str) -> None:
|
||||
"""Zeichnet ein Testbild und beendet sich - siehe Modul-Docstring, wie man damit V2
|
||||
von V3 unterscheidet."""
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
|
||||
epd_module = _load_epd_module(revision)
|
||||
epd = epd_module.EPD()
|
||||
epd.init()
|
||||
epd.Clear(0xFF)
|
||||
|
||||
width, height = 250, 122
|
||||
image = Image.new("1", (width, height), 255)
|
||||
draw = ImageDraw.Draw(image)
|
||||
draw.rectangle((0, 0, width - 1, height - 1), outline=0)
|
||||
draw.text((4, 4), "ninlight-display Test", font=ImageFont.load_default(), fill=0)
|
||||
draw.text((4, 20), f"EPD_REVISION={revision}", font=ImageFont.load_default(), fill=0)
|
||||
draw.line((4, 40, width - 4, 40), fill=0)
|
||||
# Vier Kaestchen, abwechselnd gefuellt/leer - hilft, Kontrast/Inversion am
|
||||
# angeschlossenen Panel auf einen Blick zu beurteilen.
|
||||
for i in range(4):
|
||||
x0 = 4 + i * 20
|
||||
draw.rectangle((x0, 50, x0 + 16, 90), outline=0, fill=0 if i % 2 == 0 else 255)
|
||||
|
||||
epd.display(epd.getbuffer(image))
|
||||
log.info("Testbild gezeichnet (EPD_REVISION=%s)", revision)
|
||||
|
||||
|
||||
def run_clear(revision: str) -> None:
|
||||
epd_module = _load_epd_module(revision)
|
||||
epd = epd_module.EPD()
|
||||
epd.init()
|
||||
epd.Clear(0xFF)
|
||||
epd.sleep()
|
||||
log.info("Panel geleert (EPD_REVISION=%s)", revision)
|
||||
|
||||
|
||||
def run_daemon(revision: str) -> None:
|
||||
epd_module = _load_epd_module(revision)
|
||||
epd = epd_module.EPD()
|
||||
epd.init()
|
||||
epd.Clear(0xFF)
|
||||
|
||||
last_lines: list[str] | None = None
|
||||
last_mode: object = object() # garantiert ungleich jedem echten Modus beim ersten Lauf
|
||||
last_full = time.monotonic()
|
||||
|
||||
while True:
|
||||
state = build_state()
|
||||
current_lines = render.lines(state)
|
||||
mode = state.get("mode")
|
||||
now = time.monotonic()
|
||||
need_full = (mode != last_mode) or (now - last_full >= FULL_REFRESH_INTERVAL_S)
|
||||
|
||||
if current_lines != last_lines or need_full:
|
||||
image = render.compose(state)
|
||||
_draw(epd, image, full=need_full)
|
||||
if need_full:
|
||||
last_full = now
|
||||
last_lines = current_lines
|
||||
last_mode = mode
|
||||
|
||||
time.sleep(POLL_INTERVAL_S)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
||||
|
||||
parser = argparse.ArgumentParser(description="ninlight E-Paper-Anzeige")
|
||||
group = parser.add_mutually_exclusive_group()
|
||||
group.add_argument("--test", action="store_true", help="Testbild zeichnen und beenden")
|
||||
group.add_argument("--clear", action="store_true", help="Panel leeren und beenden")
|
||||
args = parser.parse_args()
|
||||
|
||||
revision = os.environ.get("EPD_REVISION", "V3")
|
||||
|
||||
try:
|
||||
if args.test:
|
||||
run_test(revision)
|
||||
elif args.clear:
|
||||
run_clear(revision)
|
||||
else:
|
||||
run_daemon(revision)
|
||||
except DisplayUnavailable as e:
|
||||
# Genau eine Zeile, dann sys.exit(0) - siehe Modul-Docstring. Restart=on-failure
|
||||
# (statt always) in der Unit greift hier nicht, weil 0 kein Fehlschlag ist; ein
|
||||
# spaeter angestecktes Panel braucht also einen manuellen/zeitgesteuerten
|
||||
# Neustart der Unit, nicht eine Neustartschleife.
|
||||
log.info("E-Paper-Anzeige nicht verfuegbar, beende ohne Fehler: %s", e)
|
||||
sys.exit(0)
|
||||
except KeyboardInterrupt:
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user