#!/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: die beiden Migrationseintraege sind GETRENNT und namentlich unterscheidbar. "Migration: horchen" ruft "migrate on --listen" - der Zugangspunkt steht, aber es geht nichts raus, bis der Bediener eine erkannte Lampe freigibt; das ist der erste und naheliegende Eintrag. "Migration: AUTOMATIK" ruft "migrate on" ohne "--listen" und bekommt eine ANDERE, deutlichere Rueckfrage (siehe action_migrate_auto()): der Migrations-Zugangspunkt hat ein oeffentlich bekanntes Passwort (es steht in der alten Klyqa-Firmware), und im Automatikmodus wird jede alte Klyqa-Lampe in Funkreichweite ungefragt migriert, auch fremde. Die bequeme Geste bleibt damit die sichere, ohne dass der Automatikmodus - fuer eine ganze Kiste eigener Lampen am Stueck - nur noch ueber die Kommandozeile erreichbar waere, die im Zugangspunktbetrieb gerade nicht zur Verfuegung steht. 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). `confirm_title`/`confirm_details` machen die Rueckfrage bei Bedarf DEUTLICHER als die uebliche. Die Vorgabe ("RUECKFRAGE" + die zwei Bedienhinweise) reicht fuer eine Auswahl, deren Tragweite im Eintragsnamen steht ("Station aus", "Herunterfahren"). Sie reicht NICHT fuer den Automatikmodus: was dort passiert - jede alte Klyqa-Lampe in Funkreichweite wird ungefragt migriert, auch fremde - steht in keinem Eintragsnamen, und ein blosses "Wirklich?" fragt nach etwas, das der Bediener in diesem Moment gar nicht vor Augen hat.""" def __init__(self, label: str, argv=(), kind: str = "cli", confirm: bool = True, done_text: str | None = None, confirm_title: str = "RUECKFRAGE", confirm_details=None) -> None: self.label = label self.argv = tuple(argv) self.kind = kind self.confirm = confirm self.done_text = done_text or label self.confirm_title = confirm_title self.confirm_details = list(confirm_details) if confirm_details else None 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_migrate_listen() -> Action: """Der naheliegende Weg in die Migration: Zugangspunkt an, aber es geht NICHTS raus, bis der Bediener eine erkannte Lampe freigibt ("arm"). Deshalb steht dieser Eintrag vor dem Automatikmodus und traegt die uebliche Rueckfrage.""" return Action("Migration: horchen", ("migrate", "on", "--listen"), done_text="Migration: horcht") def action_migrate_auto() -> Action: """Der Automatikmodus - fuer die eigene Kiste Lampen am Stueck, ohne jede einzeln freizugeben. Er steht hier, weil das Verabreden ueber die Kommandozeile genau dann nicht geht, wenn man ihn braucht: "on" macht wlan0 zum Zugangspunkt und reisst jede SSH-Sitzung ueber WLAN ab. Seine Rueckfrage ist ABSICHTLICH eine andere als die der uebrigen Eintraege. Der Migrations-Zugangspunkt hat ein oeffentlich bekanntes Passwort (es steht in der alten Klyqa-Firmware), und im Automatikmodus wird jede alte Klyqa-Lampe in Funkreichweite ungefragt migriert - auch die des Nachbarn, der davon nichts weiss und dessen Lampe danach nicht mehr mit seiner App spricht. Ein "Wirklich?" fragt nach etwas, das der Bediener in diesem Moment gar nicht vor Augen hat; deshalb steht die Folge in der Rueckfrage selbst.""" return Action("Migration: AUTOMATIK", ("migrate", "on"), done_text="Migration: automatisch", confirm_title="ACHTUNG", confirm_details=["Flasht JEDE alte Klyqa-Lampe", "in Reichweite - auch fremde"]) 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"), # Reihenfolge ist Absicht: der horchende Modus zuerst, weil er der # naheliegende und der harmlose ist - wer nur blaettert und irgendwo # loslaesst, landet nicht im Automatikmodus. action_migrate_listen(), action_migrate_auto(), 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: # Die Bedienhinweise stehen IMMER und IMMER ZULETZT - eine Rueckfrage, bei # der man raten muss, wie man ablehnt, ist keine. Eine eigene Warnung des # Eintrags (confirm_details) kommt davor, damit sie zuerst gelesen wird. details = list(self.pending.confirm_details or []) details += ["halten (1,5 s): ja", "kurz: nein"] return _view("confirm", self.pending.confirm_title, self.pending.label, details) 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()