127 lines
3.5 KiB
Python
127 lines
3.5 KiB
Python
"""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
|