jmri_openlcb.transport.loopback 

 1from __future__ import annotations
 2
 3import asyncio
 4
 5from jmri_openlcb.transport.port import BytePort
 6
 7
 8class LoopbackPort(BytePort):
 9    def __init__(self, read_q: asyncio.Queue[int], write_q: asyncio.Queue[int]) -> None:
10        self._r = read_q
11        self._w = write_q
12        self._open = True
13
14    async def write(self, data: bytes) -> None:
15        if not self._open:
16            raise RuntimeError("port is closed")
17        for b in data:
18            await self._w.put(b)
19
20    async def read_byte(self) -> int:
21        return await self._r.get()
22
23    async def aclose(self) -> None:
24        self._open = False
25
26    @property
27    def is_open(self) -> bool:
28        return self._open
29
30
31def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]:
32    a_to_b: asyncio.Queue[int] = asyncio.Queue()
33    b_to_a: asyncio.Queue[int] = asyncio.Queue()
34    return LoopbackPort(b_to_a, a_to_b), LoopbackPort(a_to_b, b_to_a)
class LoopbackPort(jmri_openlcb.transport.port.BytePort): 
 9class LoopbackPort(BytePort):
10    def __init__(self, read_q: asyncio.Queue[int], write_q: asyncio.Queue[int]) -> None:
11        self._r = read_q
12        self._w = write_q
13        self._open = True
14
15    async def write(self, data: bytes) -> None:
16        if not self._open:
17            raise RuntimeError("port is closed")
18        for b in data:
19            await self._w.put(b)
20
21    async def read_byte(self) -> int:
22        return await self._r.get()
23
24    async def aclose(self) -> None:
25        self._open = False
26
27    @property
28    def is_open(self) -> bool:
29        return self._open

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

LoopbackPort( read_q: asyncio.queues.Queue[int], write_q: asyncio.queues.Queue[int]) 
10    def __init__(self, read_q: asyncio.Queue[int], write_q: asyncio.Queue[int]) -> None:
11        self._r = read_q
12        self._w = write_q
13        self._open = True
async def write(self, data: bytes) -> None: 
15    async def write(self, data: bytes) -> None:
16        if not self._open:
17            raise RuntimeError("port is closed")
18        for b in data:
19            await self._w.put(b)
async def read_byte(self) -> int: 
21    async def read_byte(self) -> int:
22        return await self._r.get()
async def aclose(self) -> None: 
24    async def aclose(self) -> None:
25        self._open = False
is_open: bool 
27    @property
28    def is_open(self) -> bool:
29        return self._open
def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]: 
32def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]:
33    a_to_b: asyncio.Queue[int] = asyncio.Queue()
34    b_to_a: asyncio.Queue[int] = asyncio.Queue()
35    return LoopbackPort(b_to_a, a_to_b), LoopbackPort(a_to_b, b_to_a)