jmri_loconet.transport.tcp_port

TcpPort — LocoNet over TCP (LbServer / network bridge).

 1"""``TcpPort`` — LocoNet over TCP (LbServer / network bridge)."""
 2from __future__ import annotations
 3
 4import asyncio
 5from typing import TYPE_CHECKING
 6
 7from jmri_loconet.transport.port import LocoNetPort
 8
 9if TYPE_CHECKING:
10    from asyncio import StreamReader, StreamWriter
11
12
13class TcpPort(LocoNetPort):
14    """LocoNet over a TCP socket (e.g. JMRI's LbServer or Digitrax LNWI)."""
15
16    def __init__(self, reader: StreamReader, writer: StreamWriter) -> None:
17        self._reader = reader
18        self._writer = writer
19        self._open = True
20
21    @classmethod
22    async def open(cls, host: str, port: int) -> TcpPort:
23        reader, writer = await asyncio.open_connection(host, port)
24        return cls(reader=reader, writer=writer)
25
26    async def read_byte(self) -> int:
27        data = await self._reader.readexactly(1)
28        return data[0]
29
30    async def read_exact(self, n: int) -> bytes:
31        return await self._reader.readexactly(n)
32
33    async def write_frame(self, frame: bytes) -> None:
34        if not self._open:
35            raise RuntimeError("port is closed")
36        self._writer.write(frame)
37        await self._writer.drain()
38
39    async def aclose(self) -> None:
40        if not self._open:
41            return
42        self._open = False
43        self._writer.close()
44        import contextlib  # noqa: PLC0415
45        with contextlib.suppress(Exception):
46            await self._writer.wait_closed()
47
48    @property
49    def is_open(self) -> bool:
50        return self._open
class TcpPort(jmri_loconet.transport.port.LocoNetPort): 
14class TcpPort(LocoNetPort):
15    """LocoNet over a TCP socket (e.g. JMRI's LbServer or Digitrax LNWI)."""
16
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, host: str, port: int) -> TcpPort:
24        reader, writer = await asyncio.open_connection(host, port)
25        return cls(reader=reader, writer=writer)
26
27    async def read_byte(self) -> int:
28        data = await self._reader.readexactly(1)
29        return data[0]
30
31    async def read_exact(self, n: int) -> bytes:
32        return await self._reader.readexactly(n)
33
34    async def write_frame(self, frame: bytes) -> None:
35        if not self._open:
36            raise RuntimeError("port is closed")
37        self._writer.write(frame)
38        await self._writer.drain()
39
40    async def aclose(self) -> None:
41        if not self._open:
42            return
43        self._open = False
44        self._writer.close()
45        import contextlib  # noqa: PLC0415
46        with contextlib.suppress(Exception):
47            await self._writer.wait_closed()
48
49    @property
50    def is_open(self) -> bool:
51        return self._open

LocoNet over a TCP socket (e.g. JMRI's LbServer or Digitrax LNWI).

TcpPort( reader: asyncio.streams.StreamReader, writer: asyncio.streams.StreamWriter) 
17    def __init__(self, reader: StreamReader, writer: StreamWriter) -> None:
18        self._reader = reader
19        self._writer = writer
20        self._open = True
@classmethod
async def open(cls, host: str, port: int) -> TcpPort: 
22    @classmethod
23    async def open(cls, host: str, port: int) -> TcpPort:
24        reader, writer = await asyncio.open_connection(host, port)
25        return cls(reader=reader, writer=writer)
async def read_byte(self) -> int: 
27    async def read_byte(self) -> int:
28        data = await self._reader.readexactly(1)
29        return data[0]

Read one byte. Blocks until a byte is available.

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

Read exactly n bytes.

async def write_frame(self, frame: bytes) -> None: 
34    async def write_frame(self, frame: bytes) -> None:
35        if not self._open:
36            raise RuntimeError("port is closed")
37        self._writer.write(frame)
38        await self._writer.drain()

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

async def aclose(self) -> None: 
40    async def aclose(self) -> None:
41        if not self._open:
42            return
43        self._open = False
44        self._writer.close()
45        import contextlib  # noqa: PLC0415
46        with contextlib.suppress(Exception):
47            await self._writer.wait_closed()

Close the port. Idempotent.

is_open: bool 
49    @property
50    def is_open(self) -> bool:
51        return self._open