jmri_loconet.transport.port
LocoNetPort ABC — the minimum surface a transport must provide.
The ~jmri_loconet.traffic.LocoNetTrafficController reads complete
frames using read_byte() + read_exact() (so the codec can
resync mid-stream after noise) and writes complete pre-framed bytes via
write_frame(). Closing is aclose().
1"""``LocoNetPort`` ABC — the minimum surface a transport must provide. 2 3The :class:`~jmri_loconet.traffic.LocoNetTrafficController` reads complete 4frames using :meth:`read_byte` + :meth:`read_exact` (so the codec can 5resync mid-stream after noise) and writes complete pre-framed bytes via 6:meth:`write_frame`. Closing is :meth:`aclose`. 7""" 8from __future__ import annotations 9 10from abc import ABC, abstractmethod 11 12 13class LocoNetPort(ABC): 14 """Abstract bidirectional byte-stream LocoNet port.""" 15 16 @abstractmethod 17 async def read_byte(self) -> int: 18 """Read one byte. Blocks until a byte is available.""" 19 20 @abstractmethod 21 async def read_exact(self, n: int) -> bytes: 22 """Read exactly *n* bytes.""" 23 24 @abstractmethod 25 async def write_frame(self, frame: bytes) -> None: 26 """Write a complete LocoNet frame (opcode + payload + checksum).""" 27 28 @abstractmethod 29 async def aclose(self) -> None: 30 """Close the port. Idempotent.""" 31 32 @property 33 @abstractmethod 34 def is_open(self) -> bool: ...
class
LocoNetPort(abc.ABC):
14class LocoNetPort(ABC): 15 """Abstract bidirectional byte-stream LocoNet port.""" 16 17 @abstractmethod 18 async def read_byte(self) -> int: 19 """Read one byte. Blocks until a byte is available.""" 20 21 @abstractmethod 22 async def read_exact(self, n: int) -> bytes: 23 """Read exactly *n* bytes.""" 24 25 @abstractmethod 26 async def write_frame(self, frame: bytes) -> None: 27 """Write a complete LocoNet frame (opcode + payload + checksum).""" 28 29 @abstractmethod 30 async def aclose(self) -> None: 31 """Close the port. Idempotent.""" 32 33 @property 34 @abstractmethod 35 def is_open(self) -> bool: ...
Abstract bidirectional byte-stream LocoNet port.
@abstractmethod
async def
read_byte(self) -> int:
17 @abstractmethod 18 async def read_byte(self) -> int: 19 """Read one byte. Blocks until a byte is available."""
Read one byte. Blocks until a byte is available.
@abstractmethod
async def
write_frame(self, frame: bytes) -> None:
25 @abstractmethod 26 async def write_frame(self, frame: bytes) -> None: 27 """Write a complete LocoNet frame (opcode + payload + checksum)."""
Write a complete LocoNet frame (opcode + payload + checksum).