jmri_openlcb.transport.tcp_port

TCP OpenLCB transport — for network-attached LCC bridges.

 1"""TCP OpenLCB transport — for network-attached LCC bridges."""
 2from __future__ import annotations
 3
 4import asyncio
 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
14class TcpPort(BytePort):
15    def __init__(self, reader: StreamReader, writer: StreamWriter) -> None:
16        self._reader = reader
17        self._writer = writer
18        self._open = True
19
20    @classmethod
21    async def open(cls, host: str, port: int) -> TcpPort:
22        reader, writer = await asyncio.open_connection(host, port)
23        return cls(reader=reader, writer=writer)
24
25    async def write(self, data: bytes) -> None:
26        if not self._open:
27            raise RuntimeError("port is closed")
28        self._writer.write(data)
29        await self._writer.drain()
30
31    async def read_byte(self) -> int:
32        data = await self._reader.readexactly(1)
33        return data[0]
34
35    async def aclose(self) -> None:
36        if not self._open:
37            return
38        self._open = False
39        self._writer.close()
40        with contextlib.suppress(Exception):
41            await self._writer.wait_closed()
42
43    @property
44    def is_open(self) -> bool:
45        return self._open
class TcpPort(jmri_openlcb.transport.port.BytePort): 
15class TcpPort(BytePort):
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 write(self, data: bytes) -> None:
27        if not self._open:
28            raise RuntimeError("port is closed")
29        self._writer.write(data)
30        await self._writer.drain()
31
32    async def read_byte(self) -> int:
33        data = await self._reader.readexactly(1)
34        return data[0]
35
36    async def aclose(self) -> None:
37        if not self._open:
38            return
39        self._open = False
40        self._writer.close()
41        with contextlib.suppress(Exception):
42            await self._writer.wait_closed()
43
44    @property
45    def is_open(self) -> bool:
46        return self._open

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

TcpPort( reader: asyncio.streams.StreamReader, writer: asyncio.streams.StreamWriter) 
16    def __init__(self, reader: StreamReader, writer: StreamWriter) -> None:
17        self._reader = reader
18        self._writer = writer
19        self._open = True
@classmethod
async def open(cls, host: str, port: int) -> TcpPort: 
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)
async def write(self, data: bytes) -> None: 
26    async def write(self, data: bytes) -> None:
27        if not self._open:
28            raise RuntimeError("port is closed")
29        self._writer.write(data)
30        await self._writer.drain()
async def read_byte(self) -> int: 
32    async def read_byte(self) -> int:
33        data = await self._reader.readexactly(1)
34        return data[0]
async def aclose(self) -> None: 
36    async def aclose(self) -> None:
37        if not self._open:
38            return
39        self._open = False
40        self._writer.close()
41        with contextlib.suppress(Exception):
42            await self._writer.wait_closed()
is_open: bool 
44    @property
45    def is_open(self) -> bool:
46        return self._open