commit 81e0e1a0369bb8598ba6cafbb69eded6ba7a6568 Author: Reinhard X. Fürst Date: Thu Jul 16 09:05:56 2026 +0200 First commit diff --git a/README.md b/README.md new file mode 100644 index 0000000..f3f6d74 --- /dev/null +++ b/README.md @@ -0,0 +1,138 @@ +# Großanzeige + +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. + +## Benutzung + +```bash +# Phase 1 -- Koordinaten von Hand: +python3 show_coords.py 12h34m56s +45d12m30s # Koordinaten anzeigen +python3 show_coords.py 12:34:56 -07:15:00 # Trennzeichen sind egal +python3 show_coords.py --clear # Anzeige löschen +python3 show_coords.py 12h34m56s +45d12m30s --dry-run # nur Telegramm als Hex + +# Phase 2 -- Koordinaten von der Montierung (10Micron GM4000): +python3 run_display.py # Montierung aus config.py pollen +python3 run_display.py --host 10.0.0.42 # andere IP +python3 run_display.py --mock --move # ohne Teleskop: lokaler Mock, RA läuft +python3 mount_mock.py --move # Mock separat starten + +python3 -m unittest test_migra test_lx200 # alle Tests +python3 probe.py ruler # Geometrie der Anzeige ausmessen +``` + +Einzige Abhängigkeit auf dem Mac ist `pyserial`. + +## 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. | +| `probe.py` | Testmuster zum Ausmessen von Zeilenbreite und Zeilenposition. | +| `test_migra.py` | Tests, u. a. gegen die Beispieltelegramme aus dem Handbuch. | + +## Protokoll (Kurzfassung) + +Referenz: Benutzerhandbuch `X-M31-9AXX5X-006 (V4.20)`, Kapitel 3.5–3.9. + +**Schnittstelle:** 19200 Baud, 8 Datenbits, **Even Parity**, 1 Stopbit. +9-pol. Sub-D: Pin 2 = RxD, Pin 3 = TxD, Pin 5 = GND. + +**Telegramm:** + +``` +STX | DA | SA | FC | [LEN-H LEN-L] | Data-Unit | [CHK-H CHK-L] | ETX +02 81 80 81 03 +``` + +- `DA`/`SA`: Adressen mit gesetztem Bit 7 → Anzeige = 1 (`0x81`), Host = 0 (`0x80`). +- `FC`: Bit 7 fest 1, Bit 1 = Prüfsumme, Bit 0 = Antwort anfordern. + `0x81` = ohne Prüfsumme, mit Antwort (so fuhr auch das alte MSP430-Programm). +- Ohne Prüfsumme dürfen `LEN-H/L` und `CHK-H/L` **nicht** mitgesendet werden. +- Prüfsumme = Low-Byte der Summe aus `DA, SA, FC, LEN-H, LEN-L` und allen Datenbytes, + als zwei Nibbles mit gesetztem High-Nibble (`0xF0 | nibble`). + +**Antwort:** `02 80 81 80 03`, Fehlercode `'0'` = kein Fehler. +Nach Erhalt der Antwort darf sofort das nächste Telegramm folgen — das ist die +Flusskontrolle. Ohne Antwort muss zwischen den Telegrammen gewartet werden +(Empfangstimeout + ca. 150 ms). + +**ESC-Sequenzen** (alle Zahlen ASCII-codiert, Position 123 → `'1' '2' '3'`): + +| Sequenz | Bedeutung | +|---|---| +| `ESC z NN` / `ESC Z NN` | Zeichensatz NN (`z` = Sperrschrift, gleiche Zeichenbreite) | +| `ESC C xxxyyy` | Cursor auf **Pixel**position x/y, 0/0 ist links oben | +| `ESC A ` | Attribute; Farbe 0 = schwarz, 1 = grün, 2 = rot, 3 = gelb | +| `ESC F ` | Anzeige füllen; `ESC F 0` = löschen | +| `ESC H ` | Helligkeit 0–100 % | +| `ESC T <+/->` | gespeicherten Text ein-/ausblenden | + +Folgt auf eine ESC-Sequenz ein Online-Text, muss **`0x1F` als Trennzeichen** +dazwischen. Data-Unit max. 230 Zeichen. `migra.DataUnit` setzt das Trennzeichen +selbst. + +## Offene Punkte + +- **Antworttelegramm-Pfad (RX) noch offen, absichtlich zurückgestellt.** + `WANT_RESPONSE = False` ist Default, der Normalbetrieb läuft ohne Rückkanal. + Diagnose 2026-07-15: Adapter-Loopback ok, aber über den MAX232 und am echten + Gerät kommt keine Antwort zurück → der Empfangszweig (Anzeige Pin 3 TxD → MAX232 + R_IN → R_OUT → Adapter RxD) leitet nicht. `loopback.py` hilft beim Eingrenzen. + Die Anzeige *antwortet* nachweislich (das alte MSP430-Programm nutzte die Antwort + als Handshake); es ist reine Verdrahtung. Wird der RX-Pfad fertig, `WANT_RESPONSE` + in `config.py` auf True stellen. +- **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. + +## Gradzeichen + +Der Standard-Zeichensatz 0 der Anzeige hat **kein** `°`. Der von der Sternwarte +geladene **Zeichensatz 1** enthält an Position `/` (0x2F) ein hochgestelltes +Gradzeichen (deckt sich mit dem alten Programm, `asciitaf.h`: Code 0xAF → per +`& 0x7F` = 0x2F in Zeichensatz 1). Zur Ausgabe wird für dieses eine Zeichen kurz +auf Zeichensatz 1 umgeschaltet und wieder zurück (`config.DEGREE_*`, `display._emit_degree`). + +**Fallstrick:** Ein Zeichensatzwechsel mitten im Telegramm beschädigt die Home-Zelle +(0,0) — das erste Zeichen der obersten Zeile wird zerstört. Deshalb zeichnet +`show_lines` die Zeilen **von unten nach oben** (oberste Zeile zuletzt), damit sie +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. +- 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, + die in `CHARS_PER_LINE` passt. +- **Zyklische Updates flimmerfrei:** `show_lines` löscht nicht bei jeder Aktualisierung, + sondern überschreibt die Zeilen an Ort und Stelle (auf volle Breite aufgefüllt). + `clear=True` nur beim ersten Bild. Wichtig: Zeilen dürfen **nicht breiter** als + `CHARS_PER_LINE` aufgefüllt werden, sonst bricht das Füll-Leerzeichen um und + beschädigt die andere Zeile. + +## Erledigt + +- **Phase 1 + 2 am Gerät bestätigt (2026-07-15):** Koordinaten erscheinen korrekt, + inkl. echtem Gradzeichen; im `run_display`-Betrieb aktualisieren beide Zeilen + laufend, sauber und ohne Flimmern. Protokoll, 19200 8E1, Zeichensatz und + Zeilengeometrie stimmen. +- Schrift: Sperrschrift (`ESC z`, `CHARSET_SPACED = True`), 10 Zeichen/Zeile. + Die schmale `1` wirkt dadurch etwas luftig — Font-Eigenschaft, nur per eigenem + Font (microSYST-PC-Software) änderbar. diff --git a/__pycache__/config.cpython-313.pyc b/__pycache__/config.cpython-313.pyc new file mode 100644 index 0000000..e391fe8 Binary files /dev/null and b/__pycache__/config.cpython-313.pyc differ diff --git a/__pycache__/coords.cpython-313.pyc b/__pycache__/coords.cpython-313.pyc new file mode 100644 index 0000000..e790f71 Binary files /dev/null and b/__pycache__/coords.cpython-313.pyc differ diff --git a/__pycache__/display.cpython-313.pyc b/__pycache__/display.cpython-313.pyc new file mode 100644 index 0000000..f4b0c12 Binary files /dev/null and b/__pycache__/display.cpython-313.pyc differ diff --git a/__pycache__/lx200.cpython-313.pyc b/__pycache__/lx200.cpython-313.pyc new file mode 100644 index 0000000..b57eb15 Binary files /dev/null and b/__pycache__/lx200.cpython-313.pyc differ diff --git a/__pycache__/migra.cpython-313.pyc b/__pycache__/migra.cpython-313.pyc new file mode 100644 index 0000000..29860ca Binary files /dev/null and b/__pycache__/migra.cpython-313.pyc differ diff --git a/__pycache__/mount_client.cpython-313.pyc b/__pycache__/mount_client.cpython-313.pyc new file mode 100644 index 0000000..99d4ff7 Binary files /dev/null and b/__pycache__/mount_client.cpython-313.pyc differ diff --git a/__pycache__/mount_mock.cpython-313.pyc b/__pycache__/mount_mock.cpython-313.pyc new file mode 100644 index 0000000..68f5fcb Binary files /dev/null and b/__pycache__/mount_mock.cpython-313.pyc differ diff --git a/__pycache__/test_lx200.cpython-313.pyc b/__pycache__/test_lx200.cpython-313.pyc new file mode 100644 index 0000000..1fd1d4d Binary files /dev/null and b/__pycache__/test_lx200.cpython-313.pyc differ diff --git a/__pycache__/test_migra.cpython-313.pyc b/__pycache__/test_migra.cpython-313.pyc new file mode 100644 index 0000000..9529bd2 Binary files /dev/null and b/__pycache__/test_migra.cpython-313.pyc differ diff --git a/__pycache__/transport.cpython-313.pyc b/__pycache__/transport.cpython-313.pyc new file mode 100644 index 0000000..a8574c1 Binary files /dev/null and b/__pycache__/transport.cpython-313.pyc differ diff --git a/config.py b/config.py new file mode 100644 index 0000000..df0794b --- /dev/null +++ b/config.py @@ -0,0 +1,62 @@ +"""Zentrale Einstellungen fuer die Ansteuerung der Grossanzeige.""" + +import migra + +# -- 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-1120" # Mac; auf dem ESP32-C3 spaeter die UART-Nummer +BAUDRATE = 19200 +PARITY = "E" +BYTESIZE = 8 +STOPBITS = 1 + +# -- Telegramm ------------------------------------------------------------ +DISPLAY_ADDR = 1 # Zieladresse der Anzeige (DA = 0x81) +HOST_ADDR = 0 # unsere Quelladresse (SA = 0x80) +USE_CHECKSUM = False # das alte Programm fuhr ohne Pruefsumme +# Antworttelegramm als Flusskontrolle. Derzeit False, weil die RS232-Rueckleitung +# (Anzeige Pin 3 TxD -> Adapter RxD) noch nicht verdrahtet ist. Ist der RX-Pfad +# fertig, hier wieder auf True stellen -- dann wird der Fehlercode ausgewertet und +# ohne feste Pause gesendet. +WANT_RESPONSE = False +RESPONSE_TIMEOUT = 1.0 # Sekunden (nur relevant, wenn WANT_RESPONSE = True) +QUIET_TIME = 0.4 # Pause zwischen Telegrammen, wenn ohne Antwort gefahren wird + +# -- Darstellung ---------------------------------------------------------- +# 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). +# 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) + +COLOR = migra.RED # einfarbig rote Anzeige +BACKGROUND = migra.BLACK +BRIGHTNESS = 100 # Prozent + +# Gradzeichen: Im Standard-Zeichensatz 0 der Anzeige fehlt es. Der von der +# Sternwarte geladene Zeichensatz 1 hat an Position "/" (0x2F) ein hochgestelltes +# Gradzeichen -- verifiziert 2026-07-15, deckt sich mit dem alten Programm +# (asciitaf.h: Code 0xAF, per & 0x7F = 0x2F in Zeichensatz 1). Zur Ausgabe wird +# fuer dieses eine Zeichen kurz auf DEGREE_CHARSET umgeschaltet. +DEGREE_MARKER = "°" # Platzhalter im formatierten Text (siehe coords.py) +DEGREE_CHARSET = 1 # Zeichensatz mit dem Gradzeichen; None -> Fallback nutzen +DEGREE_CHAR = "/" # Zeichen, das in DEGREE_CHARSET als Grad erscheint +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.0.50" # PLATZHALTER -- echte IP der Montierung eintragen +MOUNT_PORT = 3490 +MOUNT_TIMEOUT = 3.0 # Sekunden fuer Verbindung/Antwort +POLL_INTERVAL = 1.0 # Sekunden zwischen zwei Koordinatenabfragen + +# Anzeige, wenn die Montierung (noch) nicht erreichbar ist. +NOLINK_LINES = ("KEIN LINK", "MONTIERUNG") diff --git a/coords.py b/coords.py new file mode 100644 index 0000000..8f52ae3 --- /dev/null +++ b/coords.py @@ -0,0 +1,98 @@ +"""Rektaszension und Deklination einlesen und fuer die Anzeige formatieren. + +Die Eingabe ist bewusst tolerant, damit man beim Testen nicht auf ein Format +festgelegt ist: 12h34m56s | 12:34:56 | "12 34 56" | 12.5824 (Stunden dezimal) +Bei der Deklination entsprechend mit Vorzeichen und Grad/Bogenminuten. + +Ausgegeben wird die breiteste Schreibweise, die noch in die Zeile passt -- +die tatsaechliche Zeichenbreite der Anzeige kennen wir erst nach dem Livetest. +""" + +import config + +_SEPARATORS = "hHmMsSdD°'\"::," + + +def _split(value): + """Zahlenfelder aus einer Koordinatenangabe holen, Trennzeichen egal.""" + out, cur = [], "" + for ch in str(value).strip(): + if ch.isdigit() or ch == ".": + cur += ch + elif ch in "+-" and not cur and not out: + cur += ch + else: + if cur: + out.append(cur) + cur = "" + if cur: + out.append(cur) + if not out: + raise ValueError("keine Zahlen in %r gefunden" % (value,)) + return out + + +def _to_sexagesimal(value, unit_max, name): + """Eingabe in (Vorzeichen, Ganzteil, Minuten, Sekunden) zerlegen.""" + fields = _split(value) + negative = fields[0].startswith("-") + fields[0] = fields[0].lstrip("+-") + + if len(fields) == 1: # dezimal, z.B. 12.5824 + total = float(fields[0]) * 3600.0 + else: + parts = [float(f) for f in fields[:3]] + [0.0] * (3 - len(fields[:3])) + total = parts[0] * 3600.0 + parts[1] * 60.0 + parts[2] + + total = int(round(total)) + big, rest = divmod(total, 3600) + minutes, seconds = divmod(rest, 60) + + if big >= unit_max: + raise ValueError("%s: %s ausserhalb des gueltigen Bereichs" % (name, value)) + return negative, big, minutes, seconds + + +def parse_ra(value): + """Rektaszension -> (Stunden, Minuten, Sekunden).""" + _, h, m, s = _to_sexagesimal(value, 24, "Rektaszension") + return h, m, s + + +def parse_dec(value): + """Deklination -> (Vorzeichen als '+'/'-', Grad, Bogenminuten, Bogensekunden).""" + neg, d, m, s = _to_sexagesimal(value, 91, "Deklination") + if d > 90 or (d == 90 and (m or s)): + raise ValueError("Deklination %s liegt ausserhalb +/-90 Grad" % (value,)) + return ("-" if neg else "+"), d, m, s + + +def _fit(candidates, width): + """Erste Schreibweise nehmen, die in die Zeile passt; sonst die kuerzeste.""" + for text in candidates: + if len(text) <= width: + return text + return candidates[-1][:width] + + +def format_ra(value, width=None): + width = config.CHARS_PER_LINE if width is None else width + h, m, s = parse_ra(value) + return _fit([ + "%02dh%02dm%02ds" % (h, m, s), + "%02d:%02d:%02d" % (h, m, s), + "%02dh%02dm" % (h, m), + ], width) + + +def format_dec(value, width=None): + width = config.CHARS_PER_LINE if width is None else width + sign, d, m, s = parse_dec(value) + deg = config.DEGREE_MARKER + return _fit([ + "%s%02d%s%02d'%02d\"" % (sign, d, deg, m, s), # 10: mit Bogensekundenzeichen + "%s%02d%s%02d'%02d" % (sign, d, deg, m, s), # 9: Gradzeichen, ohne " am Ende + "%s%02d:%02d:%02d" % (sign, d, m, s), # 9: Doppelpunkte, kein Gradzeichen + "%s%02d%s%02d'" % (sign, d, deg, m), # 7: ohne Bogensekunden + "%s%02d:%02d" % (sign, d, m), # 6 + ], width) diff --git a/display.py b/display.py new file mode 100644 index 0000000..23b13c1 --- /dev/null +++ b/display.py @@ -0,0 +1,139 @@ +"""Treiber fuer die Grossanzeige: Telegramme bauen, senden, Antwort pruefen.""" + +import time + +import migra +import config + + +class Display: + """Sendet Telegramme an die Anzeige und wertet das Antworttelegramm aus. + + Das Antworttelegramm ist zugleich die Flusskontrolle: Nach seinem Erhalt + darf sofort das naechste Telegramm folgen (Handbuch 3.5). Ohne Antwort + muesste zwischen den Telegrammen "Empfangstimeout + 150 ms" gewartet werden. + """ + + def __init__(self, transport, + dest=config.DISPLAY_ADDR, + src=config.HOST_ADDR, + checksum=config.USE_CHECKSUM, + response=config.WANT_RESPONSE): + self._t = transport + self._dest = dest + self._src = src + self._checksum = checksum + self._response = response + + def send(self, data_unit): + """Data-Unit rahmen, senden und ggf. auf das Antworttelegramm warten.""" + frame = migra.build_telegram( + data_unit, + dest=self._dest, + src=self._src, + checksum=self._checksum, + response=self._response, + ) + self._t.write(frame) + if self._response: + migra.parse_response(self._t.read_response(), src=self._src) + else: + # Ohne Antwort ist die Anzeige erst nach Abarbeitung des Telegramms + # wieder empfangsbereit: Empfangstimeout + 150 ms (Handbuch 3.5). + time.sleep(config.QUIET_TIME) + + # -- Hoehere Funktionen ------------------------------------------------ + def clear(self): + self.send(migra.DataUnit().fill(config.BACKGROUND)) + + def set_brightness(self, percent): + self.send(migra.DataUnit().brightness(percent, color=config.COLOR)) + + def show_lines(self, *lines, clear=False): + """Zeilen ab Zeile 1 auf die Anzeige schreiben. + + Baut ein einziges Telegramm: pro Zeile Cursor setzen und den Text + schreiben. Das Gradzeichen (config.DEGREE_MARKER) wird ueber einen kurzen + Zeichensatzwechsel eingefuegt. + + Jede Zeile wird mit Leerzeichen auf volle Breite (CHARS_PER_LINE) + aufgefuellt und damit *an Ort und Stelle* ueberschrieben -- ohne die + Anzeige vorher komplett zu loeschen. Das vermeidet das sichtbare Blitzen + bei zyklischen Updates und haelt das Telegramm kurz (sonst kommt das + naechste zu frueh und das letzte Zeichen kann verloren gehen). + clear=True erzwingt zusaetzlich ein Vollbild-Loeschen (einmal beim Start). + + Reihenfolge: Die Zeilen werden von unten nach oben gezeichnet, damit die + oberste Zeile zuletzt kommt. Grund: Ein Zeichensatzwechsel mitten im + Telegramm beschaedigt die Home-Zelle (0,0) -- die oberste Zeile muss also + nach allen Wechseln geschrieben werden, damit sie diese Zelle ueberschreibt. + (Am Geraet verifiziert 2026-07-15.) + """ + if len(lines) > len(config.LINE_Y): + raise ValueError("Anzeige hat nur %d Zeilen" % len(config.LINE_Y)) + + du = migra.DataUnit() + if clear: + du.fill(config.BACKGROUND) + du.attributes(fg=config.COLOR, bg=config.BACKGROUND, blink=migra.STATIC) + for i in reversed(range(len(lines))): + self._emit_line(du, config.LINE_Y[i], lines[i]) + self.send(du) + + def show_line(self, index, text, clear=False): + """Nur eine einzelne Zeile aktualisieren (eigenes, kurzes Telegramm). + + Fuer zyklische Updates: die Zeile ohne Gradzeichen (also ohne + Zeichensatzwechsel) laesst sich beliebig oft neu schreiben, ohne dass die + Anzeige Bytes verschluckt. Zeilen *mit* Gradzeichen sollten dagegen selten + und moeglichst per show_lines (Vollbild) gezeichnet werden. + """ + if not 0 <= index < len(config.LINE_Y): + raise ValueError("ungueltiger Zeilenindex %d" % index) + du = migra.DataUnit() + if clear: + du.fill(config.BACKGROUND) + du.attributes(fg=config.COLOR, bg=config.BACKGROUND, blink=migra.STATIC) + self._emit_line(du, config.LINE_Y[index], text) + self.send(du) + + 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) + du.charset(config.CHARSET, spaced=config.CHARSET_SPACED) + du.cursor(0, y) + for i, part in enumerate(text.split(config.DEGREE_MARKER)): + if i > 0: + self._emit_degree(du) + if part: + du.text(part) + + def _emit_degree(self, du): + """Das Gradzeichen einfuegen -- ueber Zeichensatzwechsel oder als Fallback.""" + if config.DEGREE_CHARSET is None: + du.text(config.DEGREE_FALLBACK) + return + du.charset(config.DEGREE_CHARSET, spaced=config.CHARSET_SPACED) + du.text(config.DEGREE_CHAR) + du.charset(config.CHARSET, spaced=config.CHARSET_SPACED) + + def close(self): + self._t.close() + + +def open_display(port=config.PORT, fake=False, response=config.WANT_RESPONSE): + """Transport passend zur Plattform oeffnen und Display-Objekt liefern.""" + if fake: + from transport import FakeTransport + t = FakeTransport(src=config.HOST_ADDR, display_addr=config.DISPLAY_ADDR) + else: + from transport import SerialTransport + t = SerialTransport( + port=port, + baudrate=config.BAUDRATE, + parity=config.PARITY, + bytesize=config.BYTESIZE, + stopbits=config.STOPBITS, + timeout=config.RESPONSE_TIMEOUT, + ) + return Display(t, response=response) diff --git a/loopback.py b/loopback.py new file mode 100644 index 0000000..55aee75 --- /dev/null +++ b/loopback.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python3 +"""Loopback-Test des seriellen Adapters. + +Sendet ein paar Bytes und prueft, ob sie zurueckkommen. Dazu auf der TTL-Seite +des Adapters TXD und RXD bruecken. + + - Kommen die Bytes zurueck -> Adapter-RX + Software + Port sind in Ordnung, + das Problem liegt in der Verdrahtung MAX232 <-> Anzeige. + - Kommt nichts zurueck -> Adapter-RX, Port oder Bruecke pruefen. + + python3 loopback.py +""" + +import sys +import config +from transport import SerialTransport + + +def main(): + t = SerialTransport(port=config.PORT, baudrate=config.BAUDRATE, + parity=config.PARITY, bytesize=config.BYTESIZE, + stopbits=config.STOPBITS, timeout=1.0) + probe = bytes([0x55, 0xAA, 0x02, ord("A"), 0x03]) + print("Port:", config.PORT) + print("TX:", " ".join("%02X" % b for b in probe)) + t.write(probe) + + import time + deadline = time.monotonic() + 1.0 + buf = bytearray() + while time.monotonic() < deadline and len(buf) < len(probe): + chunk = t._port.read(len(probe)) + if chunk: + buf.extend(chunk) + t.close() + + print("RX:", " ".join("%02X" % b for b in buf) if buf else "(nichts)") + if bytes(buf) == probe: + print("OK -> Adapter-Empfang funktioniert. Fehler liegt in der Verkabelung zur Anzeige.") + return 0 + if buf: + print("Teilweise/verfaelscht -> Baudrate/Parity oder Wackelkontakt pruefen.") + return 1 + print("Nichts empfangen -> Bruecke TXD<->RXD sitzt nicht, oder Adapter-RX defekt.") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/lx200.py b/lx200.py new file mode 100644 index 0000000..7c9a2c1 --- /dev/null +++ b/lx200.py @@ -0,0 +1,46 @@ +"""LX200-Protokoll fuer die 10Micron GM4000 HPS II. + +Nur das Rahmen der Kommandos und Entpacken der Antworten. Das Parsen der +Koordinatenwerte macht coords.py (dessen Parser ist bewusst tolerant gegenueber +den verschiedenen LX200-Trennzeichen ':', '*', 0xDF, "'"). + +Reines Python, laeuft auch unter MicroPython. + +Kommando: ":" "#" z.B. :GR# +Antwort: "#" z.B. 18:36:56# (Get-Kommandos) + bzw. ein einzelnes Zeichen ohne "#" bei Steuerkommandos (hier ungenutzt) +""" + +TERMINATOR = "#" + +# Abfragekommandos (Get) -- liefern eine mit "#" abgeschlossene Antwort. +GET_RA = ":GR#" # Rektaszension des Teleskops +GET_DEC = ":GD#" # Deklination des Teleskops +GET_PRODUCT = ":GVP#" # Produktname, z.B. "10micron GM4000HPS" +GET_FIRMWARE = ":GVN#" # Firmware-Version + + +class LX200Error(Exception): + """Fehler im LX200-Protokoll.""" + + +def command(cmd): + """Einen Befehl in die Rahmenform ":#" bringen. + + Nimmt sowohl "GR" als auch bereits gerahmte ":GR#" entgegen. + """ + cmd = cmd.strip() + if cmd.startswith(":"): + cmd = cmd[1:] + if cmd.endswith(TERMINATOR): + cmd = cmd[:-1] + if not cmd: + raise LX200Error("leeres Kommando") + return ":" + cmd + TERMINATOR + + +def strip_terminator(response): + """Das abschliessende "#" entfernen und Rest liefern.""" + if not response.endswith(TERMINATOR): + raise LX200Error("Antwort ohne Abschluss '#': %r" % (response,)) + return response[:-len(TERMINATOR)] diff --git a/migra.py b/migra.py new file mode 100644 index 0000000..29bf739 --- /dev/null +++ b/migra.py @@ -0,0 +1,185 @@ +"""Protokoll-Layer fuer die microSYST migra SC/MC Seriell LED-Grossanzeige. + +Reines Python ohne Abhaengigkeiten, laeuft unter CPython und MicroPython. +Referenz: Benutzerhandbuch X-M31-9AXX5X-006 (V4.20), Kapitel 3.5 bis 3.9. + +Telegrammaufbau: STX | DA | SA | FC | [LEN-H LEN-L] | Data-Unit | [CHK-H CHK-L] | ETX +""" + +STX = 0x02 +ETX = 0x03 +ESC = 0x1B +SEP = 0x1F # Trennzeichen ESC-Sequenz -> Online-Text (Handbuch 3.9) + +MAX_DATA_UNIT = 230 + +# Farben (ESC-A, ESC-F, ESC-P, ESC-H) +BLACK = "0" +GREEN = "1" +RED = "2" +YELLOW = "3" +TRANSPARENT = "T" # nur als Hintergrundfarbe + +STATIC = "0" +BLINKING = "1" + +# Fehlercodes aus dem Antworttelegramm (Handbuch 3.8) +ERRORS = { + "0": "kein Fehler", + "1": "falsche Pruefsumme", + "2": "reserviert", + "3": "falsche Anzahl Datenbytes (LEN-H/L) oder ungueltige ESC-Sequenz", + "4": "Element nicht vorhanden oder ungueltiger Parameter", + "5": "Flash ungueltig", +} + + +class MigraError(Exception): + """Fehler im Protokoll oder von der Anzeige gemeldeter Fehlercode.""" + + +def encode_text(s): + """Text in Anzeige-Bytes wandeln. + + Die Data-Unit erlaubt 0x20-0xFF sowie 0x0A und 0x0D. Zeichen darueber + hinaus werden zu '?'. Die Abbildung entspricht Latin-1, das sich mit dem + Windows-Zeichensatz der Anzeige im Bereich 0xA0-0xFF deckt (u.a. Grad 0xB0). + """ + out = bytearray() + for ch in s: + c = ord(ch) + if c in (0x0A, 0x0D) or 0x20 <= c <= 0xFF: + out.append(c) + else: + out.append(0x3F) # '?' + return bytes(out) + + +def _digits(value, places): + """Zahl als ASCII-Ziffern fester Laenge (Handbuch 3.7: alle Werte ASCII-codiert).""" + 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") + + +class DataUnit: + """Baut eine Data-Unit aus mehreren Teiltelegrammen. + + Setzt das Trennzeichen 0x1F selbsttaetig, wenn auf eine ESC-Sequenz ein + Online-Text folgt. Nur dieser Uebergang braucht es (Handbuch 3.9). + """ + + def __init__(self): + self._buf = bytearray() + self._after_esc = False + + def __len__(self): + return len(self._buf) + + def bytes(self): + return bytes(self._buf) + + def _esc(self, *parts): + self._buf.append(ESC) + for p in parts: + self._buf.extend(p if isinstance(p, bytes) else p.encode("ascii")) + self._after_esc = True + return self + + # -- Online-Text ------------------------------------------------------ + def text(self, s): + if self._after_esc: + self._buf.append(SEP) + self._buf.extend(encode_text(s)) + self._after_esc = False + return self + + # -- Zeichensatz, Cursor, Attribute (3.7.1) --------------------------- + def charset(self, number, spaced=True): + """spaced=True -> 'z' (Sperrschrift, gleiche Breite), False -> 'Z'.""" + return self._esc("z" if spaced else "Z", _digits(number, 2)) + + def cursor(self, x, y): + """Cursor auf Pixelposition x/y. 0/0 ist links oben.""" + return self._esc("C", _digits(x, 3), _digits(y, 3)) + + def attributes(self, fg=RED, bg=BLACK, blink=STATIC): + return self._esc("A", fg, bg, blink) + + # -- Direkte Grafik (3.7.3) ------------------------------------------- + def fill(self, color=BLACK): + """Anzeige komplett fuellen. fill(BLACK) = loeschen.""" + return self._esc("F", color) + + def point(self, x, y, color=RED): + return self._esc("P", color, _digits(x, 3), _digits(y, 3)) + + # -- Gespeicherte Objekte (3.7.2) ------------------------------------- + def stored_text(self, number, show=True): + return self._esc("T", "+" if show else "-", _digits(number, 3)) + + def graphic(self, number, show=True): + return self._esc("G", "+" if show else "-", _digits(number, 3)) + + # -- Allgemeine Funktionen (3.7.4) ------------------------------------ + def brightness(self, percent, color=RED): + """Helligkeit 0-100 %. Wird geraeteintern noch mit dem HEX-Schalter multipliziert.""" + if not 0 <= percent <= 100: + raise MigraError("Helligkeit %r ausserhalb 0-100 %%" % percent) + return self._esc("H", color, _digits(percent, 3)) + + def blink_period(self, level): + """0 = 2 s bis 9 = 0,2 s.""" + return self._esc("B", _digits(level, 1)) + + +def build_telegram(data, dest=1, src=0, checksum=False, response=True): + """Data-Unit in ein vollstaendiges Telegramm rahmen. + + dest/src sind die reinen Adressen (0-126); Bit 7 wird hier gesetzt. + checksum=False bedeutet: LEN-H/L und CHK-H/L duerfen *nicht* mitgesendet + werden (Handbuch 3.5.1). + """ + if isinstance(data, DataUnit): + data = data.bytes() + if len(data) > MAX_DATA_UNIT: + raise MigraError("Data-Unit hat %d Bytes, erlaubt sind %d" % (len(data), MAX_DATA_UNIT)) + if not 0 <= dest <= 127 or not 0 <= src <= 126: + raise MigraError("Adresse ausserhalb des gueltigen Bereichs") + + da = 0x80 | dest + sa = 0x80 | src + fc = 0x80 | (0x02 if checksum else 0x00) | (0x01 if response else 0x00) + + frame = bytearray([STX, da, sa, fc]) + if checksum: + n = len(data) + frame.append(0xF0 | ((n >> 4) & 0x0F)) + frame.append(0xF0 | (n & 0x0F)) + frame.extend(data) + if checksum: + # Summe aus DA, SA, FC, LEN-H, LEN-L und allen Datenbytes; ohne STX. + chk = sum(frame[1:]) & 0xFF + frame.append(0xF0 | ((chk >> 4) & 0x0F)) + frame.append(0xF0 | (chk & 0x0F)) + frame.append(ETX) + return bytes(frame) + + +def parse_response(frame, src=0): + """Antworttelegramm auswerten und den Fehlercode als Zeichen liefern. + + Erwartet: STX | DA | SA | FC | Fehlercode | ETX (Handbuch 3.8) + Wirft MigraError, wenn die Anzeige einen Fehler ungleich '0' meldet. + """ + 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()) + if frame[1] != (0x80 | src): + raise MigraError("Antwort nicht an uns adressiert: %s" % frame.hex()) + + code = chr(frame[4]) + if code != "0": + raise MigraError("Anzeige meldet Fehler '%s': %s" % (code, ERRORS.get(code, "unbekannt"))) + return code diff --git a/mount_client.py b/mount_client.py new file mode 100644 index 0000000..f7e8159 --- /dev/null +++ b/mount_client.py @@ -0,0 +1,88 @@ +"""TCP-Client zur 10Micron-Montierung (LX200 ueber Ethernet). + +Die Socket-API (socket.socket/connect/send/recv) ist unter CPython und +MicroPython gleich, laeuft also spaeter auch auf dem ESP32-C3. +""" + +import socket + +import lx200 + + +class MountError(Exception): + """Verbindungs- oder Kommunikationsfehler mit der Montierung.""" + + +class MountClient: + """Fragt Koordinaten von der Montierung ab. + + Verbindung wird nicht automatisch gehalten -- bei einem Fehler wirft die + Abfrage MountError; der Aufrufer entscheidet ueber Reconnect (siehe + run_display.py). + """ + + def __init__(self, host, port=3490, timeout=3.0): + self.host = host + self.port = port + self.timeout = timeout + self._sock = None + + def connect(self): + self.close() + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(self.timeout) + try: + sock.connect((self.host, self.port)) + except OSError as e: + sock.close() + raise MountError("Verbindung zu %s:%d fehlgeschlagen: %s" + % (self.host, self.port, e)) + self._sock = sock + + @property + def connected(self): + return self._sock is not None + + def _recv_until_terminator(self): + buf = bytearray() + while True: + try: + chunk = self._sock.recv(64) + except OSError as e: + raise MountError("Empfang gestoert: %s" % e) + if not chunk: + raise MountError("Verbindung von der Montierung geschlossen") + buf.extend(chunk) + if buf.endswith(lx200.TERMINATOR.encode("latin-1")): + return bytes(buf) + + def query(self, framed_cmd): + """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) + raw = self._recv_until_terminator().decode("latin-1") + return lx200.strip_terminator(raw) + + def get_ra(self): + """Rektaszension als Rohstring, z.B. '18:36:56'.""" + return self.query(lx200.GET_RA) + + def get_dec(self): + """Deklination als Rohstring, z.B. '+38*47:01'.""" + return self.query(lx200.GET_DEC) + + def get_coordinates(self): + """(ra, dec) als Rohstrings in einem Rutsch.""" + return self.get_ra(), self.get_dec() + + def close(self): + if self._sock is not None: + try: + self._sock.close() + except OSError: + pass + self._sock = None diff --git a/mount_mock.py b/mount_mock.py new file mode 100644 index 0000000..3bfed0d --- /dev/null +++ b/mount_mock.py @@ -0,0 +1,195 @@ +#!/usr/bin/env python3 +"""Mock der 10Micron GM4000 HPS II -- ein TCP-Server, der LX200-Abfragen beantwortet. + +Zum Entwickeln ohne echtes Teleskop. Antwortet auf die Koordinaten- und +Versions-Abfragen; unbekannte Kommandos werden mit "#" quittiert, damit der +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) + +Nur fuer die Entwicklung (CPython). Auf dem ESP32 laeuft spaeter nur der Client. +""" + +import argparse +import socket +import sys +import threading +import time + +import lx200 + +PRODUCT = "10micron GM4000HPS" +FIRMWARE = "2.15.11" + + +def _parse_ra(text): + """RA-String (HH:MM:SS o.ae.) in Sekunden 0..86399.""" + nums = _numbers(text) + h, m, s = (nums + [0, 0, 0])[:3] + return (int(h) * 3600 + int(m) * 60 + int(s)) % 86400 + + +def _parse_dec(text): + """DEC-String (sDD*MM:SS) in vorzeichenbehaftete Bogensekunden.""" + neg = text.strip().startswith("-") + nums = _numbers(text) + d, m, s = (nums + [0, 0, 0])[:3] + total = int(d) * 3600 + int(m) * 60 + int(s) + return -total if neg else total + + +def _numbers(text): + out, cur = [], "" + for ch in text: + if ch.isdigit(): + cur += ch + elif cur: + out.append(cur) + cur = "" + if cur: + out.append(cur) + return [int(x) for x in out] + + +def _fmt_ra(total_s): + total_s %= 86400 + h, rest = divmod(total_s, 3600) + m, s = divmod(rest, 60) + return "%02d:%02d:%02d" % (h, m, s) + + +def _fmt_dec(total_arcsec): + sign = "-" if total_arcsec < 0 else "+" + total_arcsec = abs(total_arcsec) + d, rest = divmod(total_arcsec, 3600) + m, s = divmod(rest, 60) + # Gradzeichen als '*', Minuten/Sekunden mit ':' -- der coords-Parser + # kommt mit beiden Trennzeichen klar. + return "%s%02d*%02d:%02d" % (sign, d, m, s) + + +class MountMock: + """Simuliert die Montierung auf TCP-Ebene.""" + + def __init__(self, host="127.0.0.1", port=3490, + ra="18:36:56", dec="+38:47:01", move=False): + self._ra0 = _parse_ra(ra) + self._dec = _parse_dec(dec) + self._move = move + self._tick = 0 # zaehlt bei --move pro RA-Abfrage hoch + self._dectick = 0 # zaehlt bei --move pro DEC-Abfrage hoch + self._lock = threading.Lock() + self._srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._srv.bind((host, port)) + self._srv.listen(5) + self.host, self.port = self._srv.getsockname() # echter Port (bei port=0) + self._running = False + self._thread = None + + # -- aktuelle Koordinaten -------------------------------------------- + def _current_ra(self): + """RA liefern; bei --move pro Abfrage genau 1 Sekunde weiter (kein Zeit-Jitter).""" + if not self._move: + return self._ra0 + with self._lock: + ra = self._ra0 + self._tick + self._tick += 1 + return ra + + def _current_dec(self): + """DEC liefern; bei --move pro Abfrage 1 Bogensekunde weiter.""" + if not self._move: + return self._dec + with self._lock: + dec = self._dec + self._dectick + self._dectick += 1 + return dec + + def _reply_for(self, cmd): + """Antwort (inkl. '#') fuer ein Kommando ohne ':' und ohne '#'.""" + if cmd == "GR": + return _fmt_ra(self._current_ra()) + "#" + if cmd == "GD": + return _fmt_dec(self._current_dec()) + "#" + if cmd == "GVP": + return PRODUCT + "#" + if cmd == "GVN": + return FIRMWARE + "#" + # Unbekannt: leere, aber abgeschlossene Antwort, damit der Client weiterlaeuft. + return "#" + + def _handle(self, conn): + buf = "" + try: + while self._running: + data = conn.recv(64) + if not data: + break + buf += data.decode("latin-1") + # vollstaendige Kommandos (durch '#' getrennt) abarbeiten + while "#" in buf: + raw, buf = buf.split("#", 1) + cmd = raw.lstrip(":").strip() + if cmd: + conn.sendall(self._reply_for(cmd).encode("latin-1")) + except OSError: + pass + finally: + conn.close() + + def _accept_loop(self): + while self._running: + try: + conn, _ = self._srv.accept() + except OSError: + break + threading.Thread(target=self._handle, args=(conn,), daemon=True).start() + + def start(self): + """Server im Hintergrund-Thread starten.""" + self._running = True + self._thread = threading.Thread(target=self._accept_loop, daemon=True) + self._thread.start() + + def stop(self): + self._running = False + try: + self._srv.close() + except OSError: + pass + + def serve_forever(self): + self.start() + try: + while True: + time.sleep(0.5) + except KeyboardInterrupt: + pass + finally: + self.stop() + + +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("--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, + 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 "")) + print("Beenden mit Ctrl-C.") + mock.serve_forever() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/probe.py b/probe.py new file mode 100644 index 0000000..1939d21 --- /dev/null +++ b/probe.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +"""Geometrie der Anzeige ausmessen. + +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 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 +""" + +import sys +import time + +import config +import migra +from display import open_display + + +def ruler(d): + """Ziffernlineal auf EINER Zeile: die Position, an der es umbricht, ist die Breite. + + Bewusst nur eine Zeile und etwas mehr Zeichen als erwartet, damit man den + Umbruch sieht, ohne dass sich mehrere Zeilen ueberschreiben. + """ + n = config.CHARS_PER_LINE + 5 + line = "".join(str((i + 1) % 10) for i in range(n)) + du = migra.DataUnit() + du.fill(config.BACKGROUND) + du.charset(config.CHARSET, spaced=config.CHARSET_SPACED) + du.attributes(fg=config.COLOR, bg=config.BACKGROUND) + du.cursor(0, config.LINE_Y[0]) + du.text(line) + d.send(du) + print("Bis zu welcher Ziffer laeuft die obere Zeile, bevor sie umbricht?") + print("Das ist CHARS_PER_LINE (aktuell in config.py: %d)." % config.CHARS_PER_LINE) + + +def lines(d): + d.show_lines("Zeile 1", "Zeile 2") + print("Stehen beide Zeilen sauber untereinander? Sonst LINE_Y anpassen.") + + +def scan_y(d): + """Zeile schrittweise nach unten schieben, um die y-Position zu finden.""" + for y in range(0, 33, 2): + du = migra.DataUnit() + du.fill(config.BACKGROUND) + du.charset(config.CHARSET, spaced=config.CHARSET_SPACED) + du.attributes(fg=config.COLOR, bg=config.BACKGROUND) + du.cursor(0, y) + du.text("y=%d" % y) + try: + d.send(du) + except migra.MigraError as e: + print("y=%-3d -> %s" % (y, e)) + continue + print("y=%-3d -> ok" % y) + time.sleep(1.5) + + +def corners(d): + """Punkte in die Ecken setzen. Ein Fehlercode '4' heisst: ausserhalb der Anzeige.""" + for x, y in ((0, 0), (63, 0), (0, 15), (63, 15), (127, 31)): + du = migra.DataUnit().point(x, y, color=config.COLOR) + try: + d.send(du) + print("Punkt %3d/%-3d -> ok" % (x, y)) + except migra.MigraError as e: + print("Punkt %3d/%-3d -> %s" % (x, y, e)) + + +MODES = {"ruler": ruler, "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: + print(__doc__) + return 2 + response = config.WANT_RESPONSE and "--no-response" not in sys.argv + d = open_display(response=response) + try: + MODES[args[0]](d) + finally: + d.close() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/run_display.py b/run_display.py new file mode 100644 index 0000000..6a33d8c --- /dev/null +++ b/run_display.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +"""Hauptprogramm Phase 2: Koordinaten von der Montierung pollen und anzeigen. + +Fragt zyklisch RA/DEC von der 10Micron-Montierung ab und schreibt sie ueber die +Grossanzeige. Bei Verbindungsproblemen wird eine Hinweiszeile angezeigt und im +Hintergrund erneut verbunden. + + python3 run_display.py # Montierung aus config.py + python3 run_display.py --host 10.0.0.42 + python3 run_display.py --mock # lokalen Mock starten und nutzen + python3 run_display.py --mock --move --once # einmalig, mit laufender RA +""" + +import argparse +import sys +import time + +import config +import coords +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) + + +def main(argv=None): + p = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--host", default=config.MOUNT_HOST) + p.add_argument("--port", type=int, default=config.MOUNT_PORT) + p.add_argument("--interval", type=float, default=config.POLL_INTERVAL) + p.add_argument("--once", action="store_true", help="nur eine Abfrage, dann Ende") + p.add_argument("--mock", action="store_true", + help="lokalen Mock starten und gegen ihn laufen") + p.add_argument("--move", action="store_true", + help="nur mit --mock: RA laeuft mit der Zeit") + p.add_argument("--dry-run", action="store_true", + help="Anzeige nur als Hex mitschreiben (keine serielle Hardware)") + args = p.parse_args(argv) + + host, port = args.host, args.port + mock = None + if args.mock: + from mount_mock import MountMock + mock = MountMock(host="127.0.0.1", port=0, move=args.move) + mock.start() + host, port = mock.host, mock.port + print("Mock-Montierung laeuft auf %s:%d" % (host, port)) + + d = open_display(fake=args.dry_run) + client = MountClient(host, port, timeout=config.MOUNT_TIMEOUT) + last_shown = None + + try: + while True: + try: + if not client.connected: + client.connect() + 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 (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) + except KeyboardInterrupt: + print("\nbeendet") + finally: + client.close() + d.close() + if mock is not None: + mock.stop() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scope.py b/scope.py new file mode 100644 index 0000000..9c41b27 --- /dev/null +++ b/scope.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +"""Dauersenden auf der seriellen Schnittstelle, zum Messen mit dem Oszilloskop. + +Es wird ohne Antworttelegramm gesendet (die Anzeige haengt ja noch nicht dran), +d.h. FC = 0x80 statt 0x81. + +Muster: + telegram das echte Koordinaten-Telegramm (Vorgabe) + 0x55 01010101 -- bei 8E1 die schnellste Bitfolge, ideal zum + Ausmessen der Bitzeit (1 Bit = 52,08 us bei 19200 Baud) + 0xAA 10101010 + 0x00 / 0xFF konstant, zum Pruefen der Ruhepegel + count 0x00, 0x01, 0x02 ... fortlaufend + +Beispiele: + python3 scope.py # 5 min, Telegramm alle 200 ms + python3 scope.py --pattern 0x55 --interval 0.05 --minutes 10 + python3 scope.py --pattern 0x55 --interval 0 # so schnell wie moeglich +""" + +import argparse +import sys +import time + +import config +import coords +import migra +from transport import SerialTransport + +PATTERNS = {"0x55": 0x55, "0xAA": 0xAA, "0x00": 0x00, "0xFF": 0xFF} + + +def build_frame(pattern, counter): + """Die zu sendenden Bytes fuer einen Durchlauf liefern.""" + if pattern == "telegram": + du = migra.DataUnit() + du.fill(config.BACKGROUND) + du.charset(config.CHARSET) + du.attributes(fg=config.COLOR, bg=config.BACKGROUND) + du.cursor(0, config.LINE_Y[0]) + du.text(coords.format_ra("12h34m56s")) + du.cursor(0, config.LINE_Y[1]) + du.text(coords.format_dec("+45d12m30s")) + return migra.build_telegram(du, dest=config.DISPLAY_ADDR, + src=config.HOST_ADDR, + checksum=config.USE_CHECKSUM, + response=False) + if pattern == "count": + return bytes([counter & 0xFF]) + return bytes([PATTERNS[pattern]] * 16) + + +def main(argv=None): + p = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--pattern", default="telegram", + choices=["telegram", "count"] + list(PATTERNS)) + p.add_argument("--port", default=config.PORT) + p.add_argument("--interval", type=float, default=0.2, + help="Pause zwischen den Sendungen in Sekunden (Vorgabe: %(default)s)") + p.add_argument("--minutes", type=float, default=5.0, + help="Laufzeit in Minuten (Vorgabe: %(default)s)") + args = p.parse_args(argv) + + t = SerialTransport(port=args.port, baudrate=config.BAUDRATE, + parity=config.PARITY, bytesize=config.BYTESIZE, + stopbits=config.STOPBITS, timeout=0.1) + + first = build_frame(args.pattern, 0) + bit_us = 1e6 / config.BAUDRATE + print("Port : %s" % args.port) + print("Format : %d Baud, %d%s%d (1 Bit = %.2f us, 1 Byte = 11 Bit = %.1f us)" + % (config.BAUDRATE, config.BYTESIZE, config.PARITY, config.STOPBITS, + bit_us, bit_us * 11)) + print("Muster : %s, %d Byte pro Durchlauf" % (args.pattern, len(first))) + print(" %s" % " ".join("%02X" % b for b in first)) + print("Intervall : %s s, Laufzeit %s min" % (args.interval, args.minutes)) + print("Abbruch mit Ctrl-C.\n") + + deadline = time.monotonic() + args.minutes * 60 + n = 0 + try: + while time.monotonic() < deadline: + t.write(build_frame(args.pattern, n)) + n += 1 + if n % 25 == 0: + rest = deadline - time.monotonic() + print("\r%6d Sendungen, noch %4.1f min " % (n, rest / 60), end="") + sys.stdout.flush() + if args.interval: + time.sleep(args.interval) + except KeyboardInterrupt: + print("\nabgebrochen") + finally: + t.close() + + print("\n%d Sendungen abgesetzt." % n) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/show_coords.py b/show_coords.py new file mode 100644 index 0000000..5f285be --- /dev/null +++ b/show_coords.py @@ -0,0 +1,67 @@ +#!/usr/bin/env python3 +"""Testprogramm: Koordinaten von Hand uebergeben und auf der Grossanzeige ausgeben. + +Beispiele: + python3 show_coords.py 12h34m56s +45d12m30s + python3 show_coords.py 12:34:56 -07:15:00 + python3 show_coords.py 12h34m56s +45d12m30s --dry-run + python3 show_coords.py --clear +""" + +import argparse +import sys + +import config +import coords +import migra +from display import open_display + + +def main(argv=None): + p = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("ra", nargs="?", help="Rektaszension, z.B. 12h34m56s oder 12:34:56") + p.add_argument("dec", nargs="?", help="Deklination, z.B. +45d12m30s oder +45:12:30") + p.add_argument("--port", default=config.PORT, help="serieller Port (Vorgabe: %(default)s)") + p.add_argument("--brightness", type=int, help="Helligkeit 0-100 %% setzen") + p.add_argument("--clear", action="store_true", help="Anzeige loeschen und beenden") + p.add_argument("--dry-run", action="store_true", + help="nichts senden, nur die Telegramme als Hex zeigen") + p.add_argument("--no-response", action="store_true", + help="ohne Antworttelegramm senden (wenn die RS232-Rueckleitung fehlt)") + args = p.parse_args(argv) + + if not args.clear and (args.ra is None or args.dec is None): + p.error("Rektaszension und Deklination angeben (oder --clear)") + + if not args.clear: + line1 = coords.format_ra(args.ra) + line2 = coords.format_dec(args.dec) + print("Zeile 1: %r" % line1) + print("Zeile 2: %r" % line2) + for name, line in (("Zeile 1", line1), ("Zeile 2", line2)): + if len(line) > config.CHARS_PER_LINE: + print("Warnung: %s ist laenger als CHARS_PER_LINE=%d" + % (name, config.CHARS_PER_LINE), file=sys.stderr) + + d = open_display(port=args.port, fake=args.dry_run, + response=config.WANT_RESPONSE and not args.no_response) + try: + if args.brightness is not None: + d.set_brightness(args.brightness) + if args.clear: + d.clear() + else: + d.show_lines(line1, line2, clear=True) # Einzelaufruf: einmal sauber loeschen + except migra.MigraError as e: + print("Fehler: %s" % e, file=sys.stderr) + return 1 + finally: + d.close() + + print("OK") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/test_lx200.py b/test_lx200.py new file mode 100644 index 0000000..01887e7 --- /dev/null +++ b/test_lx200.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python3 +"""Tests fuer LX200-Protokoll, Mock und Client (Phase 2). + + python3 -m unittest -v test_lx200 +""" + +import unittest + +import coords +import lx200 +import mount_mock +from mount_client import MountClient, MountError + + +class TestLX200(unittest.TestCase): + + def test_command_rahmen(self): + self.assertEqual(lx200.command("GR"), ":GR#") + self.assertEqual(lx200.command(":GR#"), ":GR#") + self.assertEqual(lx200.command("GR#"), ":GR#") + self.assertEqual(lx200.command(":GR"), ":GR#") + + def test_command_leer(self): + with self.assertRaises(lx200.LX200Error): + lx200.command(":#") + + def test_strip_terminator(self): + self.assertEqual(lx200.strip_terminator("18:36:56#"), "18:36:56") + with self.assertRaises(lx200.LX200Error): + lx200.strip_terminator("18:36:56") + + +class TestAntwortParsen(unittest.TestCase): + """Die LX200-Rohantworten muessen direkt in den coords-Parser passen.""" + + def test_ra_antwort(self): + self.assertEqual(coords.parse_ra("18:36:56"), (18, 36, 56)) + + def test_dec_antwort_mit_stern(self): + self.assertEqual(coords.parse_dec("+38*47:01"), ("+", 38, 47, 1)) + + def test_dec_antwort_mit_gradzeichen_0xDF(self): + # Meade-Stil: 0xDF als Gradzeichen + self.assertEqual(coords.parse_dec("-07\xdf15:30"), ("-", 7, 15, 30)) + + +class TestMockFormat(unittest.TestCase): + + def test_ra_roundtrip(self): + self.assertEqual(mount_mock._fmt_ra(mount_mock._parse_ra("18:36:56")), "18:36:56") + + def test_dec_roundtrip(self): + s = mount_mock._parse_dec("-07*15:30") + self.assertEqual(mount_mock._fmt_dec(s), "-07*15:30") + + def test_ra_wrap(self): + # 23:59:59 + 2s -> 00:00:01 + self.assertEqual(mount_mock._fmt_ra(mount_mock._parse_ra("23:59:59") + 2), "00:00:01") + + +class TestClientGegenMock(unittest.TestCase): + """Vollstaendiger Weg: Client <-> Mock ueber echtes TCP.""" + + def setUp(self): + self.mock = mount_mock.MountMock(host="127.0.0.1", port=0, + ra="18:36:56", dec="+38:47:01") + self.mock.start() + self.client = MountClient("127.0.0.1", self.mock.port, timeout=2.0) + self.client.connect() + + def tearDown(self): + self.client.close() + self.mock.stop() + + def test_get_ra(self): + self.assertEqual(self.client.get_ra(), "18:36:56") + + def test_get_dec(self): + self.assertEqual(self.client.get_dec(), "+38*47:01") + + def test_get_coordinates_und_formatierung(self): + ra_raw, dec_raw = self.client.get_coordinates() + self.assertEqual(coords.format_ra(ra_raw), "18h36m56s") + self.assertEqual(coords.format_dec(dec_raw), "+38°47'01") # 9 Zeichen, ohne " am Ende + + def test_produktname(self): + self.assertEqual(self.client.query(lx200.GET_PRODUCT), mount_mock.PRODUCT) + + def test_unbekanntes_kommando_haengt_nicht(self): + # Muss eine (leere) abgeschlossene Antwort liefern, nicht blockieren. + self.assertEqual(self.client.query(":GXYZ#"), "") + + +class TestClientFehler(unittest.TestCase): + + def test_verbindung_verweigert(self): + # Port, auf dem nichts lauscht + c = MountClient("127.0.0.1", 1, timeout=0.5) + with self.assertRaises(MountError): + c.connect() + + def test_query_ohne_verbindung(self): + c = MountClient("127.0.0.1", 3490) + with self.assertRaises(MountError): + c.get_ra() + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/test_migra.py b/test_migra.py new file mode 100644 index 0000000..7444219 --- /dev/null +++ b/test_migra.py @@ -0,0 +1,229 @@ +#!/usr/bin/env python3 +"""Tests fuer den Protokoll-Layer, gepruefte Referenz ist das Handbuch. + + python3 -m unittest -v test_migra +""" + +import unittest + +import coords +import migra +from display import Display +from transport import FakeTransport + + +class TestTelegramm(unittest.TestCase): + + def test_handbuch_beispiel_mit_pruefsumme(self): + """Handbuch 3.10, Beispiel 2: Zeichen 'A' an Anzeige mit Adresse 1. + + Master: 02 81 80 83 F0 F1 41 FA F6 03 + """ + frame = migra.build_telegram(b"A", dest=1, src=0, checksum=True, response=True) + self.assertEqual(frame, bytes.fromhex("02818083F0F141FAF603")) + + def test_handbuch_beispiel_ohne_pruefsumme(self): + """Handbuch 3.10, Beispiel 1: ohne Pruefsumme, mit Antwort. + + Ohne Pruefsumme duerfen LEN-H/L gerade *nicht* mitgesendet werden. + """ + frame = migra.build_telegram(b"Hallo Welt", dest=1, src=0, + checksum=False, response=True) + self.assertEqual(frame[:4], bytes([0x02, 0x81, 0x80, 0x81])) + self.assertEqual(frame[4:-1], b"Hallo Welt") + self.assertEqual(frame[-1], 0x03) + + def test_frame_control_bits(self): + def fc(checksum, response): + return migra.build_telegram(b"x", checksum=checksum, response=response)[3] + + self.assertEqual(fc(False, False), 0x80) + self.assertEqual(fc(False, True), 0x81) # so faehrt das alte C-Programm + self.assertEqual(fc(True, False), 0x82) + self.assertEqual(fc(True, True), 0x83) + + def test_data_unit_zu_lang(self): + with self.assertRaises(migra.MigraError): + migra.build_telegram(b"x" * 231) + + +class TestDataUnit(unittest.TestCase): + + def test_trennzeichen_nach_esc_sequenz(self): + """Handbuch 3.9: Data-Unit = 1B 'Z01' 1B 'C002003' 1B 'A301' 1F Text.""" + du = (migra.DataUnit() + .charset(1, spaced=False) + .cursor(2, 3) + .attributes(fg=migra.YELLOW, bg=migra.BLACK, blink=migra.BLINKING) + .text("Online-Text")) + self.assertEqual( + du.bytes(), + b"\x1bZ01\x1bC002003\x1bA301\x1fOnline-Text", + ) + + def test_kein_trennzeichen_zwischen_esc_sequenzen(self): + du = migra.DataUnit().fill(migra.BLACK).charset(0) + self.assertEqual(du.bytes(), b"\x1bF0\x1bz00") + + def test_trennzeichen_nur_einmal_pro_textblock(self): + du = migra.DataUnit().cursor(0, 0).text("AB").text("CD") + self.assertEqual(du.bytes(), b"\x1bC000000\x1fABCD") + + def test_cursor_ist_ascii_codiert(self): + """Handbuch 3.7: Position 123 wird als '1' '2' '3' uebertragen.""" + self.assertEqual(migra.DataUnit().cursor(123, 45).bytes(), b"\x1bC123045") + + def test_loeschen_und_helligkeit(self): + self.assertEqual(migra.DataUnit().fill(migra.BLACK).bytes(), b"\x1bF0") + self.assertEqual(migra.DataUnit().brightness(100, migra.RED).bytes(), b"\x1bH2100") + self.assertEqual(migra.DataUnit().brightness(80, migra.RED).bytes(), b"\x1bH2080") + + def test_helligkeit_ausserhalb_bereich(self): + with self.assertRaises(migra.MigraError): + migra.DataUnit().brightness(101) + + def test_text_encoding_latin1(self): + self.assertEqual(migra.encode_text("+45°12'"), b"+45\xb012'") + self.assertEqual(migra.encode_text("✓"), b"?") # nicht darstellbar + + +class TestAntwort(unittest.TestCase): + + def test_ok(self): + frame = bytes([0x02, 0x80, 0x81, 0x80, ord("0"), 0x03]) + self.assertEqual(migra.parse_response(frame), "0") + + def test_fehlercode_wirft(self): + frame = bytes([0x02, 0x80, 0x81, 0x80, ord("3"), 0x03]) + with self.assertRaises(migra.MigraError) as ctx: + migra.parse_response(frame) + self.assertIn("ESC-Sequenz", str(ctx.exception)) + + def test_timeout_wirft(self): + with self.assertRaises(migra.MigraError): + migra.parse_response(b"") + + +class TestKoordinaten(unittest.TestCase): + + def test_ra_formate_werden_erkannt(self): + for text in ("12h34m56s", "12:34:56", "12 34 56"): + self.assertEqual(coords.parse_ra(text), (12, 34, 56)) + + def test_ra_dezimal(self): + self.assertEqual(coords.parse_ra("12.5"), (12, 30, 0)) + + def test_dec_mit_vorzeichen(self): + self.assertEqual(coords.parse_dec("+45d12m30s"), ("+", 45, 12, 30)) + self.assertEqual(coords.parse_dec("-07:15:00"), ("-", 7, 15, 0)) + + def test_dec_ohne_vorzeichen_ist_positiv(self): + self.assertEqual(coords.parse_dec("45:12:30"), ("+", 45, 12, 30)) + + def test_bereichsgrenzen(self): + with self.assertRaises(ValueError): + coords.parse_ra("24:00:01") + with self.assertRaises(ValueError): + coords.parse_dec("+91:00:00") + + def test_formatierung_passt_sich_der_breite_an(self): + self.assertEqual(coords.format_ra("12h34m56s", width=9), "12h34m56s") + self.assertEqual(coords.format_ra("12h34m56s", width=8), "12:34:56") + self.assertEqual(coords.format_dec("+45d12m30s", width=10), "+45°12'30\"") + self.assertEqual(coords.format_dec("+45d12m30s", width=9), "+45°12'30") # 9: Grad, ohne " + self.assertEqual(coords.format_dec("+45d12m30s", width=7), "+45°12'") # 7: ohne Bogensekunden + + +class TestDisplay(unittest.TestCase): + + def test_zeilen_ergeben_ein_telegramm(self): + t = FakeTransport(echo=False) + Display(t, response=True).show_lines("12:34:56", "+45:12:30", clear=True) + + self.assertEqual(len(t.frames), 1) + frame = t.frames[0] + self.assertEqual(frame[:4], bytes([0x02, 0x81, 0x80, 0x81])) + self.assertEqual(frame[-1], 0x03) + + data = frame[4:-1] + self.assertIn(b"\x1bF0", data) # mit clear=True: vorher loeschen + self.assertIn(b"\x1bC000000\x1f12:34:56", data) # Zeile 1 bei y=0 + self.assertIn(b"\x1bC000009\x1f+45:12:30", data) # Zeile 2 bei y=9 + + def test_ohne_clear_kein_fill_aber_aufgefuellt(self): + """Standard (clear=False): kein Vollbild-Loeschen, Zeilen auf volle Breite.""" + t = FakeTransport(echo=False) + Display(t, response=True).show_lines("12:34:56", clear=False) + data = t.frames[0][4:-1] + self.assertNotIn(b"\x1bF0", data) + # auf CHARS_PER_LINE mit Leerzeichen aufgefuellt + import config + padded = "12:34:56".ljust(config.CHARS_PER_LINE).encode() + self.assertIn(padded, data) + + def test_gradzeichen_via_zeichensatz(self): + """Gradzeichen wird durch kurzen Wechsel auf Zeichensatz 1 erzeugt.""" + t = FakeTransport(echo=False) + Display(t, response=True).show_lines("12h34m56s", "+45°12'30\"") + data = t.frames[0][4:-1] + # Umschalten auf Zeichensatz 1, "/", zurueck auf Zeichensatz 0 + self.assertIn(b"\x1bz01\x1f/\x1bz00", data) + + def test_oberste_zeile_wird_zuletzt_gezeichnet(self): + """Wegen der Home-Zellen-Stoerung muss y=0 nach y=9 im Telegramm stehen.""" + t = FakeTransport(echo=False) + Display(t, response=True).show_lines("12h34m56s", "+45°12'30\"") + data = t.frames[0][4:-1] + self.assertLess(data.index(b"C000009"), data.index(b"C000000")) + + def test_gradzeichen_fallback_ohne_zeichensatz(self): + """Ist DEGREE_CHARSET None, wird der Fallbacktext statt der Umschaltung genutzt.""" + import config + t = FakeTransport(echo=False) + alt = config.DEGREE_CHARSET + try: + config.DEGREE_CHARSET = None + Display(t, response=True).show_lines("+45°12'30\"") + finally: + config.DEGREE_CHARSET = alt + data = t.frames[0][4:-1] + self.assertNotIn(b"\x1bz01", data) + self.assertIn(config.DEGREE_FALLBACK.encode(), data) + + def test_fehlercode_der_anzeige_wirft(self): + t = FakeTransport(echo=False) + t._response = bytes([0x02, 0x80, 0x81, 0x80, ord("4"), 0x03]) + with self.assertRaises(migra.MigraError): + Display(t, response=True).clear() + + def test_show_line_nur_eine_zeile(self): + """show_line schreibt nur die angegebene Zeile, ohne Vollbild-Loeschen.""" + t = FakeTransport(echo=False) + Display(t, response=True).show_line(0, "18h36m56s") + data = t.frames[0][4:-1] + self.assertNotIn(b"\x1bF0", data) # kein Loeschen + self.assertIn(b"\x1bC000000", data) # Cursor auf Zeile 1 (y=0) + self.assertNotIn(b"C000009", data) # Zeile 2 unberuehrt + + def test_show_line_ungueltiger_index(self): + t = FakeTransport(echo=False) + with self.assertRaises(ValueError): + Display(t, response=True).show_line(5, "x") + + def test_ohne_antwort_fc_bit0_ist_null(self): + """Standardbetrieb ohne Antwort: FC = 0x80, kein Fehlercode wird gelesen.""" + import config + alt = config.QUIET_TIME + try: + config.QUIET_TIME = 0.0 + t = FakeTransport(echo=False) + # selbst wenn die (ignorierte) Antwort einen Fehler meldet, darf nichts fliegen + t._response = bytes([0x02, 0x80, 0x81, 0x80, ord("4"), 0x03]) + Display(t, response=False).clear() + finally: + config.QUIET_TIME = alt + self.assertEqual(t.frames[0][3], 0x80) # FC ohne Antwort-Bit + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/transport.py b/transport.py new file mode 100644 index 0000000..d315c10 --- /dev/null +++ b/transport.py @@ -0,0 +1,126 @@ +"""UART-Anbindung der Grossanzeige. + +Kapselt den Unterschied zwischen pyserial (Mac/PC) und machine.UART +(MicroPython auf dem XIAO ESP32-C3), damit display.py auf beiden laeuft. +""" + +import migra + +try: # CPython + import serial +except ImportError: + serial = None + +try: # MicroPython + from machine import UART +except ImportError: + UART = None + +try: + import time + _monotonic = time.monotonic +except AttributeError: # MicroPython kennt kein time.monotonic + _monotonic = lambda: time.ticks_ms() / 1000.0 + + +class SerialTransport: + """Serieller Port ueber pyserial (CPython).""" + + def __init__(self, port, baudrate, parity, bytesize, stopbits, timeout): + if serial is None: + raise RuntimeError("pyserial fehlt: pip install pyserial") + self._port = serial.Serial( + port=port, + baudrate=baudrate, + bytesize=bytesize, + parity=parity, # "E" = even, wie von der Anzeige erwartet + stopbits=stopbits, + timeout=timeout, + ) + self._timeout = timeout + + def write(self, data): + self._port.write(data) + self._port.flush() + + def read_response(self): + """Bytes bis einschliesslich ETX lesen, oder bis der Timeout greift.""" + deadline = _monotonic() + self._timeout + buf = bytearray() + while _monotonic() < deadline: + chunk = self._port.read(1) + if not chunk: + continue + buf.extend(chunk) + if chunk[0] == migra.ETX: + break + return bytes(buf) + + def close(self): + self._port.close() + + +class UartTransport: + """Serieller Port ueber machine.UART (MicroPython / ESP32-C3).""" + + def __init__(self, uart_id, baudrate, parity, bytesize, stopbits, timeout, + tx=None, rx=None): + if UART is None: + raise RuntimeError("machine.UART nicht verfuegbar") + # MicroPython: parity 0 = even, 1 = odd, None = keine + parity_map = {"E": 0, "O": 1, "N": None} + kwargs = dict( + baudrate=baudrate, + bits=bytesize, + parity=parity_map[parity], + stop=stopbits, + timeout=int(timeout * 1000), + ) + if tx is not None: + kwargs["tx"] = tx + if rx is not None: + kwargs["rx"] = rx + self._uart = UART(uart_id, **kwargs) + self._timeout = timeout + + def write(self, data): + self._uart.write(data) + + def read_response(self): + deadline = _monotonic() + self._timeout + buf = bytearray() + while _monotonic() < deadline: + chunk = self._uart.read(1) + if not chunk: + continue + buf.extend(chunk) + if chunk[0] == migra.ETX: + break + return bytes(buf) + + def close(self): + self._uart.deinit() + + +class FakeTransport: + """Schreibt die Telegramme nur als Hex mit, ohne Hardware. + + Fuer Trockenlauf und Tests: liefert immer ein fehlerfreies Antworttelegramm. + """ + + def __init__(self, src=0, display_addr=1, echo=True): + self.frames = [] + self._echo = echo + self._response = bytes([migra.STX, 0x80 | src, 0x80 | display_addr, + 0x80, ord("0"), migra.ETX]) + + def write(self, data): + self.frames.append(data) + if self._echo: + print("TX:", " ".join("%02X" % b for b in data)) + + def read_response(self): + return self._response + + def close(self): + pass