jmri_dccex.transport.serial_port

pyserial-asyncio DCC-EX transport. Hardware-only.

 1"""pyserial-asyncio DCC-EX transport. Hardware-only."""
 2from __future__ import annotations
 3
 4import contextlib
 5from typing import TYPE_CHECKING
 6
 7from jmri_dccex.transport.port import BytePort
 8
 9if TYPE_CHECKING:
10    from asyncio import StreamReader, StreamWriter
11
12
13DEFAULT_BAUD = 115200   # DCC-EX command stations default to 115200
14
15
16class SerialPort(BytePort):
17    def __init__(self, reader: StreamReader, writer: StreamWriter) -> None:
18        self._reader = reader
19        self._writer = writer
20        self._open = True
21
22    @classmethod
23    async def open(cls, port: str, baud: int = DEFAULT_BAUD) -> SerialPort:
24        import serial_asyncio  # noqa: PLC0415
25        reader, writer = await serial_asyncio.open_serial_connection(url=port, baudrate=baud)
26        return cls(reader=reader, writer=writer)
27
28    async def write(self, data: bytes) -> None:
29        if not self._open:
30            raise RuntimeError("port is closed")
31        self._writer.write(data)
32        await self._writer.drain()
33
34    async def read_byte(self) -> int:
35        data = await self._reader.readexactly(1)
36        return data[0]
37
38    async def aclose(self) -> None:
39        if not self._open:
40            return
41        self._open = False
42        self._writer.close()
43        with contextlib.suppress(Exception):
44            await self._writer.wait_closed()
45
46    @property
47    def is_open(self) -> bool:
48        return self._open
DEFAULT_BAUD = 115200
class SerialPort(jmri_dccex.transport.port.BytePort): 
17class SerialPort(BytePort):
18    def __init__(self, reader: StreamReader, writer: StreamWriter) -> None:
19        self._reader = reader
20        self._writer = writer
21        self._open = True
22
23    @classmethod
24    async def open(cls, port: str, baud: int = DEFAULT_BAUD) -> SerialPort:
25        import serial_asyncio  # noqa: PLC0415
26        reader, writer = await serial_asyncio.open_serial_connection(url=port, baudrate=baud)
27        return cls(reader=reader, writer=writer)
28
29    async def write(self, data: bytes) -> None:
30        if not self._open:
31            raise RuntimeError("port is closed")
32        self._writer.write(data)
33        await self._writer.drain()
34
35    async def read_byte(self) -> int:
36        data = await self._reader.readexactly(1)
37        return data[0]
38
39    async def aclose(self) -> None:
40        if not self._open:
41            return
42        self._open = False
43        self._writer.close()
44        with contextlib.suppress(Exception):
45            await self._writer.wait_closed()
46
47    @property
48    def is_open(self) -> bool:
49        return self._open

Helper class that provides a standard way to create an ABC using inheritance.

SerialPort( reader: asyncio.streams.StreamReader, writer: asyncio.streams.StreamWriter) 
18    def __init__(self, reader: StreamReader, writer: StreamWriter) -> None:
19        self._reader = reader
20        self._writer = writer
21        self._open = True
@classmethod
async def open( cls, port: str, baud: int = 115200) -> SerialPort: 
23    @classmethod
24    async def open(cls, port: str, baud: int = DEFAULT_BAUD) -> SerialPort:
25        import serial_asyncio  # noqa: PLC0415
26        reader, writer = await serial_asyncio.open_serial_connection(url=port, baudrate=baud)
27        return cls(reader=reader, writer=writer)
async def write(self, data: bytes) -> None: 
29    async def write(self, data: bytes) -> None:
30        if not self._open:
31            raise RuntimeError("port is closed")
32        self._writer.write(data)
33        await self._writer.drain()
async def read_byte(self) -> int: 
35    async def read_byte(self) -> int:
36        data = await self._reader.readexactly(1)
37        return data[0]
async def aclose(self) -> None: 
39    async def aclose(self) -> None:
40        if not self._open:
41            return
42        self._open = False
43        self._writer.close()
44        with contextlib.suppress(Exception):
45            await self._writer.wait_closed()
is_open: bool 
47    @property
48    def is_open(self) -> bool:
49        return self._open