jmri_loconet.transport.loopback

In-memory loopback transport — wires two ports together via asyncio queues.

Used by tests + the integration suite: a, b = loopback_pair() gives two LoopbackPort instances where bytes written to one appear on the other's read stream.

 1"""In-memory loopback transport — wires two ports together via asyncio queues.
 2
 3Used by tests + the integration suite: ``a, b = loopback_pair()`` gives two
 4:class:`LoopbackPort` instances where bytes written to one appear on the
 5other's read stream.
 6"""
 7from __future__ import annotations
 8
 9import asyncio
10
11from jmri_loconet.transport.port import LocoNetPort
12
13
14class LoopbackPort(LocoNetPort):
15    """One half of a loopback pair."""
16
17    def __init__(self, read_queue: asyncio.Queue[int], write_queue: asyncio.Queue[int]) -> None:
18        self._read = read_queue
19        self._write = write_queue
20        self._open = True
21
22    async def read_byte(self) -> int:
23        return await self._read.get()
24
25    async def read_exact(self, n: int) -> bytes:
26        return bytes([await self._read.get() for _ in range(n)])
27
28    async def write_frame(self, frame: bytes) -> None:
29        if not self._open:
30            raise RuntimeError("port is closed")
31        for b in frame:
32            await self._write.put(b)
33
34    async def aclose(self) -> None:
35        self._open = False
36
37    @property
38    def is_open(self) -> bool:
39        return self._open
40
41
42def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]:
43    """Return a pair of connected loopback ports."""
44    a_to_b: asyncio.Queue[int] = asyncio.Queue()
45    b_to_a: asyncio.Queue[int] = asyncio.Queue()
46    a = LoopbackPort(read_queue=b_to_a, write_queue=a_to_b)
47    b = LoopbackPort(read_queue=a_to_b, write_queue=b_to_a)
48    return a, b
class LoopbackPort(jmri_loconet.transport.port.LocoNetPort): 
15class LoopbackPort(LocoNetPort):
16    """One half of a loopback pair."""
17
18    def __init__(self, read_queue: asyncio.Queue[int], write_queue: asyncio.Queue[int]) -> None:
19        self._read = read_queue
20        self._write = write_queue
21        self._open = True
22
23    async def read_byte(self) -> int:
24        return await self._read.get()
25
26    async def read_exact(self, n: int) -> bytes:
27        return bytes([await self._read.get() for _ in range(n)])
28
29    async def write_frame(self, frame: bytes) -> None:
30        if not self._open:
31            raise RuntimeError("port is closed")
32        for b in frame:
33            await self._write.put(b)
34
35    async def aclose(self) -> None:
36        self._open = False
37
38    @property
39    def is_open(self) -> bool:
40        return self._open

One half of a loopback pair.

LoopbackPort( read_queue: asyncio.queues.Queue[int], write_queue: asyncio.queues.Queue[int]) 
18    def __init__(self, read_queue: asyncio.Queue[int], write_queue: asyncio.Queue[int]) -> None:
19        self._read = read_queue
20        self._write = write_queue
21        self._open = True
async def read_byte(self) -> int: 
23    async def read_byte(self) -> int:
24        return await self._read.get()

Read one byte. Blocks until a byte is available.

async def read_exact(self, n: int) -> bytes: 
26    async def read_exact(self, n: int) -> bytes:
27        return bytes([await self._read.get() for _ in range(n)])

Read exactly n bytes.

async def write_frame(self, frame: bytes) -> None: 
29    async def write_frame(self, frame: bytes) -> None:
30        if not self._open:
31            raise RuntimeError("port is closed")
32        for b in frame:
33            await self._write.put(b)

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

async def aclose(self) -> None: 
35    async def aclose(self) -> None:
36        self._open = False

Close the port. Idempotent.

is_open: bool 
38    @property
39    def is_open(self) -> bool:
40        return self._open
def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]: 
43def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]:
44    """Return a pair of connected loopback ports."""
45    a_to_b: asyncio.Queue[int] = asyncio.Queue()
46    b_to_a: asyncio.Queue[int] = asyncio.Queue()
47    a = LoopbackPort(read_queue=b_to_a, write_queue=a_to_b)
48    b = LoopbackPort(read_queue=a_to_b, write_queue=b_to_a)
49    return a, b

Return a pair of connected loopback ports.