jmri_dccex.traffic
DccexTrafficController — async read loop + Signal fan-out.
DCC-EX is "fire-and-respond": clients send commands, the command station
responds asynchronously with state-change messages. Multiple commands may
be in flight; we don't track request/response pairing here — the driver
just emits everything that comes back on the inbound Signal and
lets the per-bean handlers correlate.
1"""``DccexTrafficController`` — async read loop + Signal fan-out. 2 3DCC-EX is "fire-and-respond": clients send commands, the command station 4responds asynchronously with state-change messages. Multiple commands may 5be in flight; we don't track request/response pairing here — the driver 6just emits everything that comes back on the :attr:`inbound` Signal and 7lets the per-bean handlers correlate. 8""" 9from __future__ import annotations 10 11import asyncio 12import logging 13import time 14 15from jmri_core.events import Signal 16 17from jmri_dccex.codec import DccexMessage, decode, encode, read_message 18from jmri_dccex.transport.port import BytePort 19 20_log = logging.getLogger(__name__) 21 22 23class DccexTrafficController: 24 def __init__(self, port: BytePort) -> None: 25 self._port = port 26 self._send_lock = asyncio.Lock() 27 self._read_task: asyncio.Task[None] | None = None 28 self._closed = False 29 self.inbound: Signal[DccexMessage] = Signal() 30 # Timestamp of the most recent inbound message. Read by 31 # :class:`DccexConnection`'s reachability monitor. DCC-EX 32 # doesn't chatter without solicitation, so the connection also 33 # runs a periodic ``<s>`` (status) ping to guarantee traffic. 34 self.last_inbound_at: float | None = None 35 36 async def start(self) -> None: 37 if self._read_task is not None: 38 return 39 self._read_task = asyncio.create_task(self._read_loop(), name="dccex-read") 40 41 async def aclose(self) -> None: 42 self._closed = True 43 if self._read_task is not None: 44 self._read_task.cancel() 45 import contextlib # noqa: PLC0415 46 with contextlib.suppress(asyncio.CancelledError, Exception): 47 await self._read_task 48 self._read_task = None 49 await self._port.aclose() 50 51 async def send(self, msg: DccexMessage) -> None: 52 async with self._send_lock: 53 await self._port.write(encode(msg)) 54 55 async def _read_loop(self) -> None: 56 try: 57 while not self._closed: 58 try: 59 text = await read_message(self._port.read_byte) 60 except asyncio.CancelledError: 61 return 62 except Exception as e: # noqa: BLE001 63 _log.warning("DCC-EX read error: %s", e) 64 continue 65 try: 66 msg = decode(text) 67 except ValueError as e: 68 _log.warning("DCC-EX bad message %r: %s", text, e) 69 continue 70 self.last_inbound_at = time.monotonic() 71 await self.inbound.emit(msg) 72 except asyncio.CancelledError: 73 return
class
DccexTrafficController:
24class DccexTrafficController: 25 def __init__(self, port: BytePort) -> None: 26 self._port = port 27 self._send_lock = asyncio.Lock() 28 self._read_task: asyncio.Task[None] | None = None 29 self._closed = False 30 self.inbound: Signal[DccexMessage] = Signal() 31 # Timestamp of the most recent inbound message. Read by 32 # :class:`DccexConnection`'s reachability monitor. DCC-EX 33 # doesn't chatter without solicitation, so the connection also 34 # runs a periodic ``<s>`` (status) ping to guarantee traffic. 35 self.last_inbound_at: float | None = None 36 37 async def start(self) -> None: 38 if self._read_task is not None: 39 return 40 self._read_task = asyncio.create_task(self._read_loop(), name="dccex-read") 41 42 async def aclose(self) -> None: 43 self._closed = True 44 if self._read_task is not None: 45 self._read_task.cancel() 46 import contextlib # noqa: PLC0415 47 with contextlib.suppress(asyncio.CancelledError, Exception): 48 await self._read_task 49 self._read_task = None 50 await self._port.aclose() 51 52 async def send(self, msg: DccexMessage) -> None: 53 async with self._send_lock: 54 await self._port.write(encode(msg)) 55 56 async def _read_loop(self) -> None: 57 try: 58 while not self._closed: 59 try: 60 text = await read_message(self._port.read_byte) 61 except asyncio.CancelledError: 62 return 63 except Exception as e: # noqa: BLE001 64 _log.warning("DCC-EX read error: %s", e) 65 continue 66 try: 67 msg = decode(text) 68 except ValueError as e: 69 _log.warning("DCC-EX bad message %r: %s", text, e) 70 continue 71 self.last_inbound_at = time.monotonic() 72 await self.inbound.emit(msg) 73 except asyncio.CancelledError: 74 return
DccexTrafficController(port: jmri_dccex.BytePort)
25 def __init__(self, port: BytePort) -> None: 26 self._port = port 27 self._send_lock = asyncio.Lock() 28 self._read_task: asyncio.Task[None] | None = None 29 self._closed = False 30 self.inbound: Signal[DccexMessage] = Signal() 31 # Timestamp of the most recent inbound message. Read by 32 # :class:`DccexConnection`'s reachability monitor. DCC-EX 33 # doesn't chatter without solicitation, so the connection also 34 # runs a periodic ``<s>`` (status) ping to guarantee traffic. 35 self.last_inbound_at: float | None = None
inbound: jmri_core.Signal[jmri_dccex.DccexMessage]
async def
aclose(self) -> None:
42 async def aclose(self) -> None: 43 self._closed = True 44 if self._read_task is not None: 45 self._read_task.cancel() 46 import contextlib # noqa: PLC0415 47 with contextlib.suppress(asyncio.CancelledError, Exception): 48 await self._read_task 49 self._read_task = None 50 await self._port.aclose()