Martin Bauer
d44c24ec97
- event bus systen - all components are independent - preparation for web frontend
123 lines
3.9 KiB
Python
123 lines
3.9 KiB
Python
"""Serial transport with reconnect.
|
|
|
|
The old ``host_driver.py`` stopped the event loop when the USB cable was pulled and
|
|
relied on systemd to restart the whole process. Here a dropped link is just a
|
|
reconnect loop, and the device re-applies its LED state when it comes back.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from collections.abc import Callable
|
|
|
|
import serial_asyncio
|
|
|
|
from musicmouse.clock import Clock, RealClock
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
__all__ = ["SerialLink"]
|
|
|
|
_READ_CHUNK = 1024
|
|
|
|
|
|
class SerialLink:
|
|
"""Owns the serial port and keeps trying to hold it open."""
|
|
|
|
def __init__(
|
|
self,
|
|
port: str,
|
|
baudrate: int = 115200,
|
|
*,
|
|
reconnect_interval: float = 5.0,
|
|
clock: Clock | None = None,
|
|
) -> None:
|
|
self.port = port
|
|
self.baudrate = baudrate
|
|
self._reconnect_interval = reconnect_interval
|
|
self._clock = clock or RealClock()
|
|
self._writer: asyncio.StreamWriter | None = None
|
|
|
|
self._on_data: Callable[[bytes], None] = lambda _data: None
|
|
self._on_connect: Callable[[], None] | None = None
|
|
self._on_disconnect: Callable[[str], None] | None = None
|
|
|
|
def attach(
|
|
self,
|
|
on_data: Callable[[bytes], None],
|
|
*,
|
|
on_connect: Callable[[], None] | None = None,
|
|
on_disconnect: Callable[[str], None] | None = None,
|
|
) -> None:
|
|
"""Wire up the device. Separate from ``__init__`` because the device needs the
|
|
link as its transport, so one of the two has to be built first."""
|
|
self._on_data = on_data
|
|
self._on_connect = on_connect
|
|
self._on_disconnect = on_disconnect
|
|
|
|
@property
|
|
def connected(self) -> bool:
|
|
return self._writer is not None
|
|
|
|
def write(self, data: bytes) -> None:
|
|
writer = self._writer
|
|
if writer is None:
|
|
_log.debug("Dropping %d bytes: %s is not connected", len(data), self.port)
|
|
return
|
|
try:
|
|
writer.write(data)
|
|
except OSError as exc: # pragma: no cover - needs a real port dying mid-write
|
|
_log.warning("Write to %s failed: %s", self.port, exc)
|
|
|
|
async def run(self) -> None:
|
|
"""Connect, read until the link drops, wait, repeat. Runs until cancelled."""
|
|
while True:
|
|
reader = await self._connect()
|
|
if reader is None:
|
|
await self._clock.sleep(self._reconnect_interval)
|
|
continue
|
|
|
|
reason = await self._pump(reader)
|
|
|
|
self._writer = None
|
|
_log.warning("Lost connection to %s: %s", self.port, reason)
|
|
if self._on_disconnect is not None:
|
|
self._on_disconnect(reason)
|
|
await self._clock.sleep(self._reconnect_interval)
|
|
|
|
async def _connect(self) -> asyncio.StreamReader | None:
|
|
reader: asyncio.StreamReader
|
|
writer: asyncio.StreamWriter
|
|
try:
|
|
reader, writer = await serial_asyncio.open_serial_connection(
|
|
url=self.port, baudrate=self.baudrate
|
|
)
|
|
except (OSError, ValueError) as exc:
|
|
_log.warning(
|
|
"Cannot open %s (%s); retrying in %gs",
|
|
self.port,
|
|
exc,
|
|
self._reconnect_interval,
|
|
)
|
|
return None
|
|
|
|
self._writer = writer
|
|
_log.info("Connected to firmware on %s at %d baud", self.port, self.baudrate)
|
|
if self._on_connect is not None:
|
|
self._on_connect()
|
|
return reader
|
|
|
|
async def _pump(self, reader: asyncio.StreamReader) -> str:
|
|
try:
|
|
while True:
|
|
data = await reader.read(_READ_CHUNK)
|
|
if not data:
|
|
return "port closed"
|
|
self._on_data(data)
|
|
except asyncio.CancelledError:
|
|
self._writer = None
|
|
raise
|
|
except (OSError, asyncio.IncompleteReadError) as exc:
|
|
return str(exc)
|