feat(station): Panel zeigt Rettungs-Bildsatz, Lebenszeichen und klarere LAN-Zeile

display.py/_read_imageset() prueft jetzt das zum Modus passende Verzeichnispaar
(rescue-* im Rettungsmodus, migrate-* in der Migration) - vorher pruefte die
Fusszeile immer nur den Migrationssatz, auch im Rettungsmodus.

render.py haengt einen kleinen Punkt an die Servierzeile, solange
display.py:_recent_request() eine Anfrage der letzten ACTIVITY_WINDOW_S
Sekunden meldet - faehrt ueber den vorhandenen Textvergleich mit, kein
zusaetzlicher Zeichentakt fuer das E-Paper.

Aus-Zustand: "no network" hiess fuer einen Bediener wie "der Migrations-/
Rettungs-Zugangspunkt ist aus" - gemeint ist das Gegenteil, das Heimnetz.
Jetzt "LAN: <ip>" bzw. "LAN: not connected".

run_daemon() faengt jetzt jeden Fehler beim Erheben des Zustands ab
(_safe_read_state) statt die Schleife zu beenden - sonst friert das Panel
nach einem einzigen unerwarteten Lesefehler fuer immer auf dem letzten Bild
ein, ohne dass das am Geraet zu erkennen waere.
This commit is contained in:
Michael Sauer
2026-09-23 21:22:42 +02:00
parent ac8dfb1077
commit 6d0f79a6ac
3 changed files with 303 additions and 23 deletions
+109 -15
View File
@@ -34,6 +34,7 @@ import os
import subprocess import subprocess
import sys import sys
import time import time
from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
# Flacher Import wie bei server.py/products.py: dieses Verzeichnis ist kein Package, # Flacher Import wie bei server.py/products.py: dieses Verzeichnis ist kein Package,
@@ -70,6 +71,11 @@ IMAGESET_FILE = Path(os.environ.get("NINLIGHT_IMAGESET_FILE",
# Satz OHNE Zugangsdaten (dort steht die Begruendung). # Satz OHNE Zugangsdaten (dort steht die Begruendung).
IMAGESET_DIRS = {"ohne-wlan": "migrate-nocreds", "mit-wlan": "migrate-wlan"} IMAGESET_DIRS = {"ohne-wlan": "migrate-nocreds", "mit-wlan": "migrate-wlan"}
IMAGESET_DEFAULT = "ohne-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 POLL_INTERVAL_S = 5
# Solange eine Bedienung laeuft, waere ein 5-Sekunden-Takt unbenutzbar (ein Tastendruck # Solange eine Bedienung laeuft, waere ein 5-Sekunden-Takt unbenutzbar (ein Tastendruck
@@ -170,6 +176,45 @@ def _read_autooff_minutes(run_dir: Path, now: float | None = None):
return int(whole) + (1 if rest else 0) 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(run_dir: Path) -> str: def _read_mode_file(run_dir: Path) -> str:
"""Modus, in dem der Dienst laut RUN_DIR/mode tatsaechlich gestartet wurde. Fehlt die """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 Datei oder ist ihr Inhalt kein bekannter Modus, gilt "recovery" - derselbe Vorgabewert
@@ -209,19 +254,25 @@ def _service_active(service_name: str) -> bool:
return True return True
def _read_imageset(imageset_path: Path, conf: dict[str, str]) -> tuple[str, bool]: def _read_imageset(imageset_path: Path, conf: dict[str, str], mode: str) -> tuple[str, bool]:
"""(Name des eingestellten Bildsatzes, liegt er ueberhaupt auf dem Pi?). """(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 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 IMAGESET_DEFAULT - dieselbe Vorgabe wie in server.py und im CLI, sonst wuerde das
Panel etwas anderes behaupten als der Server tut. Panel etwas anderes behaupten als der Server tut.
`ok` ist False, wenn der GEWAEHLTE Satz nicht auf dem Pi liegt - nach derselben `ok` ist False, wenn der GEWAEHLTE Satz fuer den GERADE GELTENDEN Modus nicht auf dem
Regel wie server.py:migrate_set_present(): Verzeichnis vorhanden UND mindestens ein Pi liegt - nach derselben Regel wie server.py:migrate_set_present()/
Image darin. Dann liefert server.py nichts aus (Status "noimageset"), und genau das rescue_set_present(): Verzeichnis vorhanden UND mindestens ein Image darin. Welches
muss am Panel stehen, statt dass der Bediener vor einer Lampe steht, die nichts Verzeichnispaar das ist (migrate-*/rescue-*), entscheidet `mode` - Rettung und
bekommt. Ohne bekanntes IMAGE_DIR (Konfigurationsdatei nicht lesbar) wird nichts Migration liegen seit der Erweiterung des Umschalters auf die Rettung in getrennten
behauptet: dann gilt "ok", denn eine unbelegte Warnung ist schlechter als keine.""" 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: try:
value = imageset_path.read_text().strip() value = imageset_path.read_text().strip()
except OSError: except OSError:
@@ -230,7 +281,8 @@ def _read_imageset(imageset_path: Path, conf: dict[str, str]) -> tuple[str, bool
image_dir = conf.get("IMAGE_DIR") image_dir = conf.get("IMAGE_DIR")
if not image_dir: if not image_dir:
return name, True return name, True
set_dir = Path(image_dir) / IMAGESET_DIRS[name] dirs = RESCUE_IMAGESET_DIRS if mode == "recovery" else IMAGESET_DIRS
set_dir = Path(image_dir) / dirs[name]
try: try:
return name, any(p.name.endswith(".img") for p in set_dir.iterdir()) return name, any(p.name.endswith(".img") for p in set_dir.iterdir())
except OSError: except OSError:
@@ -286,9 +338,22 @@ def build_state(run_dir: Path = RUN_DIR, conf_path: Path = CONF_PATH,
Der Bildsatz ("imageset"/"imageset_ok") kommt in JEDEN Zustand, auch in "off" und in 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 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 Einschalten sehen - eine Einstellung von vorletzter Woche, die erst beim Blaettern im
Menue auffaellt, hat genau den Zweck verfehlt.""" 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
RUN_DIR/mode, 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 ueberlebt das Abschalten) 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) conf = _read_conf(conf_path)
imageset, imageset_ok = _read_imageset(imageset_path, conf) mode = _read_mode_file(run_dir)
imageset, imageset_ok = _read_imageset(imageset_path, conf, mode)
if not _service_active(service_name): if not _service_active(service_name):
# Die IP-Zeile gibt es NUR hier: ist die Station aus, ist der Pi ein gewoehnlicher # Die IP-Zeile gibt es NUR hier: ist die Station aus, ist der Pi ein gewoehnlicher
@@ -299,10 +364,10 @@ def build_state(run_dir: Path = RUN_DIR, conf_path: Path = CONF_PATH,
return {"mode": "off", "imageset": imageset, "imageset_ok": imageset_ok, return {"mode": "off", "imageset": imageset, "imageset_ok": imageset_ok,
"ip": local_ip()} "ip": local_ip()}
mode = _read_mode_file(run_dir)
status = _read_json(run_dir / "status.json") status = _read_json(run_dir / "status.json")
state: dict = {"mode": mode, "served": status.get("served"), "recent": status.get("recent"), state: dict = {"mode": mode, "served": status.get("served"), "recent": status.get("recent"),
"imageset": imageset, "imageset_ok": imageset_ok} "imageset": imageset, "imageset_ok": imageset_ok,
"activity": _recent_request(status)}
if mode == "recovery": if mode == "recovery":
state["ssid"] = conf.get("SSID") state["ssid"] = conf.get("SSID")
@@ -620,6 +685,23 @@ def run_clear(revision: str) -> None:
log.info("Panel geleert (EPD_REVISION=%s)", revision) 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, def run_daemon(revision: str, clock=time, panel: "Panel | None" = None,
read_state=build_state, read_ui=read_ui) -> None: read_state=build_state, read_ui=read_ui) -> None:
"""Endlosschleife: alle UI_POLL_INTERVAL_S Sekunden den Bedienzustand und alle """Endlosschleife: alle UI_POLL_INTERVAL_S Sekunden den Bedienzustand und alle
@@ -647,13 +729,25 @@ def run_daemon(revision: str, clock=time, panel: "Panel | None" = None,
panel = Panel(revision) panel = Panel(revision)
panel.clear() panel.clear()
planner = RefreshPlanner(clock.monotonic()) planner = RefreshPlanner(clock.monotonic())
base_state = read_state() base_state = _safe_read_state(read_state, {"mode": None})
last_state_poll = clock.monotonic() last_state_poll = clock.monotonic()
while True: while True:
now = clock.monotonic() now = clock.monotonic()
if now - last_state_poll >= POLL_INTERVAL_S: if now - last_state_poll >= POLL_INTERVAL_S:
base_state = read_state() # Jeder einzelne Baustein von build_state() faengt seine eigenen I/O-Fehler
# schon ab (siehe _read_json/_read_conf/_service_active/local_ip 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 last_state_poll = now
ui = read_ui() ui = read_ui()
if ui: if ui:
+21 -2
View File
@@ -316,9 +316,17 @@ def blocks(state: dict) -> list[tuple[str, str]]:
# die Zeile weg statt eine Adresse zu erfinden. Ist das Feld da, aber leer, sagt # die Zeile weg statt eine Adresse zu erfinden. Ist das Feld da, aber leer, sagt
# die Zeile ausdruecklich, dass kein Netz da ist: "keine Zeile" und "kein Netz" # die Zeile ausdruecklich, dass kein Netz da ist: "keine Zeile" und "kein Netz"
# sind zwei verschiedene Auskuenfte, und die zweite erklaert, warum SSH scheitert. # sind zwei verschiedene Auskuenfte, und die zweite erklaert, warum SSH scheitert.
#
# Beschriftung "LAN" statt eines nackten "IP"/"no network": Rueckmeldung eines
# Bedieners (2026-09-23) las "no network" so, als sei damit der Migrations-/
# Rettungs-Zugangspunkt gemeint (der ja tatsaechlich aus ist, die Station ist
# ausgeschaltet). Gemeint ist hier aber das GEGENTEIL - das gewoehnliche Heimnetz,
# in dem der Pi als normaler Rechner haengt, sobald die Station aus ist (siehe
# Kommentar oben). "LAN" grenzt das eindeutig vom Zugangspunkt-Netz ab, das auf
# jedem ANDEREN Statusbild als SSID steht.
ip = state.get("ip") ip = state.get("ip")
if isinstance(ip, str): if isinstance(ip, str):
out.append((f"IP {ip}" if ip else "no network", ROLE_DETAIL)) out.append((f"LAN: {ip}" if ip else "LAN: not connected", ROLE_DETAIL))
return out + footer return out + footer
if mode not in ("recovery", "migration"): if mode not in ("recovery", "migration"):
@@ -424,7 +432,18 @@ def blocks(state: dict) -> list[tuple[str, str]]:
out.append(("Auto-off pending", ROLE_DETAIL)) out.append(("Auto-off pending", ROLE_DETAIL))
served = state.get("served") served = state.get("served")
out.append((f"Served: {served if isinstance(served, int) else 0}", ROLE_DETAIL)) served_line = f"Served: {served if isinstance(served, int) else 0}"
# Lebenszeichen: ein kleiner Punkt, sobald IRGENDEINE Anfrage vor kurzem hereinkam
# (display.py:_recent_request() - auch eine erfolglose zaehlt). Angehaengt an eine
# bereits bestehende, immer vorhandene Zeile statt an einer eigenen Stelle im Bild
# gezeichnet: er braucht dadurch keine eigene Zeichenlogik in compose() und keinen
# eigenen Platz im knappen Zeilenbudget, und er faehrt ueber denselben
# Textvergleich (RefreshPlanner.decide() in display.py) mit, der ueber ein
# Neuzeichnen entscheidet - kein zusaetzlicher Takt, kein zusaetzliches Wachhalten
# des E-Papers, siehe display.py:ACTIVITY_WINDOW_S fuer die volle Begruendung.
if state.get("activity"):
served_line += " •"
out.append((served_line, ROLE_DETAIL))
recent = state.get("recent") recent = state.get("recent")
if isinstance(recent, list) and recent: if isinstance(recent, list) and recent:
+173 -6
View File
@@ -22,22 +22,27 @@ def test_off_mode_shows_the_address_the_pi_is_reachable_at():
"""Bei ausgeschalteter Station ist der Pi ein gewoehnlicher Rechner im Heimnetz - die """Bei ausgeschalteter Station ist der Pi ein gewoehnlicher Rechner im Heimnetz - die
Adresse am Panel erspart die Suche im Router, bevor man per SSH drangeht.""" Adresse am Panel erspart die Suche im Router, bevor man per SSH drangeht."""
lines = render.lines({"mode": "off", "ip": "192.168.2.96"}) lines = render.lines({"mode": "off", "ip": "192.168.2.96"})
assert any("IP 192.168.2.96" in l for l in lines) assert any("LAN: 192.168.2.96" in l for l in lines)
def test_off_mode_says_when_there_is_no_network(): def test_off_mode_says_when_there_is_no_network():
"""Leeres Feld heisst "kein Netz" und wird ausdruecklich gesagt: das erklaert, warum """Leeres Feld heisst "kein Netz" und wird ausdruecklich gesagt: das erklaert, warum
SSH gerade scheitert. Eine fehlende Zeile wuerde die Frage offen lassen.""" SSH gerade scheitert. Eine fehlende Zeile wuerde die Frage offen lassen.
Beschriftet mit "LAN", nicht mit einem nackten "no network": ein Bediener las die
frühere Formulierung so, als sei der Migrations-/Rettungs-Zugangspunkt gemeint - der
ist im Aus-Zustand ja tatsaechlich aus. Gemeint ist aber das GEGENTEIL, das
Heimnetz, in dem der Pi als gewoehnlicher Rechner haengt (siehe render.py)."""
lines = render.lines({"mode": "off", "ip": ""}) lines = render.lines({"mode": "off", "ip": ""})
assert any("no network" in l for l in lines) assert any("LAN: not connected" in l for l in lines)
assert not any("IP " in l for l in lines) assert not any("LAN: " in l and "not connected" not in l for l in lines)
def test_off_mode_without_an_address_field_claims_nothing(): def test_off_mode_without_an_address_field_claims_nothing():
"""Aeltere display.py-Fassung (oder ein von Hand gebauter Zustand): ohne das Feld """Aeltere display.py-Fassung (oder ein von Hand gebauter Zustand): ohne das Feld
bleibt die Zeile weg, statt eine Adresse zu erfinden.""" bleibt die Zeile weg, statt eine Adresse zu erfinden."""
lines = render.lines({"mode": "off"}) lines = render.lines({"mode": "off"})
assert not any("IP " in l or "no network" in l for l in lines) assert not any("LAN" in l for l in lines)
def test_the_address_line_belongs_to_the_off_screen_only(): def test_the_address_line_belongs_to_the_off_screen_only():
@@ -962,6 +967,70 @@ def test_local_ip_returns_a_string_and_never_raises():
assert isinstance(display.local_ip(), str) assert isinstance(display.local_ip(), str)
# --- Lebenszeichen: ein kleiner Punkt, sobald eine Anfrage jeder Art hereinkommt -------
#
# Bedienerwunsch (2026-09-23): "so ist es immer raetselraten und warten das der zaehler
# hochgeht um zu wissen ob es geklappt hat" - der Punkt soll VOR dem Erfolg schon zeigen,
# dass ueberhaupt jemand anklopft.
def test_recent_request_is_true_right_after_a_request():
status = {"last_request_ts": "2026-09-23T12:00:00Z"}
now = display.datetime(2026, 9, 23, 12, 0, 5, tzinfo=display.timezone.utc).timestamp()
assert display._recent_request(status, now=now) is True
def test_recent_request_expires_after_the_window():
status = {"last_request_ts": "2026-09-23T12:00:00Z"}
now = (display.datetime(2026, 9, 23, 12, 0, 0, tzinfo=display.timezone.utc).timestamp()
+ display.ACTIVITY_WINDOW_S + 1)
assert display._recent_request(status, now=now) is False
def test_recent_request_without_a_timestamp_is_false():
"""Aeltere server.py-Fassung, oder noch nie eine Anfrage seit dem Start - kein
erfundener Punkt."""
assert display._recent_request({}) is False
assert display._recent_request({"last_request_ts": None}) is False
assert display._recent_request({"last_request_ts": "kaputt"}) is False
def test_recent_request_from_the_future_is_false():
"""Eine verstellte Uhr darf den Punkt nicht auf ewig scharf halten."""
status = {"last_request_ts": "2026-09-23T12:00:10Z"}
now = display.datetime(2026, 9, 23, 12, 0, 0, tzinfo=display.timezone.utc).timestamp()
assert display._recent_request(status, now=now) is False
def test_served_line_shows_a_dot_while_activity_is_recent():
lines_active = render.lines({"mode": "recovery", "served": 3, "recent": [],
"activity": True})
lines_idle = render.lines({"mode": "recovery", "served": 3, "recent": [],
"activity": False})
active_line = next(l for l in lines_active if l.startswith("Served:"))
idle_line = next(l for l in lines_idle if l.startswith("Served:"))
assert active_line != idle_line
assert "3" in active_line and "3" in idle_line
def test_served_line_without_an_activity_field_claims_nothing():
"""Aeltere display.py-Fassung (oder ein von Hand gebauter Zustand im Test): ohne das
Feld bleibt die Zeile wie zuvor, kein erfundener Punkt."""
lines = render.lines({"mode": "recovery", "served": 3, "recent": []})
served_line = next(l for l in lines if l.startswith("Served:"))
assert served_line == "Served: 3"
def test_build_state_carries_activity_in_recovery_and_migration(tmp_path, monkeypatch):
monkeypatch.setattr(display, "_service_active", lambda service_name: True)
(tmp_path / "mode").write_text("recovery")
(tmp_path / "status.json").write_text(json.dumps({"last_request_ts": None}))
conf = tmp_path / "station.conf"
conf.write_text("")
state = display.build_state(run_dir=tmp_path, conf_path=conf, service_name="x",
imageset_path=tmp_path / "imageset")
assert state["activity"] is False
def test_build_state_mode_comes_from_mode_file_not_status_json(tmp_path, monkeypatch): def test_build_state_mode_comes_from_mode_file_not_status_json(tmp_path, monkeypatch):
"""Der Modus-Vorrang liegt bei RUN_DIR/mode, nicht bei status.json's "mode"-Feld - """Der Modus-Vorrang liegt bei RUN_DIR/mode, nicht bei status.json's "mode"-Feld -
status.json kann veraltet sein (Fixrunde 1, Kleinigkeit: der Vorrang war zuerst status.json kann veraltet sein (Fixrunde 1, Kleinigkeit: der Vorrang war zuerst
@@ -1511,6 +1580,67 @@ def _changing_every(period_s, mode="recovery"):
return lambda t: {"mode": mode, "served": int(t // period_s), "recent": []} return lambda t: {"mode": mode, "served": int(t // period_s), "recent": []}
# --- Lebenszeichen im Aus-Zustand: die IP muss auftauchen, sobald sie da ist -----------
#
# Rueckmeldung eines Bedieners (2026-09-23): am laufenden Pi blieb "no network" stehen,
# obwohl der Pi laengst im Heimnetz haengt. local_ip() liefert die Adresse dabei
# nachweislich korrekt - der Verdacht war also die Anzeigeschleife: wird der Aus-Zustand
# ueberhaupt regelmaessig neu erhoben, und fliesst die IP in den Vergleich ein, der ueber
# ein Neuzeichnen entscheidet? Dieser Test treibt run_daemon() GENAU durch diesen Fall:
# beim ersten Erheben keine Adresse, beim zweiten eine.
def test_off_mode_redraws_once_the_address_appears(monkeypatch):
"""Haelt fest: run_daemon() erhebt den Zustand auch im Aus-Zustand unbedingt alle
POLL_INTERVAL_S Sekunden neu (kein Ueberspringen mangels Moduswechsel), und die IP
ist Teil des Vergleichs, der ueber ein Neuzeichnen entscheidet (render.lines() traegt
sie, siehe render.py). Schlaegt dieser Test fehl, bleibt "LAN: not connected" auf dem
Panel stehen, obwohl der Pi laengst erreichbar ist."""
seq = [{"mode": "off", "ip": ""}, {"mode": "off", "ip": "192.168.2.96"}]
def state_at(t):
# Erste POLL_INTERVAL_S-Erhebung (bei t=0) sieht noch keine Adresse, danach schon -
# weit genug auseinander, dass zwei unterschiedliche Polls sicher dazwischenliegen.
return seq[0] if t < 2 * display.POLL_INTERVAL_S else seq[1]
panel = _drive_daemon(state_at, duration_s=4 * display.POLL_INTERVAL_S)
# Erster Aufbau zeigt "keine Adresse" (Vollaufbau beim Start). Danach MUSS
# spaetestens einer der Aufrufe die neue Adresse tatsaechlich zeichnen - das
# Panel-Doppel protokolliert keinen Bildinhalt, deshalb wird direkt geprueft, dass
# render.lines() zum Zeitpunkt des naechsten Aufbaus schon die neue Adresse traegt
# (dieselbe Quelle, die run_daemon() intern fuer den Vergleich benutzt).
assert len(panel.calls) >= 2, panel.calls
draws_after_first = [name for _, name in panel.calls[1:]]
assert any(name in ("full", "partial", "wake-full") for name in draws_after_first), (
"kein weiterer Aufbau nach dem ersten - die neue Adresse waere nie gezeichnet worden",
panel.calls,
)
assert any("LAN: 192.168.2.96" in line for line in render.lines(state_at(4 * display.POLL_INTERVAL_S)))
def test_run_daemon_survives_a_single_failed_state_read(monkeypatch):
"""Ein einzelner unerwarteter Fehler beim Erheben (ausserhalb dessen, was
build_state()s eigene Bausteine schon abfangen) darf die Endlosschleife nicht
beenden - sonst friert das Panel beim naechsten Fehlschlag fuer immer auf dem
zuletzt gezeichneten Bild ein (E-Paper vergisst nichts von selbst), und zwar OHNE
jede Fehlermeldung am Geraet. Erhebung 1 wirft, Erhebung 2 liefert einen echten
Zustand - danach muss dieser auch tatsaechlich gezeichnet sein."""
calls = {"n": 0}
good_state = {"mode": "off", "ip": "192.168.2.96"}
def flaky_state_at(t):
calls["n"] += 1
if calls["n"] == 1:
raise RuntimeError("simulierter Lesefehler (z.B. RUN_DIR mitten im Neustart)")
return good_state
panel = _drive_daemon(flaky_state_at, duration_s=3 * display.POLL_INTERVAL_S)
# Kein Absturz (sonst haette _drive_daemon die Ausnahme durchgereicht statt sie
# abzufangen) und die Schleife lief tatsaechlich weiter: mindestens der initiale
# Vollaufbau (mit dem neutralen Platzhalter) und ein weiterer, sobald der zweite
# Versuch den echten Zustand liefert.
assert len(panel.calls) >= 2, panel.calls
def test_static_content_never_triggers_another_full_refresh(): def test_static_content_never_triggers_another_full_refresh():
"""Fixrunde 3, F3 - vom Reviewer per Mutation als ungetestet belegt: nimmt man die """Fixrunde 3, F3 - vom Reviewer per Mutation als ungetestet belegt: nimmt man die
Bedingung `partials_since_full > 0` aus `stale` heraus, bleiben alle Tests gruen. Bedingung `partials_since_full > 0` aus `stale` heraus, bleiben alle Tests gruen.
@@ -1885,8 +2015,13 @@ def test_build_state_reads_the_image_set_switch(tmp_path, monkeypatch):
def test_build_state_reports_a_missing_image_set(tmp_path, monkeypatch): def test_build_state_reports_a_missing_image_set(tmp_path, monkeypatch):
""""Vorhanden" heisst: Verzeichnis da UND mindestens ein Image darin - dieselbe Regel """"Vorhanden" heisst: Verzeichnis da UND mindestens ein Image darin - dieselbe Regel
wie in server.py, sonst sagt das Panel "in Ordnung" und die Lampe bekommt nichts.""" wie in server.py, sonst sagt das Panel "in Ordnung" und die Lampe bekommt nichts.
Ausdruecklich im Migrationsmodus (RUN_DIR/mode) - das ist es, was die migrate-*-
Verzeichnisse betrifft. Der Rettungsmodus prueft ein eigenes Paar (rescue-*), siehe
test_build_state_reports_a_missing_rescue_image_set unten."""
monkeypatch.setattr(display, "_service_active", lambda service_name: True) monkeypatch.setattr(display, "_service_active", lambda service_name: True)
(tmp_path / "mode").write_text("migration")
images = tmp_path / "images" images = tmp_path / "images"
(images / "migrate-nocreds").mkdir(parents=True) (images / "migrate-nocreds").mkdir(parents=True)
(images / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG") (images / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
@@ -1909,6 +2044,37 @@ def test_build_state_reports_a_missing_image_set(tmp_path, monkeypatch):
imageset_path=switch)["imageset_ok"] is False imageset_path=switch)["imageset_ok"] is False
def test_build_state_reports_a_missing_rescue_image_set(tmp_path, monkeypatch):
"""Derselbe Umschalter gilt jetzt auch fuer die Rettung, mit einem EIGENEN
Verzeichnispaar (rescue-*, nicht migrate-*) - server.py begruendet die Trennung.
Ohne Modusdatei gilt "recovery" (dieselbe Vorgabe wie resolve_mode() in server.py),
das Panel muss also hier die rescue-*-Verzeichnisse pruefen, nicht die migrate-*."""
monkeypatch.setattr(display, "_service_active", lambda service_name: True)
images = tmp_path / "images"
(images / "rescue-nocreds").mkdir(parents=True)
(images / "rescue-nocreds" / "ninlight_g95_cw_ota.img").write_bytes(b"IMG")
# Ein voll bestueckter Migrations-Bildsatz daneben - er darf das Ergebnis nicht
# beeinflussen (falsche Zuordnung waere der Fehler, den dieser Test faengt).
(images / "migrate-wlan").mkdir(parents=True)
conf = tmp_path / "station.conf"
conf.write_text(f"IMAGE_DIR='{images}'\n")
switch = tmp_path / "imageset"
switch.write_text("ohne-wlan")
assert display.build_state(run_dir=tmp_path, conf_path=conf, service_name="x",
imageset_path=switch)["imageset_ok"] is True
switch.write_text("mit-wlan") # rescue-wlan/ gibt es nicht
state = display.build_state(run_dir=tmp_path, conf_path=conf, service_name="x",
imageset_path=switch)
assert state["imageset_ok"] is False
assert "MISSING" in _imageset_line(state)
# Und ein LEERES Verzeichnis zaehlt genauso als fehlend.
(images / "rescue-wlan").mkdir()
assert display.build_state(run_dir=tmp_path, conf_path=conf, service_name="x",
imageset_path=switch)["imageset_ok"] is False
def test_the_display_and_the_server_agree_on_the_set_names(): def test_the_display_and_the_server_agree_on_the_set_names():
"""Zwei Prozesse, eine Wahrheit: waeren die Namen oder die Vorgabe verschieden, """Zwei Prozesse, eine Wahrheit: waeren die Namen oder die Vorgabe verschieden,
behauptete das Panel etwas anderes, als der Server tut.""" behauptete das Panel etwas anderes, als der Server tut."""
@@ -1919,6 +2085,7 @@ def test_the_display_and_the_server_agree_on_the_set_names():
spec.loader.exec_module(server) spec.loader.exec_module(server)
assert display.IMAGESET_DIRS == server.IMAGE_SET_DIRS assert display.IMAGESET_DIRS == server.IMAGE_SET_DIRS
assert display.IMAGESET_DEFAULT == server.DEFAULT_IMAGE_SET assert display.IMAGESET_DEFAULT == server.DEFAULT_IMAGE_SET
assert display.RESCUE_IMAGESET_DIRS == server.RESCUE_SET_DIRS
# --- Abschlussbildschirm beim Herunterfahren und beim Neustart ----------------------- # --- Abschlussbildschirm beim Herunterfahren und beim Neustart -----------------------