jmri_nce.transport.serial_port

SerialPort — pyserial-asyncio NCE transport.

PowerCab USB and PowerPro USB present as standard USB-serial devices. Default baud is 9600 (PowerPro) but some configurations use 19200 or 57600; user is expected to set the right value in the connection settings.

 1"""``SerialPort`` — pyserial-asyncio NCE transport.
 2
 3PowerCab USB and PowerPro USB present as standard USB-serial devices.
 4Default baud is 9600 (PowerPro) but some configurations use 19200 or
 557600; user is expected to set the right value in the connection settings.
 6"""
 7from __future__ import annotations
 8
 9import contextlib
10from typing import TYPE_CHECKING
11
12from jmri_nce.transport.port import BytePort
13
14if TYPE_CHECKING:
15    from asyncio import StreamReader, StreamWriter
16
17
18DEFAULT_BAUD = 9600  # PowerPro default; PowerCab uses the same.
19
20
21class SerialPort(BytePort):
22    """NCE over an async serial port."""
23
24    def __init__(self, reader: StreamReader, writer: StreamWriter) -> None:
25        self._reader = reader
26        self._writer = writer
27        self._open = True
28
29    @classmethod
30    async def open(cls, port: str, baud: int = DEFAULT_BAUD) -> SerialPort:
31        """Open *port* (e.g. ``"COM7"`` or ``"/dev/ttyUSB0"``)."""
32        import serial_asyncio  # noqa: PLC0415
33        reader, writer = await serial_asyncio.open_serial_connection(
34            url=port, baudrate=baud
35        )
36        return cls(reader=reader, writer=writer)
37
38    async def write(self, data: bytes) -> None:
39        if not self._open:
40            raise RuntimeError("port is closed")
41        self._writer.write(data)
42        await self._writer.drain()
43
44    async def read_exact(self, n: int) -> bytes:
45        return await self._reader.readexactly(n)
46
47    async def read_pending(self) -> bytes:
48        """Return whatever bytes are already buffered without waiting.
49
50        ``StreamReader._buffer`` is private API but stable across asyncio
51        releases; ``read()`` blocks until at least one byte arrives, which
52        is the opposite of what the drain needs.
53        """
54        buf = self._reader._buffer  # type: ignore[attr-defined]
55        if not buf:
56            return b""
57        n = len(buf)
58        return await self._reader.readexactly(n)
59
60    async def aclose(self) -> None:
61        if not self._open:
62            return
63        self._open = False
64        self._writer.close()
65        with contextlib.suppress(Exception):
66            await self._writer.wait_closed()
67
68    @property
69    def is_open(self) -> bool:
70        return self._open
DEFAULT_BAUD = 9600
class SerialPort(jmri_nce.transport.port.BytePort): 
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

NCE over an async serial port.

SerialPort( reader: asyncio.streams.StreamReader, writer: asyncio.streams.StreamWriter) 
25    def __init__(self, reader: StreamReader, writer: StreamWriter) -> None:
26        self._reader = reader
27        self._writer = writer
28        self._open = True
@classmethod
async def open( cls, port: str, baud: int = 9600) -> SerialPort: 
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)

Open port (e.g. "COM7" or "/dev/ttyUSB0").

async def write(self, data: bytes) -> None: 
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()
async def read_exact(self, n: int) -> bytes: 
45    async def read_exact(self, n: int) -> bytes:
46        return await self._reader.readexactly(n)
async def read_pending(self) -> bytes: 
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)

Return whatever bytes are already buffered without waiting.

StreamReader._buffer is private API but stable across asyncio releases; read() blocks until at least one byte arrives, which is the opposite of what the drain needs.

async def aclose(self) -> None: 
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()
is_open: bool 
69    @property
70    def is_open(self) -> bool:
71        return self._open