diff --git a/tools/station/README.md b/tools/station/README.md index fb6394f..4a61299 100644 --- a/tools/station/README.md +++ b/tools/station/README.md @@ -25,6 +25,41 @@ würde dort nur mit einer Fehlermeldung enden. ssh ninlight-station status Zustand und letzte Auslieferungen ssh 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 Build-Parametern `NINLIGHT_RECOVERY_SSID` und `NINLIGHT_RECOVERY_PASSWORD` aus `config.local.cmake` passen. Weicht eines ab, findet die Lampe die Station nicht. diff --git a/tools/station/button/button.py b/tools/station/button/button.py new file mode 100644 index 0000000..6d06ec3 --- /dev/null +++ b/tools/station/button/button.py @@ -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() diff --git a/tools/station/display/display.py b/tools/station/display/display.py index d32775b..a8765a1 100644 --- a/tools/station/display/display.py +++ b/tools/station/display/display.py @@ -51,8 +51,27 @@ log = logging.getLogger("ninlight-display") RUN_DIR = Path(os.environ.get("NINLIGHT_RUN_DIR", "/run/ninlight-station")) CONF_PATH = Path(os.environ.get("NINLIGHT_STATION_CONF", "/etc/ninlight-station.conf")) SERVICE_NAME = os.environ.get("NINLIGHT_STATION_SERVICE", "ninlight-station") +# Bedienzustand des Tasterdienstes (tools/station/button/button.py). NUR GELESEN: der +# Tasterdienst fasst den SPI-Bus nicht an und dieser Dienst kein GPIO23 - zwei Prozesse +# auf denselben Leitungen enden in "lgpio.error: 'GPIO busy'". Die Datei liegt im +# RuntimeDirectory DES TASTERDIENSTES (nicht in RUN_DIR): RUN_DIR ist das +# RuntimeDirectory der Hauptunit und verschwindet beim regulaeren "off" - also +# ausgerechnet dann, wenn das Menue "Rettung starten"/"Migration starten" anbietet. +UI_FILE = Path(os.environ.get("NINLIGHT_UI_FILE", "/run/ninlight-button/ui.json")) POLL_INTERVAL_S = 5 +# Solange eine Bedienung laeuft, waere ein 5-Sekunden-Takt unbenutzbar (ein Tastendruck +# bis zu 5 s ohne Reaktion). Die Schleife tickt deshalb in diesem Takt und sieht bei +# JEDEM Tick in der Bedienzustandsdatei nach - auch im Ruhezustand, sonst waere +# ausgerechnet der erste Druck (der das Menue oeffnet) der langsamste. Das kostet eine +# kleine Dateilesung je halber Sekunde; der teure Teil des Zustands (build_state() mit +# seinem "systemctl is-active"-Fork) bleibt beim 5-Sekunden-Takt. +# +# Am Panel gemessen kommt zu dieser halben Sekunde noch der Teilaufbau (rund 0,3 s) - +# also grob 0,5-0,8 s je Druck. Das ist spuerbar, aber bedienbar: die Geste ist +# "druecken und hinsehen", nicht "tippen wie auf einer Tastatur", und die Haltegesten +# (1,5/3/10 s) liegen alle weit ueber dieser Verzoegerung. +UI_POLL_INTERVAL_S = 0.5 # E-Paper altert mit jedem Bildaufbau - ein Vollaufbau nur bei Moduswechsel und # mindestens alle 30 Minuten, sonst Teilaktualisierung (siehe Modul-Docstring der # Panel-Treiber: haeufige Teilaktualisierungen ohne gelegentlichen Vollaufbau lassen @@ -220,6 +239,21 @@ def build_state(run_dir: Path = RUN_DIR, conf_path: Path = CONF_PATH, return state +def read_ui(ui_path: Path = UI_FILE) -> dict: + """Der Bedienzustand des Tasterdienstes, oder {} wenn gerade keine Bedienung laeuft + (Datei fehlt, ist unlesbar, halb geschrieben, aus einer aelteren Fassung oder meldet + ausdruecklich "idle"). + + Jedes Feld wird defensiv gelesen: die Datei kommt aus einem FREMDEN Prozess, der + abgestuerzt, aelter oder neuer sein kann - ein Fehler hier darf die Anzeige nicht + aufhalten und schon gar nicht eine Migration stoeren.""" + data = _read_json(ui_path) + view = data.get("view") + if not data.get("active") or not isinstance(view, str) or view == "idle": + return {} + return data + + def _load_epd_module(revision: str): """Laedt den passenden Panel-Treiber. Nicht ueber ein Dict aus Modulnamen, damit ein statischer Import-Checker (und ein Mensch, der `grep -n epd2in13` tippt) beide @@ -485,30 +519,48 @@ def run_clear(revision: str) -> None: def run_daemon(revision: str, clock=time, panel: "Panel | None" = None, - read_state=build_state) -> None: - """Endlosschleife: alle POLL_INTERVAL_S Sekunden den Zustand einlesen und die von - RefreshPlanner entschiedene Aktion ausfuehren. Hier steht bewusst KEINE Zeitlogik mehr - (Fixrunde 4) - die steckt vollstaendig in RefreshPlanner und ist dort ohne Schleife, - ohne time.sleep() und ohne Hardware pruefbar. + read_state=build_state, read_ui=read_ui) -> None: + """Endlosschleife: alle UI_POLL_INTERVAL_S Sekunden den Bedienzustand und alle + POLL_INTERVAL_S Sekunden den Stationszustand einlesen und die von RefreshPlanner + entschiedene Aktion ausfuehren. Hier steht bewusst KEINE Zeitlogik zur Panelalterung + mehr (Fixrunde 4) - die steckt vollstaendig in RefreshPlanner und ist dort ohne + Schleife, ohne time.sleep() und ohne Hardware pruefbar. - `clock` (braucht monotonic() und sleep()), `panel` und `read_state` 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().""" + Die zwei Takte: der teure Teil (build_state() forkt fuer "systemctl is-active") + bleibt beim 5-Sekunden-Takt, der billige (eine kleine Datei lesen) laeuft in + UI_POLL_INTERVAL_S, damit eine Bedienung am Taster ueberhaupt benutzbar ist. Ein + laufender Bedienzustand ERSETZT das Statusbild (render.blocks() zweigt dafuer ganz + vorne ab); der Modus bleibt derselbe, sodass das Bild ueber Teilaufbauten kommt und + ein Tastendruck dem Panel keinen Vollaufbau kostet. + + `clock` (braucht monotonic() und sleep()), `panel`, `read_state` und `read_ui` sind + einschleusbar, damit die VERDRAHTUNG hier - welche Aktion welche Panel-Methode ruft, + dass die Selbst-Aufwertung von draw_partial() zurueckgemeldet wird, dass das + Einschlafen zurueckgemeldet wird, und die Reihenfolge decide -> ausfuehren -> + zurueckmelden - selbst pruefbar ist und nicht nur eine Kopie davon (Fixrunde 5, N2: + alle drei Mutationen an genau diesen Zeilen blieben vorher gruen, weil der Test eine + nachgebaute Schleife statt dieser hier trieb). Im Betrieb bleiben die Vorgaben: echte + Uhr, echtes Panel, echtes build_state()/read_ui().""" if panel is None: panel = Panel(revision) panel.clear() planner = RefreshPlanner(clock.monotonic()) + base_state = read_state() + last_state_poll = clock.monotonic() 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) mode = state.get("mode") - now = clock.monotonic() action = planner.decide(now, current_lines, mode) if action == "full": @@ -523,7 +575,7 @@ def run_daemon(revision: str, clock=time, panel: "Panel | None" = None, panel.sleep() planner.record_sleep() - clock.sleep(POLL_INTERVAL_S) + clock.sleep(UI_POLL_INTERVAL_S) def main() -> None: diff --git a/tools/station/display/render.py b/tools/station/display/render.py index fb9bc84..ae678a8 100644 --- a/tools/station/display/render.py +++ b/tools/station/display/render.py @@ -186,6 +186,16 @@ def blocks(state: dict) -> list[tuple[str, str]]: Bild - die Rolle ist ein reiner Bezeichner, die zugehoerigen Schriftgroessen kennt ausschliesslich compose().""" 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") out: list[tuple[str, str]] = [] @@ -311,6 +321,36 @@ def blocks(state: dict) -> list[tuple[str, str]]: 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]: """Nur die Texte aus blocks(), ohne Rollen - unveraenderte Schnittstelle fuer display.run_daemon(), das damit erkennt, ob sich der Inhalt geaendert hat (die Rolle diff --git a/tools/station/ninlight-button.service b/tools/station/ninlight-button.service new file mode 100644 index 0000000..a198a0e --- /dev/null +++ b/tools/station/ninlight-button.service @@ -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 diff --git a/tools/station/tests/test_button.py b/tools/station/tests/test_button.py new file mode 100644 index 0000000..1322c5b --- /dev/null +++ b/tools/station/tests/test_button.py @@ -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} diff --git a/tools/station/tests/test_render.py b/tools/station/tests/test_render.py index 57951da..6f83f15 100644 --- a/tools/station/tests/test_render.py +++ b/tools/station/tests/test_render.py @@ -1402,14 +1402,20 @@ class _RecordingPanel: 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. - `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) panel = _RecordingPanel(clock) + if ui_at is None: + def ui_at(t): + return {} try: 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: pass 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) 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]