jmri_openlcb.traffic

 1"""``OlcbTrafficController`` — read GridConnect frames, emit decoded messages.
 2
 3OpenLCB is fully async + peer-to-peer; there is no master-slave shape.
 4Every node both produces and consumes events. The traffic controller
 5just reads the byte stream, decodes frames, and emits an
 6:class:`~jmri_openlcb.events.EventReport` (for PCER frames) or the raw
 7:class:`~jmri_openlcb.codec.OlcbFrame` (for everything else).
 8"""
 9from __future__ import annotations
10
11import asyncio
12import logging
13
14from jmri_core.events import Signal
15
16from jmri_openlcb.codec import (
17    OlcbFrame,
18    decode_frame,
19    decode_pcer_if_pcer,
20    encode_frame,
21    encode_pcer,
22    read_frame,
23)
24from jmri_openlcb.events import EventId, EventReport
25from jmri_openlcb.transport.port import BytePort
26
27_log = logging.getLogger(__name__)
28
29
30class OlcbTrafficController:
31    def __init__(self, port: BytePort, *, source_alias: int = 0x123) -> None:
32        self._port = port
33        self._source_alias = source_alias
34        self._send_lock = asyncio.Lock()
35        self._read_task: asyncio.Task[None] | None = None
36        self._closed = False
37        self.events: Signal[EventReport] = Signal()
38        self.frames: Signal[OlcbFrame] = Signal()
39
40    @property
41    def source_alias(self) -> int:
42        return self._source_alias
43
44    async def start(self) -> None:
45        if self._read_task is not None:
46            return
47        self._read_task = asyncio.create_task(self._read_loop(), name="olcb-read")
48
49    async def aclose(self) -> None:
50        self._closed = True
51        if self._read_task is not None:
52            self._read_task.cancel()
53            import contextlib  # noqa: PLC0415
54            with contextlib.suppress(asyncio.CancelledError, Exception):
55                await self._read_task
56            self._read_task = None
57        await self._port.aclose()
58
59    async def produce_event(self, event: EventId) -> None:
60        """Emit a PCER for *event* with our source alias."""
61        async with self._send_lock:
62            await self._port.write(encode_pcer(event, self._source_alias))
63
64    async def send_frame(self, frame: OlcbFrame) -> None:
65        """Send a raw frame (escape hatch for unsupported MTIs)."""
66        async with self._send_lock:
67            await self._port.write(encode_frame(frame))
68
69    async def _read_loop(self) -> None:
70        try:
71            while not self._closed:
72                try:
73                    text = await read_frame(self._port.read_byte)
74                except asyncio.CancelledError:
75                    return
76                except Exception as e:  # noqa: BLE001
77                    _log.warning("OpenLCB read error: %s", e)
78                    continue
79                try:
80                    frame = decode_frame(text)
81                except ValueError as e:
82                    _log.warning("OpenLCB bad frame %r: %s", text, e)
83                    continue
84                await self.frames.emit(frame)
85                pcer = decode_pcer_if_pcer(frame)
86                if pcer is not None:
87                    await self.events.emit(pcer)
88        except asyncio.CancelledError:
89            return