jmri_loconet.transport.serial_port

SerialPort — pyserial-asyncio LocoNet transport.

Wraps a serial_asyncio StreamReader/StreamWriter pair into the LocoNetPort interface. Used for real LocoBuffer-II / PR3 / PR4 connections.

Default settings match JMRI's defaults: 16,457 baud, 8N1, no flow control.

 1"""``SerialPort`` — pyserial-asyncio LocoNet transport.
 2
 3Wraps a ``serial_asyncio`` ``StreamReader``/``StreamWriter`` pair into the
 4:class:`LocoNetPort` interface. Used for real LocoBuffer-II / PR3 / PR4
 5connections.
 6
 7Default settings match JMRI's defaults: 16,457 baud, 8N1, no flow control.
 8"""
 9from __future__ import annotations
10
11from typing import TYPE_CHECKING
12
13from jmri_loconet.transport.port import LocoNetPort
14
15if TYPE_CHECKING:
16    from asyncio import StreamReader, StreamWriter
17
18# JMRI's LocoBufferII default; lower-end LocoBuffer-I uses 19200, PR3 uses 57600.
19DEFAULT_BAUD = 16457
20
21
22class SerialPort(LocoNetPort):
23    """LocoNet 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 the serial *port* (e.g. ``"COM5"`` or ``"/dev/loconet"``)."""
33        # Import lazily so the package imports cleanly on systems without
34        # pyserial installed (tests use the loopback transport).
35        import serial_asyncio  # noqa: PLC0415
36        reader, writer = await serial_asyncio.open_serial_connection(
37            url=port, baudrate=baud
38        )
39        return cls(reader=reader, writer=writer)
40
41    async def read_byte(self) -> int:
42        data = await self._reader.readexactly(1)
43        return data[0]
44
45    async def read_exact(self, n: int) -> bytes:
46        return await self._reader.readexactly(n)
47
48    async def write_frame(self, frame: bytes) -> None:
49        if not self._open:
50            raise RuntimeError("port is closed")
51        self._writer.write(frame)
52        await self._writer.drain()
53
54    async def aclose(self) -> None:
55        if not self._open:
56            return
57        self._open = False
58        self._writer.close()
59        import contextlib  # noqa: PLC0415
60        with contextlib.suppress(Exception):  # best-effort close
61            await self._writer.wait_closed()
62
63    @property
64    def is_open(self) -> bool:
65        return self._open
DEFAULT_BAUD = 16457
class SerialPort(jmri_loconet.transport.port.LocoNetPort): 
23class SerialPort(LocoNetPort):
24    """LocoNet over an async serial port."""
25
26    def __init__(self, reader: StreamReader, writer: StreamWriter) -> None:
27        self._reader = reader
28        self._writer = writer
29        self._open = True
30
31    @classmethod
32    async def open(cls, port: str, baud: int = DEFAULT_BAUD) -> SerialPort:
33        """Open the serial *port* (e.g. ``"COM5"`` or ``"/dev/loconet"``)."""
34        # Import lazily so the package imports cleanly on systems without
35        # pyserial installed (tests use the loopback transport).
36        import serial_asyncio  # noqa: PLC0415
37        reader, writer = await serial_asyncio.open_serial_connection(
38            url=port, baudrate=baud
39        )
40        return cls(reader=reader, writer=writer)
41
42    async def read_byte(self) -> int:
43        data = await self._reader.readexactly(1)
44        return data[0]
45
46    async def read_exact(self, n: int) -> bytes:
47        return await self._reader.readexactly(n)
48
49    async def write_frame(self, frame: bytes) -> None:
50        if not self._open:
51            raise RuntimeError("port is closed")
52        self._writer.write(frame)
53        await self._writer.drain()
54
55    async def aclose(self) -> None:
56        if not self._open:
57            return
58        self._open = False
59        self._writer.close()
60        import contextlib  # noqa: PLC0415
61        with contextlib.suppress(Exception):  # best-effort close
62            await self._writer.wait_closed()
63
64    @property
65    def is_open(self) -> bool:
66        return self._open

LocoNet over an async serial port.

SerialPort( reader: asyncio.streams.StreamReader, writer: asyncio.streams.StreamWriter) 
26    def __init__(self, reader: StreamReader, writer: StreamWriter) -> None:
27        self._reader = reader
28        self._writer = writer
29        self._open = True
@classmethod
async def open( cls, port: str, baud: int = 16457) -> SerialPort: 
31    @classmethod
32    async def open(cls, port: str, baud: int = DEFAULT_BAUD) -> SerialPort:
33        """Open the serial *port* (e.g. ``"COM5"`` or ``"/dev/loconet"``)."""
34        # Import lazily so the package imports cleanly on systems without
35        # pyserial installed (tests use the loopback transport).
36        import serial_asyncio  # noqa: PLC0415
37        reader, writer = await serial_asyncio.open_serial_connection(
38            url=port, baudrate=baud
39        )
40        return cls(reader=reader, writer=writer)

Open the serial port (e.g. "COM5" or "/dev/loconet").

async def read_byte(self) -> int: 
42    async def read_byte(self) -> int:
43        data = await self._reader.readexactly(1)
44        return data[0]

Read one byte. Blocks until a byte is available.

async def read_exact(self, n: int) -> bytes: 
46    async def read_exact(self, n: int) -> bytes:
47        return await self._reader.readexactly(n)

Read exactly n bytes.

async def write_frame(self, frame: bytes) -> None: 
49    async def write_frame(self, frame: bytes) -> None:
50        if not self._open:
51            raise RuntimeError("port is closed")
52        self._writer.write(frame)
53        await self._writer.drain()

Write a complete LocoNet frame (opcode + payload + checksum).

async def aclose(self) -> None: 
55    async def aclose(self) -> None:
56        if not self._open:
57            return
58        self._open = False
59        self._writer.close()
60        import contextlib  # noqa: PLC0415
61        with contextlib.suppress(Exception):  # best-effort close
62            await self._writer.wait_closed()

Close the port. Idempotent.

is_open: bool 
64    @property
65    def is_open(self) -> bool:
66        return self._open