jmri_openlcb.transport.serial_port
pyserial-asyncio OpenLCB transport — for USB-to-CAN adapters that present as serial GridConnect.
1"""pyserial-asyncio OpenLCB transport — for USB-to-CAN adapters that present 2as serial GridConnect.""" 3from __future__ import annotations 4 5import contextlib 6from typing import TYPE_CHECKING 7 8from jmri_openlcb.transport.port import BytePort 9 10if TYPE_CHECKING: 11 from asyncio import StreamReader, StreamWriter 12 13 14DEFAULT_BAUD = 333333 # CAN bus typical; some bridges use 115200 ASCII speed 15 16 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
DEFAULT_BAUD =
333333
18class SerialPort(BytePort): 19 def __init__(self, reader: StreamReader, writer: StreamWriter) -> None: 20 self._reader = reader 21 self._writer = writer 22 self._open = True 23 24 @classmethod 25 async def open(cls, port: str, baud: int = DEFAULT_BAUD) -> SerialPort: 26 import serial_asyncio # noqa: PLC0415 27 reader, writer = await serial_asyncio.open_serial_connection(url=port, baudrate=baud) 28 return cls(reader=reader, writer=writer) 29 30 async def write(self, data: bytes) -> None: 31 if not self._open: 32 raise RuntimeError("port is closed") 33 self._writer.write(data) 34 await self._writer.drain() 35 36 async def read_byte(self) -> int: 37 data = await self._reader.readexactly(1) 38 return data[0] 39 40 async def aclose(self) -> None: 41 if not self._open: 42 return 43 self._open = False 44 self._writer.close() 45 with contextlib.suppress(Exception): 46 await self._writer.wait_closed() 47 48 @property 49 def is_open(self) -> bool: 50 return self._open
Helper class that provides a standard way to create an ABC using inheritance.