feat(station): Bedienung ueber einen Taster mit Panel-Rueckmeldung

Ein Taster auf GPIO23 (active low, 50 ms entprellt) bedient die Station am
Geraet: kurz oeffnet das Menue bzw. blaettert, gehalten waehlt (1,5 s),
bricht ab (3 s, feuert sofort) und faehrt herunter (10 s, nur ueber die
sichtbare Abbruchstufe hinweg). Menue schliesst nach 20 s, Rueckfrage
verfaellt nach 10 s als Abbruch; jede veraendernde Auswahl bekommt eine
Rueckfrage.

Noetig, weil "on"/"migrate on" wlan0 zum Zugangspunkt macht und damit jede
SSH-Sitzung abreisst - bisher blieb dort nur warten oder Stecker ziehen.

"Migration starten" startet ueber das Menue IMMER mit --listen: der
Migrations-AP hat ein oeffentlich bekanntes Passwort, ein Tastendruck im
Automatikmodus wuerde jede alte Klyqa-Lampe in Reichweite ungefragt
migrieren.

Der Tasterdienst fasst den SPI-Bus nicht an (sonst "GPIO busy"), sondern
schreibt seinen Bedienzustand atomar mit 0644 nach
/run/ninlight-button/ui.json - bewusst nicht ins RuntimeDirectory der
Hauptunit, das systemd beim "off" aufraeumt. Der Anzeigedienst liest die
Datei jede halbe Sekunde (der teure build_state()-Fork bleibt beim
5-Sekunden-Takt) und zeichnet sie als Teilaufbau, kostet dem Panel also
keinen Vollaufbau je Tastendruck.

Fehlt gpiozero oder der Taster, gibt es genau eine Protokollzeile und
sys.exit(0) - dieselbe Zusage wie bei der Anzeige.

Ausliefern/Einrichten folgt in einem eigenen Durchgang.
This commit is contained in:
Michael Sauer
2026-09-14 21:50:37 +02:00
parent f0f6894bc9
commit 82a3942554
7 changed files with 1378 additions and 19 deletions
+35
View File
@@ -25,6 +25,41 @@ würde dort nur mit einer Fehlermeldung enden.
ssh <pi-host> ninlight-station status Zustand und letzte Auslieferungen ssh <pi-host> ninlight-station status Zustand und letzte Auslieferungen
ssh <pi-host> ninlight-station off ausschalten ssh <pi-host> ninlight-station off ausschalten
## Taster am Gerät
Sobald die Station läuft, wird `wlan0` zum Zugangspunkt und **jede SSH-Sitzung reißt ab**.
Für genau diesen Zustand gibt es einen Taster auf **GPIO23**; die Rückmeldung kommt aufs
E-Paper. Ein Taster, zwei Zusammenhänge:
| | kurz (unter 1 s) | gehalten |
|----------------------|------------------|-------------------------------------------------------|
| Ruhe (Statusbild) | Menü öffnen | 3 s → Abbruch (Station aus), weiter 10 s → Herunterfahren |
| Menü offen | nächster Eintrag | 1,5 s → Eintrag auswählen |
| Rückfrage | abbrechen | 1,5 s → bestätigen |
Der Abbruch feuert **bei 3 s**, nicht erst beim Loslassen — danach läuft auf dem Panel ein
Countdown bis zum Herunterfahren, den Loslassen jederzeit abbricht. Das Menü schließt nach
20 s ohne Eingabe, eine Rückfrage verfällt nach 10 s als Abbruch. Die Einträge hängen vom
Zustand ab (Station aus / Rettung läuft / Migration läuft); jede verändernde Auswahl
bekommt eine Rückfrage.
**„Migration starten" startet über das Menü immer in der Horchphase** (`--listen`), nie
automatisch: der Migrations-Zugangspunkt hat ein öffentlich bekanntes Passwort, und ein
Tastendruck, der ihn im Automatikmodus hochfährt, würde jede alte Klyqa-Lampe in
Funkreichweite ungefragt migrieren, auch fremde. Wer den Automatikmodus will, nimmt die
Kommandozeile.
Der Taster ist nirgends Voraussetzung: fehlt `gpiozero` oder der Taster, protokolliert der
Dienst eine Zeile und beendet sich ohne Fehler. Er fasst den SPI-Bus **nicht** an (zwei
Prozesse auf denselben Leitungen ergeben `GPIO busy`), sondern schreibt seinen
Bedienzustand nach `/run/ninlight-button/ui.json`; der Anzeigedienst liest ihn und
zeichnet. Ausgeführt wird ausschließlich über das CLI.
Dienst und Unit liegen in `tools/station/button/` bzw.
`tools/station/ninlight-button.service` — das Ausliefern und Einrichten (`deploy.sh`,
`setup_pi.sh`) kommt in einem eigenen Durchgang, bis dahin installiert der Aufsetzweg den
Tasterdienst noch nicht.
SSID und Passwort stehen in `/etc/ninlight-station.conf` auf dem Pi und müssen zu den SSID und Passwort stehen in `/etc/ninlight-station.conf` auf dem Pi und müssen zu den
Build-Parametern `NINLIGHT_RECOVERY_SSID` und `NINLIGHT_RECOVERY_PASSWORD` aus Build-Parametern `NINLIGHT_RECOVERY_SSID` und `NINLIGHT_RECOVERY_PASSWORD` aus
`config.local.cmake` passen. Weicht eines ab, findet die Lampe die Station nicht. `config.local.cmake` passen. Weicht eines ab, findet die Lampe die Station nicht.
+579
View File
@@ -0,0 +1,579 @@
#!/usr/bin/env python3
"""Tasterdienst der ninlight-Station: EIN Taster auf GPIO23 (active low, interner
Pull-up), Rueckmeldung ueber das E-Paper.
Wozu das gut ist: "ninlight-station on"/"migrate on" macht wlan0 zum Zugangspunkt und
reisst damit JEDE SSH-Sitzung ueber WLAN ab. Genau in diesem Zustand hatte der Bediener
am Geraet bisher keinerlei Eingriff - es blieb warten oder Stecker ziehen, und Letzteres
hat an diesem Pi schon eine SD-Karte gekostet. Der Taster ist der Ausweg: Abbruch
(Station aus) und, ueber eine sichtbare Zwischenstufe, ein sauberes Herunterfahren.
Bedienung (ein Taster, zwei Zusammenhaenge - die Bedeutung haengt daran, ob das Menue
offen ist):
kurz (< 1 s) gehalten
Ruhe Menue oeffnen 3 s -> Abbruch, weiter 10 s -> Herunterfahren
Menue offen naechster Eintrag 1,5 s -> Eintrag auswaehlen
Rueckfrage abbrechen 1,5 s -> bestaetigen
Der Abbruch feuert BEI 3 s, nicht erst beim Loslassen: der Bediener soll die Wirkung am
Panel sehen. Haelt er weiter, laeuft dort ein Countdown bis 10 s (Herunterfahren);
Loslassen bricht diese zweite Stufe jederzeit ab. Das Herunterfahren ist damit nur ueber
eine sichtbare Zustandsaenderung hinweg erreichbar - nicht aus Versehen.
DER TASTERDIENST FASST DEN SPI-BUS NICHT AN. Zwei Prozesse auf denselben GPIO-/SPI-
Leitungen enden in "lgpio.error: 'GPIO busy'" (am Geraet passiert, als
"ninlight-station display test" neben dem laufenden Anzeigedienst lief). Deshalb schreibt
dieser Dienst nur seinen Bedienzustand nach UI_FILE, und der Anzeigedienst liest ihn und
zeichnet (siehe display.read_ui()/render.blocks()).
Alle veraendernden Aktionen laufen ueber das vorhandene CLI (/usr/local/bin/
ninlight-station), nie ueber nachgebaute Logik: nur das CLI kennt die Profilumschaltung,
die Nachpruefungen ("es darf nie mehr als ein Stationsprofil aktiv sein") und den
Zeitschalter.
SICHERHEITSREGEL: "Migration starten" startet ueber das Menue IMMER in der Horchphase
("migrate on --listen"), nie automatisch. Der Migrations-Zugangspunkt hat ein oeffentlich
bekanntes Passwort (es steht in der alten Klyqa-Firmware); ein Tastendruck, der ihn im
Automatikmodus hochfaehrt, wuerde jede alte Klyqa-Lampe in Funkreichweite ungefragt
migrieren, auch fremde. Die bequeme Geste muss die sichere sein - wer den Automatikmodus
will, nimmt die Kommandozeile, dort ist die Absicht eindeutig.
Der Taster ist NIRGENDS Voraussetzung: fehlt gpiozero, fehlt der Taster oder ist GPIO23
belegt, schreibt main() GENAU EINE Protokollzeile und beendet sich mit sys.exit(0) -
dieselbe Zusage (und derselbe Grund: Restart=on-failure darf daraus keine Neustartschleife
machen) wie bei der Anzeige, siehe display.py's Modul-Docstring. Und ein Fehler am Taster
darf niemals eine laufende Migration stoeren oder das Ein-/Ausschalten aufhalten - deshalb
faengt jeder Schreib- und Ausfuehrungspfad hier seine Fehler selbst ab.
"""
from __future__ import annotations
import json
import logging
import math
import os
import subprocess
import sys
import time
from pathlib import Path
log = logging.getLogger("ninlight-button")
# --- Zeiten und Pfade ---------------------------------------------------------------
BUTTON_PIN = int(os.environ.get("NINLIGHT_BUTTON_PIN", "23"))
# 50 ms statt der naheliegenden 20: ein sehr kurzer Tipper kam am Geraet durch eine
# 20-ms-Entprellung noch durch und oeffnete ungewollt das Menue. Die Entprellung macht
# gpiozero selbst (bounce_time), nicht diese Schleife - siehe open_button().
DEBOUNCE_S = 0.05
# Abtastrate der Hauptschleife. Deutlich feiner als die Entprellung, damit eine Flanke
# nicht erst eine halbe Entprellzeit spaeter gesehen wird; ein Durchlauf ohne Aenderung
# kostet nur einen gpiozero-Registerzugriff und einen Woerterbuchvergleich (UiWriter
# schreibt nur bei Aenderung).
TICK_S = 0.02
SHORT_PRESS_MAX_S = 1.0
SELECT_HOLD_S = 1.5 # Menue: Eintrag waehlen / Rueckfrage: bestaetigen
CANCEL_HOLD_S = 3.0 # Ruhe: Abbruch (Station aus)
SHUTDOWN_HOLD_S = 10.0 # Ruhe: Herunterfahren (nur ueber die Abbruchstufe hinweg)
MENU_TIMEOUT_S = 20.0 # ohne Eingabe zurueck aufs Statusbild
CONFIRM_TIMEOUT_S = 10.0 # Rueckfrage verfaellt als Abbruch
RESULT_S = 4.0 # wie lange die Rueckmeldung einer Aktion stehen bleibt
# Dieselben Umgebungsvariablen wie ninlight-station/display.py, damit Tests und
# Handbetrieb ohne echtes /run bzw. /etc auskommen.
RUN_DIR = Path(os.environ.get("NINLIGHT_RUN_DIR", "/run/ninlight-station"))
# BEWUSST NICHT /run/ninlight-station/ui.json (das war der Vorschlag): dieses Verzeichnis
# ist das RuntimeDirectory der Hauptunit, und systemd raeumt es beim regulaeren Stop von
# ninlight-station auf - also ausgerechnet im Zustand "Station aus", in dem das Menue am
# dringendsten gebraucht wird (dort liegen "Rettung starten"/"Migration starten"). Die
# Datei liegt deshalb im eigenen RuntimeDirectory dieses Dienstes; sie verschwindet damit
# genau dann, wenn dieser Dienst endet - und ein stehengebliebenes Menue kann den
# Anzeigedienst nicht ueberdauern.
UI_FILE = Path(os.environ.get("NINLIGHT_UI_FILE", "/run/ninlight-button/ui.json"))
STATION_CLI = os.environ.get("NINLIGHT_STATION_BIN", "/usr/local/bin/ninlight-station")
STATION_SERVICE = os.environ.get("NINLIGHT_STATION_SERVICE", "ninlight-station")
# Das Herunterfahren geht NICHT ueber das CLI (das kennt es nicht) - es ist die einzige
# Aktion dieses Dienstes, die an systemd selbst geht.
SHUTDOWN_ARGV = ("systemctl", "poweroff")
# Obergrenze fuer einen CLI-Aufruf. "recovery on"/"migrate on" schalten NetworkManager-
# Profile um und pruefen danach nach; das dauert Sekunden, nicht Minuten. Ohne Deckel
# wuerde ein haengender NetworkManager die Bedienung dauerhaft einfrieren - mit ihm
# bleibt der Taster spaetestens danach wieder bedienbar.
ACTION_TIMEOUT_S = 120
class ButtonUnavailable(Exception):
"""Kein Taster da (keine Bibliothek, kein Pin, Pin belegt) - main() macht daraus
genau eine Protokollzeile und sys.exit(0)."""
# --- Aktionen -----------------------------------------------------------------------
class Action:
"""Was ein Menueeintrag bzw. die Haltegeste ausloest.
`confirm` = braucht eine Rueckfrage. Zusage: JEDE veraendernde Auswahl bekommt eine
("Anzeige auffrischen" veraendert nichts und braucht keine)."""
def __init__(self, label: str, argv=(), kind: str = "cli", confirm: bool = True,
done_text: str | None = None) -> None:
self.label = label
self.argv = tuple(argv)
self.kind = kind
self.confirm = confirm
self.done_text = done_text or label
def __repr__(self) -> str: # nur fuer Testausgaben
return "Action({!r}, {!r}, kind={!r})".format(self.label, self.argv, self.kind)
def action_shutdown() -> Action:
return Action("Herunterfahren", kind="shutdown", confirm=True,
done_text="Faehrt herunter")
def action_cancel() -> Action:
"""Die Haltegeste in Ruhe: Station aus. Bewusst ohne Rueckfrage - sie IST die
Rueckfrage (3 s halten), und sie ist der Notausgang, der ohne Umweg wirken muss."""
return Action("Station aus", ("off",), confirm=False, done_text="Station aus")
def menu_actions(state: dict) -> list[Action]:
"""Die Menueeintraege fuer den gegebenen Stationszustand - nur das, was gerade
sinnvoll ist. Reine Funktion ueber dem von station_state() gelesenen Zustand."""
running = bool(state.get("running"))
mode = state.get("mode")
target = state.get("target")
if not running:
return [
Action("Rettung starten", ("recovery", "on"), done_text="Rettung laeuft"),
# IMMER --listen (siehe Modul-Docstring, Sicherheitsregel). Wer den
# Automatikmodus will, nimmt die Kommandozeile.
Action("Migration starten", ("migrate", "on", "--listen"),
done_text="Migration: horcht"),
action_shutdown(),
]
items: list[Action] = []
if mode == "migration" and target:
# Nur wenn last_seen wirklich ein Ziel kennt - ohne erkannte Lampe gibt es
# nichts freizugeben, und "arm" ohne Ziel wuerde das CLI ohnehin abweisen.
items.append(Action("Freigabe: {}".format(target), ("arm", str(target)),
done_text="Freigegeben: {}".format(target)))
items.append(Action("Station aus", ("off",), done_text="Station aus"))
# Veraendert nichts am Betrieb - einzige Auswahl ohne Rueckfrage.
items.append(Action("Anzeige auffrischen", ("display", "test"), confirm=False,
done_text="Anzeige aufgefrischt"))
items.append(action_shutdown())
return items
class CliActions:
"""Fuehrt eine Action aus - ueber das CLI bzw. systemctl, nie durch Nachbauen deren
Logik. Gibt zurueck, ob es geklappt hat; wirft nie: ein Fehler hier darf weder den
Tasterdienst beenden noch eine laufende Migration stoeren."""
def __init__(self, cli: str = STATION_CLI, shutdown_argv=SHUTDOWN_ARGV,
runner=subprocess.run) -> None:
self.cli = cli
self.shutdown_argv = tuple(shutdown_argv)
self.runner = runner
def argv_for(self, action: Action) -> list[str]:
if action.kind == "shutdown":
return list(self.shutdown_argv)
return [self.cli] + list(action.argv)
def run(self, action: Action) -> bool:
argv = self.argv_for(action)
log.info("Taster fuehrt aus: %s", " ".join(argv))
try:
result = self.runner(argv, check=False, timeout=ACTION_TIMEOUT_S)
except (OSError, subprocess.SubprocessError) as e:
log.warning("Aktion '%s' fehlgeschlagen: %s", action.label, e)
return False
rc = getattr(result, "returncode", 0)
if rc != 0:
log.warning("Aktion '%s' endete mit Rueckgabewert %s", action.label, rc)
return rc == 0
# --- Stationszustand (nur lesen) ----------------------------------------------------
def _read_json(path: Path) -> dict:
try:
data = json.loads(path.read_text())
except (OSError, ValueError):
return {}
return data if isinstance(data, dict) else {}
def _service_active(service_name: str = STATION_SERVICE) -> bool:
try:
result = subprocess.run(["systemctl", "is-active", "--quiet", service_name],
timeout=5, check=False)
except (OSError, subprocess.SubprocessError):
# Ohne systemd (Handlauf) lieber "laeuft" annehmen als das Menue mit
# "Rettung starten" anzubieten, waehrend die Station in Wahrheit laeuft.
return True
return result.returncode == 0
def station_state(run_dir: Path = RUN_DIR, service_name: str = STATION_SERVICE) -> dict:
"""Der Zustand, von dem die Menueeintraege abhaengen. Liest dieselben Quellen wie
display.build_state() - RUN_DIR/mode (nicht status.json's "mode"-Feld, siehe dort)
und status.json fuer das zuletzt erkannte Ziel - dupliziert aber keine Logik des
CLI: geschaltet wird ausschliesslich ueber das CLI."""
running = _service_active(service_name)
try:
mode = (run_dir / "mode").read_text().strip()
except OSError:
mode = "recovery"
if mode not in ("recovery", "migration"):
mode = "recovery"
status = _read_json(run_dir / "status.json")
last_seen = status.get("last_seen")
target = last_seen.get("target") if isinstance(last_seen, dict) else None
return {"running": running, "mode": mode,
"target": target if isinstance(target, str) and target else None}
# --- Bedienzustand in eine Datei schreiben ------------------------------------------
class UiWriter:
"""Schreibt den Bedienzustand atomar mit 0644 nach UI_FILE - dasselbe Muster und
dieselben Rechte wie status.json/control.json (Temp-Datei im selben Verzeichnis,
Rechte setzen, dann umbenennen). Beides sind Muster, deren Verletzung in diesem
Projekt schon mehrfach Fehler verursacht hat: ohne 0644 kann der Anzeigedienst die
Datei nicht lesen, ohne das Umbenennen sieht er sie halb geschrieben.
Schreibt nur bei Aenderung (die Schleife laeuft alle TICK_S) und wirft nie."""
def __init__(self, path: Path = UI_FILE) -> None:
self.path = Path(path)
self._last: dict | None = None
self._failed_logged = False
def write(self, view: dict) -> bool:
"""True, wenn tatsaechlich geschrieben wurde."""
if view == self._last:
return False
payload = json.dumps(view, ensure_ascii=False, sort_keys=True)
tmp = self.path.with_name("{}.tmp-{}".format(self.path.name, os.getpid()))
try:
self.path.parent.mkdir(parents=True, exist_ok=True)
tmp.write_text(payload)
os.chmod(tmp, 0o644)
os.replace(tmp, self.path)
except OSError as e:
# Nur einmal protokollieren: die Schleife kaeme sonst alle TICK_S wieder
# hierher und wuerde das Journal fluten. Der Taster bleibt trotzdem
# bedienbar, nur die Rueckmeldung am Panel fehlt.
if not self._failed_logged:
log.warning("Bedienzustand konnte nicht geschrieben werden (%s): %s",
self.path, e)
self._failed_logged = True
try:
tmp.unlink()
except OSError:
pass
return False
self._last = dict(view)
return True
# --- Die Zustandsfuehrung -----------------------------------------------------------
def _countdown(total_s: float, held_s: float) -> int:
"""Ganze Sekunden bis `total_s`, aufgerundet und nie negativ - "noch 6" solange auch
nur eine Zehntelsekunde davon uebrig ist."""
return max(0, int(math.ceil(total_s - held_s)))
class ButtonUI:
"""Die gesamte Bedienlogik - welcher Druck was bedeutet, welche Zeitschwelle was
ausloest, was auf dem Panel steht - als reine Zustandsmaschine ohne Schleife, ohne
time.sleep() und ohne GPIO. Dasselbe Muster wie display.RefreshPlanner, aus demselben
Grund: so laesst sich ein kompletter Bedienverlauf in Millisekunden durchspielen.
Benutzung pro Runde: update(now, pressed) -> Action oder None; eine zurueckgegebene
Action fuehrt der Aufrufer aus und meldet das Ergebnis per done() zurueck. view(now)
liefert jederzeit den Bedienzustand fuer die Anzeige.
`pressed` ist der ENTPRELLTE Pegel (gpiozero, bounce_time), nicht eine Flanke - alle
Schwellen rechnen deshalb gegen den Zeitpunkt der letzten steigenden Flanke. Das ist
auch der Grund, warum ein blockierender CLI-Aufruf mitten in einer Haltegeste nichts
kaputtmacht: nach der Rueckkehr wird derselbe Pegel wieder gegen dieselbe Uhr
gemessen."""
def __init__(self, now: float = 0.0, read_state=station_state) -> None:
self._read_state = read_state
self._pressed = False
self._press_t = now
# Diese Haltegeste hat schon etwas ausgeloest - ein spaeteres Loslassen darf
# dann nicht zusaetzlich als "kurzer Druck" gelten.
self._consumed = False
# 0 = nichts, 1 = Abbruch gefeuert, zweite Stufe (Herunterfahren) laeuft.
self._hold_stage = 0
# In welchem Zusammenhang diese Haltegeste BEGONNEN hat ("idle", "menu",
# "confirm"). Die Haltegeste wird danach beurteilt, NICHT nach dem inzwischen
# erreichten Bildschirm: waehlt man im Menue einen Eintrag (1,5 s), steht danach
# "Bitte warten"/die Rueckfrage - haelt der Bediener einfach weiter, duerfte das
# nicht plotzlich als die Ruhe-Haltegeste gelten und bei 3 s die Station
# abschalten. Genau das tat eine fruehere Fassung.
self._press_screen = "idle"
self.screen = "idle"
self.items: list[Action] = []
self.index = 0
self.pending: Action | None = None
self.deadline: float | None = None
self.busy: Action | None = None
self.result: tuple[str, bool] | None = None
self.result_until: float | None = None
# -- Eingabe ---------------------------------------------------------------------
def update(self, now: float, pressed: bool) -> Action | None:
action = None
if pressed and not self._pressed:
self._pressed = True
self._press_t = now
self._consumed = False
self._press_screen = self.screen if self.screen in ("menu", "confirm") else "idle"
elif pressed and self._pressed:
action = self._held(now)
elif self._pressed and not pressed:
self._pressed = False
held = now - self._press_t
if not self._consumed and held < SHORT_PRESS_MAX_S:
action = self._short(now)
# Loslassen bricht die zweite Stufe (Herunterfahren) jederzeit ab.
self._hold_stage = 0
self._expire(now)
else:
self._expire(now)
if action is not None:
self.busy = action
self.screen = "busy"
self.pending = None
self.deadline = None
return action
def _short(self, now: float) -> Action | None:
self.result = None
self.result_until = None
if self.screen == "menu":
if self.items:
self.index = (self.index + 1) % len(self.items)
self.deadline = now + MENU_TIMEOUT_S
return None
if self.screen == "confirm":
# Abbrechen - zurueck ins Menue, nicht aufs Statusbild: der Bediener ist
# noch am Bedienen und will meist einen anderen Eintrag.
self.pending = None
self.screen = "menu"
self.deadline = now + MENU_TIMEOUT_S
return None
# Ruhe (auch waehrend eine Rueckmeldung steht): Menue oeffnen.
self._open_menu(now)
return None
def _open_menu(self, now: float) -> None:
try:
state = self._read_state()
except Exception as e: # noqa: BLE001 - siehe Modul-Docstring: nie stoeren
log.warning("Stationszustand nicht lesbar, Menue bleibt zu: %s", e)
return
self.items = menu_actions(state)
if not self.items:
return
self.index = 0
self.screen = "menu"
self.deadline = now + MENU_TIMEOUT_S
def _held(self, now: float) -> Action | None:
held = now - self._press_t
# Nach dem Zusammenhang BEIM DRUCK beurteilt, nicht nach dem aktuellen
# Bildschirm - siehe _press_screen.
if self._press_screen == "menu":
if held >= SELECT_HOLD_S and not self._consumed and self.items:
self._consumed = True
return self._select(now, self.items[self.index])
return None
if self._press_screen == "confirm":
if held >= SELECT_HOLD_S and not self._consumed:
self._consumed = True
return self.pending
return None
# Ruhe (oder eine stehende Rueckmeldung/ein gerade gelaufener Abbruch):
# zweistufige Haltegeste.
if self._hold_stage == 1 and held >= SHUTDOWN_HOLD_S and not self._consumed:
self._consumed = True
return action_shutdown()
if self._hold_stage == 0 and held >= CANCEL_HOLD_S:
# Feuert BEI 3 s, nicht erst beim Loslassen - und setzt _consumed NICHT:
# die zweite Stufe muss aus derselben Geste noch erreichbar bleiben.
self._hold_stage = 1
self.result = None
self.result_until = None
return action_cancel()
return None
def _select(self, now: float, action: Action) -> Action | None:
if not action.confirm:
return action
self.pending = action
self.screen = "confirm"
self.deadline = now + CONFIRM_TIMEOUT_S
return None
def _expire(self, now: float) -> None:
if self.result_until is not None and now >= self.result_until:
self.result = None
self.result_until = None
if self.deadline is None or now < self.deadline:
return
if self.screen == "menu":
# Ein Geraet, das im Menue stehenbleibt, soll nicht aussehen, als sei es
# beschaeftigt - zurueck aufs Statusbild.
self.screen = "idle"
self.items = []
self.deadline = None
elif self.screen == "confirm":
# Verfaellt als ABBRUCH - eine Rueckfrage, die von selbst zustimmt, waere
# das Gegenteil einer Rueckfrage.
self.pending = None
self.screen = "idle"
self.items = []
self.deadline = None
def done(self, now: float, action: Action, ok: bool) -> None:
"""Ergebnis einer ausgefuehrten Aktion - setzt die Rueckmeldung fuers Panel."""
self.busy = None
self.screen = "idle"
self.items = []
self.pending = None
self.deadline = None
self.result = (action.done_text if ok else "{}: Fehler".format(action.label), ok)
self.result_until = now + RESULT_S
# -- Ausgabe ---------------------------------------------------------------------
def view(self, now: float) -> dict:
"""Der Bedienzustand, wie ihn der Anzeigedienst zeichnet. "active" sagt ihm, dass
gerade eine Bedienung laeuft (er sieht dann schneller nach)."""
if (self._pressed and self._press_screen == "idle"
and self.screen in ("idle", "busy")):
held = now - self._press_t
if self._hold_stage == 1:
return _view("hold", "STATION AUS",
"Herunterfahren in {} s".format(
_countdown(SHUTDOWN_HOLD_S, held)),
["Weiter halten - Loslassen bricht ab"])
if held >= SHORT_PRESS_MAX_S:
return _view("hold", "HALTEN",
"Abbruch in {} s".format(_countdown(CANCEL_HOLD_S, held)),
["Weiter halten - Loslassen bricht ab"])
if self.screen == "menu" and self.items:
selected = self.items[self.index]
rest = self.items[self.index + 1:] + self.items[:self.index]
return _view("menu", "MENUE", "> " + selected.label,
[a.label for a in rest] + ["kurz: weiter, halten: waehlen"])
if self.screen == "confirm" and self.pending is not None:
return _view("confirm", "RUECKFRAGE", self.pending.label,
["halten (1,5 s): ja", "kurz: nein"])
if self.screen == "busy" and self.busy is not None:
return _view("busy", "BITTE WARTEN", self.busy.label, [])
if self.result is not None:
text, ok = self.result
return _view("result", "FERTIG" if ok else "FEHLER", text, [])
return {"view": "idle", "active": False}
def _view(view: str, title: str, hero: str, details: list[str]) -> dict:
return {"view": view, "title": title, "hero": hero, "details": list(details),
"active": True}
# --- Hardware und Hauptschleife -----------------------------------------------------
def open_button(pin: int = BUTTON_PIN):
"""Der einzige Hardwarezugriff dieses Dienstes: ein Eingangspin. KEIN SPI, kein
Zeichenpin - siehe Modul-Docstring ("GPIO busy").
Entprellt gpiozero selbst (bounce_time): ein Tipper, der kuerzer ist als
DEBOUNCE_S, kommt gar nicht erst als Druck an."""
try:
import gpiozero
except ImportError as e:
raise ButtonUnavailable("gpiozero fehlt ({})".format(e))
try:
return gpiozero.Button(pin, pull_up=True, bounce_time=DEBOUNCE_S)
except Exception as e: # noqa: BLE001
# Bewusst breit: ein belegter Pin kommt je nach Pin-Factory als
# gpiozero.GPIOPinInUse, als roher lgpio.error ('GPIO busy') oder als OSError
# zurueck - keine gemeinsame Oberklasse, die hier greifen wuerde. Alles davon
# heisst dasselbe: kein Taster, also lautlos ohne Fehler beenden, statt eine
# Neustartschleife auszuloesen.
raise ButtonUnavailable("GPIO{} nicht verfuegbar ({})".format(pin, e))
def run_daemon(button=None, ui: "ButtonUI | None" = None, writer: "UiWriter | None" = None,
actions: "CliActions | None" = None, clock=time,
read_state=station_state) -> None:
"""Endlosschleife: Pegel lesen, ButtonUI fragen, eine zurueckgegebene Aktion
ausfuehren, Ergebnis zurueckmelden, Bedienzustand schreiben.
`button`, `ui`, `writer`, `actions` und `clock` sind einschleusbar, damit die
VERDRAHTUNG hier selbst pruefbar ist und nicht nur eine Kopie davon (display.py hatte
genau diesen Fehler: die Tests trieben eine nachgebaute Schleife, und Mutationen an
den echten Zeilen blieben gruen). Im Betrieb bleiben die Vorgaben.
Die Reihenfolge ist wesentlich: der Bedienzustand wird VOR dem (sekundenlangen)
CLI-Aufruf geschrieben, sonst steht auf dem Panel bis zu dessen Ende noch das Menue
und der Bediener haelt das Geraet fuer eingefroren."""
if button is None:
button = open_button()
if writer is None:
writer = UiWriter()
if actions is None:
actions = CliActions()
if ui is None:
ui = ButtonUI(clock.monotonic(), read_state=read_state)
log.info("Taster bereit (GPIO%s), Bedienzustand in %s", BUTTON_PIN, writer.path)
writer.write(ui.view(clock.monotonic()))
while True:
now = clock.monotonic()
action = ui.update(now, bool(button.is_pressed))
if action is not None:
writer.write(ui.view(now))
ok = actions.run(action)
ui.done(clock.monotonic(), action, ok)
writer.write(ui.view(clock.monotonic()))
clock.sleep(TICK_S)
def main() -> None:
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
try:
run_daemon()
except ButtonUnavailable as e:
# GENAU EINE Zeile, dann sys.exit(0) - siehe Modul-Docstring. Kein Fehler, damit
# Restart=on-failure daraus keine Neustartschleife macht und das Journal flutet.
log.info("Taster nicht verfuegbar, beende ohne Fehler: %s", e)
sys.exit(0)
except KeyboardInterrupt:
sys.exit(0)
if __name__ == "__main__":
main()
+68 -16
View File
@@ -51,8 +51,27 @@ log = logging.getLogger("ninlight-display")
RUN_DIR = Path(os.environ.get("NINLIGHT_RUN_DIR", "/run/ninlight-station")) 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")) CONF_PATH = Path(os.environ.get("NINLIGHT_STATION_CONF", "/etc/ninlight-station.conf"))
SERVICE_NAME = os.environ.get("NINLIGHT_STATION_SERVICE", "ninlight-station") 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"))
POLL_INTERVAL_S = 5 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 # E-Paper altert mit jedem Bildaufbau - ein Vollaufbau nur bei Moduswechsel und
# mindestens alle 30 Minuten, sonst Teilaktualisierung (siehe Modul-Docstring der # mindestens alle 30 Minuten, sonst Teilaktualisierung (siehe Modul-Docstring der
# Panel-Treiber: haeufige Teilaktualisierungen ohne gelegentlichen Vollaufbau lassen # Panel-Treiber: haeufige Teilaktualisierungen ohne gelegentlichen Vollaufbau lassen
@@ -220,6 +239,21 @@ def build_state(run_dir: Path = RUN_DIR, conf_path: Path = CONF_PATH,
return state 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): def _load_epd_module(revision: str):
"""Laedt den passenden Panel-Treiber. Nicht ueber ein Dict aus Modulnamen, damit ein """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 statischer Import-Checker (und ein Mensch, der `grep -n epd2in13` tippt) beide
@@ -485,30 +519,48 @@ def run_clear(revision: str) -> None:
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) -> None: read_state=build_state, read_ui=read_ui) -> None:
"""Endlosschleife: alle POLL_INTERVAL_S Sekunden den Zustand einlesen und die von """Endlosschleife: alle UI_POLL_INTERVAL_S Sekunden den Bedienzustand und alle
RefreshPlanner entschiedene Aktion ausfuehren. Hier steht bewusst KEINE Zeitlogik mehr POLL_INTERVAL_S Sekunden den Stationszustand einlesen und die von RefreshPlanner
(Fixrunde 4) - die steckt vollstaendig in RefreshPlanner und ist dort ohne Schleife, entschiedene Aktion ausfuehren. Hier steht bewusst KEINE Zeitlogik zur Panelalterung
ohne time.sleep() und ohne Hardware pruefbar. mehr (Fixrunde 4) - die steckt vollstaendig in RefreshPlanner und ist dort ohne
Schleife, ohne time.sleep() und ohne Hardware pruefbar.
`clock` (braucht monotonic() und sleep()), `panel` und `read_state` sind einschleusbar, Die zwei Takte: der teure Teil (build_state() forkt fuer "systemctl is-active")
damit die VERDRAHTUNG hier - welche Aktion welche Panel-Methode ruft, dass die bleibt beim 5-Sekunden-Takt, der billige (eine kleine Datei lesen) laeuft in
Selbst-Aufwertung von draw_partial() zurueckgemeldet wird, dass das Einschlafen UI_POLL_INTERVAL_S, damit eine Bedienung am Taster ueberhaupt benutzbar ist. Ein
zurueckgemeldet wird, und die Reihenfolge decide -> ausfuehren -> zurueckmelden - laufender Bedienzustand ERSETZT das Statusbild (render.blocks() zweigt dafuer ganz
selbst pruefbar ist und nicht nur eine Kopie davon (Fixrunde 5, N2: alle drei vorne ab); der Modus bleibt derselbe, sodass das Bild ueber Teilaufbauten kommt und
Mutationen an genau diesen Zeilen blieben vorher gruen, weil der Test eine nachgebaute ein Tastendruck dem Panel keinen Vollaufbau kostet.
Schleife statt dieser hier trieb). Im Betrieb bleiben die Vorgaben: echte Uhr, echtes
Panel, echtes build_state().""" `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: if panel is None:
panel = Panel(revision) panel = Panel(revision)
panel.clear() panel.clear()
planner = RefreshPlanner(clock.monotonic()) planner = RefreshPlanner(clock.monotonic())
base_state = read_state()
last_state_poll = clock.monotonic()
while True: while True:
state = read_state() now = clock.monotonic()
if now - last_state_poll >= POLL_INTERVAL_S:
base_state = read_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) current_lines = render.lines(state)
mode = state.get("mode") mode = state.get("mode")
now = clock.monotonic()
action = planner.decide(now, current_lines, mode) action = planner.decide(now, current_lines, mode)
if action == "full": if action == "full":
@@ -523,7 +575,7 @@ def run_daemon(revision: str, clock=time, panel: "Panel | None" = None,
panel.sleep() panel.sleep()
planner.record_sleep() planner.record_sleep()
clock.sleep(POLL_INTERVAL_S) clock.sleep(UI_POLL_INTERVAL_S)
def main() -> None: def main() -> None:
+40
View File
@@ -186,6 +186,16 @@ def blocks(state: dict) -> list[tuple[str, str]]:
Bild - die Rolle ist ein reiner Bezeichner, die zugehoerigen Schriftgroessen kennt Bild - die Rolle ist ein reiner Bezeichner, die zugehoerigen Schriftgroessen kennt
ausschliesslich compose().""" ausschliesslich compose()."""
state = state if isinstance(state, dict) else {} state = state if isinstance(state, dict) else {}
# Laeuft gerade eine Bedienung am Taster, ERSETZT sie das Statusbild komplett - eine
# Bedienung, die neben dem Statusbild um Platz kaempft, waere auf 250x122 unlesbar,
# und in diesem Moment ist der Bedienzustand die einzige Auskunft, auf die der
# Bediener wartet. display.read_ui() liefert {} (und damit kein "ui"), sobald nichts
# laeuft - dann geht es unveraendert weiter wie zuvor.
ui = state.get("ui")
if isinstance(ui, dict) and ui.get("view") not in (None, "", "idle"):
return _ui_blocks(ui)
mode = state.get("mode") mode = state.get("mode")
out: list[tuple[str, str]] = [] out: list[tuple[str, str]] = []
@@ -311,6 +321,36 @@ def blocks(state: dict) -> list[tuple[str, str]]:
return out return out
def _ui_blocks(ui: dict) -> list[tuple[str, str]]:
"""Der Bedienzustand des Tasterdienstes in Zeilen mit Rolle. Bewusst DUMM: welcher
Text bei welcher Geste erscheint, entscheidet allein der Tasterdienst (dort ist es
ohne Hardware pruefbar) - hier wird er nur noch den vorhandenen Rollen zugeordnet,
damit Kopfbalken, Schriftgroessen, Breitenkuerzung und unterer Rand dieselben bleiben
wie im Statusbild.
Jedes Feld wird auf seinen Typ geprueft: die Datei kommt aus einem anderen Prozess
(der aelter, neuer oder halb geschrieben sein kann) - genau wie status.json darf sie
die Anzeige nie zum Absturz bringen."""
out: list[tuple[str, str]] = []
title = ui.get("title")
if isinstance(title, str) and title:
out.append((title, ROLE_HEADER))
hero = ui.get("hero")
if isinstance(hero, str) and hero:
out.append((hero, ROLE_HERO))
details = ui.get("details")
if isinstance(details, list):
for detail in details:
if isinstance(detail, str) and detail:
out.append((detail, ROLE_DETAIL))
if not out:
# Es laeuft eine Bedienung, aber die Datei traegt keinen brauchbaren Text
# (abgeschnitten, fremde Fassung). Lieber gross sagen, dass bedient wird, als
# ein leeres Panel zu zeigen, das nach "haengt" aussieht.
out.append(("Bedienung", ROLE_BIG))
return out
def lines(state: dict) -> list[str]: def lines(state: dict) -> list[str]:
"""Nur die Texte aus blocks(), ohne Rollen - unveraenderte Schnittstelle fuer """Nur die Texte aus blocks(), ohne Rollen - unveraenderte Schnittstelle fuer
display.run_daemon(), das damit erkennt, ob sich der Inhalt geaendert hat (die Rolle display.run_daemon(), das damit erkennt, ob sich der Inhalt geaendert hat (die Rolle
+75
View File
@@ -0,0 +1,75 @@
[Unit]
Description=ninlight Taster (Bedienung der Station ueber GPIO23)
# Wie beim Anzeigedienst nur After=, NICHT Requires=/BindsTo=/Wants=: ein fehlender oder
# abgestuerzter Taster darf server.py niemals mitreissen, und server.py muss starten
# koennen, auch wenn dieser Dienst fehlt oder deaktiviert ist. KEIN
# Wants=ninlight-station.service - das wuerde die absichtlich deaktivierte Station bei
# jedem Boot mit hochziehen (Fund der Hardware-Abnahme, ausfuehrlich in
# ninlight-display.service begruendet).
After=ninlight-station.service
# Gehoeren seit systemd 229 ins [Unit], nicht ins [Service] (dort werden sie mit
# "Unknown key name" ignoriert, und die Deckelung unten greift dann nie).
StartLimitIntervalSec=60
StartLimitBurst=5
[Service]
Type=simple
Environment=NINLIGHT_RUN_DIR=/run/ninlight-station
Environment=NINLIGHT_UI_FILE=/run/ninlight-button/ui.json
# "-"-Praefix VOR dem Dateipfad (nicht vor dem Schluesselnamen - so herum ist es ein
# ungueltiger Schluessel und systemd verwirft die Zeile ganz): ohne die
# Konfigurationsdatei startet der Taster trotzdem.
EnvironmentFile=-/etc/ninlight-station.conf
# Eigenes RuntimeDirectory, aus zwei Gruenden:
#
# 1. Hier liegt ui.json. Es darf NICHT in /run/ninlight-station liegen (dem
# RuntimeDirectory der Hauptunit): systemd raeumt das beim regulaeren Stop von
# ninlight-station auf - also ausgerechnet im Zustand "Station aus", in dem das Menue
# ("Rettung starten"/"Migration starten") am dringendsten gebraucht wird. So
# verschwindet ui.json genau dann, wenn dieser Dienst endet, und ein
# stehengebliebenes Menue kann ihn nicht ueberdauern.
# 2. lgpio (gpiozero's bevorzugte Pin-Factory) legt beim Start "$HOME/.lgd-nfy0" an.
# Unter systemd ist HOME per Vorgabe LEER und das Arbeitsverzeichnis "/" - der Pfad
# wird "//.lgd-nfy0" und scheitert. Am Geraet fuer den Anzeigedienst nachgemessen,
# siehe dort; derselbe Stolperstein trifft jeden gpiozero-Dienst.
RuntimeDirectory=ninlight-button
RuntimeDirectoryMode=0755
WorkingDirectory=/run/ninlight-button
Environment=HOME=/run/ninlight-button
ExecStart=/usr/bin/python3 /opt/ninlight-station/button/button.py
# button.py faengt einen fehlenden Taster (keine Bibliothek, kein Pin, Pin belegt) als
# ButtonUnavailable ab, protokolliert eine Zeile und beendet sich mit sys.exit(0) - das
# ist nie ein "Fehlschlag" im systemd-Sinne und loest kein Restart= aus. on-failure
# (nicht always) deckelt damit genau das Richtige: echte Abstuerze werden neu versucht,
# ein dauerhaft fehlender Taster nicht.
Restart=on-failure
RestartSec=5
# BEWUSST KEIN NoNewPrivileges=/ProtectSystem=strict/ProtectHome= hier - anders als beim
# Anzeigedienst, und das ist kein Versehen:
#
# Dieser Dienst ruft /usr/local/bin/ninlight-station auf, und dessen einziger Zweck ist
# es, NetworkManager-Profile umzuschalten, Units zu starten/stoppen und nach
# /run/ninlight-station zu schreiben - alles ueber "sudo". Kindprozesse erben den
# Namensraum und das no_new_privs-Flag dieser Unit: mit ProtectSystem=strict waere
# /run/ninlight-station fuer das CLI schreibgeschuetzt (mode/control.json liessen sich
# nicht mehr schreiben), und no_new_privs laesst kein setuid-Binary mehr zu (sudo). Eine
# Absicherung, die genau die Aktionen unmoeglich macht, wegen derer es den Dienst gibt,
# ist keine. Die Angriffsflaeche ist stattdessen eng gehalten: EIN Eingangspin, kein
# Netzwerk, keine Eingabe ausser der Taste selbst, und ausgefuehrt wird ausschliesslich
# die feste Liste von CLI-Unterbefehlen aus button.py.
MemoryMax=64M
# Laeuft als root (wie Haupt- und Anzeigedienst) fuer GPIO23 und fuer das CLI - kein
# SupplementaryGroups=gpio: auf reinem Debian gibt es die Gruppe oft gar nicht, und eine
# fehlende Gruppe ist fuer systemd ein harter Startfehler (216/GROUP), ausgerechnet in
# einer Unit, die nie stoeren darf.
# KEIN Zugriff auf SPI und keine Zeichenleitung: der Tasterdienst fasst den Panel-Bus
# nicht an (zwei Prozesse darauf ergeben "lgpio.error: 'GPIO busy'"). Er schreibt seinen
# Bedienzustand nach /run/ninlight-button/ui.json, der Anzeigedienst liest ihn und
# zeichnet.
[Install]
WantedBy=multi-user.target
+461
View File
@@ -0,0 +1,461 @@
"""Tests fuer den Tasterdienst (tools/station/button/button.py).
Die Bedienlogik ist reine Zustandsfuehrung und wird hier OHNE Hardware geprueft - aber
bewusst NICHT, indem der Ablauf im Test nachgebaut wird: getrieben wird die ECHTE
Hauptschleife button.run_daemon() mit einer Fake-Uhr, einem nach Zeitplan gedrueckten
Taster-Doppel, dem ECHTEN CliActions (nur dessen subprocess.run ist ersetzt) und einem
protokollierenden Schreiber.
Der Grund steht in display.py/test_render.py: bei der Zeitsteuerung der Anzeige trieben
die Tests eine KOPIE der Verdrahtung zwischen Zustandsmaschine und Ausfuehrung, und man
konnte die eigentliche Aktion durch eine falsche ersetzen, ohne dass ein Test rot wurde.
Deshalb laeuft hier alles ueber dieselbe Schleife, die auch auf dem Pi laeuft."""
import json
import os
import subprocess
import sys
import types
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "button"))
import button # noqa: E402
# --- Zustaende, von denen die Menueeintraege abhaengen -------------------------------
STATE_OFF = {"running": False, "mode": "recovery", "target": None}
STATE_RECOVERY = {"running": True, "mode": "recovery", "target": None}
STATE_MIGRATION = {"running": True, "mode": "migration", "target": "gu10_rgb"}
STATE_MIGRATION_BLIND = {"running": True, "mode": "migration", "target": None}
# --- Fake-Uhr, Taster-Doppel, Protokoll ---------------------------------------------
class _StopDaemon(Exception):
"""Bricht run_daemon()s Endlosschleife ab, sobald die simulierte Zeit abgelaufen ist."""
class _FakeClock:
def __init__(self, until_s: float) -> None:
self.t = 0.0
self.until_s = until_s
def monotonic(self) -> float:
return self.t
def advance(self, seconds: float) -> None:
self.t += seconds
def sleep(self, seconds: float) -> None:
self.t += seconds
if self.t > self.until_s:
raise _StopDaemon
class _Recorder:
"""Gemeinsame Zeitleiste von Schreiber und Ausfuehrung - nur so faellt auf, wenn die
Reihenfolge in run_daemon() kippt (z.B. das "Bitte warten" erst NACH dem
sekundenlangen CLI-Aufruf geschrieben wuerde)."""
def __init__(self, clock: _FakeClock) -> None:
self.clock = clock
self.events: list[tuple[float, str, object]] = []
def note(self, kind: str, payload: object) -> None:
self.events.append((self.clock.monotonic(), kind, payload))
def views(self) -> list[tuple[float, dict]]:
return [(t, p) for t, k, p in self.events if k == "view"]
def runs(self) -> list[tuple[float, list]]:
return [(t, p) for t, k, p in self.events if k == "run"]
def argvs(self) -> list[list]:
return [p for _t, p in self.runs()]
def kinds(self) -> list[str]:
return [v.get("view") for _t, v in self.views()]
def first_view(self, name: str):
for t, v in self.views():
if v.get("view") == name:
return t, v
return None
class _ScriptedButton:
"""Taster-Doppel: gedrueckt genau waehrend der angegebenen Zeitfenster. `is_pressed`
ist wie bei gpiozero ein PEGEL (die Entprellung macht dort bounce_time), keine
Flanke - genau das, was run_daemon() abfragt."""
def __init__(self, clock: _FakeClock, presses) -> None:
self.clock = clock
self.presses = list(presses)
@property
def is_pressed(self) -> bool:
t = self.clock.monotonic()
return any(start <= t < end for start, end in self.presses)
class _RecordingWriter:
"""Schreiber-Doppel: protokolliert jeden tatsaechlich geschriebenen Bedienzustand.
Dedupliziert wie der echte UiWriter, damit die Zeitleiste die ANZEIGEWECHSEL zeigt
und nicht 50 identische Zeilen je Sekunde."""
def __init__(self, rec: _Recorder) -> None:
self.rec = rec
self.path = Path("/dev/null")
self._last = None
def write(self, view: dict) -> bool:
if view == self._last:
return False
self._last = dict(view)
self.rec.note("view", dict(view))
return True
def _fake_runner(clock: _FakeClock, rec: _Recorder, duration_s: float = 0.0, rc: int = 0):
"""Ersetzt subprocess.run in CliActions - der Rest von CliActions (und damit die
ZUSAMMENSETZUNG der Kommandozeile, auf die es sicherheitsrelevant ankommt) ist echt."""
def run(argv, check=False, timeout=None):
rec.note("run", list(argv))
# Ein CLI-Aufruf dauert Sekunden (Profilumschaltung) - die simulierte Uhr muss
# das mitmachen, sonst pruefen die Tests eine Schleife, die es so nicht gibt.
clock.advance(duration_s)
return types.SimpleNamespace(returncode=rc)
return run
def _drive(presses, duration_s: float, state=STATE_OFF, action_duration_s: float = 0.0,
rc: int = 0) -> _Recorder:
"""Laesst das echte button.run_daemon() `duration_s` simulierte Sekunden laufen."""
clock = _FakeClock(duration_s)
rec = _Recorder(clock)
actions = button.CliActions(cli="/usr/local/bin/ninlight-station",
runner=_fake_runner(clock, rec, action_duration_s, rc))
try:
button.run_daemon(button=_ScriptedButton(clock, presses),
writer=_RecordingWriter(rec), actions=actions, clock=clock,
read_state=lambda: dict(state))
except _StopDaemon:
pass
return rec
def _short(at: float):
"""Ein kurzer Druck (deutlich unter SHORT_PRESS_MAX_S, deutlich ueber DEBOUNCE_S)."""
return (at, at + 0.2)
# --- Menueaufbau --------------------------------------------------------------------
def _labels(state):
return [a.label for a in button.menu_actions(state)]
def test_menu_off_offers_starting_and_shutdown():
assert _labels(STATE_OFF) == ["Rettung starten", "Migration starten", "Herunterfahren"]
def test_menu_in_recovery_offers_stopping_refreshing_and_shutdown():
assert _labels(STATE_RECOVERY) == ["Station aus", "Anzeige auffrischen",
"Herunterfahren"]
def test_menu_in_migration_offers_the_detected_target():
assert _labels(STATE_MIGRATION) == ["Freigabe: gu10_rgb", "Station aus",
"Anzeige auffrischen", "Herunterfahren"]
def test_menu_in_migration_without_a_detected_target_offers_no_release():
"""Ohne last_seen gibt es nichts freizugeben - der Eintrag faellt ganz weg, statt
eine Freigabe ins Leere anzubieten (die das CLI ohnehin abweisen wuerde)."""
assert "Freigabe: " not in " ".join(_labels(STATE_MIGRATION_BLIND))
assert _labels(STATE_MIGRATION_BLIND) == ["Station aus", "Anzeige auffrischen",
"Herunterfahren"]
def test_only_the_refresh_entry_skips_the_confirmation():
"""Zusage: JEDE veraendernde Auswahl bekommt eine Rueckfrage. "Anzeige auffrischen"
veraendert nichts und braucht keine."""
for state in (STATE_OFF, STATE_RECOVERY, STATE_MIGRATION):
for action in button.menu_actions(state):
assert action.confirm is (action.label != "Anzeige auffrischen"), action.label
def test_actions_run_through_the_station_cli():
"""Die Aktionen gehen ueber das vorhandene CLI, nicht durch nachgebaute Logik - und
das Herunterfahren als einzige an systemd selbst."""
actions = button.CliActions(cli="/usr/local/bin/ninlight-station")
assert actions.argv_for(button.menu_actions(STATE_OFF)[0]) == [
"/usr/local/bin/ninlight-station", "recovery", "on"]
assert actions.argv_for(button.action_shutdown()) == ["systemctl", "poweroff"]
# --- Bedienung, getrieben durch die echte Hauptschleife ------------------------------
def test_a_short_press_opens_the_menu():
rec = _drive([_short(1.0)], 4.0)
assert rec.argvs() == []
t, view = rec.first_view("menu")
assert view["hero"] == "> Rettung starten"
assert view["active"] is True
# Spuerbar, aber bedienbar: das Menue steht binnen eines Schleifentakts nach dem
# Loslassen (die Verzoegerung am Panel kommt aus dessen eigenem Takt, siehe
# display.UI_POLL_INTERVAL_S).
assert 1.2 <= t <= 1.3
def test_a_short_press_steps_to_the_next_entry():
rec = _drive([_short(1.0), _short(2.0)], 4.0)
assert rec.argvs() == []
assert rec.views()[-1][1]["hero"] == "> Migration starten"
def test_a_short_press_wraps_around_the_last_entry():
rec = _drive([_short(1.0), _short(2.0), _short(3.0), _short(4.0)], 6.0)
assert rec.views()[-1][1]["hero"] == "> Rettung starten"
def test_holding_one_and_a_half_seconds_selects_the_entry():
rec = _drive([_short(1.0), (2.0, 4.0)], 6.0)
t, view = rec.first_view("confirm")
assert view["hero"] == "Rettung starten"
assert 3.5 <= t <= 3.6
# Die Auswahl allein fuehrt noch nichts aus - dafuer ist die Rueckfrage da.
assert rec.argvs() == []
def test_holding_in_the_confirmation_executes_the_entry():
rec = _drive([_short(1.0), (2.0, 4.0), (5.0, 7.0)], 9.0)
assert rec.argvs() == [["/usr/local/bin/ninlight-station", "recovery", "on"]]
assert 6.5 <= rec.runs()[0][0] <= 6.6
def test_a_short_press_in_the_confirmation_cancels_and_returns_to_the_menu():
rec = _drive([_short(1.0), (2.0, 4.0), _short(5.0)], 8.0)
assert rec.argvs() == []
assert rec.views()[-1][1]["view"] == "menu"
def test_the_refresh_entry_runs_without_a_confirmation():
rec = _drive([_short(1.0), _short(2.0), (3.0, 5.0)], 7.0, state=STATE_RECOVERY)
assert rec.argvs() == [["/usr/local/bin/ninlight-station", "display", "test"]]
assert "confirm" not in rec.kinds()
def test_starting_the_migration_always_listens():
"""Sicherheitsregel, nicht verhandelbar: der Migrations-Zugangspunkt hat ein
oeffentlich bekanntes Passwort - ein Tastendruck, der ihn im Automatikmodus
hochfaehrt, wuerde jede alte Klyqa-Lampe in Funkreichweite ungefragt migrieren, auch
fremde."""
rec = _drive([_short(1.0), _short(2.0), (3.0, 5.0), (6.0, 8.0)], 10.0)
assert rec.argvs() == [["/usr/local/bin/ninlight-station", "migrate", "on",
"--listen"]]
assert "--listen" in rec.argvs()[0]
def test_the_cancel_fires_at_three_seconds_while_the_button_is_still_held():
"""Der Abbruch feuert BEI 3 s, nicht erst beim Loslassen - der Bediener soll die
Wirkung am Panel sehen, waehrend er noch haelt."""
rec = _drive([(1.0, 8.0)], 10.0)
assert rec.argvs() == [["/usr/local/bin/ninlight-station", "off"]]
t = rec.runs()[0][0]
assert 4.0 <= t <= 4.1
# ... und danach laeuft am Panel der Countdown der zweiten Stufe.
later = [v for vt, v in rec.views() if vt > t]
assert any("Herunterfahren in" in v.get("hero", "") for v in later)
def test_releasing_between_three_and_ten_seconds_does_not_shut_down():
rec = _drive([(1.0, 7.0)], 12.0)
assert rec.argvs() == [["/usr/local/bin/ninlight-station", "off"]]
assert ["systemctl", "poweroff"] not in rec.argvs()
def test_holding_ten_seconds_shuts_down():
rec = _drive([(1.0, 12.0)], 14.0)
assert rec.argvs() == [["/usr/local/bin/ninlight-station", "off"],
["systemctl", "poweroff"]]
assert 11.0 <= rec.runs()[1][0] <= 11.1
def test_the_shutdown_countdown_is_visible_before_it_happens():
"""Das Herunterfahren ist nur ueber eine SICHTBARE Zustandsaenderung hinweg
erreichbar - der Countdown muss also vorher auf dem Panel gestanden haben."""
rec = _drive([(1.0, 12.0)], 14.0)
heroes = [v.get("hero", "") for t, v in rec.views() if t < 11.0]
assert "Herunterfahren in 6 s" in heroes
assert "Herunterfahren in 1 s" in heroes
def test_a_blocking_action_does_not_break_the_hold_gesture():
"""Der Abbruch-Aufruf blockiert die Schleife sekundenlang (Profilumschaltung). Danach
wird derselbe PEGEL gegen dieselbe Uhr gemessen - die zweite Stufe muss trotzdem
erreichbar bleiben."""
rec = _drive([(1.0, 12.0)], 14.0, action_duration_s=2.0)
assert rec.argvs()[-1] == ["systemctl", "poweroff"]
def test_the_waiting_screen_is_written_before_the_slow_call():
rec = _drive([_short(1.0), _short(2.0), (3.0, 5.0)], 8.0, state=STATE_RECOVERY,
action_duration_s=4.0)
kinds = [k for _t, k, _p in rec.events]
run_at = kinds.index("run")
assert rec.events[run_at - 1][2].get("view") == "busy"
def test_the_menu_closes_after_twenty_seconds_without_input():
rec = _drive([_short(1.0)], 26.0)
assert rec.kinds()[-1] == "idle"
closed_at = [t for t, v in rec.views() if v.get("view") == "idle" and t > 2.0][0]
assert 21.2 <= closed_at <= 21.4
def test_the_confirmation_expires_after_ten_seconds_as_a_cancel():
rec = _drive([_short(1.0), (2.0, 4.0)], 18.0)
assert rec.argvs() == []
assert rec.kinds()[-1] == "idle"
expired_at = [t for t, v in rec.views() if v.get("view") == "idle" and t > 4.0][0]
assert 13.5 <= expired_at <= 13.7
def test_an_expired_confirmation_leaves_nothing_pending():
"""Der Abbruch muss die vorgemerkte Aktion auch wirklich vergessen - eine Aktion, die
nach dem Verfall noch in der Maschine liegt, ist genau das, was eine Rueckfrage
verhindern soll."""
ui = button.ButtonUI(0.0, read_state=lambda: dict(STATE_OFF))
ui.update(1.0, True)
ui.update(1.2, False) # kurzer Druck: Menue auf
ui.update(2.0, True)
ui.update(3.6, True) # 1,6 s gehalten: Eintrag gewaehlt -> Rueckfrage
ui.update(3.8, False)
assert ui.screen == "confirm" and ui.pending is not None
ui.update(3.6 + button.CONFIRM_TIMEOUT_S + 0.1, False)
assert ui.screen == "idle"
assert ui.pending is None
def test_a_failed_action_says_so_on_the_panel():
rec = _drive([_short(1.0), (2.0, 4.0), (5.0, 7.0)], 9.0, rc=1)
assert any(v.get("view") == "result" and v.get("title") == "FEHLER"
for _t, v in rec.views())
def test_nothing_happens_without_any_press():
rec = _drive([], 30.0)
assert rec.argvs() == []
assert rec.kinds() == ["idle"]
# --- Bedienzustandsdatei ------------------------------------------------------------
def test_the_ui_file_is_written_atomically_with_0644(tmp_path):
"""Dasselbe Muster und dieselben Rechte wie status.json/control.json - ohne 0644
kann der Anzeigedienst die Datei nicht lesen, ohne das Umbenennen sieht er sie halb
geschrieben."""
path = tmp_path / "run" / "ui.json"
writer = button.UiWriter(path)
view = {"view": "menu", "title": "MENUE", "hero": "> Station aus", "details": [],
"active": True}
# Unter einer strengen umask geschrieben: die Rechte muessen ausdruecklich gesetzt
# werden, nicht zufaellig aus der umask des Dienstes fallen - sonst laege die Zusage
# "der Anzeigedienst kann die Datei lesen" an einer Umgebungsvariablen des Systems.
old_umask = os.umask(0o077)
try:
assert writer.write(view) is True
finally:
os.umask(old_umask)
assert json.loads(path.read_text()) == view
assert oct(os.stat(path).st_mode & 0o777) == "0o644"
# Keine Temp-Datei zurueckgelassen (sie wuerde sonst mit jedem Neustart ein weiteres
# Bruchstueck im RuntimeDirectory hinterlassen).
assert [p.name for p in path.parent.iterdir()] == ["ui.json"]
def test_the_ui_file_is_only_written_on_change(tmp_path):
writer = button.UiWriter(tmp_path / "ui.json")
view = {"view": "idle", "active": False}
assert writer.write(view) is True
assert writer.write(dict(view)) is False
def test_an_unwritable_ui_file_is_survived_and_logged_once(tmp_path, caplog):
"""Ein Fehler am Taster darf niemals eine laufende Migration stoeren - und eine Zeile
je Schleifentakt wuerde das Journal fluten."""
blocker = tmp_path / "blocker"
blocker.write_text("kein Verzeichnis")
writer = button.UiWriter(blocker / "ui.json")
with caplog.at_level("WARNING", logger="ninlight-button"):
assert writer.write({"view": "menu", "active": True}) is False
assert writer.write({"view": "confirm", "active": True}) is False
assert len(caplog.records) == 1
# --- Der Taster ist nirgends Voraussetzung ------------------------------------------
def test_missing_gpiozero_logs_exactly_one_line_and_exits_zero(monkeypatch, caplog):
"""Dieselbe Zusage wie bei der Anzeige (dort seit fuenf Fixrunden): kein Taster ->
GENAU EINE Protokollzeile und sys.exit(0), kein Fehler - sonst startet systemd
(Restart=on-failure) den Dienst endlos neu und flutet das Journal."""
monkeypatch.setitem(sys.modules, "gpiozero", None)
with caplog.at_level("INFO", logger="ninlight-button"):
with pytest.raises(SystemExit) as excinfo:
button.main()
assert excinfo.value.code == 0
assert len(caplog.records) == 1
assert "gpiozero" in caplog.records[0].getMessage()
def test_a_busy_pin_is_treated_like_a_missing_button(monkeypatch, caplog):
""""GPIO busy" kommt je nach Pin-Factory als gpiozero-Fehler, als roher lgpio.error
oder als OSError - alles davon heisst "kein Taster", nicht "abstuerzen"."""
class _Busy(Exception):
pass
fake = types.SimpleNamespace(
Button=lambda *a, **k: (_ for _ in ()).throw(_Busy("GPIO busy")))
monkeypatch.setitem(sys.modules, "gpiozero", fake)
with caplog.at_level("INFO", logger="ninlight-button"):
with pytest.raises(SystemExit) as excinfo:
button.main()
assert excinfo.value.code == 0
assert len(caplog.records) == 1
def test_a_failing_cli_call_never_escapes(monkeypatch):
"""subprocess-Fehler (Binary fehlt, Timeout) duerfen den Dienst nicht beenden."""
def boom(argv, check=False, timeout=None):
raise OSError("kein CLI da")
actions = button.CliActions(runner=boom)
assert actions.run(button.action_cancel()) is False
def slow(argv, check=False, timeout=None):
raise subprocess.TimeoutExpired(argv, timeout)
assert button.CliActions(runner=slow).run(button.action_cancel()) is False
def test_an_unreadable_station_state_keeps_the_menu_closed(caplog):
"""Menueeintraege haengen am Stationszustand. Ist der nicht zu ermitteln, bleibt das
Menue zu, statt einen falschen Zustand zu behaupten - und der Dienst laeuft weiter."""
ui = button.ButtonUI(0.0, read_state=lambda: (_ for _ in ()).throw(RuntimeError("weg")))
with caplog.at_level("WARNING", logger="ninlight-button"):
ui.update(0.0, True)
assert ui.update(0.2, False) is None
assert ui.view(0.3) == {"view": "idle", "active": False}
+120 -3
View File
@@ -1402,14 +1402,20 @@ class _RecordingPanel:
self.awake_s += end_t - self._awake_since self.awake_s += end_t - self._awake_since
def _drive_daemon(state_at, duration_s: float) -> _RecordingPanel: def _drive_daemon(state_at, duration_s: float, ui_at=None) -> _RecordingPanel:
"""Laesst das echte display.run_daemon() `duration_s` simulierte Sekunden laufen. """Laesst das echte display.run_daemon() `duration_s` simulierte Sekunden laufen.
`state_at(t)` liefert den Zustand, den build_state() zum Zeitpunkt t liefern wuerde.""" `state_at(t)` liefert den Zustand, den build_state() zum Zeitpunkt t liefern wuerde,
`ui_at(t)` den Bedienzustand, den read_ui() dort liefern wuerde (Vorgabe: keiner -
dann verhaelt sich die Schleife wie ohne Taster)."""
clock = _FakeClock(duration_s) clock = _FakeClock(duration_s)
panel = _RecordingPanel(clock) panel = _RecordingPanel(clock)
if ui_at is None:
def ui_at(t):
return {}
try: try:
display.run_daemon("V3", clock=clock, panel=panel, display.run_daemon("V3", clock=clock, panel=panel,
read_state=lambda: state_at(clock.monotonic())) read_state=lambda: state_at(clock.monotonic()),
read_ui=lambda: ui_at(clock.monotonic()))
except _StopDaemon: except _StopDaemon:
pass pass
panel._finish(clock.monotonic()) panel._finish(clock.monotonic())
@@ -1570,3 +1576,114 @@ def test_planner_draws_fully_on_mode_change():
planner.record_draw(0.0, ["RETTUNG"], "recovery", was_full=True) planner.record_draw(0.0, ["RETTUNG"], "recovery", was_full=True)
assert planner.decide(5.0, ["MIGRATION"], "migration") == "full" assert planner.decide(5.0, ["MIGRATION"], "migration") == "full"
# --- Kopplung an den Tasterdienst ----------------------------------------------------
#
# Der Tasterdienst (tools/station/button/button.py) fasst den SPI-Bus NICHT an - zwei
# Prozesse auf denselben GPIO-/SPI-Leitungen ergeben "lgpio.error: 'GPIO busy'". Er
# schreibt stattdessen seinen Bedienzustand in eine Datei, die hier gelesen und gezeichnet
# wird. Geprueft wird deshalb beides: dass ein Bedienzustand das Statusbild ersetzt, und
# dass die ECHTE Schleife ihn schnell genug aufgreift, um bedienbar zu sein.
_MENU_UI = {"view": "menu", "title": "MENUE", "hero": "> Station aus",
"details": ["Anzeige auffrischen", "Herunterfahren"], "active": True}
def test_the_ui_file_lives_outside_the_stations_runtime_directory():
"""Nicht in /run/ninlight-station: systemd raeumt dieses RuntimeDirectory beim
regulaeren Stop der Hauptunit auf - also ausgerechnet im Zustand "Station aus", in dem
das Menue "Rettung starten"/"Migration starten" anbietet."""
assert str(display.UI_FILE) == "/run/ninlight-button/ui.json"
assert not str(display.UI_FILE).startswith("/run/ninlight-station/")
def test_a_running_operation_replaces_the_status_screen():
lines = render.lines({"mode": "recovery", "ssid": "ninlight-recovery", "served": 7,
"recent": [], "ui": _MENU_UI})
assert lines == ["MENUE", "> Station aus", "Anzeige auffrischen", "Herunterfahren"]
assert not any("ninlight-recovery" in l for l in lines)
def test_the_menu_uses_the_same_roles_as_the_status_screen():
"""Kopfbalken, grosse Zeile, Beiwerk - damit Schriftgroessen, Breitenkuerzung und
unterer Rand dieselben bleiben wie im Statusbild."""
roles = [role for _text, role in render.blocks({"mode": "recovery", "ui": _MENU_UI})]
assert roles[:2] == [render.ROLE_HEADER, render.ROLE_HERO]
assert set(roles[2:]) <= {render.ROLE_DETAIL}
def test_the_menu_is_actually_drawn_on_the_panel():
image = render.compose({"mode": "recovery", "ssid": "ninlight-recovery",
"ui": _MENU_UI})
reference = render.compose({"mode": "recovery", "ssid": "ninlight-recovery"})
assert list(image.getdata()) != list(reference.getdata())
assert _ink_box(image) is not None
def test_a_broken_ui_file_does_not_break_the_display():
"""Die Datei kommt aus einem fremden Prozess, der abgestuerzt, aelter oder neuer sein
kann - genau wie status.json darf sie die Anzeige nie zum Absturz bringen."""
lines = render.lines({"mode": "recovery", "ui": {"view": "menu", "title": 17,
"hero": None, "details": "nope"}})
assert lines == ["Bedienung"]
render.compose({"mode": "recovery", "ui": {"view": "menu"}})
def test_read_ui_ignores_a_missing_broken_or_idle_file(tmp_path):
path = tmp_path / "ui.json"
assert display.read_ui(path) == {}
path.write_text("{kein json")
assert display.read_ui(path) == {}
path.write_text(json.dumps({"view": "idle", "active": False}))
assert display.read_ui(path) == {}
# Auch ein Menue, das sich selbst nicht mehr als laufend meldet, zaehlt nicht - so
# kann ein stehengebliebener Bedienzustand das Statusbild nicht dauerhaft verdecken.
path.write_text(json.dumps({"view": "menu", "active": False}))
assert display.read_ui(path) == {}
path.write_text(json.dumps(_MENU_UI))
assert display.read_ui(path)["hero"] == "> Station aus"
def test_the_daemon_shows_an_operation_within_half_a_second():
"""Solange eine Bedienung laeuft, muss die Anzeige schneller nachsehen als in ihrem
ueblichen 5-Sekunden-Takt - sonst ist das Menue unbenutzbar."""
panel = _drive_daemon(lambda t: {"mode": "recovery", "served": 0, "recent": []},
duration_s=20.0,
ui_at=lambda t: _MENU_UI if t >= 3.0 else {})
drawn = [t for t, name in panel.calls if name in ("partial", "full", "wake-full")
and t >= 3.0]
assert drawn, "das Menue wurde ueberhaupt nicht gezeichnet"
assert drawn[0] <= 3.5
def test_a_keypress_costs_the_panel_no_full_refresh():
"""Alterungsschutz: Vollaufbauten sind teuer, und das Panel soll nicht bei jedem
Tastendruck einen bekommen. Der Modus bleibt ueber eine Bedienung hinweg derselbe,
also kommt jeder Bedienschritt als Teilaufbau."""
views = [dict(_MENU_UI, hero="> Eintrag {}".format(i)) for i in range(4)]
def ui_at(t):
if t < 3.0:
return {}
return views[min(int((t - 3.0) // 2), len(views) - 1)]
panel = _drive_daemon(lambda t: {"mode": "recovery", "served": 0, "recent": []},
duration_s=12.0, ui_at=ui_at)
assert len(_times(panel, "partial")) >= 4
# Genau einer: der allererste Bildaufbau nach dem Start.
assert _times(panel, "full") == [0.0]
def test_the_expensive_station_state_is_still_only_read_every_five_seconds():
"""Der schnelle Takt gilt NUR fuer die kleine Bedienzustandsdatei. build_state()
forkt fuer "systemctl is-active" - das zehnmal so oft zu tun waere auf einem
ARMv6-Kern spuerbar und voellig unnoetig."""
calls = []
def state_at(t):
calls.append(t)
return {"mode": "recovery", "served": 0, "recent": []}
_drive_daemon(state_at, duration_s=10.0)
assert calls == [0.0, 5.0, 10.0]