jmri_ncemonitor.monitor
CabBusMonitor — passive read loop + Signal fan-out.
Reads Cab Bus messages from the transport, decodes them, and emits each
on the inbound Signal. There is no send path. There is no Lock.
The whole module is shorter than either of the bidirectional drivers
because passivity removes most of the complexity.
1"""``CabBusMonitor`` — passive read loop + Signal fan-out. 2 3Reads Cab Bus messages from the transport, decodes them, and emits each 4on the :attr:`inbound` Signal. There is no send path. There is no Lock. 5The whole module is shorter than either of the bidirectional drivers 6because passivity removes most of the complexity. 7""" 8from __future__ import annotations 9 10import asyncio 11import logging 12 13from jmri_core.events import Signal 14 15from jmri_ncemonitor.codec import ( 16 BodyOverflowError, 17 PeekableReader, 18 read_message_peekable, 19) 20from jmri_ncemonitor.messages import CabBusMessage, decode 21from jmri_ncemonitor.transport.port import BytePort 22 23_log = logging.getLogger(__name__) 24 25 26class CabBusMonitor: 27 """Owns one :class:`BytePort` and a background read task. 28 29 Listeners connect to :attr:`inbound` and receive every decoded message 30 in the order it appeared on the bus. 31 """ 32 33 def __init__(self, port: BytePort) -> None: 34 self._port = port 35 self._reader = PeekableReader(port.read_byte) 36 self._read_task: asyncio.Task[None] | None = None 37 self._closed = False 38 self.inbound: Signal[CabBusMessage] = Signal() 39 40 async def start(self) -> None: 41 if self._read_task is not None: 42 return 43 self._read_task = asyncio.create_task(self._read_loop(), name="cabbus-monitor") 44 45 async def aclose(self) -> None: 46 self._closed = True 47 if self._read_task is not None: 48 self._read_task.cancel() 49 import contextlib # noqa: PLC0415 50 with contextlib.suppress(asyncio.CancelledError, Exception): 51 await self._read_task 52 self._read_task = None 53 await self._port.aclose() 54 55 async def _read_loop(self) -> None: 56 try: 57 while not self._closed: 58 try: 59 frame = await read_message_peekable(self._reader) 60 except asyncio.CancelledError: 61 return 62 except BodyOverflowError as e: 63 # A stuck bus buffered past the cap. Log the 64 # partial for diagnostics and drop it — the next 65 # read_message call resyncs on the wire. 66 _log.warning( 67 "Cab Bus body overflow (%d bytes buffered): %s", 68 len(e.partial), e.partial.hex(), 69 ) 70 continue 71 except Exception as e: # noqa: BLE001 72 _log.warning("Cab Bus read error: %s", e) 73 continue 74 try: 75 msg = decode(frame) 76 except ValueError as e: 77 _log.warning("Cab Bus bad frame %s: %s", frame.hex(), e) 78 continue 79 await self.inbound.emit(msg) 80 except asyncio.CancelledError: 81 return
class
CabBusMonitor:
27class CabBusMonitor: 28 """Owns one :class:`BytePort` and a background read task. 29 30 Listeners connect to :attr:`inbound` and receive every decoded message 31 in the order it appeared on the bus. 32 """ 33 34 def __init__(self, port: BytePort) -> None: 35 self._port = port 36 self._reader = PeekableReader(port.read_byte) 37 self._read_task: asyncio.Task[None] | None = None 38 self._closed = False 39 self.inbound: Signal[CabBusMessage] = Signal() 40 41 async def start(self) -> None: 42 if self._read_task is not None: 43 return 44 self._read_task = asyncio.create_task(self._read_loop(), name="cabbus-monitor") 45 46 async def aclose(self) -> None: 47 self._closed = True 48 if self._read_task is not None: 49 self._read_task.cancel() 50 import contextlib # noqa: PLC0415 51 with contextlib.suppress(asyncio.CancelledError, Exception): 52 await self._read_task 53 self._read_task = None 54 await self._port.aclose() 55 56 async def _read_loop(self) -> None: 57 try: 58 while not self._closed: 59 try: 60 frame = await read_message_peekable(self._reader) 61 except asyncio.CancelledError: 62 return 63 except BodyOverflowError as e: 64 # A stuck bus buffered past the cap. Log the 65 # partial for diagnostics and drop it — the next 66 # read_message call resyncs on the wire. 67 _log.warning( 68 "Cab Bus body overflow (%d bytes buffered): %s", 69 len(e.partial), e.partial.hex(), 70 ) 71 continue 72 except Exception as e: # noqa: BLE001 73 _log.warning("Cab Bus read error: %s", e) 74 continue 75 try: 76 msg = decode(frame) 77 except ValueError as e: 78 _log.warning("Cab Bus bad frame %s: %s", frame.hex(), e) 79 continue 80 await self.inbound.emit(msg) 81 except asyncio.CancelledError: 82 return
Owns one BytePort and a background read task.
Listeners connect to inbound and receive every decoded message
in the order it appeared on the bus.
CabBusMonitor(port: jmri_ncemonitor.BytePort)
async def
aclose(self) -> None:
46 async def aclose(self) -> None: 47 self._closed = True 48 if self._read_task is not None: 49 self._read_task.cancel() 50 import contextlib # noqa: PLC0415 51 with contextlib.suppress(asyncio.CancelledError, Exception): 52 await self._read_task 53 self._read_task = None 54 await self._port.aclose()