From 42fc7f52d4be37273346b046679c770487281ca9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Reinhard=20X=2E=20F=C3=BCrst?= Date: Sun, 26 Jul 2026 18:51:45 +0200 Subject: [PATCH] Portierung auf den XIAO ESP32-C3 (MicroPython) Die Kette laeuft auf dem Geraet gegen den Mock auf dem Mac. Getestet mit MicroPython 1.28.0 (ESP32_GENERIC_C3): alle Module importieren, Telegrammbau und Koordinaten-Formatierung sind byte-identisch zum Mac (inkl. Gradzeichen), Poll-Schleife 22 Abfragen in 20 s ohne Aussetzer, Ausfall und Reconnect verhalten sich wie vorgesehen. Offen bleibt allein die RS232-Strecke zur Anzeige -- der Steckbrettaufbau mit MAX3232 fehlt noch. Gemeinsamer Kern statt zweier Programme: - displayloop.py: die Poll-Schleife, herausgeloest aus run_display. Mac und ESP32 fahren jetzt denselben Ablauf; drumherum liegt nur die Kommandozeile (run_display.py) bzw. das WLAN-Setup (run_esp32.py). - ticks.py: Timeouts fuer beide Plattformen. time.ticks_ms() laeuft nach ~12 Tagen ueber; das bisherige "jetzt + Timeout" haette eine Warteschleife danach tagelang haengen lassen. Jetzt ueber ticks_diff. - wifi.py, run_esp32.py, wifi_config.py (gitignored, Vorlage dabei), deploy.sh. Pinbelegung aus dem KiCad-Schaltplan uebernommen, nicht geraten: UART1 mit TX=GPIO21 (Pad 7/D6) und RX=GPIO20 (Pad 8/D7) ueber den MAX3232, LDR an GPIO2 (Pad 1/A0). UART0 bleibt die REPL. LDR-Kennlinie ist auf der Platine umgekehrt: der Teiler lautet 3V3-[2k]-A0-[LDR]-GND, die Spannung faellt also mit steigender Helligkeit -- andersherum als beim MSP430, auf den die Schwellen zugeschnitten sind. Wird im Code gespiegelt (config.LDR_INVERTED, brightness.raw_to_ldr). Die Schwellen selbst muessen am realen Aufbau noch nachgemessen werden. Am Geraet gefundene Fehler, die den Attrappen-Tests entgehen mussten, weil CPython sie nicht hat: - MicroPython kennt kein str.ljust (auch kein rjust/center/zfill). display._emit_line fuellt von Hand auf. - NOLINK_LINES hatte 10 Zeichen bei 9 Zeichen Zeilenbreite und erschien als "MONTIERUN". Jetzt "KEIN LINK"/"TELESKOP". Dagegen: TestMicropythonVertraeglichkeit prueft die deployten Dateien statisch auf solche Methoden und auf modulweites argparse/threading und haelt die Dateiliste mit deploy.sh synchron; test_display.py prueft die Breite fester Anzeigetexte. Weitere Portabilitaet: mount_client nutzt getaddrinfo und eine Sendeschleife statt sendall (MicroPython laesst dessen Verhalten mit Timeout offen); argparse in brightness wandert in die Funktion; migra._digits und die Hex-Ausgabe verlassen sich nicht mehr auf %0*d bzw. bytes.hex(). Beides kann 1.28 zwar doch -- am Geraet nachgeprueft, die Kommentare sagen es jetzt richtig. Widerspruch in der Dokumentation bereinigt: die Zeichenbreite stand an vier Stellen als 9 oder 10. Gemessen sind 9; die Erklaerung duerfte sein, dass die Sperrschrift 7 px pro Zelle braucht (9*7 = 63 von 64 px) und nicht 6, wie es eine 5x7-Matrix mit 1 px Luecke waere. Nachgeprueft ist das nicht -- dafuer gibt es jetzt "probe.py pitch", das den Auto-Vorschub gegen feste Pixelabstaende stellt, statt am Umbruch abzuzaehlen. mount_mock.py --lan bindet auf 0.0.0.0 und nennt die einzutragende LAN-IP, damit der ESP32 den Mock erreicht. 92 Tests gruen. Co-Authored-By: Claude Opus 5 (1M context) --- .gitignore | 5 +- README.md | 252 ++++++++++++++--- brightness.py | 25 +- config.py | 66 ++++- deploy.sh | 50 ++++ display.py | 26 +- displayloop.py | 112 ++++++++ migra.py | 24 +- mount_client.py | 27 +- mount_mock.py | 36 ++- probe.py | 63 ++++- run_display.py | 67 +---- run_esp32.py | 142 ++++++++++ test_display.py | 91 +++++++ test_esp32.py | 605 +++++++++++++++++++++++++++++++++++++++++ ticks.py | 45 +++ transport.py | 59 ++-- wifi.py | 126 +++++++++ wifi_config_example.py | 10 + 19 files changed, 1702 insertions(+), 129 deletions(-) create mode 100755 deploy.sh create mode 100644 displayloop.py create mode 100644 run_esp32.py create mode 100644 test_display.py create mode 100644 test_esp32.py create mode 100644 ticks.py create mode 100644 wifi.py create mode 100644 wifi_config_example.py diff --git a/.gitignore b/.gitignore index 31ca500..6686b65 100644 --- a/.gitignore +++ b/.gitignore @@ -10,4 +10,7 @@ venv/ .DS_Store #Hardware-Sachern nicht ist Git -Hardware/ \ No newline at end of file +Hardware/ + +# WLAN-Zugangsdaten des ESP32 (Vorlage: wifi_config_example.py) +wifi_config.py \ No newline at end of file diff --git a/README.md b/README.md index 08ad200..9bfb17f 100644 --- a/README.md +++ b/README.md @@ -3,10 +3,10 @@ Ansteuerung einer microSYST **migra SC/MC Seriell** LED-Großanzeige mit den Koordinaten (Rektaszension / Deklination) einer 10Micron GM4000 HPS II. -Endziel ist ein XIAO ESP32-C3, der die Montierung per WLAN abfragt und die -Koordinaten über UART/RS232 auf die Anzeige schreibt. **Phase 1** (dieser Stand) -ist das Testprogramm auf dem Mac: Koordinaten von Hand übergeben, Ausgabe über -einen USB-RS232-Wandler. +Zielgerät ist ein XIAO ESP32-C3, der die Montierung per WLAN abfragt und die +Koordinaten über UART/RS232 auf die Anzeige schreibt. Entwickelt und getestet +wird auf dem Mac (Koordinaten von Hand bzw. von der Montierung, Ausgabe über +einen USB-RS232-Wandler); dieselben Module laufen auf dem ESP32. ## Benutzung @@ -27,28 +27,49 @@ python3 mount_mock.py --move # Mock separat starten # Helligkeit aus einem LDR-Wert setzen: python3 brightness.py --ldr 1500 # LDR-Rohwert -> Stufe/Prozent -> senden -python3 -m unittest test_migra test_lx200 test_brightness # alle Tests +python3 -m unittest test_migra test_lx200 test_brightness test_display test_esp32 python3 probe.py ruler # Geometrie der Anzeige ausmessen ``` -Einzige Abhängigkeit auf dem Mac ist `pyserial`. +Auf dem ESP32-C3 (siehe Abschnitt [ESP32-C3](#esp32-c3)): + +```bash +cp wifi_config_example.py wifi_config.py # WLAN-Zugangsdaten eintragen +./deploy.sh # Dateien auf den ESP32 kopieren +mpremote repl # dann: import run_esp32 + # run_esp32.main() +``` + +Einzige Abhängigkeit auf dem Mac ist `pyserial`, fürs Übertragen zusätzlich +`mpremote`. Auf dem ESP32 laufen die Module ohne jede Fremdbibliothek. ## Aufbau -| Datei | Zweck | -|---|---| -| `migra.py` | Protokoll: Telegrammrahmen, Prüfsumme, ESC-Sequenzen. Ohne Abhängigkeiten, **läuft unverändert unter MicroPython**. | -| `transport.py` | UART: `pyserial` (Mac), `machine.UART` (ESP32-C3), `FakeTransport` (Trockenlauf). | -| `display.py` | Treiber: Telegramm senden, Antwort auswerten. | -| `coords.py` | Koordinaten einlesen und auf die Zeilenbreite formatieren. | -| `config.py` | Port, Adresse, Geometrie, Helligkeit — alles an einer Stelle. | -| `show_coords.py` | Das Testprogramm (Koordinaten von Hand). | -| `brightness.py` | Helligkeitsregelung: LDR-Wert → Stufe/Prozent (portabel + CLI). | -| `run_display.py` | Hauptschleife Phase 2: Montierung pollen, anzeigen, Helligkeit regeln. | -| `mount_client.py` / `mount_mock.py` | TCP-Client zur GM4000 bzw. Simulator (LX200). | -| `lx200.py` | LX200-Kommandorahmen. | -| `probe.py` | Testmuster zum Ausmessen von Zeilenbreite und Zeilenposition. | -| `test_migra.py` / `test_lx200.py` / `test_brightness.py` | Tests. | +Alles, was auf beiden Plattformen läuft, ist bewusst abhängigkeitsfrei +gehalten. Was sich unterscheidet, steckt in `transport.py` (UART) und in den +beiden Startprogrammen. + +| Datei | Zweck | ESP32 | +|---|---|---| +| `migra.py` | Protokoll: Telegrammrahmen, Prüfsumme, ESC-Sequenzen. | ✓ | +| `transport.py` | UART: `pyserial` (Mac), `machine.UART` (ESP32-C3), `FakeTransport` (Trockenlauf). | ✓ | +| `ticks.py` | Timeouts, die unter CPython und MicroPython gleich funktionieren. | ✓ | +| `display.py` | Treiber: Telegramm senden, Antwort auswerten. | ✓ | +| `coords.py` | Koordinaten einlesen und auf die Zeilenbreite formatieren. | ✓ | +| `config.py` | Port/Pins, Adresse, Geometrie, Helligkeit, WLAN — alles an einer Stelle. | ✓ | +| `displayloop.py` | Die Poll-Schleife: abfragen, formatieren, anzeigen, Fehler abfangen. | ✓ | +| `brightness.py` | Helligkeitsregelung: LDR-Wert → Stufe/Prozent. | ✓ | +| `mount_client.py` | TCP-Client zur GM4000 (LX200). | ✓ | +| `lx200.py` | LX200-Kommandorahmen. | ✓ | +| `wifi.py` | WLAN-Anbindung des ESP32 (verbinden, Verbindung halten). | ✓ | +| `run_esp32.py` | Startprogramm auf dem ESP32: WLAN, Anzeige, Schleife. | ✓ | +| `wifi_config.py` | WLAN-Zugangsdaten, **nicht im Git** (Vorlage `wifi_config_example.py`). | ✓ | +| `run_display.py` | Startprogramm auf dem Mac: dieselbe Schleife mit Kommandozeile. | | +| `show_coords.py` | Testprogramm (Koordinaten von Hand). | | +| `mount_mock.py` | Simulator der Montierung (LX200) fürs Testen ohne Teleskop. | | +| `probe.py` | Testmuster zum Ausmessen von Zeilenbreite und Zeilenposition. | | +| `deploy.sh` | Kopiert die ESP32-Dateien per `mpremote` auf das Gerät. | | +| `test_migra.py` / `test_lx200.py` / `test_brightness.py` / `test_display.py` / `test_esp32.py` | Tests. | | ## Protokoll (Kurzfassung) @@ -91,14 +112,150 @@ Folgt auf eine ESC-Sequenz ein Online-Text, muss **`0x1F` als Trennzeichen** dazwischen. Data-Unit max. 230 Zeichen. `migra.DataUnit` setzt das Trennzeichen selbst. +## ESP32-C3 + +### Verdrahtung + +Die Pins stammen aus dem Schaltplan (`Hardware/KiCad/Grossanzeige`) und stehen +in `config.py`. Am XIAO ESP32-C3 ist UART1 frei zuordenbar; UART0 bleibt für die +REPL-Konsole über USB reserviert. + +| XIAO | GPIO | Netz | weiter zu | +|---|---|---|---| +| Pad 7 (D6) | 21 | TX | MAX3232 `T2IN` → `T2OUT` → J5.3 → Anzeige RxD (Sub-D Pin 2) | +| Pad 8 (D7) | 20 | RX | MAX3232 `R2OUT` ← `R2IN` ← J5.2 ← Anzeige TxD (Sub-D Pin 3) | +| Pad 1 (D0/A0) | 2 | LDR | Spannungsteiler J2, siehe unten | +| Pad 12 | — | 3V3 | MAX3232 `VCC` | +| Pad 13 | — | GND | gemeinsame Masse | + +Die Anzeige will RS232-Pegel, der ESP32-C3 kann nur TTL — der MAX3232 dazwischen +ist zwingend. + +### Einrichten + +MicroPython für den ESP32-C3 flashen (einmalig; Firmware von +[micropython.org/download/ESP32_GENERIC_C3](https://micropython.org/download/ESP32_GENERIC_C3/)): + +```bash +pip install esptool mpremote +esptool.py --chip esp32c3 --port /dev/cu.usbmodem1101 erase_flash +esptool.py --chip esp32c3 --port /dev/cu.usbmodem1101 --baud 460800 \ + write_flash -z 0x0 ESP32_GENERIC_C3-*.bin +``` + +Dann die Zugangsdaten anlegen und alles übertragen: + +```bash +cp wifi_config_example.py wifi_config.py # SSID und Passwort eintragen +./deploy.sh # oder ./deploy.sh /dev/cu.usbmodem1101 +``` + +`wifi_config.py` steht in `.gitignore` — das WLAN-Passwort gehört nicht ins +Repository. + +### Starten + +```bash +mpremote repl +>>> import run_esp32 +>>> run_esp32.main() # ohne Helligkeitsregelung +>>> run_esp32.main(with_ldr=True) # mit LDR über den ADC +``` + +Bewusst **kein** `main.py`: MicroPython startet eine Datei dieses Namens nach +jedem Reset von selbst, und solange am Aufbau gearbeitet wird, ist der manuelle +Start angenehmer. Für den Dauerbetrieb in der Sternwarte reicht später eine +`main.py` mit den zwei Zeilen von oben. + +### Was auf beiden Seiten gleich läuft + +`displayloop.poll_loop` ist die gemeinsame Hauptschleife — Mac und ESP32 fahren +denselben Ablauf, drumherum liegt nur die Kommandozeile (`run_display.py`) bzw. +das WLAN-Setup (`run_esp32.py`). Fällt das WLAN aus, zeigt die Anzeige +`NOWIFI_LINES`, fällt die Montierung aus, `NOLINK_LINES`; in beiden Fällen läuft +die Schleife weiter und verbindet sich neu. + +Stolpersteine, die beim Portieren aufgefallen sind und in den Tests +festgehalten sind (`test_esp32.py`, `test_display.py`): + +- **MicroPython hat kein `str.ljust`** (auch kein `rjust`/`center`/`zfill`). + Das fiel erst am Gerät auf — auf dem Mac gibt es die Methode, alle Tests + waren grün. `display._emit_line` füllt jetzt von Hand auf. + `TestMicropythonVertraeglichkeit` sucht solche Aufrufe seitdem statisch. +- **`time.ticks_ms()` läuft nach ~12 Tagen über.** Ein einfaches + „jetzt + Timeout" ließe eine Warteschleife danach tagelang hängen. `ticks.py` + rechnet deshalb über `ticks_diff`, das den Überlauf richtig behandelt. +- **Der UART braucht denselben Puffer-Flush wie pyserial.** `UartTransport.write` + leert den Empfangspuffer vor jedem Senden, sonst sieht `read_response` alte + Störbytes statt der frischen Antwort. +- **Feste Anzeigetexte müssen in `CHARS_PER_LINE` passen.** `NOLINK_LINES` hieß + zuerst „MONTIERUNG" (10 Zeichen) und erschien als „MONTIERUN"; `_emit_line` + schneidet stillschweigend ab. `test_display.py` prüft die Breite jetzt. + +Was MicroPython 1.28 entgegen der Erwartung **doch** kann (am Gerät nachgeprüft): +`%0*d` mit dynamischer Feldbreite, `%-11s`, `bytes.hex()`, `str.strip`/`lstrip`, +`str.isdigit`. + +Ohne Hardware lässt sich der ESP32-Teil trotzdem weitgehend prüfen: +`test_esp32.py` hängt Attrappen für `machine` und `network` ein und fährt die +ganze Kette gegen einen echten Mock — UART-Transport, WLAN, ADC-Auswertung, +Zeitrechnung und die Poll-Schleife samt Ausfallverhalten. + +### Gegen den Mock auf dem Mac testen + +Solange die Montierung nicht erreichbar ist, spielt der Mock ihre Rolle. Er muss +dafür auf allen Schnittstellen lauschen, sonst kommt der ESP32 nicht heran: + +```bash +python3 mount_mock.py --lan --move # gibt die IP aus, die einzutragen ist +``` + +Auf dem ESP32 dann: + +```python +import run_esp32 +run_esp32.selftest(host="192.168.178.26") # Bring-up prüfen +run_esp32.main(host="192.168.178.26", fake_display=True) # Dauerschleife +``` + +`selftest` geht die drei Stufen einzeln durch — WLAN, Montierung, Telegrammbau — +und sagt, an welcher es klemmt. `fake_display=True` schreibt die Telegramme nur +als Hex in die REPL, es braucht also weder Pegelwandler noch Anzeige. + +Wenn keine Verbindung zustande kommt, obwohl der Mock lauscht: macOS fragt beim +ersten Mal, ob `python3` eingehende Verbindungen annehmen darf — das muss +erlaubt sein. + +### Am Gerät bestätigt (2026-07-26) + +XIAO ESP32-C3 mit MicroPython 1.28.0 (`ESP32_GENERIC_C3`), gegen den Mock auf dem +Mac: + +- Alle zwölf Module importieren auf dem Gerät. +- Telegrammbau und Koordinaten-Formatierung sind **byte-identisch** zum Mac, + inklusive Gradzeichen über den Zeichensatzwechsel. +- WLAN verbindet sich, `selftest` geht durch alle drei Stufen. +- Die Poll-Schleife lief 20 s im Sekundentakt, 22 Abfragen ohne Aussetzer. +- **Ausfall und Erholung:** Mock gestoppt → „KEIN LINK / TELESKOP" erscheint, die + Schleife läuft weiter; Mock zurück → automatischer Reconnect, und der erste + Wiedereinstieg löscht einmal das Vollbild, sodass kein Rest stehen bleibt. +- UART1 öffnet auf GPIO21/20 mit 19200 8E1, sendet, und der Lese-Timeout greift + sauber (kein Hänger) — allerdings noch **ohne Gegenstelle**. +- ADC auf GPIO2 mit 11 dB Dämpfung liefert Werte. + ## Offene Punkte -- **Echte Montierungs-IP:** In `config.py` steht `MOUNT_HOST` noch als Platzhalter. - Sobald die GM4000 im Netz erreichbar ist, dort die IP eintragen (Port 3490 ist - Vorgabe). Bis dahin läuft alles gegen den Mock (`run_display.py --mock`). -- Portierung auf den XIAO ESP32-C3 (MicroPython): `migra.py`, `lx200.py`, - `mount_client.py` sind bewusst abhängigkeitsfrei; nur `transport` (UART) und das - WLAN-Setup kommen hinzu. +- **RS232-Strecke zur Anzeige.** Als einziges Stück der Kette noch nicht am + ESP32 erprobt: Der Aufbau mit MAX3232 und Anzeige fehlt. Alles davor ist + bestätigt, die Telegramme stimmen byteweise — es geht nur noch um die Pegel. +- **LDR-Schwellen nachmessen.** Die Grenzen in `brightness.py` stammen vom alten + MSP430-Aufbau. Sobald der LDR am ESP32 hängt, die Rohwerte bei Tag und Nacht + ablesen und `HELL_LOW`/`HELL_HIGH`/`HELL_MAX` anpassen. Bis dahin läuft die + Anzeige mit fester Helligkeit (`run_esp32.main()` ohne `with_ldr`). +- **Netzwerk zur echten Montierung.** Die GM4000 steht in `192.168.1.115`, das + Heimnetz des ESP32 ist `192.168.178.x` — der Mac erreicht sie per VPN, der + ESP32 so nicht. Muss geklärt werden, bevor es an die echte Montierung geht. +- **Dauerbetrieb:** Autostart (`main.py`) und ein Watchdog fehlen noch bewusst. ## Gradzeichen @@ -116,7 +273,28 @@ diese Zelle überschreibt. ## Geometrie und Darstellung - **9 Zeichen pro Zeile** in Sperrschrift (`CHARSET_SPACED = True`, am Gerät - ausgemessen). `probe.py ruler` misst es nach. + ausgemessen 2026-07-15). + + Das widerspricht der naheliegenden Rechnung: Die Anzeige ist 64 px breit, und + eine 5×7-Matrix mit 1 px Lücke wäre 6 px pro Zelle — das gäbe 10 Zeichen + (60 px). Die Erklärung ist vermutlich, dass die Sperrschrift eben sperrt: + bei 2 px Lücke sind es 7 px pro Zelle und damit 9 Zeichen (63 von 64 px). + Dazu passt auch die Beobachtung, dass die *proportionale* Schrift `ESC Z` + rund 11 Zeichen schafft (≈ 5,8 px/Zelle). + + **Nachgeprüft ist das nicht.** `probe.py pitch` entscheidet es direkt, statt + am Umbruch abzuzählen: Es schreibt zehn Ziffern einmal mit dem Auto-Vorschub + der Anzeige und einmal auf feste Pixelpositionen `x = i·cell`. Ein Unterschied + von 1 px summiert sich über zehn Zeichen auf 9 px und ist unübersehbar. + + ```bash + python3 probe.py pitch 6 # stehen beide Zeilen untereinander -> 10 Zeichen + python3 probe.py pitch 7 # ... -> 9 Zeichen + ``` + + Sollten es doch 6 px sein, passt die Deklination mit abschließendem `"` in + voller Form (`+38°47'01"`, 10 Zeichen) — `coords.format_dec` wählt sie dann + von selbst, es genügt `CHARS_PER_LINE = 10`. - Die Deklination wird deshalb als **9-Zeichen-Form ohne abschließendes `"`** ausgegeben (`+38°47'01`) — Gradzeichen und Bogenminuten bleiben, die Bogensekunden stehen als Ziffern da. `coords.format_dec` wählt automatisch die breiteste Form, @@ -168,9 +346,19 @@ stammen aus dem alten Code**; die obere Grenze `HELL_MAX = 3500` für 100 % ist reale LDR-Wertebereich feststeht. Ein Helligkeitstelegramm wird nur bei **Stufenwechsel** gesendet (`ESC H 2 `), wie im Original. -Der LDR-Wert kommt derzeit per Kommandozeile (`--ldr`); auf dem ESP32-C3 liefert -ihn später der ADC. `run_display.py --ldr ` bindet die Regelung in die -Hauptschleife ein (`BrightnessController`). +Auf dem Mac kommt der LDR-Wert per Kommandozeile (`run_display.py --ldr `), +auf dem ESP32-C3 vom ADC an GPIO2 (`run_esp32.main(with_ldr=True)`). + +**Die Kennlinie ist auf der Platine umgekehrt.** Der Teiler lautet +`3V3 —[R2 2k]— A0 —[LDR]— GND`; der LDR wird bei mehr Licht niederohmiger, die +Spannung an A0 sinkt also mit steigender Helligkeit — genau andersherum als beim +alten MSP430-Aufbau, auf den die Schwellen oben zugeschnitten sind. Der Rohwert +wird deshalb gespiegelt: `config.LDR_INVERTED = True`, angewendet in +`brightness.raw_to_ldr`. Wer die Hardware lieber umbaut (LDR nach 3V3, R2 nach +GND), setzt das Flag auf `False`. + +Die Schwellen selbst sind noch die alten und **müssen am realen Aufbau +nachgemessen werden**, siehe Offene Punkte. ## Antwortleitung (RX) @@ -196,6 +384,8 @@ gegeneinander (verschliffener ~3-V-Pegel am Oszi). `loopback.py` grenzt so etwas inkl. echtem Gradzeichen; im `run_display`-Betrieb aktualisieren beide Zeilen laufend, sauber und ohne Flimmern, mit Antwort-Handshake der Anzeige. Protokoll, 19200 8E1, Zeichensatz und Zeilengeometrie stimmen. -- Schrift: Sperrschrift (`ESC z`, `CHARSET_SPACED = True`), 10 Zeichen/Zeile. +- Schrift: Sperrschrift (`ESC z`, `CHARSET_SPACED = True`), 9 Zeichen/Zeile. + (Hier stand ursprünglich „10 Zeichen/Zeile" — das war die Annahme vor der + Messung vom 15.07., siehe *Geometrie und Darstellung*.) Die schmale `1` wirkt dadurch etwas luftig — Font-Eigenschaft, nur per eigenem Font (microSYST-PC-Software) änderbar. diff --git a/brightness.py b/brightness.py index 758a214..203863a 100644 --- a/brightness.py +++ b/brightness.py @@ -13,8 +13,9 @@ Reines Python, laeuft auch unter MicroPython (dort spaeter vom ADC gespeist). python3 brightness.py --ldr 200 --dry-run """ -import argparse -import sys +# argparse und sys.stderr gibt es unter MicroPython nicht -- beides wird nur vom +# CLI-Teil ganz unten gebraucht und darum erst dort importiert. So laesst sich +# dieses Modul auf dem ESP32 einfach importieren. # -- Werte aus dem alten Programm (main.h) -------------------------------- HELL_LOW = 300 # Stufengrenze 0<->1 (aus dem alten Code) @@ -34,6 +35,22 @@ _CENTERS = (HELL_LOW, HELL_HIGH, HELL_MAX) _HYSTS = (HELL_HYST, HELL_HYST * 5, HELL_HYST * 5) +def raw_to_ldr(raw, inverted=None, full_scale=None): + """ADC-Rohwert in die Zaehlrichtung bringen, die die Schwellen erwarten. + + Die Schwellen oben stammen vom alten MSP430-Aufbau: hoher Wert = helle + Umgebung. Auf der ESP32-Platine liegt der LDR gegen Masse und der 2k gegen + 3V3, dort faellt die Spannung an A0 mit steigender Helligkeit. Der Wert wird + deshalb gespiegelt, wenn config.LDR_INVERTED gesetzt ist. + """ + import config + if inverted is None: + inverted = config.LDR_INVERTED + if full_scale is None: + full_scale = config.LDR_MAX + return (full_scale - raw) if inverted else raw + + class BrightnessController: """Rechnet aus LDR-Rohwerten die Anzeigehelligkeit -- wie DoDimmen_V. @@ -80,6 +97,9 @@ class BrightnessController: def main(argv=None): + import argparse + import sys + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument("--ldr", type=int, required=True, help="LDR-Rohwert (0..4095)") @@ -106,4 +126,5 @@ def main(argv=None): if __name__ == "__main__": + import sys sys.exit(main()) diff --git a/config.py b/config.py index 0af0f45..1aa493a 100644 --- a/config.py +++ b/config.py @@ -1,16 +1,38 @@ -"""Zentrale Einstellungen fuer die Ansteuerung der Grossanzeige.""" +"""Zentrale Einstellungen fuer die Ansteuerung der Grossanzeige. + +Laeuft unveraendert unter CPython (Mac) und MicroPython (XIAO ESP32-C3); was +sich zwischen beiden unterscheidet, haengt an MICROPYTHON. +""" + +import sys import migra +# True auf dem ESP32-C3, False auf dem Mac. Steuert, welchen Transport +# display.open_display waehlt und ob der LDR ueber den ADC gelesen werden kann. +MICROPYTHON = sys.implementation.name == "micropython" + # -- Serielle Schnittstelle ----------------------------------------------- # Auslieferzustand der Anzeige laut Handbuch (Kap. 4.1) und identisch zur # UART-Init des alten MSP430-Programms: 19200 8E1. -PORT = "/dev/cu.usbserial-120" # Mac; auf dem ESP32-C3 spaeter die UART-Nummer BAUDRATE = 19200 PARITY = "E" BYTESIZE = 8 STOPBITS = 1 +# CPython: Geraetedatei des USB-RS232-Wandlers. +PORT = "/dev/cu.usbserial-120" + +# ESP32-C3: UART1 mit frei zugeordneten Pins. Die Nummern stammen aus dem +# Schaltplan (Hardware/KiCad/Grossanzeige) -- am XIAO ist Pad 7 = D6 = GPIO21 +# und Pad 8 = D7 = GPIO20. Von dort geht es an den MAX3232: +# GPIO21 (TX) -> T2IN, T2OUT -> J5 Pin 3 -> Anzeige RxD (Sub-D Pin 2) +# GPIO20 (RX) <- R2OUT, R2IN <- J5 Pin 2 <- Anzeige TxD (Sub-D Pin 3) +# UART0 bleibt frei, das ist die REPL-Konsole ueber USB. +UART_ID = 1 +UART_TX = 21 +UART_RX = 20 + # -- Telegramm ------------------------------------------------------------ DISPLAY_ADDR = 1 # Zieladresse der Anzeige (DA = 0x81) HOST_ADDR = 0 # unsere Quelladresse (SA = 0x80) @@ -26,14 +48,21 @@ QUIET_TIME = 0.4 # Pause zwischen Telegrammen, wenn ohne Antwort gefahren # Zeichensatz 0, Zeile 1 ab Pixelzeile 0, Zeile 2 ab Pixelzeile 9 (aus dem alten # C-Programm). Am echten Geraet ausgemessen (2026-07-15): # CHARSET_SPACED=True -> ESC z (Sperrschrift, alle Zeichen gleich breit) -- vom -# Nutzer bevorzugt, weil sich sonst schmale Ziffern beruehren. In dieser Schrift -# passen mind. 10 Zeichen pro Zeile (RA/DEC brauchen max. 10). +# Nutzer bevorzugt, weil sich sonst schmale Ziffern beruehren. # CHARSET_SPACED=False -> ESC Z (proportional, schmaler, ~11 Zeichen, aber die "1" # klebt an Nachbarn). CHARSET = 0 CHARSET_SPACED = True LINE_Y = (0, 9) -CHARS_PER_LINE = 9 # in Sperrschrift am Geraet ausgemessen (2026-07-15) + +# 9 Zeichen: am Geraet ausgemessen (2026-07-15). Die Anzeige ist 64 px breit; +# eine reine 5x7-Matrix mit 1 px Luecke waere 6 px pro Zelle und ergaebe 10 +# Zeichen. Dass es nur 9 sind, passt zu 7 px pro Zelle -- die Sperrschrift +# sperrt eben, sie hat 2 px Luecke (9*7 = 63 von 64 px). Das ist die Erklaerung, +# nicht die Messung: "probe.py pitch 6" bzw. "pitch 7" entscheidet es direkt, +# indem es den Auto-Vorschub gegen feste Pixelabstaende stellt. +# Faende sich doch 6 px, passt die DEC-Form mit abschliessendem " (10 Zeichen). +CHARS_PER_LINE = 9 COLOR = migra.RED # einfarbig rote Anzeige BACKGROUND = migra.BLACK @@ -52,7 +81,7 @@ DEGREE_FALLBACK = "d" # Ersatz, falls DEGREE_CHARSET = None # -- Montierung (Phase 2) ------------------------------------------------- # 10Micron GM4000 HPS II: LX200-Protokoll ueber TCP. Die IP der Montierung # eintragen, sobald sie erreichbar ist; Port 3490 ist die Vorgabe von 10Micron. -MOUNT_HOST = "192.168.1.115" # PLATZHALTER -- echte IP der Montierung eintragen +MOUNT_HOST = "192.168.1.115" # echte IP der Montierung (eingetragen 2026-07-18) MOUNT_PORT = 3490 MOUNT_TIMEOUT = 3.0 # Sekunden fuer Verbindung/Antwort POLL_INTERVAL = 1.0 # Sekunden zwischen zwei Koordinatenabfragen @@ -65,4 +94,27 @@ POLL_INTERVAL = 1.0 # Sekunden zwischen zwei Koordinatenabfragen HIGH_PRECISION = False # Anzeige, wenn die Montierung (noch) nicht erreichbar ist. -NOLINK_LINES = ("KEIN LINK", "MONTIERUNG") +# Achtung: hoechstens CHARS_PER_LINE Zeichen, sonst wird abgeschnitten -- +# "MONTIERUNG" stand hier zuerst und erschien am Geraet als "MONTIERUN". +NOLINK_LINES = ("KEIN LINK", "TELESKOP") + +# -- WLAN (nur ESP32-C3) -------------------------------------------------- +# Zugangsdaten stehen in wifi_config.py (nicht im Git, siehe +# wifi_config_example.py). Hier nur das Zeitverhalten. +WIFI_TIMEOUT = 20.0 # Sekunden aufs Verbinden warten +WIFI_RETRY = 5.0 # Pause vor einem neuen Verbindungsversuch + +# Anzeige, solange noch kein WLAN steht. +NOWIFI_LINES = ("KEIN WLAN", " ... ") + +# -- Helligkeitssensor (LDR, nur ESP32-C3) -------------------------------- +# LDR an Pad 1 (D0/A0) = GPIO2, auf dem C3 ein Kanal von ADC1. +LDR_PIN = 2 + +# Der Spannungsteiler auf der Platine ist 3V3 --[R2 2k]-- A0 --[LDR]-- GND. +# Der LDR wird bei mehr Licht niederohmiger, die Spannung an A0 sinkt also mit +# steigender Helligkeit -- genau andersherum als beim alten MSP430-Aufbau, auf +# den die Schwellen in brightness.py zugeschnitten sind. Deshalb wird der +# Rohwert gespiegelt (siehe brightness.raw_to_ldr). +LDR_INVERTED = True +LDR_MAX = 4095 # 12-bit-ADC, voller Ausschlag diff --git a/deploy.sh b/deploy.sh new file mode 100755 index 0000000..1443d7e --- /dev/null +++ b/deploy.sh @@ -0,0 +1,50 @@ +#!/bin/sh +# Die Projektdateien auf den XIAO ESP32-C3 kopieren. +# +# ./deploy.sh # alle Dateien kopieren +# ./deploy.sh /dev/cu.usbmodem1101 # anderen Port erzwingen +# +# Braucht mpremote: pip install mpremote +# Danach am REPL (mpremote repl, mit Ctrl-] wieder raus): +# +# import run_esp32 +# run_esp32.main() + +set -e + +PORT="${1:-auto}" + +# Alles, was auf dem Geraet gebraucht wird. run_display.py, show_coords.py, +# probe.py, mount_mock.py und die Tests bleiben auf dem Mac -- sie brauchen +# argparse bzw. threading. +FILES=" +ticks.py +migra.py +config.py +transport.py +display.py +coords.py +lx200.py +mount_client.py +brightness.py +displayloop.py +wifi.py +run_esp32.py +" + +if [ ! -f wifi_config.py ]; then + echo "wifi_config.py fehlt." >&2 + echo " cp wifi_config_example.py wifi_config.py und ausfuellen" >&2 + exit 1 +fi + +for f in $FILES wifi_config.py; do + echo "-> $f" + mpremote connect "$PORT" fs cp "$f" ":$f" +done + +echo +echo "Fertig. Starten mit:" +echo " mpremote connect $PORT repl" +echo " >>> import run_esp32" +echo " >>> run_esp32.main()" diff --git a/display.py b/display.py index 23b13c1..2d4dc74 100644 --- a/display.py +++ b/display.py @@ -99,7 +99,9 @@ class Display: def _emit_line(self, du, y, text): """Eine Zeile bei Pixelzeile y ausgeben, auf volle Breite aufgefuellt.""" - text = text[:config.CHARS_PER_LINE].ljust(config.CHARS_PER_LINE) + # Ohne str.ljust: das kennt MicroPython nicht (am Geraet aufgefallen). + text = text[:config.CHARS_PER_LINE] + text += " " * (config.CHARS_PER_LINE - len(text)) du.charset(config.CHARSET, spaced=config.CHARSET_SPACED) du.cursor(0, y) for i, part in enumerate(text.split(config.DEGREE_MARKER)): @@ -121,15 +123,31 @@ class Display: self._t.close() -def open_display(port=config.PORT, fake=False, response=config.WANT_RESPONSE): - """Transport passend zur Plattform oeffnen und Display-Objekt liefern.""" +def open_display(port=None, fake=False, response=config.WANT_RESPONSE): + """Transport passend zur Plattform oeffnen und Display-Objekt liefern. + + Auf dem ESP32-C3 wird machine.UART genommen (Pins aus config), auf dem Mac + pyserial mit config.PORT. fake=True haengt stattdessen den Mitschreiber ein. + """ if fake: from transport import FakeTransport t = FakeTransport(src=config.HOST_ADDR, display_addr=config.DISPLAY_ADDR) + elif config.MICROPYTHON: + from transport import UartTransport + t = UartTransport( + uart_id=config.UART_ID, + baudrate=config.BAUDRATE, + parity=config.PARITY, + bytesize=config.BYTESIZE, + stopbits=config.STOPBITS, + timeout=config.RESPONSE_TIMEOUT, + tx=config.UART_TX, + rx=config.UART_RX, + ) else: from transport import SerialTransport t = SerialTransport( - port=port, + port=config.PORT if port is None else port, baudrate=config.BAUDRATE, parity=config.PARITY, bytesize=config.BYTESIZE, diff --git a/displayloop.py b/displayloop.py new file mode 100644 index 0000000..9c75c0d --- /dev/null +++ b/displayloop.py @@ -0,0 +1,112 @@ +"""Die Poll-Schleife: Koordinaten holen, formatieren, anzeigen. + +Herausgeloest aus run_display.py, damit Mac und ESP32-C3 wirklich denselben +Ablauf fahren. Auf dem Mac kommt drumherum die Kommandozeile (run_display.py), +auf dem ESP32 das WLAN-Setup (run_esp32.py) -- die Schleife selbst ist gleich. + +Reines Python ohne Abhaengigkeiten ausser den Projektmodulen; argparse und +sys.stderr bleiben bewusst draussen, beides gibt es unter MicroPython nicht. +""" + +import time + +import config +import coords +import migra +from mount_client import MountError + + +class LinkError(Exception): + """Die Netzverbindung fehlt (auf dem ESP32: kein WLAN).""" + + +def format_coordinates(ra_raw, dec_raw): + """Rohstrings der Montierung fuer die Anzeige aufbereiten.""" + return coords.format_ra(ra_raw), coords.format_dec(dec_raw) + + +def poll_loop(display, client, interval=None, once=False, + read_ldr=None, dimmer=None, ensure_link=None, log=None): + """Zyklisch die Montierung abfragen und die Anzeige nachfuehren. + + display offenes Display-Objekt + client MountClient (wird bei Bedarf selbst verbunden) + interval Sekunden zwischen zwei Abfragen (Vorgabe config.POLL_INTERVAL) + once nur ein Durchlauf, dann zurueck + read_ldr optional: Funktion ohne Argumente, liefert einen LDR-Rohwert + in der Zaehlrichtung "hoch = hell" (siehe brightness.raw_to_ldr) + dimmer optional: BrightnessController; noetig, wenn read_ldr gesetzt ist + ensure_link optional: Funktion, die vor jeder Abfrage die Netzverbindung + sicherstellt (auf dem ESP32 das WLAN) + log optional: Ausgabefunktion, Vorgabe print; None-freies Logging + erlaubt es, auf dem ESP32 still zu laufen (log=lambda *a: None) + + Laeuft bis KeyboardInterrupt oder -- bei once=True -- genau einmal. Fehler + von Anzeige und Montierung werden abgefangen: die Schleife meldet sie und + versucht es beim naechsten Durchlauf erneut, statt abzustuerzen. + """ + if interval is None: + interval = config.POLL_INTERVAL + if log is None: + log = print + + last_shown = None + while True: + try: + if ensure_link is not None: + ensure_link() + if not client.connected: + client.connect() + + # Helligkeit nachfuehren -- nur senden, wenn sich die Stufe aendert. + if read_ldr is not None and dimmer is not None: + raw = read_ldr() + level, percent, changed = dimmer.update(raw) + if changed: + display.set_brightness(percent) + log("LDR %d -> Helligkeit Stufe %d = %d %%" % (raw, level, percent)) + + ra_raw, dec_raw = client.get_coordinates() + ra_line, dec_line = format_coordinates(ra_raw, dec_raw) + + if (ra_line, dec_line) != last_shown: + # Bei jeder Aenderung beide Zeilen neu zeichnen. Kein Vollbild- + # Loeschen (kein Flimmern); die umgekehrte Zeilenreihenfolge in + # show_lines schuetzt die Home-Zelle vor dem Gradzeichen- + # Zeichensatzwechsel, auch bei staendiger Aktualisierung. + first = last_shown is None or last_shown in (config.NOLINK_LINES, + config.NOWIFI_LINES) + display.show_lines(ra_line, dec_line, clear=first) + last_shown = (ra_line, dec_line) + log("RA %-11s DEC %-12s -> %s | %s" + % (ra_raw, dec_raw, ra_line, dec_line)) + except LinkError as e: + # Kein Netz (WLAN): die Montierung gar nicht erst versuchen. + log("Netz: %s" % e) + client.close() + last_shown = _show_notice(display, config.NOWIFI_LINES, last_shown, log) + except migra.MigraError as e: + # Anzeige-/Antwortfehler (z.B. gestoerte Antwort): nicht abstuerzen, + # nur melden und beim naechsten Durchlauf neu zeichnen. + log("Anzeige: %s" % e) + last_shown = None + except (MountError, ValueError) as e: + log("Montierung: %s" % e) + client.close() + last_shown = _show_notice(display, config.NOLINK_LINES, last_shown, log) + + if once: + return last_shown + time.sleep(interval) + + +def _show_notice(display, lines, last_shown, log): + """Hinweiszeilen anzeigen, wenn sie nicht schon stehen.""" + if lines == last_shown: + return last_shown + try: + display.show_lines(lines[0], lines[1], clear=True) + except migra.MigraError as e: + log("Anzeige: %s" % e) + return None + return lines diff --git a/migra.py b/migra.py index 29bf739..bd0bbd6 100644 --- a/migra.py +++ b/migra.py @@ -38,6 +38,16 @@ class MigraError(Exception): """Fehler im Protokoll oder von der Anzeige gemeldeter Fehlercode.""" +def _hex(data): + """Bytes als Hexstring, ohne sich auf bytes.hex() zu verlassen. + + MicroPython 1.28 hat bytes.hex() (am Geraet nachgeprueft), aber nicht jede + Firmware ist damit gebaut. Wird nur in Fehlermeldungen gebraucht -- ein + AttributeError ausgerechnet dort wuerde den eigentlichen Fehler verdecken. + """ + return "".join("%02x" % b for b in data) + + def encode_text(s): """Text in Anzeige-Bytes wandeln. @@ -56,10 +66,16 @@ def encode_text(s): def _digits(value, places): - """Zahl als ASCII-Ziffern fester Laenge (Handbuch 3.7: alle Werte ASCII-codiert).""" + """Zahl als ASCII-Ziffern fester Laenge (Handbuch 3.7: alle Werte ASCII-codiert). + + Von Hand aufgefuellt statt mit "%0*d". MicroPython 1.28 beherrscht die + dynamische Feldbreite zwar (am Geraet nachgeprueft), sie stand aber lange + als nicht unterstuetzt in den Docs -- so bleibt es unabhaengig davon. + """ if value < 0 or value >= 10 ** places: raise MigraError("Wert %d passt nicht in %d Stellen" % (value, places)) - return ("%0*d" % (places, value)).encode("ascii") + s = "%d" % value + return ("0" * (places - len(s)) + s).encode("ascii") class DataUnit: @@ -175,9 +191,9 @@ def parse_response(frame, src=0): if not frame: raise MigraError("keine Antwort von der Anzeige (Timeout)") if len(frame) < 6 or frame[0] != STX or frame[-1] != ETX: - raise MigraError("unplausibles Antworttelegramm: %s" % frame.hex()) + raise MigraError("unplausibles Antworttelegramm: %s" % _hex(frame)) if frame[1] != (0x80 | src): - raise MigraError("Antwort nicht an uns adressiert: %s" % frame.hex()) + raise MigraError("Antwort nicht an uns adressiert: %s" % _hex(frame)) code = chr(frame[4]) if code != "0": diff --git a/mount_client.py b/mount_client.py index f7e8159..63d9a89 100644 --- a/mount_client.py +++ b/mount_client.py @@ -29,10 +29,18 @@ class MountClient: def connect(self): self.close() + # Ueber getaddrinfo statt direkt mit (host, port): MicroPython kann ein + # Adresstupel mit Hostnamen nicht selbst aufloesen, mit der aufgeloesten + # Adresse laeuft connect() auf beiden Seiten gleich. + try: + addr = socket.getaddrinfo(self.host, self.port)[0][-1] + except (OSError, IndexError) as e: + raise MountError("Adresse %s:%d nicht aufloesbar: %s" + % (self.host, self.port, e)) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(self.timeout) try: - sock.connect((self.host, self.port)) + sock.connect(addr) except OSError as e: sock.close() raise MountError("Verbindung zu %s:%d fehlgeschlagen: %s" @@ -60,10 +68,19 @@ class MountClient: """Ein gerahmtes Kommando senden und die Antwort ohne '#' zurueckgeben.""" if not self.connected: raise MountError("nicht verbunden") - try: - self._sock.sendall(framed_cmd.encode("latin-1")) - except OSError as e: - raise MountError("Senden gestoert: %s" % e) + # Von Hand schleifen statt sendall(): MicroPython laesst dessen Verhalten + # auf Sockets mit Timeout ausdruecklich offen. Die Kommandos sind vier + # Bytes lang, die Schleife laeuft also praktisch immer nur einmal. + data = framed_cmd.encode("latin-1") + sent = 0 + while sent < len(data): + try: + n = self._sock.send(data[sent:]) + except OSError as e: + raise MountError("Senden gestoert: %s" % e) + if not n: + raise MountError("Verbindung von der Montierung geschlossen") + sent += n raw = self._recv_until_terminator().decode("latin-1") return lx200.strip_terminator(raw) diff --git a/mount_mock.py b/mount_mock.py index 3bfed0d..e9c7793 100644 --- a/mount_mock.py +++ b/mount_mock.py @@ -8,8 +8,13 @@ Client nicht haengt. python3 mount_mock.py # feste Koordinaten, Port 3490 python3 mount_mock.py --ra 18:36:56 --dec +38:47:01 python3 mount_mock.py --move # RA laeuft (Sekunden ticken hoch) + python3 mount_mock.py --lan --move # auch vom ESP32 aus erreichbar -Nur fuer die Entwicklung (CPython). Auf dem ESP32 laeuft spaeter nur der Client. +Vorgabe ist 127.0.0.1, also nur lokal. Damit der ESP32 den Mock uebers WLAN +erreicht, braucht es --lan (bindet auf 0.0.0.0); die Adresse, die dann auf dem +ESP32 einzutragen ist, wird beim Start ausgegeben. + +Nur fuer die Entwicklung (CPython). Auf dem ESP32 laeuft nur der Client. """ import argparse @@ -172,20 +177,47 @@ class MountMock: self.stop() +def lan_address(): + """Die eigene LAN-Adresse ermitteln -- die, die der ESP32 ansprechen muss. + + Ueber einen UDP-Socket ans Gateway: der verschickt nichts, verraet aber, + welche lokale Adresse das Betriebssystem fuer den Weg nach draussen nimmt. + """ + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + s.connect(("192.168.1.1", 1)) + return s.getsockname()[0] + except OSError: + return None + finally: + s.close() + + def main(argv=None): p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument("--host", default="127.0.0.1") + p.add_argument("--lan", action="store_true", + help="auf allen Schnittstellen lauschen (0.0.0.0), damit der " + "ESP32 den Mock uebers WLAN erreicht") p.add_argument("--port", type=int, default=3490) p.add_argument("--ra", default="18:36:56") p.add_argument("--dec", default="+38:47:01") p.add_argument("--move", action="store_true", help="RA laeuft mit der Zeit hoch") args = p.parse_args(argv) - mock = MountMock(host=args.host, port=args.port, ra=args.ra, dec=args.dec, + host = "0.0.0.0" if args.lan else args.host + mock = MountMock(host=host, port=args.port, ra=args.ra, dec=args.dec, move=args.move) print("Mock-Montierung auf %s:%d (RA=%s DEC=%s%s)" % (mock.host, mock.port, args.ra, args.dec, ", move" if args.move else "")) + if host == "0.0.0.0": + ip = lan_address() + if ip: + print("Vom ESP32 aus erreichbar unter: %s:%d" % (ip, mock.port)) + print(" run_esp32.main(host=\"%s\", fake_display=True)" % ip) + print("Falls keine Verbindung zustande kommt: macOS fragt beim ersten Mal,") + print("ob python3 eingehende Verbindungen annehmen darf -- das muss erlaubt sein.") print("Beenden mit Ctrl-C.") mock.serve_forever() return 0 diff --git a/probe.py b/probe.py index 1939d21..40b0cb7 100644 --- a/probe.py +++ b/probe.py @@ -5,6 +5,8 @@ LINE_Y und CHARS_PER_LINE in config.py sind aus dem alten C-Programm abgeleitet, nicht aus einem Datenblatt. Dieses Skript hilft, die echten Werte zu finden. python3 probe.py ruler Ziffernlineal 12345678901234... -> Zeichen zaehlen + python3 probe.py pitch 6 Zellenbreite messen: Auto-Vorschub gegen 6 px + python3 probe.py pitch 7 dasselbe gegen 7 px python3 probe.py lines Zeile 1 und Zeile 2 an den Positionen aus config.py python3 probe.py scan-y dieselbe Zeile schrittweise nach unten schieben python3 probe.py corners Punkte in die vier Ecken -> Pixelaufloesung pruefen @@ -37,6 +39,51 @@ def ruler(d): print("Das ist CHARS_PER_LINE (aktuell in config.py: %d)." % config.CHARS_PER_LINE) +def pitch(d, cell=6): + """Zellenbreite direkt messen, statt Zeichen zu zaehlen. + + Der Cursor arbeitet auf *Pixel*positionen -- das laesst sich ausnutzen: + + Zeile 1: zehn Ziffern als ein Text ab x=0. Die Anzeige rueckt selbst vor, + mit der echten Zellenbreite. + Zeile 2: dieselben Ziffern einzeln auf x = i*cell gesetzt. + + Stehen beide Zeilen exakt untereinander, ist `cell` die Zellenbreite. Ein + Unterschied von nur 1 px summiert sich ueber zehn Zeichen auf 9 px auf und + ist nicht zu uebersehen -- viel eindeutiger als das Abzaehlen am Umbruch. + + 5x7-Matrix mit 1 px Luecke waere cell=6 (dann passen 10 Zeichen in 64 px), + Sperrschrift mit 2 px Luecke waere cell=7 (dann 9 Zeichen). + """ + ziffern = "0123456789" + + du = migra.DataUnit() + du.fill(config.BACKGROUND) + du.attributes(fg=config.COLOR, bg=config.BACKGROUND) + du.charset(config.CHARSET, spaced=config.CHARSET_SPACED) + + # Untere Zeile zuerst -- der Zeichensatzwechsel wuerde sonst die Home-Zelle + # der oberen Zeile beschaedigen (siehe display.show_lines). + for i, ch in enumerate(ziffern): + x = i * cell + if x + cell > 64: + break + du.cursor(x, config.LINE_Y[1]) + du.text(ch) + + du.cursor(0, config.LINE_Y[0]) + du.text(ziffern) + d.send(du) + + print("Obere Zeile: Anzeige rueckt selbst vor.") + print("Untere Zeile: Ziffern fest auf x = i*%d gesetzt." % cell) + print("") + print("Stehen sie exakt untereinander? -> Zellenbreite ist %d px," % cell) + print("also CHARS_PER_LINE = %d bei 64 px Breite." % (64 // cell)) + print("Laeuft die obere Zeile nach rechts weg? -> Zelle ist breiter als %d px." % cell) + print("Dann mit einem groesseren Wert wiederholen: probe.py pitch %d" % (cell + 1)) + + def lines(d): d.show_lines("Zeile 1", "Zeile 2") print("Stehen beide Zeilen sauber untereinander? Sonst LINE_Y anpassen.") @@ -71,18 +118,28 @@ def corners(d): print("Punkt %3d/%-3d -> %s" % (x, y, e)) -MODES = {"ruler": ruler, "lines": lines, "scan-y": scan_y, "corners": corners} +MODES = {"ruler": ruler, "pitch": pitch, "lines": lines, "scan-y": scan_y, + "corners": corners} def main(): args = [a for a in sys.argv[1:] if not a.startswith("-")] - if len(args) != 1 or args[0] not in MODES: + if not args or args[0] not in MODES: print(__doc__) return 2 + + extra = [] + if args[0] == "pitch" and len(args) > 1: + try: + extra = [int(args[1])] + except ValueError: + print("pitch braucht eine Pixelzahl, z.B.: probe.py pitch 6") + return 2 + response = config.WANT_RESPONSE and "--no-response" not in sys.argv d = open_display(response=response) try: - MODES[args[0]](d) + MODES[args[0]](d, *extra) finally: d.close() return 0 diff --git a/run_display.py b/run_display.py index eb37ef5..16e409b 100644 --- a/run_display.py +++ b/run_display.py @@ -13,19 +13,12 @@ Hintergrund erneut verbunden. import argparse import sys -import time import config -import coords -import migra +import displayloop from brightness import BrightnessController from display import open_display -from mount_client import MountClient, MountError - - -def _format(ra_raw, dec_raw): - """Rohstrings der Montierung fuer die Anzeige aufbereiten.""" - return coords.format_ra(ra_raw), coords.format_dec(dec_raw) +from mount_client import MountClient def main(argv=None): @@ -56,54 +49,20 @@ def main(argv=None): d = open_display(fake=args.dry_run) client = MountClient(host, port, timeout=config.MOUNT_TIMEOUT) - last_shown = None - # Helligkeitsregelung: derzeit ein fester LDR-Wert von der Kommandozeile; - # spaeter (ESP32) wird hier pro Durchlauf der ADC gelesen. - dimmer = BrightnessController() if args.ldr is not None else None + # Helligkeitsregelung: hier ein fester LDR-Wert von der Kommandozeile; auf + # dem ESP32 liest run_esp32.py stattdessen den ADC. + dimmer = read_ldr = None + if args.ldr is not None: + dimmer = BrightnessController() + read_ldr = lambda: args.ldr try: - while True: - try: - if not client.connected: - client.connect() - - # Helligkeit nachfuehren (nur senden, wenn sich die Stufe aendert) - if dimmer is not None: - level, percent, changed = dimmer.update(args.ldr) - if changed: - d.set_brightness(percent) - print("LDR %d -> Helligkeit Stufe %d = %d %%" - % (args.ldr, level, percent)) - - ra_raw, dec_raw = client.get_coordinates() - ra_line, dec_line = _format(ra_raw, dec_raw) - - if (ra_line, dec_line) != last_shown: - # Bei jeder Aenderung beide Zeilen neu zeichnen. Kein Vollbild- - # Loeschen (kein Flimmern); die umgekehrte Zeilenreihenfolge in - # show_lines schuetzt die Home-Zelle vor dem Gradzeichen- - # Zeichensatzwechsel, auch bei staendiger Aktualisierung. - first = last_shown is None or last_shown == config.NOLINK_LINES - d.show_lines(ra_line, dec_line, clear=first) - last_shown = (ra_line, dec_line) - print("RA %-11s DEC %-12s -> %s | %s" - % (ra_raw, dec_raw, ra_line, dec_line)) - except migra.MigraError as e: - # Anzeige-/Antwortfehler (z.B. gestoerte Antwort): nicht abstuerzen, - # nur melden und beim naechsten Durchlauf neu zeichnen. - print("Anzeige: %s" % e, file=sys.stderr) - last_shown = None - except (MountError, ValueError) as e: - print("Montierung: %s" % e, file=sys.stderr) - client.close() - if config.NOLINK_LINES != last_shown: - d.show_lines(*config.NOLINK_LINES, clear=True) - last_shown = config.NOLINK_LINES - - if args.once: - break - time.sleep(args.interval) + displayloop.poll_loop(d, client, + interval=args.interval, + once=args.once, + read_ldr=read_ldr, + dimmer=dimmer) except KeyboardInterrupt: print("\nbeendet") finally: diff --git a/run_esp32.py b/run_esp32.py new file mode 100644 index 0000000..47026cd --- /dev/null +++ b/run_esp32.py @@ -0,0 +1,142 @@ +"""Hauptprogramm auf dem XIAO ESP32-C3 (MicroPython). + +Das Gegenstueck zu run_display.py: dort die Kommandozeile auf dem Mac, hier das +WLAN-Setup auf dem Geraet. Die Poll-Schleife selbst ist dieselbe (displayloop). + +Am REPL starten: + + import run_esp32 + run_esp32.selftest(host="192.168.178.26") # Bring-up pruefen + run_esp32.main(host="192.168.178.26", fake_display=True) # gegen den Mock + run_esp32.main() # echte Montierung, echte Anzeige + +Absichtlich *nicht* main.py genannt: MicroPython startet eine Datei dieses +Namens nach jedem Reset von selbst. Solange wir am Aufbau arbeiten, ist der +manuelle Start angenehmer -- der Autostart ist spaeter eine main.py mit den +zwei Zeilen von oben. + +Verdrahtung (aus dem Schaltplan, siehe Hardware/KiCad/Grossanzeige): + + XIAO Pad 7 (D6/GPIO21) --> MAX3232 T2IN, T2OUT --> Anzeige RxD (Sub-D 2) + XIAO Pad 8 (D7/GPIO20) <-- MAX3232 R2OUT, R2IN <-- Anzeige TxD (Sub-D 3) + XIAO Pad 1 (D0/GPIO2) <-- LDR-Teiler (3V3 --[2k]-- A0 --[LDR]-- GND) + MAX3232 VCC = 3V3 vom XIAO, GND gemeinsam +""" + +import config +import displayloop +import wifi +from brightness import BrightnessController, raw_to_ldr +from display import open_display +from mount_client import MountClient + +try: # MicroPython + from machine import ADC, Pin +except ImportError: + ADC = Pin = None + + +def make_ldr_reader(): + """Funktion liefern, die den LDR als Wert in Zaehlrichtung "hoch = hell" gibt. + + Gibt None zurueck, wenn kein ADC verfuegbar ist -- dann laeuft die Anzeige + einfach mit fester Helligkeit weiter. + """ + if ADC is None: + return None + adc = ADC(Pin(config.LDR_PIN)) + # Voller Eingangsbereich bis ~3,3 V; ohne Daempfung endet der ADC bei ~1 V. + try: + adc.atten(ADC.ATTN_11DB) + except AttributeError: + pass + + def read(): + # read_u16() gibt es auf allen Ports; auf 12 bit zurueckskalieren, weil + # die Schwellen in brightness.py aus dem 12-bit-ADC des MSP430 stammen. + raw = adc.read_u16() * config.LDR_MAX // 65535 + return raw_to_ldr(raw) + + return read + + +def selftest(host=None, port=None, log=print): + """Bring-up in einem Rutsch pruefen, ohne die Dauerschleife zu starten. + + WLAN aufbauen, IP zeigen, einmal die Montierung (oder den Mock auf dem Mac) + abfragen und ein Anzeige-Telegramm bauen -- letzteres nur als Hex, ohne + serielle Hardware. Damit laesst sich Stufe fuer Stufe eingrenzen, wo es + klemmt, bevor Anzeige und Pegelwandler ueberhaupt dranhaengen. + + import run_esp32 + run_esp32.selftest(host="192.168.178.26") + """ + log("1) WLAN ...") + cfg = wifi.connect(log=log) + log(" IP %s, Gateway %s" % (cfg[0], cfg[2])) + + host = config.MOUNT_HOST if host is None else host + port = config.MOUNT_PORT if port is None else port + log("2) Montierung %s:%d ..." % (host, port)) + client = MountClient(host, port, timeout=config.MOUNT_TIMEOUT) + try: + client.connect() + ra_raw, dec_raw = client.get_coordinates() + log(" RA %s DEC %s" % (ra_raw, dec_raw)) + finally: + client.close() + + log("3) Anzeige-Telegramm (nur Hex, keine Hardware) ...") + d = open_display(fake=True) + try: + d.show_lines(*displayloop.format_coordinates(ra_raw, dec_raw), clear=True) + finally: + d.close() + log("Selbsttest bestanden.") + + +def main(host=None, port=None, with_ldr=False, fake_display=False, + once=False, log=print): + """WLAN aufbauen, Anzeige oeffnen und die Poll-Schleife fahren. + + host/port Montierung; Vorgabe aus config.py. Zum Testen gegen den Mock + auf dem Mac dessen LAN-Adresse angeben. + with_ldr Helligkeitsregelung ueber den ADC dazuschalten. Vorgabe aus, + solange die Schwellen am realen LDR nicht nachgemessen sind + (siehe README, Abschnitt Helligkeitsregelung). + fake_display Anzeige nur als Hex in die REPL schreiben, statt ueber den + UART zu senden -- fuer den Test ohne Pegelwandler und Anzeige. + once nur ein Durchlauf, dann zurueck. + """ + d = open_display(fake=fake_display) + client = MountClient(config.MOUNT_HOST if host is None else host, + config.MOUNT_PORT if port is None else port, + timeout=config.MOUNT_TIMEOUT) + + dimmer = read_ldr = None + if with_ldr: + read_ldr = make_ldr_reader() + if read_ldr is None: + log("LDR: kein ADC verfuegbar, Helligkeitsregelung bleibt aus") + else: + dimmer = BrightnessController() + + try: + # Einmal vorab verbinden, damit ein Fehler gleich sichtbar wird; danach + # haelt ensure_link die Verbindung in der Schleife. + try: + wifi.connect(log=log) + except wifi.WifiError as e: + log("WLAN: %s" % e) + + displayloop.poll_loop(d, client, + once=once, + read_ldr=read_ldr, + dimmer=dimmer, + ensure_link=lambda: wifi.ensure(log=log), + log=log) + except KeyboardInterrupt: + log("\nbeendet") + finally: + client.close() + d.close() diff --git a/test_display.py b/test_display.py new file mode 100644 index 0000000..419300e --- /dev/null +++ b/test_display.py @@ -0,0 +1,91 @@ +"""Tests fuer den Anzeigetreiber -- Zeilenaufbau und die festen Hinweistexte. + + python3 -m unittest test_display +""" + +import unittest + +import config +import migra +from display import Display +from transport import FakeTransport + + +def _lesbar(frame): + """Telegramm als lesbarer Text; Steuerzeichen werden zu '.'.""" + return "".join(chr(b) if 32 <= b < 127 else "." for b in frame) + + +class TestFesteTexte(unittest.TestCase): + """Die Hinweiszeilen muessen in die Anzeige passen. + + Am Geraet aufgefallen: "MONTIERUNG" hat 10 Zeichen, die Anzeige fasst 9 -- + angezeigt wurde "MONTIERUN". _emit_line schneidet stillschweigend ab, also + faellt so etwas sonst erst auf der Anzeige auf. + """ + + def test_hinweiszeilen_passen_in_die_zeilenbreite(self): + for name in ("NOLINK_LINES", "NOWIFI_LINES"): + for zeile in getattr(config, name): + self.assertLessEqual( + len(zeile), config.CHARS_PER_LINE, + "config.%s: %r hat %d Zeichen, die Anzeige fasst %d" + % (name, zeile, len(zeile), config.CHARS_PER_LINE)) + + def test_es_gibt_nicht_mehr_hinweiszeilen_als_zeilen(self): + for name in ("NOLINK_LINES", "NOWIFI_LINES"): + self.assertLessEqual(len(getattr(config, name)), len(config.LINE_Y)) + + +class TestZeilenaufbau(unittest.TestCase): + + def setUp(self): + self.t = FakeTransport(echo=False) + self.d = Display(self.t, response=True) + + def _letztes(self): + return self.t.frames[-1] + + def test_zeile_wird_auf_volle_breite_aufgefuellt(self): + """Ohne Auffuellen blieben Reste der vorigen Anzeige stehen. + + Gleichzeitig die Regressionsprobe fuer str.ljust, das MicroPython + nicht kennt (am Geraet aufgefallen). + """ + self.d.show_lines("AB", "CD") + text = _lesbar(self._letztes()) + self.assertIn("AB" + " " * (config.CHARS_PER_LINE - 2), text) + self.assertIn("CD" + " " * (config.CHARS_PER_LINE - 2), text) + + def test_zu_lange_zeile_wird_abgeschnitten(self): + """Zu breit aufgefuellt bricht das Fuell-Leerzeichen um und + beschaedigt die andere Zeile -- deshalb hart abschneiden.""" + lang = "X" * (config.CHARS_PER_LINE + 5) + self.d.show_lines(lang) + text = _lesbar(self._letztes()) + self.assertIn("X" * config.CHARS_PER_LINE, text) + self.assertNotIn("X" * (config.CHARS_PER_LINE + 1), text) + + def test_oberste_zeile_wird_zuletzt_gezeichnet(self): + """Schutz der Home-Zelle: ein Zeichensatzwechsel mitten im Telegramm + zerstoert (0,0), die oberste Zeile muss danach kommen.""" + self.d.show_lines("OBEN", "UNTEN") + text = _lesbar(self._letztes()) + self.assertLess(text.index("UNTEN"), text.index("OBEN")) + + def test_clear_erzeugt_ein_vollbild_loeschen(self): + self.d.show_lines("A", "B", clear=True) + mit = self._letztes() + self.d.show_lines("A", "B", clear=False) + ohne = self._letztes() + fuellen = bytes([migra.ESC]) + b"F" + config.BACKGROUND.encode("ascii") + self.assertIn(fuellen, mit) + self.assertNotIn(fuellen, ohne) + + def test_mehr_zeilen_als_die_anzeige_hat(self): + with self.assertRaises(ValueError): + self.d.show_lines(*(["X"] * (len(config.LINE_Y) + 1))) + + +if __name__ == "__main__": + unittest.main() diff --git a/test_esp32.py b/test_esp32.py new file mode 100644 index 0000000..65409df --- /dev/null +++ b/test_esp32.py @@ -0,0 +1,605 @@ +"""Tests fuer den MicroPython-Teil -- ohne ESP32, mit Ersatzmodulen. + +machine und network gibt es auf dem Mac nicht. Hier werden sie durch Attrappen +ersetzt, damit sich UART-Transport, WLAN-Anbindung, ADC-Auswertung und die +ueberlaufsichere Zeitrechnung auch ohne Geraet pruefen lassen. Getestet wird +damit die Logik, nicht die Hardware -- ob 19200 8E1 auf dem realen UART +ankommen, zeigt erst der Aufbau. + + python3 -m unittest test_esp32 +""" + +import importlib +import sys +import types +import unittest + +import migra +import ticks + + +# -- Ersatzmodule ---------------------------------------------------------- + +class FakeUART: + """machine.UART-Attrappe: merkt sich Parameter und Gesendetes.""" + + last = None + + def __init__(self, uart_id, **kwargs): + self.uart_id = uart_id + self.kwargs = kwargs + self.rx = bytearray() # was die Anzeige "zurueckschickt" + self.written = [] + self.deinit_called = False + FakeUART.last = self + + def any(self): + return len(self.rx) + + def read(self, n=None): + if n is None: + n = len(self.rx) + chunk = bytes(self.rx[:n]) + del self.rx[:n] + return chunk or None + + def write(self, data): + self.written.append(bytes(data)) + return len(data) + + def deinit(self): + self.deinit_called = True + + +class FakePin: + def __init__(self, number): + self.number = number + + def __eq__(self, other): + return isinstance(other, FakePin) and other.number == self.number + + def __repr__(self): + return "FakePin(%d)" % self.number + + +class FakeADC: + ATTN_11DB = 3 + + last = None + + def __init__(self, pin): + self.pin = pin + self.value = 0 + self.attenuation = None + FakeADC.last = self + + def atten(self, value): + self.attenuation = value + + def read_u16(self): + return self.value + + +class FakeWLAN: + """network.WLAN-Attrappe. connect() gelingt, wenn succeed gesetzt ist.""" + + succeed = True + last = None + + def __init__(self, mode): + self.mode = mode + self._active = False + self._connected = False + self.credentials = None + self.disconnect_calls = 0 + FakeWLAN.last = self + + def active(self, value=None): + if value is not None: + self._active = value + return self._active + + def connect(self, ssid, password): + self.credentials = (ssid, password) + self._connected = FakeWLAN.succeed + + def isconnected(self): + return self._connected + + def disconnect(self): + self.disconnect_calls += 1 + self._connected = False + + def ifconfig(self): + return ("192.168.1.77", "255.255.255.0", "192.168.1.1", "192.168.1.1") + + +def _install_machine(): + """Ein Ersatz-machine bereitstellen und transport neu laden.""" + m = types.ModuleType("machine") + m.UART = FakeUART + m.Pin = FakePin + m.ADC = FakeADC + sys.modules["machine"] = m + return m + + +def _install_network(): + n = types.ModuleType("network") + n.STA_IF = 0 + n.WLAN = FakeWLAN + sys.modules["network"] = n + return n + + +def _remove(*names): + for name in names: + sys.modules.pop(name, None) + + +# -- UART-Transport -------------------------------------------------------- + +class TestUartTransport(unittest.TestCase): + + def setUp(self): + _install_machine() + import transport + self.transport = importlib.reload(transport) + import config + self.config = config + + def tearDown(self): + _remove("machine") + import transport + importlib.reload(transport) # zurueck auf UART = None + + def _open(self, timeout=1.0): + return self.transport.UartTransport( + uart_id=self.config.UART_ID, + baudrate=self.config.BAUDRATE, + parity=self.config.PARITY, + bytesize=self.config.BYTESIZE, + stopbits=self.config.STOPBITS, + timeout=timeout, + tx=self.config.UART_TX, + rx=self.config.UART_RX, + ) + + def test_uart_parameter_aus_config(self): + """19200 8E1 auf UART1, Pins wie im Schaltplan (GPIO21/GPIO20).""" + self._open() + u = FakeUART.last + self.assertEqual(u.uart_id, 1) + self.assertEqual(u.kwargs["baudrate"], 19200) + self.assertEqual(u.kwargs["bits"], 8) + self.assertEqual(u.kwargs["parity"], 0) # 0 = even + self.assertEqual(u.kwargs["stop"], 1) + self.assertEqual(u.kwargs["tx"], FakePin(21)) + self.assertEqual(u.kwargs["rx"], FakePin(20)) + + def test_write_leert_erst_den_empfangspuffer(self): + """Stoerbytes vor dem Senden verwerfen -- wie SerialTransport.""" + t = self._open() + FakeUART.last.rx.extend(b"\xff\x00Muell") + t.write(b"\x02\x81\x80\x81\x03") + self.assertEqual(FakeUART.last.any(), 0) + self.assertEqual(FakeUART.last.written, [b"\x02\x81\x80\x81\x03"]) + + def test_read_response_liest_bis_etx(self): + t = self._open() + FakeUART.last.rx.extend(b"\x02\x80\x81\x800\x03") + frame = t.read_response() + self.assertEqual(frame, b"\x02\x80\x81\x800\x03") + self.assertEqual(migra.parse_response(frame), "0") + + def test_read_response_schneidet_nachlauf_ab(self): + """Nach ETX ist Schluss; Nachlauf wuerde parse_response scheitern lassen.""" + t = self._open() + FakeUART.last.rx.extend(b"\x02\x80\x81\x800\x03\xff\xfe") + self.assertEqual(t.read_response(), b"\x02\x80\x81\x800\x03") + + def test_read_response_timeout_liefert_leer(self): + t = self._open(timeout=0.05) + self.assertEqual(t.read_response(), b"") + + def test_close_gibt_den_uart_frei(self): + t = self._open() + t.close() + self.assertTrue(FakeUART.last.deinit_called) + + +class TestOpenDisplayAufMicropython(unittest.TestCase): + """open_display muss auf dem ESP32 den UART nehmen, nicht pyserial.""" + + def setUp(self): + _install_machine() + import transport + importlib.reload(transport) + import config + self.config = config + self._orig = config.MICROPYTHON + config.MICROPYTHON = True + + def tearDown(self): + self.config.MICROPYTHON = self._orig + _remove("machine") + import transport + importlib.reload(transport) + + def test_waehlt_uart_transport(self): + import display + d = display.open_display() + self.assertEqual(FakeUART.last.uart_id, self.config.UART_ID) + d.close() + + +# -- Zeitrechnung ---------------------------------------------------------- + +class FakeTicks: + """time.ticks_*-Attrappe von MicroPython, inklusive Ueberlauf bei 2^30.""" + + MASK = (1 << 30) - 1 + + def __init__(self, now=0): + self.now = now + + def ticks_ms(self): + return self.now + + def ticks_add(self, t, delta): + return (t + delta) & self.MASK + + def ticks_diff(self, a, b): + d = (a - b) & self.MASK + if d > self.MASK // 2: + d -= self.MASK + 1 + return d + + +class TestTicks(unittest.TestCase): + + def setUp(self): + self._orig_time = ticks.time + self.clock = FakeTicks() + ticks.time = self.clock + + def tearDown(self): + ticks.time = self._orig_time + + def test_frist_laeuft_normal_ab(self): + self.clock.now = 1000 + end = ticks.ticks_deadline(1.5) + self.assertFalse(ticks.ticks_expired(end)) + self.clock.now = 2499 + self.assertFalse(ticks.ticks_expired(end)) + self.clock.now = 2500 + self.assertTrue(ticks.ticks_expired(end)) + + def test_frist_ueberlebt_den_ueberlauf(self): + """Der Kern: 500 ms vor dem Ueberlauf eine 1-s-Frist setzen. + + Mit "jetzt + timeout" und einfachem Vergleich haenge die Warteschleife + hier fast 12 Tage, weil der Zaehler zurueckspringt. Mit ticks_diff + laeuft die Frist wie erwartet nach 1000 ms ab. + """ + self.clock.now = FakeTicks.MASK - 500 + end = ticks.ticks_deadline(1.0) + self.assertLess(end, self.clock.now) # der Zaehler ist umgelaufen + + self.assertFalse(ticks.ticks_expired(end)) # sofort: noch nicht + self.clock.now = 400 # 900 ms spaeter + self.assertFalse(ticks.ticks_expired(end)) + self.clock.now = 499 # genau 1000 ms + self.assertTrue(ticks.ticks_expired(end)) + self.clock.now = 600 + self.assertTrue(ticks.ticks_expired(end)) + + +class TestTicksCPython(unittest.TestCase): + """Eigene Klasse: hier laeuft die echte Uhr, nicht die Attrappe.""" + + def test_monotonic_variante(self): + self.assertTrue(ticks.monotonic_expired(ticks.monotonic_deadline(0.0))) + self.assertFalse(ticks.monotonic_expired(ticks.monotonic_deadline(30.0))) + + def test_auf_dem_mac_ist_monotonic_ausgewaehlt(self): + self.assertIs(ticks.deadline, ticks.monotonic_deadline) + self.assertIs(ticks.expired, ticks.monotonic_expired) + + +# -- Helligkeit ------------------------------------------------------------ + +class TestLdrKennlinie(unittest.TestCase): + """Der Teiler auf der Platine dreht die Kennlinie um (siehe config).""" + + def test_invertiert_spiegelt_den_rohwert(self): + import brightness + self.assertEqual(brightness.raw_to_ldr(0, inverted=True, full_scale=4095), 4095) + self.assertEqual(brightness.raw_to_ldr(4095, inverted=True, full_scale=4095), 0) + self.assertEqual(brightness.raw_to_ldr(1000, inverted=True, full_scale=4095), 3095) + + def test_nicht_invertiert_laesst_durch(self): + import brightness + self.assertEqual(brightness.raw_to_ldr(1234, inverted=False), 1234) + + def test_vorgabe_kommt_aus_config(self): + import brightness + import config + self.assertEqual(brightness.raw_to_ldr(0), + config.LDR_MAX if config.LDR_INVERTED else 0) + + def test_helle_umgebung_ergibt_hohe_stufe(self): + """Viel Licht -> niedrige ADC-Spannung -> nach Spiegelung hoher Wert.""" + import brightness + hell = brightness.raw_to_ldr(100, inverted=True, full_scale=4095) + dunkel = brightness.raw_to_ldr(4000, inverted=True, full_scale=4095) + c = brightness.BrightnessController() + self.assertEqual(c.update(hell)[0], len(brightness.PERCENTS) - 1) + c = brightness.BrightnessController() + self.assertEqual(c.update(dunkel)[0], 0) + + +class TestLdrLeser(unittest.TestCase): + """run_esp32.make_ldr_reader: ADC lesen, auf 12 bit bringen, spiegeln.""" + + def setUp(self): + _install_machine() + import transport + importlib.reload(transport) + _remove("run_esp32") + import run_esp32 + self.run_esp32 = run_esp32 + + def tearDown(self): + _remove("machine", "run_esp32") + import transport + importlib.reload(transport) + + def test_liest_den_konfigurierten_pin(self): + import config + self.run_esp32.make_ldr_reader() + self.assertEqual(FakeADC.last.pin, FakePin(config.LDR_PIN)) + self.assertEqual(FakeADC.last.attenuation, FakeADC.ATTN_11DB) + + def test_skaliert_und_spiegelt(self): + import config + read = self.run_esp32.make_ldr_reader() + FakeADC.last.value = 0 # dunkelste ADC-Spannung + self.assertEqual(read(), config.LDR_MAX) + FakeADC.last.value = 65535 # voller Ausschlag + self.assertEqual(read(), 0) + + +# -- WLAN ------------------------------------------------------------------ + +class TestWifi(unittest.TestCase): + + def setUp(self): + _install_network() + FakeWLAN.succeed = True + _remove("wifi", "wifi_config") + import wifi + self.wifi = wifi + # Zugangsdaten unterschieben, damit der Test ohne wifi_config.py laeuft + cfg = types.ModuleType("wifi_config") + cfg.SSID = "Sternwarte" + cfg.PASSWORD = "geheim" + sys.modules["wifi_config"] = cfg + + def tearDown(self): + _remove("network", "wifi", "wifi_config") + + def _quiet(self, *args): + pass + + def test_connect_meldet_sich_mit_den_zugangsdaten_an(self): + cfg = self.wifi.connect(log=self._quiet) + self.assertEqual(FakeWLAN.last.credentials, ("Sternwarte", "geheim")) + self.assertTrue(FakeWLAN.last.active()) + self.assertEqual(cfg[0], "192.168.1.77") + + def test_connect_wirft_wifierror_bei_zeitablauf(self): + FakeWLAN.succeed = False + with self.assertRaises(self.wifi.WifiError): + self.wifi.connect(timeout=0.2, log=self._quiet) + # Der haengende Versuch muss abgebrochen werden, sonst kollidiert er + # mit dem naechsten connect(). + self.assertEqual(FakeWLAN.last.disconnect_calls, 1) + + def test_wifierror_ist_ein_linkerror(self): + """Damit die Poll-Schleife den Ausfall selbst abfaengt.""" + import displayloop + self.assertTrue(issubclass(self.wifi.WifiError, displayloop.LinkError)) + + def test_ensure_verbindet_nur_wenn_noetig(self): + self.wifi.connect(log=self._quiet) + wlan = FakeWLAN.last + wlan.credentials = None + self.wifi.ensure(log=self._quiet) + self.assertIsNone(wlan.credentials) # kein zweiter connect() + + def test_fehlende_zugangsdaten_melden_sich_deutlich(self): + # None in sys.modules laesst "import wifi_config" mit ImportError + # scheitern -- unabhaengig davon, ob auf diesem Rechner schon eine + # echte wifi_config.py liegt. Ein blosses _remove() wuerde die Datei + # einfach neu laden und der Test haenge am Zustand des Entwicklerrechners. + sys.modules["wifi_config"] = None + with self.assertRaises(self.wifi.WifiError) as ctx: + self.wifi.credentials() + self.assertIn("wifi_config", str(ctx.exception)) + + + +# -- Vertraeglichkeit mit MicroPython -------------------------------------- + +# Die Dateien, die auf dem Geraet landen -- muss zu deploy.sh passen. +DEPLOYED = ( + "ticks.py", "migra.py", "config.py", "transport.py", "display.py", + "coords.py", "lx200.py", "mount_client.py", "brightness.py", + "displayloop.py", "wifi.py", "run_esp32.py", +) + + +class TestMicropythonVertraeglichkeit(unittest.TestCase): + """Statische Pruefung dessen, was CPython kann und MicroPython nicht. + + Solche Fehler entgehen den ueblichen Tests, weil sie hier auf CPython + laufen: str.ljust gibt es auf dem Mac, auf dem ESP32 nicht -- aufgefallen + ist das erst am Geraet. Diese Liste haelt fest, was dort nachgeprueft + wurde (MicroPython 1.28, ESP32_GENERIC_C3). + """ + + # Auf dem Geraet mit hasattr geprueft: fehlen. + FEHLENDE_STRINGMETHODEN = (".ljust(", ".rjust(", ".center(", + ".zfill(", ".casefold(", ".expandtabs(") + + # Module, die es unter MicroPython nicht gibt. Sie duerfen in den + # deployten Dateien hoechstens innerhalb einer Funktion importiert werden + # (wie in brightness.main), nicht auf Modulebene. + FEHLENDE_MODULE = ("argparse", "threading", "subprocess", "unittest") + + def _quelle(self, name): + with open(name, encoding="utf-8") as f: + return f.read() + + def test_keine_fehlenden_stringmethoden(self): + for name in DEPLOYED: + src = self._quelle(name) + for methode in self.FEHLENDE_STRINGMETHODEN: + self.assertNotIn( + methode, src, + "%s benutzt %s -- das gibt es unter MicroPython nicht" % (name, methode)) + + def test_keine_fehlenden_module_auf_modulebene(self): + import ast + for name in DEPLOYED: + baum = ast.parse(self._quelle(name), filename=name) + for knoten in baum.body: # nur die oberste Ebene + if isinstance(knoten, ast.Import): + namen = [a.name.split(".")[0] for a in knoten.names] + elif isinstance(knoten, ast.ImportFrom): + namen = [(knoten.module or "").split(".")[0]] + else: + continue + for modul in namen: + self.assertNotIn( + modul, self.FEHLENDE_MODULE, + "%s importiert %s auf Modulebene -- unter MicroPython " + "scheitert damit schon der Import" % (name, modul)) + + def test_deploy_sh_kopiert_genau_diese_dateien(self): + """Sonst laeuft die Liste hier und die im Deploy-Skript auseinander.""" + with open("deploy.sh", encoding="utf-8") as f: + skript = f.read() + block = skript.split('FILES="', 1)[1].split('"', 1)[0] + im_skript = tuple(z.strip() for z in block.split() if z.strip()) + self.assertEqual(sorted(im_skript), sorted(DEPLOYED)) + + +# -- Gesamtkette ----------------------------------------------------------- + +class TestEsp32GegenMock(unittest.TestCase): + """Der ESP32-Pfad gegen einen echten Mock, ueber eine echte TCP-Verbindung. + + Alles ausser der Hardware ist hier zusammengeschaltet: WLAN-Attrappe, + mount_client, displayloop, Telegrammbau. Das ist derselbe Ablauf, der auf + dem Geraet gegen den Mock auf dem Mac laufen soll -- nur eben auf einem + Rechner und mit Attrappe statt Funk. + """ + + def setUp(self): + _install_machine() + _install_network() + FakeWLAN.succeed = True + _remove("wifi", "wifi_config", "run_esp32") + cfg = types.ModuleType("wifi_config") + cfg.SSID = "Sternwarte" + cfg.PASSWORD = "geheim" + sys.modules["wifi_config"] = cfg + + import transport + importlib.reload(transport) + + from mount_mock import MountMock + self.mock = MountMock(host="127.0.0.1", port=0, move=True) + self.mock.start() + + import run_esp32 + self.run_esp32 = run_esp32 + self.log = [] + + def tearDown(self): + self.mock.stop() + _remove("machine", "network", "wifi", "wifi_config", "run_esp32") + import transport + importlib.reload(transport) + + def _run(self, fn, **kwargs): + """Aufrufen und dabei die Hex-Ausgabe des FakeTransport wegfangen.""" + import contextlib + import io + with contextlib.redirect_stdout(io.StringIO()) as out: + fn(host=self.mock.host, port=self.mock.port, + log=self.log.append, **kwargs) + return out.getvalue() + + def test_selftest_geht_durch_alle_stufen(self): + self._run(self.run_esp32.selftest) + text = "\n".join(self.log) + self.assertIn("WLAN", text) + self.assertIn("192.168.1.77", text) # IP der WLAN-Attrappe + self.assertIn("18:36:56", text) # RA vom Mock + self.assertIn("Selbsttest bestanden", text) + + def test_main_once_zeigt_die_koordinaten(self): + self._run(self.run_esp32.main, fake_display=True, once=True) + text = "\n".join(self.log) + self.assertIn("18h36m56s", text) # RA, fuer die Anzeige formatiert + self.assertIn("+38", text) # DEC + + def test_main_once_baut_ein_gueltiges_telegramm(self): + hexdump = self._run(self.run_esp32.main, fake_display=True, once=True) + self.assertIn("TX: 02 81 80 81", hexdump) # STX, DA, SA, FC + self.assertTrue(hexdump.strip().endswith("03")) # ETX + + def test_ohne_mock_zeigt_die_anzeige_kein_link(self): + """Faellt die Gegenstelle aus, laeuft die Schleife weiter.""" + import config + import contextlib + import io + self.mock.stop() + with contextlib.redirect_stdout(io.StringIO()) as out: + self.run_esp32.main(host="127.0.0.1", port=self.mock.port, + fake_display=True, once=True, log=self.log.append) + self.assertTrue(any("Montierung" in line for line in self.log)) + # Die Hinweiszeilen stehen wirklich im Telegramm -- ohne den Text hier + # zu wiederholen, sonst bricht der Test bei jeder Umformulierung. + gesendet = "".join(chr(int(b, 16)) if 32 <= int(b, 16) < 127 else "." + for zeile in out.getvalue().splitlines() + if zeile.startswith("TX:") + for b in zeile.split(":", 1)[1].split()) + for zeile in config.NOLINK_LINES: + self.assertIn(zeile, gesendet) + + def test_ohne_wlan_zeigt_die_anzeige_kein_wlan(self): + import config + FakeWLAN.succeed = False + # Den WLAN-Timeout kurz drehen, sonst wartet der Test zweimal die + # vollen config.WIFI_TIMEOUT Sekunden ab (vorab-connect und ensure). + orig = config.WIFI_TIMEOUT + config.WIFI_TIMEOUT = 0.2 + try: + import contextlib + import io + with contextlib.redirect_stdout(io.StringIO()): + self.run_esp32.main(host=self.mock.host, port=self.mock.port, + fake_display=True, once=True, log=self.log.append) + finally: + config.WIFI_TIMEOUT = orig + text = "\n".join(self.log) + self.assertIn("WLAN", text) + self.assertIn("Netz", text) # LinkError wurde in der Schleife gefangen + + +if __name__ == "__main__": + unittest.main() diff --git a/ticks.py b/ticks.py new file mode 100644 index 0000000..b70e62c --- /dev/null +++ b/ticks.py @@ -0,0 +1,45 @@ +"""Timeouts, die unter CPython und MicroPython gleich funktionieren. + +CPython rechnet mit time.monotonic() in Sekunden. MicroPython hat das nicht, +sondern time.ticks_ms(); dessen Zaehler laeuft nach etwa 12 Tagen ueber. Ein +blosses "jetzt + timeout" laege danach fuer Tage in der Zukunft und wuerde eine +Warteschleife entsprechend lange haengen lassen -- fuer ein Geraet, das in der +Sternwarte monatelang durchlaeuft, ein echter Stolperstein. Deshalb dort die +Differenz immer ueber ticks_diff bilden, das den Ueberlauf richtig behandelt. + + end = deadline(1.5) + while not expired(end): + ... + +Beide Varianten sind immer definiert, ausgewaehlt wird beim Import. So laesst +sich die MicroPython-Variante auch auf dem Mac testen (test_esp32.py haengt +dafuer ein Ersatz-time an dieses Modul). +""" + +import time + + +# -- CPython --------------------------------------------------------------- +def monotonic_deadline(seconds): + return time.monotonic() + seconds + + +def monotonic_expired(end): + return time.monotonic() >= end + + +# -- MicroPython ----------------------------------------------------------- +def ticks_deadline(seconds): + return time.ticks_add(time.ticks_ms(), int(seconds * 1000)) + + +def ticks_expired(end): + return time.ticks_diff(end, time.ticks_ms()) <= 0 + + +if hasattr(time, "monotonic"): + deadline = monotonic_deadline + expired = monotonic_expired +else: + deadline = ticks_deadline + expired = ticks_expired diff --git a/transport.py b/transport.py index 4577ef1..5a07b08 100644 --- a/transport.py +++ b/transport.py @@ -12,15 +12,22 @@ except ImportError: serial = None try: # MicroPython - from machine import UART + from machine import UART, Pin except ImportError: UART = None + Pin = None -try: - import time - _monotonic = time.monotonic -except AttributeError: # MicroPython kennt kein time.monotonic - _monotonic = lambda: time.ticks_ms() / 1000.0 +from ticks import deadline as _deadline, expired as _expired + + +def _truncate_at_etx(buf): + """Alles ab dem ersten ETX abschneiden (einschliesslich). + + Nach ETX ist das Antworttelegramm zu Ende; alles danach waere Stoerung und + wuerde die Plausibilitaetspruefung in migra.parse_response scheitern lassen. + """ + idx = buf.find(b"\x03") + return bytes(buf if idx < 0 else buf[:idx + 1]) class SerialTransport: @@ -49,9 +56,9 @@ class SerialTransport: def read_response(self): """Bytes bis einschliesslich ETX lesen, oder bis der Timeout greift.""" - deadline = _monotonic() + self._timeout + deadline = _deadline(self._timeout) buf = bytearray() - while _monotonic() < deadline: + while not _expired(deadline): chunk = self._port.read(1) if not chunk: continue @@ -65,7 +72,12 @@ class SerialTransport: class UartTransport: - """Serieller Port ueber machine.UART (MicroPython / ESP32-C3).""" + """Serieller Port ueber machine.UART (MicroPython / ESP32-C3). + + Auf dem XIAO ESP32-C3 haengt die Anzeige ueber einen MAX3232 an UART1: + TX = GPIO21 (Pad 7 / D6) -> T2IN, RX = GPIO20 (Pad 8 / D7) <- R2OUT. + Die Pins stehen in config.py, weil UART1 auf dem C3 frei zuordenbar ist. + """ def __init__(self, uart_id, baudrate, parity, bytesize, stopbits, timeout, tx=None, rx=None): @@ -80,27 +92,42 @@ class UartTransport: stop=stopbits, timeout=int(timeout * 1000), ) + # Die Pins als Pin-Objekt uebergeben -- die esp32-Portierung akzeptiert + # zwar meist auch die nackte GPIO-Nummer, Pin() ist aber ueberall gueltig. if tx is not None: - kwargs["tx"] = tx + kwargs["tx"] = Pin(tx) if Pin is not None else tx if rx is not None: - kwargs["rx"] = rx + kwargs["rx"] = Pin(rx) if Pin is not None else rx self._uart = UART(uart_id, **kwargs) self._timeout = timeout + def _drain(self): + """Empfangspuffer leeren (wie SerialTransport.reset_input_buffer).""" + n = self._uart.any() + while n: + self._uart.read(n) + n = self._uart.any() + def write(self, data): + # Wie bei pyserial: vor dem Senden den Empfangspuffer leeren, damit + # read_response nur die frische Antwort auf DIESES Telegramm sieht. + self._drain() self._uart.write(data) def read_response(self): - deadline = _monotonic() + self._timeout + deadline = _deadline(self._timeout) buf = bytearray() - while _monotonic() < deadline: - chunk = self._uart.read(1) + while not _expired(deadline): + n = self._uart.any() + if not n: + continue + chunk = self._uart.read(n) if not chunk: continue buf.extend(chunk) - if chunk[0] == migra.ETX: + if migra.ETX in chunk: break - return bytes(buf) + return _truncate_at_etx(buf) def close(self): self._uart.deinit() diff --git a/wifi.py b/wifi.py new file mode 100644 index 0000000..6ee2591 --- /dev/null +++ b/wifi.py @@ -0,0 +1,126 @@ +"""WLAN-Anbindung des XIAO ESP32-C3 (MicroPython). + +Die Zugangsdaten stehen in wifi_config.py, das nicht im Git liegt -- Vorlage ist +wifi_config_example.py. Zeitverhalten kommt aus config.py. + + import wifi + wifi.connect() # einmal verbinden, wirft WifiError bei Misserfolg + wifi.ensure() # vor jeder Abfrage: verbunden halten + +WifiError erbt von displayloop.LinkError, damit die Poll-Schleife den Ausfall +selbst abfaengt und "KEIN WLAN" anzeigt, statt abzustuerzen. +""" + +import time + +import config +import ticks +from displayloop import LinkError + +try: # MicroPython + import network +except ImportError: # auf dem Mac nicht vorhanden + network = None + + +class WifiError(LinkError): + """WLAN nicht verfuegbar oder Zugangsdaten fehlen.""" + + +_wlan = None + + +def credentials(): + """(SSID, Passwort) aus wifi_config.py holen.""" + try: + import wifi_config + except ImportError: + raise WifiError("wifi_config.py fehlt -- wifi_config_example.py " + "kopieren, ausfuellen und auf den ESP32 kopieren") + try: + return wifi_config.SSID, wifi_config.PASSWORD + except AttributeError: + raise WifiError("wifi_config.py braucht SSID und PASSWORD") + + +def station(): + """Das WLAN-Interface im Station-Modus, aktiviert.""" + global _wlan + if network is None: + raise WifiError("network fehlt -- laeuft dieser Code auf dem ESP32?") + if _wlan is None: + _wlan = network.WLAN(network.STA_IF) + if not _wlan.active(): + _wlan.active(True) + return _wlan + + +def connected(): + """True, wenn eine WLAN-Verbindung steht.""" + if network is None or _wlan is None: + return False + return _wlan.isconnected() + + +def connect(timeout=None, log=None): + """Verbinden und warten, bis eine IP steht. Liefert die ifconfig-Tupel.""" + if timeout is None: + timeout = config.WIFI_TIMEOUT + if log is None: + log = print + + wlan = station() + if wlan.isconnected(): + return wlan.ifconfig() + + ssid, password = credentials() + log("WLAN: verbinde mit %s ..." % ssid) + wlan.connect(ssid, password) + + end = ticks.deadline(timeout) + while not wlan.isconnected(): + # Kurz schlafen statt hart zu pollen: der Verbindungsaufbau laeuft im + # Hintergrund und braucht die Rechenzeit selbst. + time.sleep(0.2) + if ticks.expired(end): + # Die Verbindung abbrechen, sonst versucht die Firmware im + # Hintergrund weiter und der naechste connect() kollidiert damit. + try: + wlan.disconnect() + except OSError: + pass + raise WifiError("keine Verbindung zu %s nach %g s" % (ssid, timeout)) + + cfg = wlan.ifconfig() + log("WLAN: verbunden, IP %s" % cfg[0]) + return cfg + + +def ensure(timeout=None, log=None): + """Sicherstellen, dass das WLAN steht -- fuer displayloop.ensure_link. + + Steht die Verbindung schon, kostet der Aufruf praktisch nichts. Sonst wird + neu verbunden; scheitert das, fliegt WifiError (= LinkError) und die + Poll-Schleife zeigt "KEIN WLAN" an und versucht es beim naechsten Durchlauf + erneut. + """ + if connected(): + return + connect(timeout=timeout, log=log) + + +def ifconfig(): + """(IP, Netzmaske, Gateway, DNS) oder None, wenn nicht verbunden.""" + return _wlan.ifconfig() if connected() else None + + +def disconnect(): + """Verbindung trennen und das Interface abschalten.""" + global _wlan + if _wlan is not None: + try: + _wlan.disconnect() + _wlan.active(False) + except OSError: + pass + _wlan = None diff --git a/wifi_config_example.py b/wifi_config_example.py new file mode 100644 index 0000000..8d7e196 --- /dev/null +++ b/wifi_config_example.py @@ -0,0 +1,10 @@ +"""Vorlage fuer die WLAN-Zugangsdaten des ESP32-C3. + +Kopieren nach wifi_config.py, ausfuellen und mit auf den ESP32 kopieren. +wifi_config.py steht in .gitignore -- das Passwort gehoert nicht ins Repository. + + cp wifi_config_example.py wifi_config.py +""" + +SSID = "NetzwerkName" +PASSWORD = "geheim"