grossanzeige/run_esp32.py

"""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, resolve_host

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 = resolve_host(host, log=log)
    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 make_watchdog(timeout=None, log=None):
    """Watchdog starten und seine feed-Funktion liefern.

    Liefert None, wenn es keinen gibt (CPython). **Ein gestarteter Watchdog
    laesst sich nicht mehr abschalten** -- deshalb ruft nur main.py das hier
    auf, nicht der Handbetrieb ueber die REPL: Sonst startet das Geraet mitten
    im Entwickeln neu, sobald man an der REPL nachdenkt.
    """
    log = log or (lambda *a: None)
    try:
        from machine import WDT
    except ImportError:
        log("Watchdog: machine.WDT nicht verfuegbar")
        return None
    sekunden = config.WDT_TIMEOUT if timeout is None else timeout
    wdt = WDT(timeout=int(sekunden * 1000))
    log("Watchdog: aktiv, %d s" % sekunden)
    return wdt.feed


def make_on_change(display, client, dimmer=None, log=None):
    """Rueckruf fuer mqtt.MqttBridge: geaenderte Einstellungen sofort wirksam
    machen. settings.json ist zu diesem Zeitpunkt schon geschrieben.

    **Helligkeit:** Der Dimmer muss die neuen Werte nicht nur lesen -- die
    Helligkeit muss auch neu an die Anzeige. Sonst bliebe eine gesetzte
    Helligkeit unsichtbar, solange die Stufe dieselbe bleibt: ein Telegramm geht
    sonst nur beim *Wechsel* raus. (Am Geraet aufgefallen, 2026-07-27.)

    **Adresse der Montierung:** Die wirkt erst, wenn die bestehende Verbindung
    faellt. Hier wird darum nur getrennt -- neu verbindet die Poll-Schleife im
    naechsten Durchlauf von selbst, mit derselben Fehlerbehandlung wie sonst
    auch: Antwortet unter der neuen Adresse niemand, steht "KEIN LINK", und ein
    weiteres set/mount_host holt das Geraet wieder zurueck.

    Eigene Funktion und nicht als Verschachtelung in main(), damit die Tests
    genau diese Logik pruefen koennen statt einer Abschrift davon.
    """
    log = log or (lambda *a: None)

    def uebernehmen(werte):
        try:
            if dimmer is not None:
                dimmer.reload()
                if dimmer.level is not None:
                    display.set_brightness(dimmer.percent(dimmer.level))
            else:
                # Ohne LDR gibt es keine Stufen -- dann gilt "hell".
                display.set_brightness(werte["hell_prozent"])
        except Exception as e:
            log("Anzeige: %s" % e)

        neuer_host = resolve_host(werte.get("mount_host"))
        if neuer_host != client.host:
            log("Montierung: Adresse %s -> %s" % (client.host, neuer_host))
            client.host = neuer_host
            client.close()

    return uebernehmen


def make_on_ota(feed=None, log=None):
    """Rueckruf fuer mqtt.MqttBridge: OTA-Update anstossen (cmd/ota).

    Die Payload ist die Basis-URL des Dateiservers, z.B.
    "http://192.168.178.50:8000". Bei leerer Nachricht gilt config.OTA_URL --
    so laesst sich per MQTT auch ohne Payload eine feste Vorgabe anstossen,
    genau wie set/mount_host mit leerer Nachricht auf config.MOUNT_HOST
    zurueckfaellt.

    Eigene Funktion statt Verschachtelung in main(), aus demselben Grund wie
    make_on_change: Tests pruefen genau diese Logik.
    """
    log = log or (lambda *a: None)

    def ausloesen(payload):
        import ota
        url = payload or getattr(config, "OTA_URL", "")
        if not url:
            log("OTA: keine URL angegeben und config.OTA_URL nicht gesetzt")
            return
        ota.update(url, feed=feed, log=log)

    return ausloesen


def main(host=None, port=None, with_ldr=False, fake_display=False,
         once=False, with_mqtt=True, feed=None, log=print):
    """WLAN aufbauen, Anzeige oeffnen und die Poll-Schleife fahren.

    host/port    Montierung. Ohne host gilt die per MQTT gesetzte Adresse aus
                 settings.json, sonst config.MOUNT_HOST (siehe
                 mount_client.resolve_host). Zum Testen gegen den Mock auf dem
                 Mac dessen LAN-Adresse angeben.
    with_ldr     Helligkeitsregelung ueber den ADC dazuschalten. Die Schwellen
                 sind am realen LDR gemessen (2026-07-27, siehe brightness.py),
                 aber am Schreibtisch, nicht in der Sternwarte -- deshalb bleibt
                 die Vorgabe aus, bis sie vor Ort bestaetigt sind.
    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.
    with_mqtt    MQTT-Anbindung versuchen. Ohne mqtt_config.py passiert nichts,
                 deshalb ist die Vorgabe an: fehlt die Datei, laeuft alles wie
                 bisher weiter.
    feed         Watchdog-Funktion aus make_watchdog(). Vorgabe None: Von Hand
                 gestartet laeuft die Schleife ohne Watchdog, sonst wuerde das
                 Geraet beim Nachdenken an der REPL neu starten. main.py legt
                 ihn an.
    """
    d = open_display(fake=fake_display)
    client = MountClient(resolve_host(host, log=log),
                         config.MOUNT_PORT if port is None else port,
                         timeout=config.MOUNT_TIMEOUT)
    log("Montierung: %s:%d" % (client.host, client.port))

    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()

    bridge = None
    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)

        if with_mqtt:
            import mqtt
            bridge = mqtt.connect_from_config(
                log=log,
                on_change=make_on_change(d, client, dimmer, log=log),
                on_ota=make_on_ota(feed=feed, log=log))

        displayloop.poll_loop(d, client,
                              once=once,
                              read_ldr=read_ldr,
                              dimmer=dimmer,
                              ensure_link=lambda: wifi.ensure(log=log),
                              bridge=bridge,
                              feed=feed,
                              log=log)
    except KeyboardInterrupt:
        log("\nbeendet")
    finally:
        if bridge is not None:
            bridge.close()
        client.close()
        d.close()