jmri_nce.transport

12class BytePort(ABC):
13    """Bidirectional byte-stream port for NCE."""
14
15    @abstractmethod
16    async def write(self, data: bytes) -> None: ...
17
18    @abstractmethod
19    async def read_exact(self, n: int) -> bytes: ...
20
21    @abstractmethod
22    async def read_pending(self) -> bytes:
23        """Read any bytes already buffered, without waiting for more.
24
25        Used by the traffic controller to drain stale bytes between
26        commands. Returns ``b""`` if nothing is currently buffered.
27        """
28
29    @abstractmethod
30    async def aclose(self) -> None: ...
31
32    @property
33    @abstractmethod
34    def is_open(self) -> bool: ...

16class LoopbackPort(BytePort):
17    """One half of a connected pair."""
18
19    def __init__(self, read_q: asyncio.Queue[int], write_q: asyncio.Queue[int]) -> None:
20        self._r = read_q
21        self._w = write_q
22        self._open = True
23
24    async def write(self, data: bytes) -> None:
25        if not self._open:
26            raise RuntimeError("port is closed")
27        for b in data:
28            await self._w.put(b)
29
30    async def read_exact(self, n: int) -> bytes:
31        out = bytearray()
32        for _ in range(n):
33            out.append(await self._r.get())
34        return bytes(out)
35
36    async def read_pending(self) -> bytes:
37        """Drain anything currently sitting in the receive queue."""
38        out = bytearray()
39        while not self._r.empty():
40            out.append(self._r.get_nowait())
41        return bytes(out)
42
43    async def aclose(self) -> None:
44        self._open = False
45
46    @property
47    def is_open(self) -> bool:
48        return self._open

22class SerialPort(BytePort):
23    """NCE over an async serial port."""
24
25    def __init__(self, reader: StreamReader, writer: StreamWriter) -> None:
26        self._reader = reader
27        self._writer = writer
28        self._open = True
29
30    @classmethod
31    async def open(cls, port: str, baud: int = DEFAULT_BAUD) -> SerialPort:
32        """Open *port* (e.g. ``"COM7"`` or ``"/dev/ttyUSB0"``)."""
33        import serial_asyncio  # noqa: PLC0415
34        reader, writer = await serial_asyncio.open_serial_connection(
35            url=port, baudrate=baud
36        )
37        return cls(reader=reader, writer=writer)
38
39    async def write(self, data: bytes) -> None:
40        if not self._open:
41            raise RuntimeError("port is closed")
42        self._writer.write(data)
43        await self._writer.drain()
44
45    async def read_exact(self, n: int) -> bytes:
46        return await self._reader.readexactly(n)
47
48    async def read_pending(self) -> bytes:
49        """Return whatever bytes are already buffered without waiting.
50
51        ``StreamReader._buffer`` is private API but stable across asyncio
52        releases; ``read()`` blocks until at least one byte arrives, which
53        is the opposite of what the drain needs.
54        """
55        buf = self._reader._buffer  # type: ignore[attr-defined]
56        if not buf:
57            return b""
58        n = len(buf)
59        return await self._reader.readexactly(n)
60
61    async def aclose(self) -> None:
62        if not self._open:
63            return
64        self._open = False
65        self._writer.close()
66        with contextlib.suppress(Exception):
67            await self._writer.wait_closed()
68
69    @property
70    def is_open(self) -> bool:
71        return self._open

51def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]:
52    """Two connected loopback ports — A writes to B's read stream and vice versa."""
53    a_to_b: asyncio.Queue[int] = asyncio.Queue()
54    b_to_a: asyncio.Queue[int] = asyncio.Queue()
55    return LoopbackPort(b_to_a, a_to_b), LoopbackPort(a_to_b, b_to_a)