jmri_loconet.traffic

  1"""``LocoNetTrafficController`` — async read loop + send queue fan-out.
  2
  3Owns one :class:`LocoNetPort` and a background task that reads complete
  4frames, decodes them, and emits the typed message on the ``inbound`` Signal.
  5The :meth:`send` method serializes a typed message to bytes, attaches the
  6checksum, and writes to the port.
  7
  8LocoNet runs at ~16.6 kbaud, so back-to-back writes need a small interframe
  9gap; this is configurable via ``min_send_interval``.
 10"""
 11from __future__ import annotations
 12
 13import asyncio
 14import logging
 15import time
 16
 17from jmri_core.events import Signal
 18
 19from jmri_loconet.codec import frame_length, read_frame
 20from jmri_loconet.messages import LocoNetMessage, decode, encode
 21from jmri_loconet.transport.port import LocoNetPort
 22
 23_log = logging.getLogger(__name__)
 24
 25
 26class LocoNetTrafficController:
 27    """The protocol-level conduit between transport bytes and typed messages.
 28
 29    Listeners connect to :attr:`inbound`; senders call :meth:`send`. The
 30    read loop runs as a background asyncio task started by :meth:`start`
 31    and cancelled by :meth:`aclose`.
 32    """
 33
 34    def __init__(
 35        self,
 36        port: LocoNetPort,
 37        *,
 38        min_send_interval: float = 0.005,
 39    ) -> None:
 40        self._port = port
 41        self._min_send_interval = min_send_interval
 42        self._send_lock = asyncio.Lock()
 43        self._last_send: float = 0.0
 44        self._read_task: asyncio.Task[None] | None = None
 45        self.inbound: Signal[LocoNetMessage] = Signal()
 46        self._closed = False
 47        # Timestamp (``time.monotonic()``) of the most recent inbound
 48        # frame. Read by :class:`LocoNetConnection`'s reachability
 49        # monitor to flip CS-power state. LocoNet is chatty — the CS
 50        # emits ``OPC_IDLE`` and other frames periodically — so a
 51        # multi-second silence is a strong signal the bus is dark.
 52        self.last_inbound_at: float | None = None
 53
 54    async def start(self) -> None:
 55        if self._read_task is not None:
 56            return
 57        self._read_task = asyncio.create_task(self._read_loop(), name="loconet-read")
 58
 59    async def aclose(self) -> None:
 60        self._closed = True
 61        if self._read_task is not None:
 62            self._read_task.cancel()
 63            import contextlib  # noqa: PLC0415
 64            with contextlib.suppress(asyncio.CancelledError, Exception):
 65                await self._read_task
 66            self._read_task = None
 67        await self._port.aclose()
 68
 69    async def send(self, msg: LocoNetMessage) -> None:
 70        """Encode and write *msg* to the port, with interframe pacing."""
 71        frame = encode(msg)
 72        async with self._send_lock:
 73            now = asyncio.get_event_loop().time()
 74            wait = self._min_send_interval - (now - self._last_send)
 75            if wait > 0:
 76                await asyncio.sleep(wait)
 77            await self._port.write_frame(frame)
 78            self._last_send = asyncio.get_event_loop().time()
 79
 80    # ----- internals --------------------------------------------------------
 81
 82    async def _read_loop(self) -> None:
 83        try:
 84            while not self._closed:
 85                try:
 86                    frame = await read_frame(
 87                        self._port.read_byte, self._port.read_exact
 88                    )
 89                except asyncio.CancelledError:
 90                    return
 91                except Exception as e:  # noqa: BLE001
 92                    _log.warning("LocoNet read error: %s", e)
 93                    continue
 94
 95                expected = frame_length(frame)
 96                if expected is None or expected != len(frame):
 97                    _log.warning(
 98                        "LocoNet frame length mismatch: got %d bytes for %s",
 99                        len(frame), frame.hex()
100                    )
101                    continue
102
103                try:
104                    msg = decode(frame)
105                except ValueError as e:
106                    _log.warning("LocoNet bad frame %s: %s", frame.hex(), e)
107                    continue
108
109                # Any well-formed inbound frame counts as evidence the
110                # CS is alive — even a bare OPC_IDLE. The reachability
111                # monitor watches this timestamp.
112                self.last_inbound_at = time.monotonic()
113                await self.inbound.emit(msg)
114        except asyncio.CancelledError:
115            return