grossanzeige/transport.py

"""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, Pin
except ImportError:
    UART = None
    Pin = None

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:
    """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):
        # Empfangspuffer vor dem Senden leeren: eventuelle Stoerbytes zwischen
        # den Telegrammen verwerfen, damit read_response nur die frische Antwort
        # auf DIESES Telegramm sieht.
        self._port.reset_input_buffer()
        self._port.write(data)
        self._port.flush()

    def read_response(self):
        """Bytes bis einschliesslich ETX lesen, oder bis der Timeout greift."""
        deadline = _deadline(self._timeout)
        buf = bytearray()
        while not _expired(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).

    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):
        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),
        )
        # 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"] = Pin(tx) if Pin is not None else tx
        if rx is not None:
            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 = _deadline(self._timeout)
        buf = bytearray()
        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 migra.ETX in chunk:
                break
        return _truncate_at_etx(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