jmri_ncemonitor.transport

14class BytePort(ABC):
15    @abstractmethod
16    async def write(self, data: bytes) -> None: ...
17
18    @abstractmethod
19    async def read_byte(self) -> int: ...
20
21    @abstractmethod
22    async def aclose(self) -> None: ...
23
24    @property
25    @abstractmethod
26    def is_open(self) -> bool: ...

10class LoopbackPort(BytePort):
11    def __init__(self, read_q: asyncio.Queue[int], write_q: asyncio.Queue[int]) -> None:
12        self._r = read_q
13        self._w = write_q
14        self._open = True
15
16    async def write(self, data: bytes) -> None:
17        if not self._open:
18            raise RuntimeError("port is closed")
19        for b in data:
20            await self._w.put(b)
21
22    async def read_byte(self) -> int:
23        return await self._r.get()
24
25    async def aclose(self) -> None:
26        self._open = False
27
28    @property
29    def is_open(self) -> bool:
30        return self._open

 39class SerialPort(BytePort):
 40    def __init__(self, reader: StreamReader, writer: StreamWriter) -> None:
 41        self._reader = reader
 42        self._writer = writer
 43        self._open = True
 44
 45    @classmethod
 46    async def open(
 47        cls,
 48        port: str,
 49        baud: int = DEFAULT_BAUD,
 50        *,
 51        parity: str = "N",
 52        stopbits: float = 1.0,
 53        bytesize: int = 8,
 54    ) -> SerialPort:
 55        """Open a serial port for Cab Bus monitoring.
 56
 57        ``parity`` / ``stopbits`` / ``bytesize`` accept the standard
 58        pyserial values — the default of ``N``, ``1``, ``8`` matches
 59        the high-bit-marker convention that JMRI's reference USB
 60        adapter uses.
 61
 62        For true 9-bit adapters that discriminate address bytes via
 63        mark/space parity, operators typically set
 64        ``parity="M", stopbits=1, bytesize=8`` on the address-byte
 65        side of the bus, and rely on the underlying UART's parity
 66        detection to flag the boundary. The exact combination
 67        depends on the adapter's firmware — see the README.
 68        """
 69        import serial_asyncio  # noqa: PLC0415
 70        reader, writer = await serial_asyncio.open_serial_connection(
 71            url=port,
 72            baudrate=baud,
 73            parity=parity,
 74            stopbits=stopbits,
 75            bytesize=bytesize,
 76        )
 77        return cls(reader=reader, writer=writer)
 78
 79    async def write(self, data: bytes) -> None:
 80        # The monitor never writes; we expose this only because the ABC
 81        # requires it. Surfacing a clear error guards against misuse.
 82        raise RuntimeError(
 83            "NCE Cab Bus monitor is passive — writes are not supported"
 84        )
 85
 86    async def read_byte(self) -> int:
 87        data = await self._reader.readexactly(1)
 88        return data[0]
 89
 90    async def aclose(self) -> None:
 91        if not self._open:
 92            return
 93        self._open = False
 94        self._writer.close()
 95        with contextlib.suppress(Exception):
 96            await self._writer.wait_closed()
 97
 98    @property
 99    def is_open(self) -> bool:
100        return self._open

33def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]:
34    a_to_b: asyncio.Queue[int] = asyncio.Queue()
35    b_to_a: asyncio.Queue[int] = asyncio.Queue()
36    return LoopbackPort(b_to_a, a_to_b), LoopbackPort(a_to_b, b_to_a)