diff --git a/tools/station/display/HERKUNFT.md b/tools/station/display/HERKUNFT.md new file mode 100644 index 0000000..d35b173 --- /dev/null +++ b/tools/station/display/HERKUNFT.md @@ -0,0 +1,46 @@ +# Herkunft der Panel-Treiber + +`epd2in13_V2.py` und `epd2in13_V3.py` sind unveraendert aus dem offiziellen +Waveshare-Repository uebernommen: + +- Quelle: https://github.com/waveshareteam/e-Paper +- Pfad: `RaspberryPi_JetsonNano/python/lib/waveshare_epd/epd2in13_V2.py` bzw. + `.../epd2in13_V3.py` +- Commit: `85802dee4b938212b0ae069a34492c54a3d99878` (2023-02-15), abgerufen am + 2026-09-13 per `git log --follow` auf diesen beiden Dateien im Repository. +- `epd2in13_V2.py`: Treiber-Version laut Dateikopf V4.0 (2019-06-20). +- `epd2in13_V3.py`: Treiber-Version laut Dateikopf V1.2 (2022-08-09). + +## Was geaendert wurde + +Ausschliesslich die Importzeile: + +``` +from . import epdconfig +``` + +wurde zu + +``` +import epdconfig +``` + +weil `display/` in diesem Repository kein Python-Package ist (kein `__init__.py`, +`sys.path`-Import wie bei `products.py` in `server.py`), sondern flach importiert wird - +sowohl von `display.py` als auch von `tools/station/tests/test_render.py`. Ein relativer +Import wuerde in dieser Umgebung mit `ImportError: attempted relative import with no known +parent package` fehlschlagen. + +Init-Sequenz, LUT-Tabellen, Timings und der Urheberrechtsvermerk der Originaldateien +bleiben unangetastet. Ein kurzer Herkunftsvermerk mit denselben Angaben wie oben steht +zusaetzlich im Kopf jeder der beiden Dateien. + +## Was NICHT uebernommen wurde + +`epdconfig.py` in diesem Verzeichnis ist eine eigene Neuimplementierung (siehe deren +Modul-Docstring), keine Uebernahme aus dem Waveshare-Repository: Waveshares eigenes +`epdconfig.py` fuer den Raspberry Pi setzt `RPi.GPIO` voraus, das auf Debian 13 (trixie) +mit Python 3.13 nicht mehr baut/importiert. Unser `epdconfig.py` implementiert dieselbe +von den beiden Treibern erwartete Schnittstelle (Pin-Konstanten, `digital_write`, +`digital_read`, `delay_ms`, `spi_writebyte`, `spi_writebyte2`, `module_init`, +`module_exit`) gegen `gpiozero`/`spidev`, die auf dem Ziel-Pi vorhanden sind. diff --git a/tools/station/display/display.py b/tools/station/display/display.py new file mode 100644 index 0000000..8db5677 --- /dev/null +++ b/tools/station/display/display.py @@ -0,0 +1,286 @@ +#!/usr/bin/env python3 +"""Daemon fuer die E-Paper-Anzeige der ninlight-Station: alle 5 s status.json, +control.json und die Konfigurationsdatei lesen, daraus einen Text zeichnen und - nur bei +Aenderung - aufs Panel bringen. + +Panel: Waveshare 2,13" (250x122 px), SPI. Ob das verbaute Panel Revision V2 oder V3 ist, +laesst sich am Geraet selbst nicht auslesen - die Revision kommt deshalb aus der +Umgebungsvariable EPD_REVISION (Vorgabe "V3", von setup_pi.sh in +/etc/ninlight-station.conf geschrieben und von der systemd-Unit als Environment= +weitergereicht). + +So bestimmst du die richtige Revision empirisch: erst + EPD_REVISION=V3 python3 display.py --test +laufen lassen und das Panel ansehen. Zeigt es ein sauberes, seitenrichtiges Testbild ohne +Geisterbilder/Streifen, ist V3 richtig. Zeigt es nichts, ein invertiertes/verschobenes Bild +oder haengt beim SWRESET (ReadBusy() kehrt nie zurueck), mit + EPD_REVISION=V2 python3 display.py --test +wiederholen. Zwischen den beiden Versuchen kurz die Stromversorgung des Panels trennen +(--clear reicht nicht immer, falsche Revision kann das Panel in einem Zwischenzustand +lassen). + +Die Anzeige ist nirgends Voraussetzung fuer die Migration: fehlt das Panel, SPI oder die +gpiozero/spidev-Bibliothek, faengt main() das als DisplayUnavailable ab, protokolliert +GENAU EINE Zeile und beendet sich mit sys.exit(0) - nicht mit einem Fehler, sonst wuerde +systemd (Restart=on-failure) den Dienst endlos neu starten und das Journal fluten. Ein +Display-Fehler darf niemals eine Migration verhindern. +""" +from __future__ import annotations + +import argparse +import json +import logging +import os +import subprocess +import sys +import time +from pathlib import Path + +# Flacher Import wie bei server.py/products.py: dieses Verzeichnis ist kein Package, +# display.py kann aber von ueberall aus gestartet werden (die systemd-Unit ruft es mit +# vollem Pfad auf). +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +import render # noqa: E402 +from epdconfig import DisplayUnavailable # noqa: E402 + +log = logging.getLogger("ninlight-display") + +# Dieselben Umgebungsvariablen wie tools/station/ninlight-station verwendet, damit Tests +# und Handbetrieb ohne echtes /run bzw. /etc auskommen. +RUN_DIR = Path(os.environ.get("NINLIGHT_RUN_DIR", "/run/ninlight-station")) +CONF_PATH = Path(os.environ.get("NINLIGHT_STATION_CONF", "/etc/ninlight-station.conf")) +SERVICE_NAME = os.environ.get("NINLIGHT_STATION_SERVICE", "ninlight-station") + +POLL_INTERVAL_S = 5 +# E-Paper altert mit jedem Bildaufbau - ein Vollaufbau nur bei Moduswechsel und +# mindestens alle 30 Minuten, sonst Teilaktualisierung (siehe Modul-Docstring der +# Panel-Treiber: haeufige Teilaktualisierungen ohne gelegentlichen Vollaufbau lassen +# Geisterbilder stehen). +FULL_REFRESH_INTERVAL_S = 30 * 60 + + +def _read_json(path: Path) -> dict: + """Geparster Inhalt von `path`, oder {} wenn er fehlt/unlesbar/kaputt/kein Objekt + ist - status.json und control.json koennen jederzeit in genau diesem Zustand sein + (frisch aufgesetzter Pi, Schreibfehler mitten im Schreiben).""" + try: + data = json.loads(path.read_text()) + except (OSError, ValueError): + return {} + return data if isinstance(data, dict) else {} + + +def _read_conf(path: Path) -> dict[str, str]: + """Minimaler Parser fuer /etc/ninlight-station.conf: KEY='VALUE'-Zeilen, wie + setup_pi.sh sie schreibt. Kein echtes Shell-Sourcing (das CLI braucht das fuer $- + und `-Escaping in Passwoertern, siehe dessen Kommentar - display.py liest hier nur + SSID/MIGRATE_SSID, reine Bezeichner ohne Sonderzeichen).""" + values: dict[str, str] = {} + try: + text = path.read_text() + except OSError: + return values + for raw_line in text.splitlines(): + line = raw_line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, _, value = line.partition("=") + key = key.strip() + value = value.strip() + if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'): + value = value[1:-1] + if key: + values[key] = value + return values + + +def _read_mode_file(run_dir: Path) -> str: + """Modus, in dem der Dienst laut RUN_DIR/mode tatsaechlich gestartet wurde. Fehlt die + Datei oder ist ihr Inhalt kein bekannter Modus, gilt "recovery" - derselbe Vorgabewert + wie in server.py:resolve_mode().""" + try: + value = (run_dir / "mode").read_text().strip() + except OSError: + return "recovery" + return value if value in ("recovery", "migration") else "recovery" + + +def _service_active(service_name: str) -> bool: + """Laeuft der Hauptdienst (server.py) gerade? "ninlight-station off" stoppt ihn und + faehrt den Zugangspunkt herunter, laesst status.json dabei aber unangetastet liegen - + ein reines Lesen von status.json wuerde nach "off" also faelschlich weiter den letzten + Modus vor dem Abschalten zeigen. Deshalb ist die Dienstaktivitaet selbst (nicht das + Vorhandensein von status.json) die Quelle fuer den Anzeigemodus "off".""" + try: + result = subprocess.run( + ["systemctl", "is-active", "--quiet", service_name], + timeout=5, check=False) + return result.returncode == 0 + except (OSError, subprocess.SubprocessError): + # systemctl selbst nicht erreichbar (z.B. Handlauf ohne systemd) - dann anhand + # der Statusdatei entscheiden statt "off" zu behaupten, ohne es zu wissen. + return True + + +def build_state(run_dir: Path = RUN_DIR, conf_path: Path = CONF_PATH, + service_name: str = SERVICE_NAME) -> dict: + """Baut den zusammengefuehrten Zustand, den render.lines()/compose() erwarten - + status.json (served/recent/last_seen), control.json (phase/armed_target, nur im + Migrationsmodus massgeblich), die SSID aus der Konfigurationsdatei und "off", wenn + der Hauptdienst gar nicht laeuft.""" + if not _service_active(service_name): + return {"mode": "off"} + + status = _read_json(run_dir / "status.json") + mode = status.get("mode") + if mode not in ("recovery", "migration"): + # status.json fehlt/ist kaputt/stammt (noch) nicht vom laufenden Prozess - + # RUN_DIR/mode ist das, was "ninlight-station on"/"migrate on" zuletzt + # geschrieben hat und ist unabhaengig von status.json verfuegbar. + mode = _read_mode_file(run_dir) + + conf = _read_conf(conf_path) + state: dict = {"mode": mode, "served": status.get("served"), "recent": status.get("recent")} + + if mode == "recovery": + state["ssid"] = conf.get("SSID") + else: # migration + state["ssid"] = conf.get("MIGRATE_SSID") + state["last_seen"] = status.get("last_seen") + # Die Phase kommt bewusst aus control.json, nicht aus status.json: server.py + # schreibt status.json nur beim Start und nach einer Auslieferung, waehrend + # "ninlight-station arm"/"listen" nur die Steuerdatei anfasst (Spec/CLI + # status_cmd folgt demselben Muster). Und bewusst nur HIER gelesen - im + # Rettungsmodus ignoriert server.py control.json ebenfalls, ein + # liegengebliebener Inhalt aus einer frueheren Migration darf dort nicht + # faelschlich "Phase: FREIGEGEBEN" zeigen. + control = _read_json(run_dir / "control.json") + phase = control.get("phase") + state["phase"] = phase if phase in ("auto", "listen", "armed") else "auto" + state["armed_target"] = control.get("armed_target") + + return state + + +def _load_epd_module(revision: str): + """Laedt den passenden Panel-Treiber. Nicht ueber ein Dict aus Modulnamen, damit ein + statischer Import-Checker (und ein Mensch, der `grep -n epd2in13` tippt) beide + moeglichen Importe findet.""" + if revision == "V2": + import epd2in13_V2 as epd_module + elif revision == "V3": + import epd2in13_V3 as epd_module + else: + raise DisplayUnavailable( + f"unbekannte EPD_REVISION '{revision}' - erwartet 'V2' oder 'V3'") + return epd_module + + +def _draw(epd, image, full: bool) -> None: + buf = epd.getbuffer(image) + if full: + # epd.init() laedt die volle LUT neu - noetig, weil displayPartial() unten immer + # die Teil-LUT setzt und die naechste display()-Auslieferung sonst mit der + # falschen Wellenform liefe. + epd.init() + epd.display(buf) + else: + epd.displayPartial(buf) + + +def run_test(revision: str) -> None: + """Zeichnet ein Testbild und beendet sich - siehe Modul-Docstring, wie man damit V2 + von V3 unterscheidet.""" + from PIL import Image, ImageDraw, ImageFont + + epd_module = _load_epd_module(revision) + epd = epd_module.EPD() + epd.init() + epd.Clear(0xFF) + + width, height = 250, 122 + image = Image.new("1", (width, height), 255) + draw = ImageDraw.Draw(image) + draw.rectangle((0, 0, width - 1, height - 1), outline=0) + draw.text((4, 4), "ninlight-display Test", font=ImageFont.load_default(), fill=0) + draw.text((4, 20), f"EPD_REVISION={revision}", font=ImageFont.load_default(), fill=0) + draw.line((4, 40, width - 4, 40), fill=0) + # Vier Kaestchen, abwechselnd gefuellt/leer - hilft, Kontrast/Inversion am + # angeschlossenen Panel auf einen Blick zu beurteilen. + for i in range(4): + x0 = 4 + i * 20 + draw.rectangle((x0, 50, x0 + 16, 90), outline=0, fill=0 if i % 2 == 0 else 255) + + epd.display(epd.getbuffer(image)) + log.info("Testbild gezeichnet (EPD_REVISION=%s)", revision) + + +def run_clear(revision: str) -> None: + epd_module = _load_epd_module(revision) + epd = epd_module.EPD() + epd.init() + epd.Clear(0xFF) + epd.sleep() + log.info("Panel geleert (EPD_REVISION=%s)", revision) + + +def run_daemon(revision: str) -> None: + epd_module = _load_epd_module(revision) + epd = epd_module.EPD() + epd.init() + epd.Clear(0xFF) + + last_lines: list[str] | None = None + last_mode: object = object() # garantiert ungleich jedem echten Modus beim ersten Lauf + last_full = time.monotonic() + + while True: + state = build_state() + current_lines = render.lines(state) + mode = state.get("mode") + now = time.monotonic() + need_full = (mode != last_mode) or (now - last_full >= FULL_REFRESH_INTERVAL_S) + + if current_lines != last_lines or need_full: + image = render.compose(state) + _draw(epd, image, full=need_full) + if need_full: + last_full = now + last_lines = current_lines + last_mode = mode + + time.sleep(POLL_INTERVAL_S) + + +def main() -> None: + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + + parser = argparse.ArgumentParser(description="ninlight E-Paper-Anzeige") + group = parser.add_mutually_exclusive_group() + group.add_argument("--test", action="store_true", help="Testbild zeichnen und beenden") + group.add_argument("--clear", action="store_true", help="Panel leeren und beenden") + args = parser.parse_args() + + revision = os.environ.get("EPD_REVISION", "V3") + + try: + if args.test: + run_test(revision) + elif args.clear: + run_clear(revision) + else: + run_daemon(revision) + except DisplayUnavailable as e: + # Genau eine Zeile, dann sys.exit(0) - siehe Modul-Docstring. Restart=on-failure + # (statt always) in der Unit greift hier nicht, weil 0 kein Fehlschlag ist; ein + # spaeter angestecktes Panel braucht also einen manuellen/zeitgesteuerten + # Neustart der Unit, nicht eine Neustartschleife. + log.info("E-Paper-Anzeige nicht verfuegbar, beende ohne Fehler: %s", e) + sys.exit(0) + except KeyboardInterrupt: + sys.exit(0) + + +if __name__ == "__main__": + main() diff --git a/tools/station/display/epd2in13_V2.py b/tools/station/display/epd2in13_V2.py new file mode 100644 index 0000000..ea49350 --- /dev/null +++ b/tools/station/display/epd2in13_V2.py @@ -0,0 +1,324 @@ +# ***************************************************************************** +# * | File : epd2in13_V2.py +# * | Author : Waveshare team +# * | Function : Electronic paper driver +# * | Info : +# *---------------- +# * | This version: V4.0 +# * | Date : 2019-06-20 +# # | Info : python demo +# ----------------------------------------------------------------------------- +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documnetation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS OR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +# + +# --- Herkunftsvermerk (ninlight-firmware) --------------------------------- +# Unveraendert uebernommen aus waveshareteam/e-Paper (RaspberryPi_JetsonNano/ +# python/lib/waveshare_epd/epd2in13_V2.py), Commit 85802dee (2023-02-15). +# Einzige Aenderung: 'from . import epdconfig' -> 'import epdconfig', damit das +# Modul flach importierbar ist (kein Package). Alles andere - Init-Sequenz, LUTs, +# Timings - unveraendert. Details: tools/station/display/HERKUNFT.md +# ----------------------------------------------------------------------------- + + +import logging +import epdconfig + +# Display resolution +EPD_WIDTH = 122 +EPD_HEIGHT = 250 + +logger = logging.getLogger(__name__) + +class EPD: + def __init__(self): + self.reset_pin = epdconfig.RST_PIN + self.dc_pin = epdconfig.DC_PIN + self.busy_pin = epdconfig.BUSY_PIN + self.cs_pin = epdconfig.CS_PIN + self.width = EPD_WIDTH + self.height = EPD_HEIGHT + + FULL_UPDATE = 0 + PART_UPDATE = 1 + lut_full_update= [ + 0x80,0x60,0x40,0x00,0x00,0x00,0x00, #LUT0: BB: VS 0 ~7 + 0x10,0x60,0x20,0x00,0x00,0x00,0x00, #LUT1: BW: VS 0 ~7 + 0x80,0x60,0x40,0x00,0x00,0x00,0x00, #LUT2: WB: VS 0 ~7 + 0x10,0x60,0x20,0x00,0x00,0x00,0x00, #LUT3: WW: VS 0 ~7 + 0x00,0x00,0x00,0x00,0x00,0x00,0x00, #LUT4: VCOM: VS 0 ~7 + + 0x03,0x03,0x00,0x00,0x02, # TP0 A~D RP0 + 0x09,0x09,0x00,0x00,0x02, # TP1 A~D RP1 + 0x03,0x03,0x00,0x00,0x02, # TP2 A~D RP2 + 0x00,0x00,0x00,0x00,0x00, # TP3 A~D RP3 + 0x00,0x00,0x00,0x00,0x00, # TP4 A~D RP4 + 0x00,0x00,0x00,0x00,0x00, # TP5 A~D RP5 + 0x00,0x00,0x00,0x00,0x00, # TP6 A~D RP6 + + 0x15,0x41,0xA8,0x32,0x30,0x0A, + ] + + lut_partial_update = [ #20 bytes + 0x00,0x00,0x00,0x00,0x00,0x00,0x00, #LUT0: BB: VS 0 ~7 + 0x80,0x00,0x00,0x00,0x00,0x00,0x00, #LUT1: BW: VS 0 ~7 + 0x40,0x00,0x00,0x00,0x00,0x00,0x00, #LUT2: WB: VS 0 ~7 + 0x00,0x00,0x00,0x00,0x00,0x00,0x00, #LUT3: WW: VS 0 ~7 + 0x00,0x00,0x00,0x00,0x00,0x00,0x00, #LUT4: VCOM: VS 0 ~7 + + 0x0A,0x00,0x00,0x00,0x00, # TP0 A~D RP0 + 0x00,0x00,0x00,0x00,0x00, # TP1 A~D RP1 + 0x00,0x00,0x00,0x00,0x00, # TP2 A~D RP2 + 0x00,0x00,0x00,0x00,0x00, # TP3 A~D RP3 + 0x00,0x00,0x00,0x00,0x00, # TP4 A~D RP4 + 0x00,0x00,0x00,0x00,0x00, # TP5 A~D RP5 + 0x00,0x00,0x00,0x00,0x00, # TP6 A~D RP6 + + 0x15,0x41,0xA8,0x32,0x30,0x0A, + ] + + # Hardware reset + def reset(self): + epdconfig.digital_write(self.reset_pin, 1) + epdconfig.delay_ms(200) + epdconfig.digital_write(self.reset_pin, 0) + epdconfig.delay_ms(5) + epdconfig.digital_write(self.reset_pin, 1) + epdconfig.delay_ms(200) + + def send_command(self, command): + epdconfig.digital_write(self.dc_pin, 0) + epdconfig.digital_write(self.cs_pin, 0) + epdconfig.spi_writebyte([command]) + epdconfig.digital_write(self.cs_pin, 1) + + def send_data(self, data): + epdconfig.digital_write(self.dc_pin, 1) + epdconfig.digital_write(self.cs_pin, 0) + epdconfig.spi_writebyte([data]) + epdconfig.digital_write(self.cs_pin, 1) + + # send a lot of data + def send_data2(self, data): + epdconfig.digital_write(self.dc_pin, 1) + epdconfig.digital_write(self.cs_pin, 0) + epdconfig.spi_writebyte2(data) + epdconfig.digital_write(self.cs_pin, 1) + + def ReadBusy(self): + while(epdconfig.digital_read(self.busy_pin) == 1): # 0: idle, 1: busy + epdconfig.delay_ms(100) + + def TurnOnDisplay(self): + self.send_command(0x22) + self.send_data(0xC7) + self.send_command(0x20) + self.ReadBusy() + + def TurnOnDisplayPart(self): + self.send_command(0x22) + self.send_data(0x0c) + self.send_command(0x20) + self.ReadBusy() + + def init(self, update): + if (epdconfig.module_init() != 0): + return -1 + # EPD hardware init start + self.reset() + if(update == self.FULL_UPDATE): + self.ReadBusy() + self.send_command(0x12) # soft reset + self.ReadBusy() + + self.send_command(0x74) #set analog block control + self.send_data(0x54) + self.send_command(0x7E) #set digital block control + self.send_data(0x3B) + + self.send_command(0x01) #Driver output control + self.send_data(0xF9) + self.send_data(0x00) + self.send_data(0x00) + + self.send_command(0x11) #data entry mode + self.send_data(0x01) + + self.send_command(0x44) #set Ram-X address start/end position + self.send_data(0x00) + self.send_data(0x0F) #0x0C-->(15+1)*8=128 + + self.send_command(0x45) #set Ram-Y address start/end position + self.send_data(0xF9) #0xF9-->(249+1)=250 + self.send_data(0x00) + self.send_data(0x00) + self.send_data(0x00) + + self.send_command(0x3C) #BorderWavefrom + self.send_data(0x03) + + self.send_command(0x2C) #VCOM Voltage + self.send_data(0x55) # + + self.send_command(0x03) + self.send_data(self.lut_full_update[70]) + + self.send_command(0x04) # + self.send_data(self.lut_full_update[71]) + self.send_data(self.lut_full_update[72]) + self.send_data(self.lut_full_update[73]) + + self.send_command(0x3A) #Dummy Line + self.send_data(self.lut_full_update[74]) + self.send_command(0x3B) #Gate time + self.send_data(self.lut_full_update[75]) + + self.send_command(0x32) + for count in range(70): + self.send_data(self.lut_full_update[count]) + + self.send_command(0x4E) # set RAM x address count to 0 + self.send_data(0x00) + self.send_command(0x4F) # set RAM y address count to 0X127 + self.send_data(0xF9) + self.send_data(0x00) + self.ReadBusy() + else: + self.send_command(0x2C) #VCOM Voltage + self.send_data(0x26) + + self.ReadBusy() + + self.send_command(0x32) + for count in range(70): + self.send_data(self.lut_partial_update[count]) + + self.send_command(0x37) + self.send_data(0x00) + self.send_data(0x00) + self.send_data(0x00) + self.send_data(0x00) + self.send_data(0x40) + self.send_data(0x00) + self.send_data(0x00) + + self.send_command(0x22) + self.send_data(0xC0) + self.send_command(0x20) + self.ReadBusy() + + self.send_command(0x3C) #BorderWavefrom + self.send_data(0x01) + return 0 + + def getbuffer(self, image): + if self.width%8 == 0: + linewidth = int(self.width/8) + else: + linewidth = int(self.width/8) + 1 + + buf = [0xFF] * (linewidth * self.height) + image_monocolor = image.convert('1') + imwidth, imheight = image_monocolor.size + pixels = image_monocolor.load() + + if(imwidth == self.width and imheight == self.height): + logger.debug("Vertical") + for y in range(imheight): + for x in range(imwidth): + if pixels[x, y] == 0: + x = imwidth - x + buf[int(x / 8) + y * linewidth] &= ~(0x80 >> (x % 8)) + elif(imwidth == self.height and imheight == self.width): + logger.debug("Horizontal") + for y in range(imheight): + for x in range(imwidth): + newx = y + newy = self.height - x - 1 + if pixels[x, y] == 0: + newy = imwidth - newy - 1 + buf[int(newx / 8) + newy*linewidth] &= ~(0x80 >> (y % 8)) + return buf + + + def display(self, image): + self.send_command(0x24) + self.send_data2(image) + self.TurnOnDisplay() + + def displayPartial(self, image): + if self.width%8 == 0: + linewidth = int(self.width/8) + else: + linewidth = int(self.width/8) + 1 + + buf = [0x00] * self.height * linewidth + for j in range(0, self.height): + for i in range(0, linewidth): + buf[i + j * linewidth] = ~image[i + j * linewidth] + + self.send_command(0x24) + self.send_data2(image) + + + self.send_command(0x26) + self.send_data2(buf) + self.TurnOnDisplayPart() + + def displayPartBaseImage(self, image): + self.send_command(0x24) + self.send_data2(image) + + self.send_command(0x26) + self.send_data2(image) + self.TurnOnDisplay() + + def Clear(self, color=0xFF): + if self.width%8 == 0: + linewidth = int(self.width/8) + else: + linewidth = int(self.width/8) + 1 + # logger.debug(linewidth) + + buf = [0x00] * self.height * linewidth + for j in range(0, self.height): + for i in range(0, linewidth): + buf[i + j * linewidth] = color + + self.send_command(0x24) + self.send_data2(buf) + + # self.send_command(0x26) + # for j in range(0, self.height): + # for i in range(0, linewidth): + # self.send_data(color) + + self.TurnOnDisplay() + + def sleep(self): + # self.send_command(0x22) #POWER OFF + # self.send_data(0xC3) + # self.send_command(0x20) + + self.send_command(0x10) #enter deep sleep + self.send_data(0x03) + epdconfig.delay_ms(2000) + epdconfig.module_exit() + +### END OF FILE ### + diff --git a/tools/station/display/epd2in13_V3.py b/tools/station/display/epd2in13_V3.py new file mode 100644 index 0000000..45734ee --- /dev/null +++ b/tools/station/display/epd2in13_V3.py @@ -0,0 +1,395 @@ +# ***************************************************************************** +# * | File : epd2in13_V3.py +# * | Author : Waveshare team +# * | Function : Electronic paper driver +# * | Info : +# *---------------- +# * | This version: V1.2 +# * | Date : 2022-08-9 +# # | Info : python demo +# ----------------------------------------------------------------------------- +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documnetation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS OR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +# + +# --- Herkunftsvermerk (ninlight-firmware) --------------------------------- +# Unveraendert uebernommen aus waveshareteam/e-Paper (RaspberryPi_JetsonNano/ +# python/lib/waveshare_epd/epd2in13_V3.py), Commit 85802dee (2023-02-15). +# Einzige Aenderung: 'from . import epdconfig' -> 'import epdconfig', damit das +# Modul flach importierbar ist (kein Package). Alles andere - Init-Sequenz, LUTs, +# Timings - unveraendert. Details: tools/station/display/HERKUNFT.md +# ----------------------------------------------------------------------------- + + +import logging +import epdconfig + +# Display resolution +EPD_WIDTH = 122 +EPD_HEIGHT = 250 + +logger = logging.getLogger(__name__) + +class EPD: + def __init__(self): + self.reset_pin = epdconfig.RST_PIN + self.dc_pin = epdconfig.DC_PIN + self.busy_pin = epdconfig.BUSY_PIN + self.cs_pin = epdconfig.CS_PIN + self.width = EPD_WIDTH + self.height = EPD_HEIGHT + + lut_partial_update= [ + 0x0,0x40,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x80,0x80,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x40,0x40,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x80,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x14,0x0,0x0,0x0,0x0,0x0,0x0, + 0x1,0x0,0x0,0x0,0x0,0x0,0x0, + 0x1,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x22,0x22,0x22,0x22,0x22,0x22,0x0,0x0,0x0, + 0x22,0x17,0x41,0x00,0x32,0x36, + ] + + lut_full_update = [ + 0x80,0x4A,0x40,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x40,0x4A,0x80,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x80,0x4A,0x40,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x40,0x4A,0x80,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0xF,0x0,0x0,0x0,0x0,0x0,0x0, + 0xF,0x0,0x0,0xF,0x0,0x0,0x2, + 0xF,0x0,0x0,0x0,0x0,0x0,0x0, + 0x1,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x0,0x0,0x0,0x0,0x0,0x0,0x0, + 0x22,0x22,0x22,0x22,0x22,0x22,0x0,0x0,0x0, + 0x22,0x17,0x41,0x0,0x32,0x36, + ] + + ''' + function :Hardware reset + parameter: + ''' + def reset(self): + epdconfig.digital_write(self.reset_pin, 1) + epdconfig.delay_ms(20) + epdconfig.digital_write(self.reset_pin, 0) + epdconfig.delay_ms(2) + epdconfig.digital_write(self.reset_pin, 1) + epdconfig.delay_ms(20) + + ''' + function :send command + parameter: + command : Command register + ''' + def send_command(self, command): + epdconfig.digital_write(self.dc_pin, 0) + epdconfig.digital_write(self.cs_pin, 0) + epdconfig.spi_writebyte([command]) + epdconfig.digital_write(self.cs_pin, 1) + + ''' + function :send data + parameter: + data : Write data + ''' + def send_data(self, data): + epdconfig.digital_write(self.dc_pin, 1) + epdconfig.digital_write(self.cs_pin, 0) + epdconfig.spi_writebyte([data]) + epdconfig.digital_write(self.cs_pin, 1) + + # send a lot of data + def send_data2(self, data): + epdconfig.digital_write(self.dc_pin, 1) + epdconfig.digital_write(self.cs_pin, 0) + epdconfig.spi_writebyte2(data) + epdconfig.digital_write(self.cs_pin, 1) + + ''' + function :Wait until the busy_pin goes LOW + parameter: + ''' + def ReadBusy(self): + logger.debug("e-Paper busy") + while(epdconfig.digital_read(self.busy_pin) == 1): # 0: idle, 1: busy + epdconfig.delay_ms(10) + logger.debug("e-Paper busy release") + + ''' + function : Turn On Display + parameter: + ''' + def TurnOnDisplay(self): + self.send_command(0x22) # Display Update Control + self.send_data(0xC7) + self.send_command(0x20) # Activate Display Update Sequence + self.ReadBusy() + + ''' + function : Turn On Display Part + parameter: + ''' + def TurnOnDisplayPart(self): + self.send_command(0x22) # Display Update Control + self.send_data(0x0f) # fast:0x0c, quality:0x0f, 0xcf + self.send_command(0x20) # Activate Display Update Sequence + self.ReadBusy() + + ''' + function : Set lut + parameter: + lut : lut data + ''' + def Lut(self, lut): + self.send_command(0x32) + for i in range(0, 153): + self.send_data(lut[i]) + self.ReadBusy() + + ''' + function : Send lut data and configuration + parameter: + lut : lut data + ''' + def SetLut(self, lut): + self.Lut(lut) + self.send_command(0x3f) + self.send_data(lut[153]) + self.send_command(0x03) # gate voltage + self.send_data(lut[154]) + self.send_command(0x04) # source voltage + self.send_data(lut[155]) # VSH + self.send_data(lut[156]) # VSH2 + self.send_data(lut[157]) # VSL + self.send_command(0x2c) # VCOM + self.send_data(lut[158]) + + ''' + function : Setting the display window + parameter: + xstart : X-axis starting position + ystart : Y-axis starting position + xend : End position of X-axis + yend : End position of Y-axis + ''' + def SetWindow(self, x_start, y_start, x_end, y_end): + self.send_command(0x44) # SET_RAM_X_ADDRESS_START_END_POSITION + # x point must be the multiple of 8 or the last 3 bits will be ignored + self.send_data((x_start>>3) & 0xFF) + self.send_data((x_end>>3) & 0xFF) + + self.send_command(0x45) # SET_RAM_Y_ADDRESS_START_END_POSITION + self.send_data(y_start & 0xFF) + self.send_data((y_start >> 8) & 0xFF) + self.send_data(y_end & 0xFF) + self.send_data((y_end >> 8) & 0xFF) + + ''' + function : Set Cursor + parameter: + x : X-axis starting position + y : Y-axis starting position + ''' + def SetCursor(self, x, y): + self.send_command(0x4E) # SET_RAM_X_ADDRESS_COUNTER + # x point must be the multiple of 8 or the last 3 bits will be ignored + self.send_data(x & 0xFF) + + self.send_command(0x4F) # SET_RAM_Y_ADDRESS_COUNTER + self.send_data(y & 0xFF) + self.send_data((y >> 8) & 0xFF) + + ''' + function : Initialize the e-Paper register + parameter: + ''' + def init(self): + if (epdconfig.module_init() != 0): + return -1 + # EPD hardware init start + self.reset() + + self.ReadBusy() + self.send_command(0x12) #SWRESET + self.ReadBusy() + + self.send_command(0x01) #Driver output control + self.send_data(0xf9) + self.send_data(0x00) + self.send_data(0x00) + + self.send_command(0x11) #data entry mode + self.send_data(0x03) + + self.SetWindow(0, 0, self.width-1, self.height-1) + self.SetCursor(0, 0) + + self.send_command(0x3c) + self.send_data(0x05) + + self.send_command(0x21) # Display update control + self.send_data(0x00) + self.send_data(0x80) + + self.send_command(0x18) + self.send_data(0x80) + + self.ReadBusy() + + self.SetLut(self.lut_full_update) + return 0 + + ''' + function : Display images + parameter: + image : Image data + ''' + def getbuffer(self, image): + img = image + imwidth, imheight = img.size + if(imwidth == self.width and imheight == self.height): + img = img.convert('1') + elif(imwidth == self.height and imheight == self.width): + # image has correct dimensions, but needs to be rotated + img = img.rotate(90, expand=True).convert('1') + else: + logger.warning("Wrong image dimensions: must be " + str(self.width) + "x" + str(self.height)) + # return a blank buffer + return [0x00] * (int(self.width/8) * self.height) + + buf = bytearray(img.tobytes('raw')) + return buf + + ''' + function : Sends the image buffer in RAM to e-Paper and displays + parameter: + image : Image data + ''' + def display(self, image): + if self.width%8 == 0: + linewidth = int(self.width/8) + else: + linewidth = int(self.width/8) + 1 + + self.send_command(0x24) + for j in range(0, self.height): + for i in range(0, linewidth): + self.send_data(image[i + j * linewidth]) + self.TurnOnDisplay() + + ''' + function : Sends the image buffer in RAM to e-Paper and partial refresh + parameter: + image : Image data + ''' + def displayPartial(self, image): + epdconfig.digital_write(self.reset_pin, 0) + epdconfig.delay_ms(1) + epdconfig.digital_write(self.reset_pin, 1) + + self.SetLut(self.lut_partial_update) + self.send_command(0x37) + self.send_data(0x00) + self.send_data(0x00) + self.send_data(0x00) + self.send_data(0x00) + self.send_data(0x00) + self.send_data(0x40) + self.send_data(0x00) + self.send_data(0x00) + self.send_data(0x00) + self.send_data(0x00) + + self.send_command(0x3C) #BorderWavefrom + self.send_data(0x80) + + self.send_command(0x22) + self.send_data(0xC0) + self.send_command(0x20) + self.ReadBusy() + + self.SetWindow(0, 0, self.width - 1, self.height - 1) + self.SetCursor(0, 0) + + self.send_command(0x24) # WRITE_RAM + # for j in range(0, self.height): + # for i in range(0, linewidth): + # self.send_data(image[i + j * linewidth]) + self.send_data2(image) + self.TurnOnDisplayPart() + + ''' + function : Refresh a base image + parameter: + image : Image data + ''' + def displayPartBaseImage(self, image): + self.send_command(0x24) + self.send_data2(image) + + self.send_command(0x26) + self.send_data2(image) + self.TurnOnDisplay() + + ''' + function : Clear screen + parameter: + ''' + def Clear(self, color=0xFF): + if self.width%8 == 0: + linewidth = int(self.width/8) + else: + linewidth = int(self.width/8) + 1 + # logger.debug(linewidth) + + self.send_command(0x24) + self.send_data2([color] * int(self.height * linewidth)) + self.TurnOnDisplay() + + ''' + function : Enter sleep mode + parameter: + ''' + def sleep(self): + self.send_command(0x10) #enter deep sleep + self.send_data(0x01) + + epdconfig.delay_ms(2000) + epdconfig.module_exit() + +### END OF FILE ### + diff --git a/tools/station/display/epdconfig.py b/tools/station/display/epdconfig.py new file mode 100644 index 0000000..8eb93b8 --- /dev/null +++ b/tools/station/display/epdconfig.py @@ -0,0 +1,139 @@ +"""GPIO/SPI-Unterbau fuer die vendored Waveshare-Panel-Treiber (epd2in13_V2.py, +epd2in13_V3.py), geschrieben gegen gpiozero + spidev statt gegen Waveshares eigenes +epdconfig.py, das RPi.GPIO voraussetzt - ein Modul, das auf Debian 13 (trixie) mit +Python 3.13 nicht mehr baut/importiert. gpiozero, lgpio (gpiozero's Pin-Factory auf +trixie) und spidev sind auf dem Ziel-Pi vorhanden. + +Stellt genau die Schnittstelle bereit, die die beiden Treiber erwarten: die Pin-Konstanten +RST_PIN/DC_PIN/CS_PIN/BUSY_PIN sowie digital_write, digital_read, delay_ms, spi_writebyte, +spi_writebyte2, module_init und module_exit als Modulfunktionen (kein Klassen-Wrapper - +die Treiber greifen direkt auf `epdconfig.` zu). + +CS_PIN wird NICHT als GPIO geschaltet: spidev.SpiDev(0, 0) steuert die Hardware-CS0-Leitung +selbst bei jeder Übertragung, ein zusaetzliches Software-Toggle wuerde nur mit der Hardware +um die Leitung konkurrieren. digital_write()/digital_read() sind fuer CS_PIN deshalb +No-ops - genau das Muster, das Waveshares eigene, neuere epdconfig.py-Fassung fuer +Raspberry Pi inzwischen ebenfalls verwendet (dort ist die GPIO-Instanziierung fuer CS_PIN +auskommentiert). + +Ein fehlender Import (gpiozero/spidev nicht installiert) oder ein fehlendes +/dev/spidev0.0 wirft DisplayUnavailable - display.py faengt das ab, protokolliert eine +Zeile und beendet sich sauber (sys.exit(0)), statt dass ein fehlendes Panel eine Migration +verhindern koennte (siehe Modul-Docstring von display.py).""" +from __future__ import annotations + +import os +import time + +try: + import gpiozero +except ImportError: # pragma: no cover - auf dem Pi immer vorhanden, hier nur defensiv + gpiozero = None + +try: + import spidev +except ImportError: # pragma: no cover - dito + spidev = None + + +class DisplayUnavailable(Exception): + """Kein Panel ansprechbar: Bibliothek fehlt, /dev/spidev0.0 fehlt, oder der + Hardwarezugriff selbst ist fehlgeschlagen (z.B. Berechtigung, Pin belegt).""" + + +# Dieselben Pin-Nummern wie in Waveshares eigenem epdconfig.py fuer den Raspberry Pi. +RST_PIN = 17 +DC_PIN = 25 +CS_PIN = 8 +BUSY_PIN = 24 + +_SPI_BUS = 0 +_SPI_DEVICE = 0 +_SPI_HZ = 4_000_000 + +_rst_pin = None +_dc_pin = None +_busy_pin = None +_spi = None + + +def digital_write(pin: int, value: int) -> None: + if pin == RST_PIN: + _rst_pin.value = bool(value) + elif pin == DC_PIN: + _dc_pin.value = bool(value) + elif pin == CS_PIN: + pass # Hardware-CS ueber spidev, siehe Modul-Docstring. + # Andere Pins fragen die Treiber nicht ab - kein else-Zweig noetig. + + +def digital_read(pin: int) -> int: + if pin == BUSY_PIN: + return int(bool(_busy_pin.value)) + if pin == RST_PIN: + return int(bool(_rst_pin.value)) + if pin == DC_PIN: + return int(bool(_dc_pin.value)) + return 0 + + +def delay_ms(ms: int) -> None: + time.sleep(ms / 1000.0) + + +def spi_writebyte(data) -> None: + _spi.writebytes(data) + + +def spi_writebyte2(data) -> None: + _spi.writebytes2(data) + + +def module_init() -> int: + """Oeffnet GPIO-Pins und SPI-Geraet. Wirft DisplayUnavailable statt irgendeiner + Bibliotheksausnahme, damit display.py genau eine Ausnahmeklasse abfangen muss.""" + global _rst_pin, _dc_pin, _busy_pin, _spi + + if gpiozero is None or spidev is None: + raise DisplayUnavailable("gpiozero/spidev nicht installiert") + + spi_path = f"/dev/spidev{_SPI_BUS}.{_SPI_DEVICE}" + if not os.path.exists(spi_path): + raise DisplayUnavailable(f"{spi_path} fehlt - ist SPI aktiviert (raspi-config)?") + + try: + _rst_pin = gpiozero.DigitalOutputDevice(RST_PIN) + _dc_pin = gpiozero.DigitalOutputDevice(DC_PIN) + _busy_pin = gpiozero.DigitalInputDevice(BUSY_PIN, pull_up=False) + + _spi = spidev.SpiDev() + _spi.open(_SPI_BUS, _SPI_DEVICE) + _spi.max_speed_hz = _SPI_HZ + _spi.mode = 0b00 + except Exception as e: # noqa: BLE001 - alles hier wird zu DisplayUnavailable + module_exit() + raise DisplayUnavailable(f"GPIO/SPI-Initialisierung fehlgeschlagen: {e}") from e + + return 0 + + +def module_exit() -> None: + """Gibt GPIO-Pins und SPI-Geraet frei. Verzeiht, wenn module_init() nie (vollstaendig) + gelaufen ist - display.py ruft das u.a. im Fehlerpfad von module_init() selbst auf.""" + global _rst_pin, _dc_pin, _busy_pin, _spi + + if _spi is not None: + try: + _spi.close() + except Exception: # noqa: BLE001 - Aufraeumen darf nicht seinerseits scheitern + pass + _spi = None + + for attr_name in ("_rst_pin", "_dc_pin", "_busy_pin"): + device = globals()[attr_name] + if device is not None: + try: + device.close() + except Exception: # noqa: BLE001 + pass + globals()[attr_name] = None diff --git a/tools/station/display/render.py b/tools/station/display/render.py new file mode 100644 index 0000000..c89c00e --- /dev/null +++ b/tools/station/display/render.py @@ -0,0 +1,133 @@ +"""Textzerlegung und Bildaufbau fuer die E-Paper-Anzeige der ninlight-Station. + +lines() ist bewusst ohne Pillow und ohne Hardwarezugriff importierbar und testbar (siehe +tools/station/tests/test_render.py) - compose() importiert Pillow erst innerhalb der +Funktion, damit ein fehlendes PIL (das erst per "apt install python3-pil" auf den Pi +kommt, siehe setup_pi.sh) render.py selbst nicht unbenutzbar macht. + +Das Panel ist 250x122 Pixel, quer verwendet. MAX_COLS begrenzt jede Textzeile so, dass sie +bei der in compose() verwendeten Schriftgroesse nicht umgebrochen werden muss - ueberlange +Werte (z.B. ein Produktname aus der Lampenerkennung) werden mit "..." gekuerzt, nie +umgebrochen (Spec: eine zu lange Zeile darf das Layout nicht sprengen). + +Der Zustand, den lines()/compose() bekommen, ist bereits zusammengefuehrt: status.json +(vom Server geschrieben) traegt kein "ssid"-Feld - die SSID ist Konfiguration aus +/etc/ninlight-station.conf. display.py liest beides und reicht ein gemeinsames dict +herein. Jedes Feld wird ueber .get() mit Vorgabewert gelesen: status.json, +control.json und die Konfigurationsdatei koennen jederzeit fehlen, halb geschrieben oder +aus einer aelteren Fassung sein (frisch aufgesetzter Pi, Schreibfehler mitten im +Schreiben, Neustart) - siehe test_missing_fields_do_not_raise.""" +from __future__ import annotations + +MAX_COLS = 42 + +MODE_LABELS = {"recovery": "RETTUNG", "migration": "MIGRATION"} +# "armed" wird gesondert behandelt (braucht armed_target im Text), deshalb hier nicht +# gelistet. +PHASE_LABELS = {"auto": "AUTO", "listen": "HORCHT"} + + +def _cut(text: object) -> str: + """Kuerzt auf MAX_COLS Zeichen, mit "..." am Ende statt Umbruch.""" + text = str(text) + if len(text) <= MAX_COLS: + return text + if MAX_COLS <= 1: + return text[:MAX_COLS] + return text[: MAX_COLS - 1] + "…" + + +def _hhmm(ts: object) -> str: + """"2026-09-13T13:04:02Z" -> "13:04". Alles andere (fehlt, falsch geformt, aus einer + aelteren Serverfassung ohne Zeitstempel) -> "--:--".""" + if isinstance(ts, str) and len(ts) >= 16 and ts[10:11] == "T": + return ts[11:16] + return "--:--" + + +def lines(state: dict) -> list[str]: + """Zerlegt den zusammengefuehrten Zustand in Textzeilen fuers Panel. Wirft nie - + fehlende/falsch geformte Felder werden ausgelassen statt eine Ausnahme zu werfen.""" + state = state if isinstance(state, dict) else {} + mode = state.get("mode") + out: list[str] = [] + + if mode == "off": + # Die Station ist komplett aus (kein status.json vorhanden) - display.py setzt + # diesen Modus selbst, server.py kennt ihn nicht. Bewusst keine weiteren Zeilen + # (insbesondere keine SSID - die waere aus einer alten Konfigurationsdatei und + # gerade nicht aktiv). + out.append("Station aus") + return out + + if mode not in ("recovery", "migration"): + # Unbekannter oder fehlender Modus, z.B. ein frisch aufgesetzter Pi, auf dem noch + # nichts geschrieben wurde. Lieber ehrlich "unbekannt" zeigen als einen Modus zu + # unterstellen. + out.append(_cut(f"Status unbekannt ({mode})") if mode else "Status unbekannt") + return out + + out.append(MODE_LABELS[mode]) + + ssid = state.get("ssid") + if ssid: + out.append(_cut(f"SSID: {ssid}")) + + if mode == "migration": + # Phase kommt aus control.json, nicht aus status.json (das schreibt der Server + # nur beim Start und nach einer Auslieferung) - display.py reicht sie hier als + # Teil des zusammengefuehrten Zustands herein. Im Rettungsmodus gibt es keine + # Phase, deshalb dieser Block nur hier drin. + phase = state.get("phase") + if phase == "armed": + target = state.get("armed_target") or "?" + out.append(_cut(f"FREIGEGEBEN: {target}")) + elif phase in PHASE_LABELS: + out.append(PHASE_LABELS[phase]) + + last_seen = state.get("last_seen") + if not isinstance(last_seen, dict): + last_seen = {} + product_name = last_seen.get("product_name") + if product_name: + out.append(_cut(product_name)) + target = last_seen.get("target") + variant = last_seen.get("variant") + if target or variant: + out.append(_cut(f"{target or '?'} ({variant or '?'})")) + lamp_version = last_seen.get("lamp_version") + if lamp_version: + out.append(_cut(f"FW: {lamp_version}")) + + served = state.get("served") + out.append(_cut(f"Ausgeliefert: {served if isinstance(served, int) else 0}")) + + recent = state.get("recent") + if isinstance(recent, list) and recent: + last = recent[-1] + if isinstance(last, dict): + out.append(_cut(f"{_hhmm(last.get('ts'))} {last.get('target') or '?'} " + f"{last.get('status') or '?'}")) + + return out + + +def compose(state: dict): + """Baut das Panel-Bild (250x122, 1-Bit) aus lines(). Importiert Pillow erst hier - + siehe Modul-Docstring.""" + from PIL import Image, ImageDraw, ImageFont + + width, height = 250, 122 + image = Image.new("1", (width, height), 255) + draw = ImageDraw.Draw(image) + font = ImageFont.load_default() + + line_height = 14 + y = 2 + for line in lines(state): + if y > height - line_height: + break + draw.text((2, y), line, font=font, fill=0) + y += line_height + + return image diff --git a/tools/station/ninlight-display.service b/tools/station/ninlight-display.service new file mode 100644 index 0000000..91c0bac --- /dev/null +++ b/tools/station/ninlight-display.service @@ -0,0 +1,49 @@ +[Unit] +Description=ninlight E-Paper-Anzeige (Status der Migrations-/Rettungsstation) +# Nur eine Anzeige fuer den Hauptdienst - Wants=/After=, NICHT Requires=/BindsTo=: ein +# fehlendes oder abgestuerztes Panel darf server.py niemals mitreissen, und server.py +# darf umgekehrt starten, auch wenn dieser Dienst fehlt oder deaktiviert ist. Siehe +# display.py's Modul-Docstring: die Anzeige ist nirgends Voraussetzung fuer die +# Migration. +After=ninlight-station.service +Wants=ninlight-station.service + +[Service] +Type=simple +# Reihenfolge wichtig: systemd wendet Environment=/EnvironmentFile= in der Reihenfolge +# an, in der sie hier stehen, und eine spaetere Zuweisung ueberschreibt eine gleichnamige +# fruehere. EPD_REVISION=V3 hier ist nur der Vorgabewert, falls die Konfigurationsdatei +# (unten, per EnvironmentFile=) den Schluessel (noch) nicht enthaelt - z.B. auf einem Pi, +# der vor Task 6 aufgesetzt wurde. Steht er dort, gewinnt er. +Environment=EPD_REVISION=V3 +Environment=NINLIGHT_RUN_DIR=/run/ninlight-station +Environment=NINLIGHT_STATION_CONF=/etc/ninlight-station.conf +EnvironmentFile=/etc/ninlight-station.conf +ExecStart=/usr/bin/python3 /opt/ninlight-station/display/display.py + +# display.py faengt ein fehlendes/abgestuerztes Panel als DisplayUnavailable ab, +# protokolliert eine Zeile und beendet sich mit sys.exit(0) - das ist also nie ein +# "Fehlschlag" im systemd-Sinne und loest kein Restart= aus. Restart=on-failure (nicht +# always) greift nur fuer echte Abstuerze (z.B. ein Python-Traceback aus einem +# Programmierfehler); ohne den Deckel wuerde ein dauerhaft fehlendes Panel sonst NICHT +# staendig neu versucht (gut so), waehrend ein echter Bug endlos neu starten wuerde und +# so das Journal fluten koennte. +Restart=on-failure +RestartSec=5 +StartLimitIntervalSec=60 +StartLimitBurst=5 + +NoNewPrivileges=yes +ProtectSystem=strict +ProtectHome=yes +PrivateTmp=yes +ReadOnlyPaths=/run/ninlight-station +RestrictAddressFamilies=AF_UNIX AF_NETLINK +MemoryMax=64M + +# Fuer /dev/spidev0.0 und die GPIO-Zeichenleitung (per lgpio/gpiozero) noetig - laeuft +# deshalb nicht mit DynamicUser=, sondern als root wie der Hauptdienst. +SupplementaryGroups=spi gpio + +[Install] +WantedBy=multi-user.target diff --git a/tools/station/tests/test_render.py b/tools/station/tests/test_render.py new file mode 100644 index 0000000..7ca1f01 --- /dev/null +++ b/tools/station/tests/test_render.py @@ -0,0 +1,56 @@ +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "display")) + +import render # noqa: E402 + + +def test_off_mode_says_so_and_shows_nothing_else(): + lines = render.lines({"mode": "off"}) + assert any("Station aus" in l for l in lines) + assert not any("SSID" in l for l in lines) + + +def test_migration_shows_detected_lamp(): + lines = render.lines({ + "mode": "migration", "phase": "auto", "served": 3, + "ssid": "qconnex-firmware", + "last_seen": {"product_name": "Klyqa GU10 RGB CW/WW", "target": "gu10_rgb", + "variant": "prod", "lamp_version": "V1.1.0.1138"}, + "recent": [{"target": "gu10_rgb", "status": "ok", "ts": "2026-09-13T13:04:02Z"}]}) + text = "\n".join(lines) + assert "MIGRATION" in text + assert "Klyqa GU10 RGB CW/WW" in text + assert "V1.1.0.1138" in text + assert "prod" in text + assert "13:04" in text + + +def test_recovery_mode_without_traffic(): + lines = render.lines({"mode": "recovery", "ssid": "ninlight-recovery", + "served": 0, "recent": []}) + text = "\n".join(lines) + assert "RETTUNG" in text + assert "ninlight-recovery" in text + + +def test_long_product_name_is_truncated_not_wrapped(): + """Das Panel ist 250 px breit - eine zu lange Zeile darf das Layout nicht sprengen.""" + lines = render.lines({"mode": "migration", "phase": "listen", "served": 0, "recent": [], + "ssid": "qconnex-firmware", + "last_seen": {"product_name": "X" * 90, "target": "e27_rgb", + "variant": "dev", "lamp_version": "V1"}}) + assert all(len(l) <= render.MAX_COLS for l in lines) + + +def test_missing_fields_do_not_raise(): + """status.json kann jederzeit halb geschrieben oder aus einer aelteren Version sein.""" + for state in ({}, {"mode": "migration"}, {"mode": "migration", "last_seen": {}}): + assert render.lines(state) + + +def test_armed_phase_is_visible(): + lines = render.lines({"mode": "migration", "phase": "armed", "armed_target": "g95_cw", + "served": 0, "recent": [], "ssid": "qconnex-firmware"}) + assert "FREIGEGEBEN" in "\n".join(lines)