Michael Sauer
58cdc73afc
Welche Version die Station gerade ausliefert, war am Geraet nur per SSH zu beantworten. Jetzt sagt es die Anzeige selbst, je nach Bild an der Stelle, wo Platz ist: - Aus-Zustand: rechtsbuendig neben der Adresse (neue Rolle ROLE_DETAIL_RIGHT, nach dem Muster der Phase neben dem Modus) - Rettung: rechtsbuendig in der Kopfzeile, dort steht keine Phase im Weg - Migration: als beschriftete Zeile "liefert: <version>" direkt hinter "FW:", NICHT in der Kopfzeile - daneben bleiben nur rund 112 px, daran ist schon "FREIGEGEBEN: <ziel>" gescheitert, und zwei nackte Versionsnummern nebeneinander waeren eine Verwechslungsfalle (FW: ist die der Lampe) Quelle ist der GEWAEHLTE Bildsatz: im Migrationsmodus manifest.json (die Migrationsabzuege tragen keinen ninlight-Anhaenger), im Rettungsmodus der Anhaenger der _ota.img. Weichen die Eintraege voneinander ab, steht da "gemischt" - ein zusammengefuehrter Deploy kann Alt-Eintraege mit anderer Version enthalten, und eine davon herauszugreifen waere geraten. Fehlt die Quelle, bleibt die Angabe weg statt eine Version zu erfinden.
949 lines
49 KiB
Python
949 lines
49 KiB
Python
#!/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 eine
|
|
Bibliothek (gpiozero/spidev/Pillow), faengt main() das als DisplayUnavailable bzw.
|
|
ImportError 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 datetime import datetime, timezone
|
|
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")
|
|
# Bedienzustand des Tasterdienstes (tools/station/button/button.py). NUR GELESEN: der
|
|
# Tasterdienst fasst den SPI-Bus nicht an und dieser Dienst kein GPIO23 - zwei Prozesse
|
|
# auf denselben Leitungen enden in "lgpio.error: 'GPIO busy'". Die Datei liegt im
|
|
# RuntimeDirectory DES TASTERDIENSTES (nicht in RUN_DIR): RUN_DIR ist das
|
|
# RuntimeDirectory der Hauptunit und verschwindet beim regulaeren "off" - also
|
|
# ausgerechnet dann, wenn das Menue "Rettung starten"/"Migration starten" anbietet.
|
|
UI_FILE = Path(os.environ.get("NINLIGHT_UI_FILE", "/run/ninlight-button/ui.json"))
|
|
# Der Umschalter "Migrationsimages mit oder ohne WLAN-Zugangsdaten" (siehe
|
|
# ninlight-station, bildsatz_cmd, und server.py:read_image_set). NUR GELESEN, und
|
|
# absichtlich aus derselben Datei wie der Server - nicht aus status.json: der Zustand
|
|
# muss auch dann auf dem Panel stehen, wenn die Station AUS ist. Genau davor soll er
|
|
# schuetzen ("man soll vor jedem Einschalten sehen, was gilt"), und ausgeschaltet
|
|
# schreibt server.py nichts mehr.
|
|
IMAGESET_FILE = Path(os.environ.get("NINLIGHT_IMAGESET_FILE",
|
|
"/var/lib/ninlight-station/imageset"))
|
|
# Der Modus, den "recovery on"/"migrate on" zuletzt angefordert haben - fuer die
|
|
# Fusszeile GENAUSO dauerhaft gebraucht wie der Bildsatz-Umschalter oben, aus demselben
|
|
# Grund und nach demselben Muster (eigene Datei unter /var/lib, NICHT unter RUN_DIR).
|
|
#
|
|
# Fixrunde 1, Befund 3: RUN_DIR/mode taugt dafuer NICHT. Die Unit setzt
|
|
# RuntimeDirectoryPreserve=restart - das erhaelt /run/ninlight-station nur ueber einen
|
|
# "systemctl restart" hinweg (das, was "recovery on"/"migrate on" fuer einen bereits
|
|
# laufenden Dienst ausloesen). "ninlight-station off" aber STOPPT den Dienst
|
|
# (systemctl stop), und GENAU DANN raeumt systemd das RuntimeDirectory leer, "restart"
|
|
# deckt einen echten Stop nicht ab. Ein frueherer Docstring hier behauptete das
|
|
# Gegenteil ("die Datei ueberlebt das Abschalten") - das war schlicht falsch, und der
|
|
# Fehler war unauffaellig, weil eine fehlende Datei auf "recovery" faellt: genau der
|
|
# Modus, den viele Tests ohnehin erwarteten, ohne dass die Faelligkeit der Datei
|
|
# je geprueft wurde. Die Folge am Geraet waere die falsche Beruhigung, gegen die die
|
|
# Fusszeile ueberhaupt gebaut ist - "in Ordnung" vor einer geplanten Migration, obwohl
|
|
# der Migrationssatz leer ist, weil das Panel mangels Datei den Rettungssatz geprueft
|
|
# haette.
|
|
#
|
|
# write_mode() im CLI schreibt deshalb NEBEN der ephemeren RUN_DIR/mode (die
|
|
# ausschliesslich server.py's naechsten Start betrifft) auch hierher - dieselbe
|
|
# Zeichenkette, zwei Ablagen fuer zwei verschiedene Leser mit verschiedenen
|
|
# Haltbarkeitsanforderungen.
|
|
LAST_MODE_FILE = Path(os.environ.get("NINLIGHT_LAST_MODE_FILE",
|
|
"/var/lib/ninlight-station/mode"))
|
|
# Dieselben Namen und dieselbe Vorgabe wie in server.py und im CLI - im Zweifel der
|
|
# Satz OHNE Zugangsdaten (dort steht die Begruendung).
|
|
IMAGESET_DIRS = {"ohne-wlan": "migrate-nocreds", "mit-wlan": "migrate-wlan"}
|
|
IMAGESET_DEFAULT = "ohne-wlan"
|
|
# Derselbe Umschalter gilt jetzt auch fuer die RETTUNG (server.py: rescue_dir()) - mit
|
|
# eigenen Verzeichnissen, denn Rettungs- und Migrationsabzuege sind verschieden hergestellt
|
|
# (siehe server.py fuer die Begruendung). Welches Paar am Panel zu pruefen ist, haengt vom
|
|
# Modus ab: _read_imageset() bekommt ihn deshalb als Parameter statt fest verdrahtet.
|
|
RESCUE_IMAGESET_DIRS = {"ohne-wlan": "rescue-nocreds", "mit-wlan": "rescue-wlan"}
|
|
|
|
POLL_INTERVAL_S = 5
|
|
# Solange eine Bedienung laeuft, waere ein 5-Sekunden-Takt unbenutzbar (ein Tastendruck
|
|
# bis zu 5 s ohne Reaktion). Die Schleife tickt deshalb in diesem Takt und sieht bei
|
|
# JEDEM Tick in der Bedienzustandsdatei nach - auch im Ruhezustand, sonst waere
|
|
# ausgerechnet der erste Druck (der das Menue oeffnet) der langsamste. Das kostet eine
|
|
# kleine Dateilesung je halber Sekunde; der teure Teil des Zustands (build_state() mit
|
|
# seinem "systemctl is-active"-Fork) bleibt beim 5-Sekunden-Takt.
|
|
#
|
|
# Am Panel gemessen kommt zu dieser halben Sekunde noch der Teilaufbau (rund 0,3 s) -
|
|
# also grob 0,5-0,8 s je Druck. Das ist spuerbar, aber bedienbar: die Geste ist
|
|
# "druecken und hinsehen", nicht "tippen wie auf einer Tastatur", und die Haltegesten
|
|
# (1,5/3/10 s) liegen alle weit ueber dieser Verzoegerung.
|
|
UI_POLL_INTERVAL_S = 0.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
|
|
|
|
# Panel erst schlafen legen, wenn seit dem letzten Bildaufbau schon eine Weile nichts
|
|
# mehr passiert ist - NICHT nach jedem einzelnen Bild (Fixrunde 2, N1/N2/N3: das liess
|
|
# V3 beim naechsten Teilaufbau abstuerzen und machte V2s Deep-Sleep-Aufwachsequenz
|
|
# unbrauchbar, siehe Panel-Klasse). Migrationsereignisse kommen oft in kurzen Buendeln
|
|
# (mehrere Lampen kurz nacheinander) - das Panel dafuer jedesmal schlafen zu legen und
|
|
# wieder aufzuwecken waere nur zusaetzlicher Verschleiss ohne Nutzen fuer die
|
|
# Bias-Alterung, die I4 eigentlich vermeiden soll. 10 Minuten ohne Aenderung sind lang
|
|
# genug, dass ein Buendel zusammenhaengender Ereignisse das Panel nicht staendig
|
|
# schlafen und wieder aufwachen laesst, aber kurz genug, dass es zwischen zwei
|
|
# tatsaechlich seltenen Ereignissen (z.B. stundenlanger Rettungsmodus ohne Anfrage)
|
|
# nicht dauerhaft unter Bias steht.
|
|
SLEEP_AFTER_IDLE_S = 10 * 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, optional mit einem "export "-Praefix (setup_pi.sh prueft
|
|
beim Nachtragen fehlender Schluessel ausdruecklich auf "^(export[[:space:]]+)?KEY=" -
|
|
ein bereits vorhandenes "export SSID=..." muss also erkannt werden, nicht nur die
|
|
unpraefigierte Form). 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
|
|
if line.startswith("export ") or line.startswith("export\t"):
|
|
line = line[len("export"):].lstrip()
|
|
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_autooff_minutes(run_dir: Path, now: float | None = None):
|
|
"""Restzeit des optionalen Zeitschalters in ganzen Minuten (aufgerundet), oder None,
|
|
wenn gerade keiner laeuft.
|
|
|
|
Quelle ist RUN_DIR/autooff.json, die "ninlight-station --for/renew" schreibt und
|
|
"hold"/"off" wieder loescht (siehe dort). Ein Zeitschalter, den man am Panel nicht
|
|
sieht, waere schlimmer als keiner: man glaubt, die Station laeuft, und sie ist
|
|
laengst aus.
|
|
|
|
"deadline" ist eine Unix-Zeit (der Zeitschalter selbst haengt an systemds monotoner
|
|
Uhr, die Anzeige rechnet in Wanduhrzeit) - eine Zeitverstellung verschiebt also die
|
|
angezeigte Restzeit, nicht das tatsaechliche Abschalten. Rundet auf: solange noch
|
|
eine Sekunde uebrig ist, steht dort "1 min" und nicht "0 min"."""
|
|
data = _read_json(run_dir / "autooff.json")
|
|
deadline = data.get("deadline")
|
|
if isinstance(deadline, bool) or not isinstance(deadline, (int, float)):
|
|
return None
|
|
remaining_s = deadline - (time.time() if now is None else now)
|
|
if remaining_s <= 0:
|
|
# Der Termin ist da (oder der Lauf verschiebt gerade) - "0 min" ist hier die
|
|
# ehrliche Auskunft, nicht das Verschwinden der Zeile.
|
|
return 0
|
|
whole, rest = divmod(remaining_s, 60)
|
|
return int(whole) + (1 if rest else 0)
|
|
|
|
|
|
# Lebenszeichen: ein kleiner Punkt, der erscheint, sobald irgendeine Anfrage hereinkommt
|
|
# (auch eine erfolglose - "notfound"/"listen"/"unknown"/"noimageset" zaehlen genauso wie
|
|
# eine erfolgreiche Auslieferung), und nach ACTIVITY_WINDOW_S Sekunden von selbst wieder
|
|
# verschwindet. Der Wunsch dahinter (Rueckmeldung eines Bedieners, 2026-09-23): bisher war
|
|
# nur am Servierzaehler ("Served: N") zu sehen, ob es geklappt HAT - bis dahin war offen,
|
|
# ob ueberhaupt jemand anklopft. server.py schreibt den Zeitpunkt der letzten Anfrage
|
|
# jeder Art schon nach status.json (Station.last_activity); hier wird daraus nur noch ein
|
|
# Ja/Nein fuer "gerade eben".
|
|
#
|
|
# Bewusst KEIN eigener Zeittakt/Timer: das Panel pollt status.json ohnehin schon alle
|
|
# POLL_INTERVAL_S Sekunden, und render.blocks() haengt den Punkt an eine bestehende
|
|
# Textzeile an (siehe dort) - er faehrt also ueber genau denselben Textvergleich
|
|
# (RefreshPlanner.decide()) mit, der auch jede andere Aenderung erkennt. Ein Erscheinen
|
|
# und ein Verschwinden kosten damit je einen gewoehnlichen Teilaufbau, keinen eigenen
|
|
# Mechanismus und kein zusaetzliches Wachhalten des E-Papers - fuegt sich also in die
|
|
# vorhandene, fuer E-Paper sparsame Taktung ein, statt eine eigene, potenziell haeufigere
|
|
# aufzumachen.
|
|
ACTIVITY_WINDOW_S = 15
|
|
|
|
|
|
def _recent_request(status: dict, now: float | None = None) -> bool:
|
|
"""War die letzte Anfrage (jeder Art) hoechstens ACTIVITY_WINDOW_S Sekunden her?
|
|
|
|
"last_request_ts" fehlt bei einer aelteren server.py-Fassung oder wenn noch nie eine
|
|
Anfrage einging - dann gilt False, kein Punkt erfunden. Ein Zeitstempel aus der
|
|
Zukunft (Uhr verstellt, oder Server- und Anzeigerechner-Uhr laufen leicht auseinander)
|
|
zaehlt ebenfalls als "nicht mehr aktuell" statt als potenziell dauerhaft aktiv."""
|
|
ts = status.get("last_request_ts")
|
|
if not isinstance(ts, str):
|
|
return False
|
|
try:
|
|
parsed = datetime.strptime(ts, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
return False
|
|
now_s = time.time() if now is None else now
|
|
age_s = now_s - parsed.timestamp()
|
|
return 0 <= age_s <= ACTIVITY_WINDOW_S
|
|
|
|
|
|
def _read_mode_file(path: Path) -> str:
|
|
"""Zuletzt angeforderter Modus laut `path`. Fehlt die Datei, ist sie unlesbar
|
|
(inklusive ungueltigem UTF-8 - eine halb geschriebene/beschaedigte Datei ist nicht
|
|
nur per fehlendem Inhalt kaputt), oder ist ihr Inhalt kein bekannter Modus, gilt
|
|
"recovery" - derselbe Vorgabewert wie in server.py:resolve_mode(). Wirft nie: eine
|
|
kaputte Datei darf weder das Panel noch "off" aufhalten.
|
|
|
|
`path` ist bewusst eine Datei, kein Verzeichnis (anders als frueher, wo hier
|
|
`run_dir / "mode"` gelesen wurde) - siehe LAST_MODE_FILE oben, warum die Fusszeile
|
|
NICHT mehr aus RUN_DIR liest."""
|
|
try:
|
|
value = path.read_text().strip()
|
|
except (OSError, UnicodeDecodeError):
|
|
return "recovery"
|
|
return value if value in ("recovery", "migration") else "recovery"
|
|
|
|
|
|
# Nur beim ersten Mal protokollieren, dass systemctl selbst nicht erreichbar war (siehe
|
|
# _service_active) - das passiert bei jedem der Aufrufe alle 5 s erneut, und eine Zeile
|
|
# alle 5 s wuerde das Journal fluten, obwohl es sich nie aendert (Fixrunde 1, Kleinigkeit).
|
|
_service_active_fallback_logged = False
|
|
|
|
|
|
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"."""
|
|
global _service_active_fallback_logged
|
|
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.
|
|
if not _service_active_fallback_logged:
|
|
log.warning("systemctl nicht erreichbar - Anzeige raet den Modus anhand von "
|
|
"RUN_DIR/mode statt der echten Dienstaktivitaet")
|
|
_service_active_fallback_logged = True
|
|
return True
|
|
|
|
|
|
def _read_imageset(imageset_path: Path, conf: dict[str, str], mode: str) -> tuple[str, bool]:
|
|
"""(Name des eingestellten Bildsatzes, liegt der fuer `mode` PASSENDE Satz ueberhaupt
|
|
auf dem Pi?).
|
|
|
|
Fehlt die Datei, ist sie unlesbar oder steht etwas Unbekanntes darin, gilt
|
|
IMAGESET_DEFAULT - dieselbe Vorgabe wie in server.py und im CLI, sonst wuerde das
|
|
Panel etwas anderes behaupten als der Server tut.
|
|
|
|
`ok` ist False, wenn der GEWAEHLTE Satz fuer den GERADE GELTENDEN Modus nicht auf dem
|
|
Pi liegt - nach derselben Regel wie server.py:migrate_set_present()/
|
|
rescue_set_present(): Verzeichnis vorhanden UND mindestens ein Image darin. Welches
|
|
Verzeichnispaar das ist (migrate-*/rescue-*), entscheidet `mode` - Rettung und
|
|
Migration liegen seit der Erweiterung des Umschalters auf die Rettung in getrennten
|
|
Verzeichnissen (server.py begruendet das). Eine falsche Zuordnung hier wuerde das
|
|
Panel in GENAU DEM Modus luegen lassen, fuer den es gerade steht: "ok", obwohl die
|
|
Rettung nichts ausliefern kann, oder "FEHLT", obwohl sie es koennte.
|
|
|
|
Ohne bekanntes IMAGE_DIR (Konfigurationsdatei nicht lesbar) wird nichts behauptet:
|
|
dann gilt "ok", denn eine unbelegte Warnung ist schlechter als keine."""
|
|
try:
|
|
value = imageset_path.read_text().strip()
|
|
except (OSError, UnicodeDecodeError):
|
|
value = ""
|
|
name = value if value in IMAGESET_DIRS else IMAGESET_DEFAULT
|
|
image_dir = conf.get("IMAGE_DIR")
|
|
if not image_dir:
|
|
return name, True
|
|
dirs = RESCUE_IMAGESET_DIRS if mode == "recovery" else IMAGESET_DIRS
|
|
set_dir = Path(image_dir) / dirs[name]
|
|
try:
|
|
return name, any(p.name.endswith(".img") for p in set_dir.iterdir())
|
|
except OSError:
|
|
return name, False
|
|
|
|
|
|
def _trailer_version(path: Path) -> str:
|
|
"""Firmware-Version aus dem NINL-Anhaenger eines ninlight-Abzugs, sonst "".
|
|
|
|
Nur die letzten 512 Byte werden gelesen: die Rettungsabzuege liegen als
|
|
_ota.img vor und tragen ihre Version im Anhaenger (Geraetetyp bei Offset 5,
|
|
Version bei 21) - das Entpacken des gzip-Rumpfs waere dafuer unnoetige Arbeit auf
|
|
einem Pi, der alle paar Sekunden neu zeichnet.
|
|
"""
|
|
try:
|
|
with path.open("rb") as f:
|
|
f.seek(-512, 2)
|
|
anhaenger = f.read(512)
|
|
except OSError:
|
|
return ""
|
|
if len(anhaenger) != 512 or anhaenger[0:4] != b"NINL":
|
|
return ""
|
|
return anhaenger[21:37].split(b"\x00", 1)[0].decode("ascii", errors="replace")
|
|
|
|
|
|
def delivered_version(conf: dict, imageset: str, mode: str) -> str:
|
|
"""Die Firmware-Version, die die Station gerade ausliefern wuerde, sonst "".
|
|
|
|
Quelle ist der GEWAEHLTE Bildsatz, nicht irgendein Verzeichnis: wer am Taster
|
|
zwischen "mit WLAN" und "ohne WLAN" umschaltet, bekommt sonst eine Zahl angezeigt,
|
|
die zum anderen Satz gehoert.
|
|
|
|
Im Migrationsmodus steht die Version in `manifest.json` - die Migrationsabzuege
|
|
selbst tragen keinen ninlight-Anhaenger (siehe server.py:image_version). Weichen
|
|
die Eintraege voneinander ab, ist das Ergebnis "gemischt": ein zusammengefuehrter
|
|
Deploy kann Alt-Eintraege mit anderer Version enthalten, und eine davon
|
|
herauszugreifen waere geraten. Im Rettungsmodus liegen gewoehnliche _ota.img, deren
|
|
Anhaenger die Version tragen.
|
|
|
|
Fehlt die Quelle oder ist sie unlesbar, bleibt das Ergebnis leer - dann zeigt die
|
|
Anzeige gar nichts, statt eine Version zu erfinden.
|
|
"""
|
|
image_dir = conf.get("IMAGE_DIR")
|
|
if not image_dir:
|
|
return ""
|
|
if mode == "recovery":
|
|
verzeichnis = Path(image_dir) / RESCUE_IMAGESET_DIRS.get(imageset, "")
|
|
try:
|
|
abzuege = sorted(p for p in verzeichnis.iterdir() if p.name.endswith(".img"))
|
|
except OSError:
|
|
return ""
|
|
versionen = {v for v in (_trailer_version(p) for p in abzuege) if v}
|
|
else:
|
|
pfad = Path(image_dir) / IMAGESET_DIRS.get(imageset, "") / "manifest.json"
|
|
try:
|
|
daten = json.loads(pfad.read_text())
|
|
except (OSError, ValueError):
|
|
return ""
|
|
if not isinstance(daten, dict):
|
|
return ""
|
|
versionen = set()
|
|
eintraege = daten.get("images")
|
|
if isinstance(eintraege, list):
|
|
versionen = {
|
|
e["ninlight_version"]
|
|
for e in eintraege
|
|
if isinstance(e, dict) and isinstance(e.get("ninlight_version"), str)
|
|
and e["ninlight_version"]
|
|
}
|
|
if not versionen:
|
|
oben = daten.get("ninlight_version")
|
|
versionen = {oben} if isinstance(oben, str) and oben else set()
|
|
if not versionen:
|
|
return ""
|
|
return versionen.pop() if len(versionen) == 1 else "gemischt"
|
|
|
|
|
|
def _route_lookup() -> str:
|
|
"""Die Rohauskunft von "ip -4 route get" zur Standardroute, oder "" wenn es keine gibt.
|
|
|
|
Eigene Funktion, damit die Tests sie ersetzen koennen, ohne einen Unterprozess zu
|
|
starten - und damit local_address() nur noch das Auswerten macht."""
|
|
import subprocess
|
|
|
|
try:
|
|
fertig = subprocess.run(
|
|
# Bewusst eine dokumentierte Testadresse (RFC 5737) statt eines echten Ziels
|
|
# wie 8.8.8.8: "route get" fragt nur den Kernel, es geht kein Paket hinaus,
|
|
# aber es soll auch beim Mitlesen nichts Fremdes anklingen.
|
|
["ip", "-4", "route", "get", "192.0.2.1"],
|
|
capture_output=True, text=True, timeout=2, check=False)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return ""
|
|
return fertig.stdout if fertig.returncode == 0 else ""
|
|
|
|
|
|
def local_address() -> tuple[str, str]:
|
|
"""(Geraet, Adresse), unter der der Pi im Heimnetz erreichbar ist - ("", "") ohne Netz.
|
|
|
|
Ueber "ip route get", NICHT ueber einen verbundenen UDP-Socket: die Unit erlaubt dem
|
|
Anzeigedienst nur AF_UNIX und AF_NETLINK (ninlight-display.service:
|
|
RestrictAddressFamilies). Ein AF_INET-Socket scheitert dort mit EAFNOSUPPORT, und
|
|
genau daran ist die Adresszeile am echten Panel gescheitert, waehrend derselbe
|
|
Aufruf von Hand - ohne Sandbox - die Adresse lieferte (Rueckmeldung 2026-09-23).
|
|
"ip" spricht Netlink und laeuft damit innerhalb der Erlaubnis. Der Unterprozess
|
|
kostet einen fork alle POLL_INTERVAL_S Sekunden und faellt nur im Aus-Zustand an,
|
|
dem einzigen Statusbild mit dieser Zeile.
|
|
|
|
Das Geraet kommt mit, weil "LAN" am Panel eine Falschauskunft war: dieser Pi haengt
|
|
per WLAN im Netz, nicht am Kabel. Der Kernel nennt in derselben Auskunft beides -
|
|
"192.0.2.1 dev wlan0 src 192.168.2.96 uid 1000".
|
|
|
|
Wirft nie: ohne Route gibt es nichts auszuwerten, dann liefert der Aufruf ("", "")
|
|
und die Anzeige sagt das - eine Ausnahme aus der Adressermittlung darf das Panel
|
|
nicht aufhalten."""
|
|
worte = _route_lookup().split()
|
|
geraet = adresse = ""
|
|
for i, w in enumerate(worte):
|
|
if w == "dev" and i + 1 < len(worte) and not geraet:
|
|
geraet = worte[i + 1]
|
|
elif w == "src" and i + 1 < len(worte) and not adresse:
|
|
adresse = worte[i + 1]
|
|
# Ohne Adresse ist auch der Geraetename wertlos - er wuerde ein Netz vortaeuschen,
|
|
# das nicht steht.
|
|
return (geraet, adresse) if adresse else ("", "")
|
|
|
|
|
|
def build_state(run_dir: Path = RUN_DIR, conf_path: Path = CONF_PATH,
|
|
service_name: str = SERVICE_NAME,
|
|
imageset_path: Path = IMAGESET_FILE,
|
|
last_mode_path: Path = LAST_MODE_FILE) -> dict:
|
|
"""Baut den zusammengefuehrten Zustand, den render.lines()/compose() erwarten.
|
|
|
|
Modus kommt vorrangig aus `last_mode_path` (Vorgabe "recovery", wie
|
|
server.py:resolve_mode()) - das ist das, was "ninlight-station on"/"migrate on"
|
|
zuletzt geschrieben haben, dauerhaft (siehe LAST_MODE_FILE oben - NICHT die ephemere
|
|
RUN_DIR/mode, die server.py fuer den naechsten Start liest) und unabhaengig davon
|
|
verfuegbar, ob/wann der Serverprozess seit dem letzten Start schon einmal
|
|
status.json geschrieben hat. status.json selbst liefert nur served/recent/
|
|
last_seen; sein "mode"-Feld wird bewusst NICHT als Modusquelle verwendet
|
|
(Fixrunde 1: die urspruengliche Fassung hatte den Vorrang umgekehrt).
|
|
|
|
Phase/armed_target kommen aus control.json, nur im Migrationsmodus massgeblich - der
|
|
Rettungsmodus ignoriert control.json ebenso wie server.py das tut. "off" ueberschreibt
|
|
alles, wenn der Hauptdienst gar nicht laeuft.
|
|
|
|
Der Bildsatz ("imageset"/"imageset_ok") kommt in JEDEN Zustand, auch in "off" und in
|
|
den Rettungsmodus: er ist ein dauerhafter Umschalter, und man soll ihn VOR dem
|
|
Einschalten sehen - eine Einstellung von vorletzter Woche, die erst beim Blaettern im
|
|
Menue auffaellt, hat genau den Zweck verfehlt.
|
|
|
|
Ob "imageset_ok" den Rettungs- oder den Migrations-Bildsatz prueft, richtet sich nach
|
|
dem MODUS, in dem die Station beim naechsten Einschalten anlaufen wuerde - also
|
|
`last_mode_path`, nicht danach, ob sie gerade laeuft. Deshalb wird der Modus HIER,
|
|
vor der "ist der Dienst aktiv"-Abfrage, gelesen: er ist auch im Zustand "off"
|
|
verfuegbar (die Datei ist dauerhaft, siehe LAST_MODE_FILE) und genau dort soll die
|
|
Vorschau ja gelten - "off" zeigt ja noch keinen Modus im engeren Sinn, aber der
|
|
naechste Start faehrt in genau diesen Modus, und die Fusszeile soll VORHER zeigen,
|
|
ob dessen Bildsatz bereitliegt.
|
|
|
|
"activity" (nur in Recovery/Migration, siehe _recent_request()) sagt, ob die letzte
|
|
Anfrage jeder Art hoechstens ACTIVITY_WINDOW_S Sekunden her ist - das Lebenszeichen,
|
|
das render.blocks() an die Servierzeile anhaengt."""
|
|
conf = _read_conf(conf_path)
|
|
mode = _read_mode_file(last_mode_path)
|
|
imageset, imageset_ok = _read_imageset(imageset_path, conf, mode)
|
|
|
|
if not _service_active(service_name):
|
|
# Die IP-Zeile gibt es NUR hier: ist die Station aus, ist der Pi ein gewoehnlicher
|
|
# Rechner im Heimnetz, und die einzige Auskunft, die man dann am Panel braucht,
|
|
# ist die Adresse, unter der man per SSH drankommt. Im Betrieb waere sie sinnlos
|
|
# (wlan0 traegt dann 192.168.8.1, das steht ohnehin auf jedem Statusbild) und
|
|
# wuerde den knappen Platz fuer Betriebsauskuenfte wegnehmen.
|
|
geraet, adresse = local_address()
|
|
# Auch im Aus-Zustand: welche Version die Station ausliefern WUERDE, ist genau
|
|
# dann interessant, wenn man sie noch nicht eingeschaltet hat.
|
|
return {"mode": "off", "imageset": imageset, "imageset_ok": imageset_ok,
|
|
"ip": adresse, "iface": geraet,
|
|
"image_version": delivered_version(conf, imageset, "migration")}
|
|
|
|
status = _read_json(run_dir / "status.json")
|
|
state: dict = {"mode": mode, "served": status.get("served"), "recent": status.get("recent"),
|
|
"imageset": imageset, "imageset_ok": imageset_ok,
|
|
"activity": _recent_request(status),
|
|
"image_version": delivered_version(conf, imageset, mode)}
|
|
|
|
if mode == "recovery":
|
|
state["ssid"] = conf.get("SSID")
|
|
else: # migration
|
|
state["ssid"] = conf.get("MIGRATE_SSID")
|
|
state["last_seen"] = status.get("last_seen")
|
|
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")
|
|
|
|
# Der Zeitschalter ist optional und gilt in beiden Modi gleichermassen - ohne ihn
|
|
# bleibt das Feld weg und render.blocks() zeichnet keine Zeile dafuer.
|
|
autooff_minutes = _read_autooff_minutes(run_dir)
|
|
if autooff_minutes is not None:
|
|
state["autooff_minutes"] = autooff_minutes
|
|
|
|
return state
|
|
|
|
|
|
def read_ui(ui_path: Path = UI_FILE) -> dict:
|
|
"""Der Bedienzustand des Tasterdienstes, oder {} wenn gerade keine Bedienung laeuft
|
|
(Datei fehlt, ist unlesbar, halb geschrieben, aus einer aelteren Fassung oder meldet
|
|
ausdruecklich "idle").
|
|
|
|
Jedes Feld wird defensiv gelesen: die Datei kommt aus einem FREMDEN Prozess, der
|
|
abgestuerzt, aelter oder neuer sein kann - ein Fehler hier darf die Anzeige nicht
|
|
aufhalten und schon gar nicht eine Migration stoeren."""
|
|
data = _read_json(ui_path)
|
|
view = data.get("view")
|
|
if not data.get("active") or not isinstance(view, str) or view == "idle":
|
|
return {}
|
|
return data
|
|
|
|
|
|
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
|
|
|
|
|
|
class Panel:
|
|
"""Kapselt die V2/V3-Unterschiede der Waveshare-Treiber hinter einer einheitlichen
|
|
Schnittstelle (Fixrunde 1, C1): epd2in13_V2.init() braucht ein Pflichtargument
|
|
(FULL_UPDATE/PART_UPDATE), epd2in13_V3.init() braucht keins.
|
|
|
|
Verfolgt zusaetzlich selbst, ob das Panel gerade "wach" ist (Register initialisiert,
|
|
Referenzpuffer fuer Teilaufbauten bekannt) oder zuletzt sleep() gerufen wurde
|
|
(Fixrunde 2, N1/N3): epd.sleep() schickt das Panel in einen Zustand, aus dem heraus
|
|
KEINE Revision direkt einen Teilaufbau vertraegt - V2s sleep() ist Deep Sleep Mode 2
|
|
(RAM/Register verworfen; ein nachfolgendes init(PART_UPDATE) setzt nur einen Teil der
|
|
Register, die ein Teilaufbau braucht - 0x01/0x11/0x44/0x45/0x4E/0x4F setzt
|
|
ausschliesslich der FULL_UPDATE-Zweig), V3s sleep() ruft epdconfig.module_exit() und
|
|
schliesst damit die GPIO-Pins, die epd2in13_V3.displayPartial() als Erstes anfasst
|
|
(roher AttributeError auf ein geschlossenes Pin-Objekt). Ein Teilaufbau nach dem
|
|
Schlafen wird deshalb hier, nicht beim Aufrufer, transparent zu einem Vollaufbau
|
|
aufgewertet - unabhaengig davon, in welcher Reihenfolge draw_full()/draw_partial()/
|
|
sleep() aufgerufen werden.
|
|
|
|
draw_full() (und clear(), Fixrunde 3, F2) verwendet fuer BEIDE Revisionen
|
|
displayPartBaseImage() statt display()/Clear() (Fixrunde 2, N2): das schreibt beide
|
|
RAM-Puffer des Controllers (0x24 UND 0x26) und legt damit erst die Referenz an, gegen
|
|
die ein nachfolgender Teilaufbau diffen kann - display() allein haette nur 0x24
|
|
beschrieben (und Clear() ebenso - bei V2 ist das 0x26-Schreiben in Clear() sogar
|
|
auskommentiert) und den ersten Teilaufbau danach gegen einen veralteten/undefinierten
|
|
Puffer laufen lassen. self._awake wird deshalb nur dort auf True gesetzt, wo
|
|
tatsaechlich ein gueltiger Referenzpuffer entstanden ist - vorher (Fixrunde 2) setzte
|
|
clear() es, obwohl es nur epd.Clear() (kein 0x26) rief, und die im Klassen-Docstring
|
|
behauptete Invariante stimmte nach clear() nicht."""
|
|
|
|
def __init__(self, revision: str) -> None:
|
|
self.revision = revision
|
|
self._epd_module = _load_epd_module(revision)
|
|
self.epd = self._epd_module.EPD()
|
|
# Vor dem ersten clear()/draw_full() ist nichts initialisiert - siehe Klassen-
|
|
# Docstring, warum draw_partial() das genauso wie ein sleep() davor behandeln
|
|
# muss (beides: kein gueltiger Referenzpuffer, keine vollstaendige Registerlage).
|
|
self._awake = False
|
|
|
|
def _full_init(self) -> None:
|
|
if self.revision == "V2":
|
|
self.epd.init(self.epd.FULL_UPDATE)
|
|
else:
|
|
self.epd.init()
|
|
|
|
def _blank_buffer(self) -> list[int]:
|
|
"""Ein reiner Weiss-Puffer in der Groesse, die epd.getbuffer() fuer ein
|
|
vollformatiges Bild liefern wuerde - dieselbe Formel wie in epd.Clear() selbst
|
|
(siehe epd2in13_V2.py/epd2in13_V3.py), nur ohne ueber Pillow ein echtes weisses
|
|
Bild zu erzeugen."""
|
|
width, height = self.epd.width, self.epd.height
|
|
linewidth = width // 8 if width % 8 == 0 else width // 8 + 1
|
|
return [0xFF] * (linewidth * height)
|
|
|
|
def clear(self) -> None:
|
|
self._full_init()
|
|
# displayPartBaseImage() statt epd.Clear() (Fixrunde 3, F2): dieselbe sichtbare
|
|
# Wirkung (Panel wird weiss), aber schreibt zusaetzlich den 0x26-Referenzpuffer,
|
|
# den ein nachfolgender draw_partial() braucht - epd.Clear() allein wuerde
|
|
# self._awake=True vortaeuschen, ohne die im Klassen-Docstring beschriebene
|
|
# Invariante wirklich herzustellen.
|
|
self.epd.displayPartBaseImage(self._blank_buffer())
|
|
self._awake = True
|
|
|
|
def draw_full(self, image) -> None:
|
|
buf = self.epd.getbuffer(image)
|
|
self._full_init()
|
|
self.epd.displayPartBaseImage(buf)
|
|
self._awake = True
|
|
|
|
def draw_partial(self, image) -> bool:
|
|
"""Zeichnet teilweise - oder, falls das Panel nicht wach ist, vollstaendig.
|
|
Gibt zurueck, ob intern zu einem Vollaufbau aufgewertet wurde (Fixrunde 3, F4):
|
|
der Aufrufer muss das wie einen regulaeren Vollaufbau behandeln (z.B. fuer die
|
|
30-Minuten-Uhr in run_daemon()), sonst kommt der naechste dortige Vollaufbau zu
|
|
frueh und altert das Panel zusaetzlich, ohne dass ein Teilaufbau ueberhaupt
|
|
stattgefunden haette."""
|
|
if not self._awake:
|
|
# Siehe Klassen-Docstring (Fixrunde 2, N1/N3): ein Teilaufbau direkt nach dem
|
|
# Schlafen (oder vor dem allerersten Vollaufbau) ist auf keiner Revision eine
|
|
# gueltige Sequenz - stattdessen einen vollwertigen Vollaufbau ausfuehren.
|
|
self.draw_full(image)
|
|
return True
|
|
buf = self.epd.getbuffer(image)
|
|
if self.revision == "V2":
|
|
self.epd.init(self.epd.PART_UPDATE)
|
|
self.epd.displayPartial(buf)
|
|
return False
|
|
|
|
def sleep(self) -> None:
|
|
# Idempotent: ein zweiter sleep()-Aufruf, waehrend das Panel schon schlaeft,
|
|
# wuerde sonst versuchen, ueber bereits von module_exit() geschlossene GPIO-Pins
|
|
# SPI-Befehle zu schicken (derselbe AttributeError-Mechanismus wie N1, nur
|
|
# selbst herbeigefuehrt statt durch einen Teilaufbau).
|
|
if not self._awake:
|
|
return
|
|
self.epd.sleep()
|
|
self._awake = False
|
|
|
|
|
|
class RefreshPlanner:
|
|
"""Die gesamte Zeitsteuerung der Anzeige - wann voll aufgebaut, wann teilweise, wann
|
|
geschlafen wird - als reine Zustandsmaschine ohne Schleife, ohne time.sleep() und ohne
|
|
Hardware.
|
|
|
|
Vorher steckte diese Logik direkt in run_daemon()s Endlosschleife und war dadurch
|
|
ungetestet (Fixrunde 4): drei Korrekturen aus Fixrunde 3 (die stale-Bedingung, die
|
|
Rueckmeldung einer intern aufgewerteten Teilaktualisierung, die Zeilenumbruchbehandlung
|
|
in compose()) liessen sich per Mutation aendern, ohne dass ein einziger Test rot wurde.
|
|
Hier laesst sich stattdessen ein kompletter Tagesverlauf in Millisekunden durchspielen
|
|
(siehe test_render.py, _run_schedule()).
|
|
|
|
Benutzung: pro Runde einmal decide() fragen, die zurueckgegebene Aktion ausfuehren und
|
|
das Ergebnis per record_draw()/record_sleep() zurueckmelden. decide() selbst aendert
|
|
nichts - der Aufrufer meldet, was tatsaechlich passiert ist (insbesondere, ob
|
|
Panel.draw_partial() sich selbst zu einem Vollaufbau aufgewertet hat).
|
|
|
|
Aktionen:
|
|
"full" - kompletter Bildaufbau (Moduswechsel oder faelliger Ghosting-Ausgleich)
|
|
"partial" - Teilaktualisierung (Inhalt hat sich geaendert)
|
|
"sleep" - Panel schlafen legen (laenger nichts passiert)
|
|
"idle" - nichts tun
|
|
"""
|
|
|
|
_NO_MODE = object() # garantiert ungleich jedem echten Modus beim ersten Lauf
|
|
|
|
def __init__(self, now: float,
|
|
full_refresh_interval_s: float | None = None,
|
|
sleep_after_idle_s: float | None = None) -> None:
|
|
# Bewusst None statt der Modulkonstanten als Vorgabewert (Fixrunde 5): ein
|
|
# Default-Argument wird zur Importzeit ausgewertet, ein spaeteres
|
|
# monkeypatch.setattr(display, "FULL_REFRESH_INTERVAL_S", ...) haette dann keine
|
|
# Wirkung mehr auf neu gebaute Planer - die alte Schleife las die Konstanten pro
|
|
# Runde. Die Properties unten lesen sie weiterhin bei jeder Entscheidung.
|
|
self._full_refresh_interval_s = full_refresh_interval_s
|
|
self._sleep_after_idle_s = sleep_after_idle_s
|
|
self.last_lines: list[str] | None = None
|
|
self.last_mode: object = self._NO_MODE
|
|
self.last_full = now
|
|
self.last_draw = now
|
|
# Zaehlt Teilaufbauten seit dem letzten Vollaufbau. full_refresh_interval_s
|
|
# existiert nur, um Geisterbilder aus AUFEINANDERFOLGENDEN Teilaufbauten
|
|
# auszugleichen - ohne mindestens einen Teilaufbau seit dem letzten Vollaufbau gibt
|
|
# es nichts auszugleichen, und ein Vollaufbau des UNVERAENDERTEN Bildes waere
|
|
# reine, nutzlose Alterung (Fixrunde 3, F3): ohne diese Absicherung entstand im
|
|
# eingeschwungenen Ruhezustand ein endloser 10-Minuten-schlafen/10-Minuten-wach-
|
|
# Zyklus (33% Bias-Anteil) rein aus der 30-Minuten-Uhr, obwohl sich der Inhalt nie
|
|
# aenderte.
|
|
#
|
|
# Der Zaehler wird beim Einschlafen BEWUSST NICHT zurueckgesetzt (Fixrunde 4, N1):
|
|
# das tat die Fassung aus Fixrunde 3 und liess den faelligen Ghosting-Ausgleich
|
|
# damit dauerhaft ausfallen - nach einem Buendel Teilaufbauten (z.B. 30 Lampen in
|
|
# 2,5 Minuten) schlief das Panel nach 10 Minuten mit dem entstandenen Geisterbild
|
|
# ein und zeigte genau dieses Bild stundenlang weiter, weil stale wegen
|
|
# partials_since_full == 0 nie mehr feuern konnte. So bleibt der Ausgleich
|
|
# ausstehend und wird zum 30-Minuten-Termin nachgeholt (das weckt das Panel einmal
|
|
# kurz auf; danach ist der Zaehler 0 und der eingeschwungene Ruhezustand bleibt
|
|
# wie in F3 gewuenscht bei 3% Wachanteil). Die Alternative - vor dem Einschlafen
|
|
# einmal draw_full() ziehen - kostet denselben einen Vollaufbau, aber JEDES Mal,
|
|
# wenn ein Buendel endet, auch wenn kurz darauf ohnehin neue Inhalte kommen (die
|
|
# das Panel nach dem Schlafen sowieso ueber draw_partial()s Aufwertung voll
|
|
# aufbauen). Diese Variante altert das Panel im Betriebsfall also mehr, nicht
|
|
# weniger, und der Ausgleich hier kommt spaetestens zum ohnehin vorgesehenen
|
|
# Termin.
|
|
self.partials_since_full = 0
|
|
self.asleep = False
|
|
|
|
@property
|
|
def full_refresh_interval_s(self) -> float:
|
|
return (FULL_REFRESH_INTERVAL_S if self._full_refresh_interval_s is None
|
|
else self._full_refresh_interval_s)
|
|
|
|
@property
|
|
def sleep_after_idle_s(self) -> float:
|
|
return (SLEEP_AFTER_IDLE_S if self._sleep_after_idle_s is None
|
|
else self._sleep_after_idle_s)
|
|
|
|
def decide(self, now: float, lines: list[str], mode: object) -> str:
|
|
"""Was jetzt zu tun ist. Aendert nichts - das Ergebnis wird per record_*()
|
|
zurueckgemeldet."""
|
|
mode_changed = mode != self.last_mode
|
|
stale = (self.partials_since_full > 0
|
|
and (now - self.last_full) >= self.full_refresh_interval_s)
|
|
|
|
if lines != self.last_lines or mode_changed or stale:
|
|
return "full" if (mode_changed or stale) else "partial"
|
|
if not self.asleep and (now - self.last_draw) >= self.sleep_after_idle_s:
|
|
# Nach laengerer Ruhe schlafen legen (Fixrunde 1, I4: die Pixel bleiben -
|
|
# e-Paper ist bistabil - aber ohne dauerhafte Spannung altert das Panel
|
|
# deutlich weniger).
|
|
return "sleep"
|
|
return "idle"
|
|
|
|
def record_draw(self, now: float, lines: list[str], mode: object,
|
|
was_full: bool) -> None:
|
|
"""`was_full` ist True fuer einen regulaeren Vollaufbau UND dann, wenn
|
|
Panel.draw_partial() sich selbst zu einem Vollaufbau aufgewertet hat (Fixrunde 3,
|
|
F4) - sonst kaeme der naechste "stale"-Vollaufbau zu frueh und altert das Panel
|
|
zusaetzlich, obwohl gerade erst vollstaendig aufgebaut wurde."""
|
|
self.last_lines = lines
|
|
self.last_mode = mode
|
|
self.last_draw = now
|
|
self.asleep = False
|
|
if was_full:
|
|
self.last_full = now
|
|
self.partials_since_full = 0
|
|
else:
|
|
self.partials_since_full += 1
|
|
|
|
def record_sleep(self) -> None:
|
|
"""Kein `now`-Parameter (Fixrunde 5): das Einschlafen verschiebt weder die
|
|
30-Minuten-Uhr noch last_draw - es merkt sich nur, dass ein weiteres
|
|
panel.sleep() nichts mehr zu tun haette."""
|
|
self.asleep = True
|
|
|
|
|
|
def _test_image(revision: str):
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
|
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)
|
|
return image
|
|
|
|
|
|
def run_test(revision: str) -> None:
|
|
"""Zeichnet ein Testbild und beendet sich - siehe Modul-Docstring, wie man damit V2
|
|
von V3 unterscheidet."""
|
|
panel = Panel(revision)
|
|
panel.clear()
|
|
panel.draw_full(_test_image(revision))
|
|
panel.sleep()
|
|
log.info("Testbild gezeichnet (EPD_REVISION=%s)", revision)
|
|
|
|
|
|
def run_final(revision: str, kind: str) -> None:
|
|
"""Zeichnet den Abschlussbildschirm ("poweroff"/"reboot") und legt das Panel schlafen.
|
|
|
|
Das ist das Bild, das nach dem Herunterfahren DAUERHAFT stehen bleibt - E-Paper ist
|
|
bistabil, es haelt sein Bild auch ohne Strom. Bewusst ein VOLLAUFBAU: ein Teilaufbau
|
|
liesse Reste des vorigen Bildes stehen, und dieses eine Bild ueberdauert alle anderen.
|
|
|
|
Aufgerufen wird das ausschliesslich ueber "ninlight-station display final <kind>" aus
|
|
einer systemd-Unit heraus, und zwar ERST, nachdem der Anzeigedienst gestoppt ist - er
|
|
haelt sonst die GPIO-/SPI-Leitungen ("lgpio.error: 'GPIO busy'"). Diese Uebergabe
|
|
macht das CLI (display_final()), nicht diese Funktion."""
|
|
panel = Panel(revision)
|
|
panel.clear()
|
|
panel.draw_full(render.compose({"mode": "final", "final": kind}))
|
|
panel.sleep()
|
|
log.info("Abschlussbild gezeichnet (%s, EPD_REVISION=%s)", kind, revision)
|
|
|
|
|
|
def run_clear(revision: str) -> None:
|
|
panel = Panel(revision)
|
|
panel.clear()
|
|
panel.sleep()
|
|
log.info("Panel geleert (EPD_REVISION=%s)", revision)
|
|
|
|
|
|
def _safe_read_state(read_state, fallback: dict) -> dict:
|
|
"""Ruft `read_state()` auf und faengt JEDE Ausnahme ab - nicht nur die, die die
|
|
einzelnen build_state()-Bausteine schon selbst abfangen (siehe _read_json() etc.
|
|
oben). run_daemon() laeuft als Endlosschleife, und main() fängt darum nur
|
|
(DisplayUnavailable, ImportError) ab; ein einzelner, unvorhergesehener Fehler beim
|
|
Erheben wuerde diese Schleife sonst beenden - und weil E-Paper sein zuletzt
|
|
gezeichnetes Bild dauerhaft behaelt, saehe das Panel einfach so aus, als sei nichts
|
|
passiert, nur eben fuer immer eingefroren. `fallback` (der vorherige Zustand, oder
|
|
beim allerersten Aufruf ein neutraler Platzhalter) gilt dann fuer diesen einen Takt;
|
|
beim naechsten wird es erneut versucht."""
|
|
try:
|
|
return read_state()
|
|
except Exception as e: # noqa: BLE001 - bewusst pauschal, siehe Docstring
|
|
log.warning("Zustand konnte nicht gelesen werden, vorheriger Stand bleibt: %s", e)
|
|
return fallback
|
|
|
|
|
|
def run_daemon(revision: str, clock=time, panel: "Panel | None" = None,
|
|
read_state=build_state, read_ui=read_ui) -> None:
|
|
"""Endlosschleife: alle UI_POLL_INTERVAL_S Sekunden den Bedienzustand und alle
|
|
POLL_INTERVAL_S Sekunden den Stationszustand einlesen und die von RefreshPlanner
|
|
entschiedene Aktion ausfuehren. Hier steht bewusst KEINE Zeitlogik zur Panelalterung
|
|
mehr (Fixrunde 4) - die steckt vollstaendig in RefreshPlanner und ist dort ohne
|
|
Schleife, ohne time.sleep() und ohne Hardware pruefbar.
|
|
|
|
Die zwei Takte: der teure Teil (build_state() forkt fuer "systemctl is-active")
|
|
bleibt beim 5-Sekunden-Takt, der billige (eine kleine Datei lesen) laeuft in
|
|
UI_POLL_INTERVAL_S, damit eine Bedienung am Taster ueberhaupt benutzbar ist. Ein
|
|
laufender Bedienzustand ERSETZT das Statusbild (render.blocks() zweigt dafuer ganz
|
|
vorne ab); der Modus bleibt derselbe, sodass das Bild ueber Teilaufbauten kommt und
|
|
ein Tastendruck dem Panel keinen Vollaufbau kostet.
|
|
|
|
`clock` (braucht monotonic() und sleep()), `panel`, `read_state` und `read_ui` sind
|
|
einschleusbar, damit die VERDRAHTUNG hier - welche Aktion welche Panel-Methode ruft,
|
|
dass die Selbst-Aufwertung von draw_partial() zurueckgemeldet wird, dass das
|
|
Einschlafen zurueckgemeldet wird, und die Reihenfolge decide -> ausfuehren ->
|
|
zurueckmelden - selbst pruefbar ist und nicht nur eine Kopie davon (Fixrunde 5, N2:
|
|
alle drei Mutationen an genau diesen Zeilen blieben vorher gruen, weil der Test eine
|
|
nachgebaute Schleife statt dieser hier trieb). Im Betrieb bleiben die Vorgaben: echte
|
|
Uhr, echtes Panel, echtes build_state()/read_ui()."""
|
|
if panel is None:
|
|
panel = Panel(revision)
|
|
panel.clear()
|
|
planner = RefreshPlanner(clock.monotonic())
|
|
base_state = _safe_read_state(read_state, {"mode": None})
|
|
last_state_poll = clock.monotonic()
|
|
|
|
while True:
|
|
now = clock.monotonic()
|
|
if now - last_state_poll >= POLL_INTERVAL_S:
|
|
# Jeder einzelne Baustein von build_state() faengt seine eigenen I/O-Fehler
|
|
# schon ab (siehe _read_json/_read_conf/_service_active/local_address oben) - aber
|
|
# main() faengt um run_daemon() herum nur (DisplayUnavailable, ImportError)
|
|
# ab, nicht Exception im Allgemeinen. Ein einzelner unerwarteter Fehler beim
|
|
# Erheben (z.B. ein kaputtes RUN_DIR mitten in einem Neustart) wuerde diese
|
|
# Endlosschleife sonst ganz beenden - und E-Paper behaelt sein zuletzt
|
|
# gezeichnetes Bild fuer immer (siehe Modul-Docstring): die Anzeige wuerde
|
|
# dann OHNE jede Fehlermeldung am Geraet einfach stehen bleiben, ununterscheidbar
|
|
# von einem echten, andauernden Zustand. read_state() bleibt deshalb hier
|
|
# abgesichert - ein einzelner Fehlschlag ueberspringt genau diese Erhebung
|
|
# (der vorherige Zustand bleibt stehen) statt den ganzen Dienst zu beenden;
|
|
# beim naechsten Takt in POLL_INTERVAL_S wird es erneut versucht.
|
|
base_state = _safe_read_state(read_state, base_state)
|
|
last_state_poll = now
|
|
ui = read_ui()
|
|
if ui:
|
|
state = dict(base_state)
|
|
state["ui"] = ui
|
|
else:
|
|
state = base_state
|
|
current_lines = render.lines(state)
|
|
mode = state.get("mode")
|
|
|
|
action = planner.decide(now, current_lines, mode)
|
|
if action == "full":
|
|
panel.draw_full(render.compose(state))
|
|
planner.record_draw(now, current_lines, mode, was_full=True)
|
|
elif action == "partial":
|
|
# Panel.draw_partial() wertet sich selbst zu einem Vollaufbau auf, wenn das
|
|
# Panel nicht wach ist (siehe Panel-Klasse) - das meldet es hier zurueck.
|
|
upgraded = panel.draw_partial(render.compose(state))
|
|
planner.record_draw(now, current_lines, mode, was_full=upgraded)
|
|
elif action == "sleep":
|
|
panel.sleep()
|
|
planner.record_sleep()
|
|
|
|
clock.sleep(UI_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")
|
|
group.add_argument("--final", choices=sorted(render.FINAL_LABELS),
|
|
help="Abschlussbild zeichnen und beenden (beim Herunterfahren "
|
|
"bzw. Neustart; die Texte unterscheiden sich)")
|
|
args = parser.parse_args()
|
|
|
|
revision = os.environ.get("EPD_REVISION", "V3")
|
|
|
|
try:
|
|
if args.test:
|
|
run_test(revision)
|
|
elif args.clear:
|
|
run_clear(revision)
|
|
elif args.final:
|
|
run_final(revision, args.final)
|
|
else:
|
|
run_daemon(revision)
|
|
except (DisplayUnavailable, ImportError) as e:
|
|
# Genau eine Zeile, dann sys.exit(0) - siehe Modul-Docstring. Das deckt sowohl
|
|
# Hardwareprobleme (DisplayUnavailable: kein Panel/SPI) als auch eine fehlende
|
|
# Bibliothek ab - allen voran Pillow, das laut Spec auf dem Ziel-Pi absichtlich
|
|
# erst per "apt install python3-pil" nachkommt (Fixrunde 1, C4): render.compose()
|
|
# importiert Pillow erst hier, ein fehlendes Paket zeigt sich also erst als
|
|
# ModuleNotFoundError (eine ImportError-Unterklasse) beim ersten Zeichnen, nicht
|
|
# schon beim Start. Restart=on-failure (statt always) in der Unit greift hier
|
|
# nicht, weil 0 kein Fehlschlag ist; ein spaeter angestecktes Panel bzw.
|
|
# nachinstalliertes Paket 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()
|