jmri_openlcb

 7class BytePort(ABC):
 8    @abstractmethod
 9    async def write(self, data: bytes) -> None: ...
10    @abstractmethod
11    async def read_byte(self) -> int: ...
12    @abstractmethod
13    async def aclose(self) -> None: ...
14    @property
15    @abstractmethod
16    def is_open(self) -> bool: ...

145@dataclass(frozen=True, slots=True)
146class Datagram:
147    """A fully-reassembled datagram — the body plus the two node aliases."""
148
149    source_alias: int
150    dest_alias: int
151    data: bytes

181class DatagramEngine:
182    """Reassembles inbound datagrams, sends outbound ones with
183    fragmentation, and pairs sends with the peer's ``Received OK`` /
184    ``Rejected`` reply.
185
186    The reader hooks into an :class:`OlcbTrafficController`'s frames
187    signal via :meth:`attach`; the caller is responsible for
188    :meth:`detach` (or calling ``aclose``) at shutdown so the signal
189    listener doesn't outlive the engine.
190    """
191
192    def __init__(
193        self, tc: OlcbTrafficController,
194    ) -> None:
195        self._tc = tc
196        self._unsubscribe: Callable[[], None] | None = None
197        # Per-peer reassembly buffers keyed by source alias.
198        self._partial: dict[int, bytearray] = {}
199        # Pending sends keyed by destination alias.
200        self._pending: dict[int, _Pending] = {}
201        # Fires with every fully-received *inbound* datagram after the
202        # engine has already sent Received OK to the peer.
203        self.inbound: Signal[Datagram] = Signal()
204
205    async def attach(self) -> None:
206        if self._unsubscribe is None:
207            self._unsubscribe = self._tc.frames.connect(self._on_frame)
208
209    async def detach(self) -> None:
210        if self._unsubscribe is not None:
211            self._unsubscribe()
212            self._unsubscribe = None
213        # Fail any outstanding sends so callers don't hang forever.
214        for pending in list(self._pending.values()):
215            if not pending.fut.done():
216                pending.fut.set_exception(
217                    RuntimeError("datagram engine closed"),
218                )
219        self._pending.clear()
220        self._partial.clear()
221
222    # ---- send -----------------------------------------------------------
223
224    async def send(
225        self,
226        dest_alias: int,
227        data: bytes,
228        *,
229        timeout: float = 3.0,
230        wait_for_reply: bool = True,
231    ) -> None:
232        """Send *data* to *dest_alias*. Awaits the peer's Datagram
233        Received OK when ``wait_for_reply`` is True (default). Raises
234        :class:`DatagramRejected` on Rejected, :class:`TimeoutError`
235        on no reply, :class:`ValueError` on bad inputs.
236
237        Data over 8 bytes is fragmented into First / Middle... / Final
238        frames per S-9.7.4.1. Zero-byte payloads are permitted and
239        travel as a single Complete frame with an empty data field.
240        """
241        if not 0 <= dest_alias <= 0xFFF:
242            raise ValueError(f"dest alias must be 12-bit: {dest_alias:#x}")
243        if len(data) > MAX_DATAGRAM_BYTES:
244            raise ValueError(
245                f"datagram body {len(data)} exceeds "
246                f"{MAX_DATAGRAM_BYTES}-byte cap",
247            )
248        if dest_alias in self._pending:
249            raise RuntimeError(
250                f"a datagram to alias {dest_alias:#x} is already in flight",
251            )
252
253        source_alias = self._tc.source_alias
254
255        # Arm the reply-waiter BEFORE sending so a very fast peer
256        # can't beat us to it. Skipped when the caller opts out.
257        pending: _Pending | None = None
258        if wait_for_reply:
259            loop = asyncio.get_event_loop()
260            pending = _Pending(fut=loop.create_future(), dest_alias=dest_alias)
261            self._pending[dest_alias] = pending
262
263        try:
264            await self._transmit(source_alias, dest_alias, data)
265            if pending is not None:
266                await asyncio.wait_for(pending.fut, timeout=timeout)
267        finally:
268            if pending is not None:
269                self._pending.pop(dest_alias, None)
270
271    async def _transmit(
272        self, source_alias: int, dest_alias: int, data: bytes,
273    ) -> None:
274        """Write out the datagram frames. Single Complete when body ≤ 8,
275        First + N × Middle + Final otherwise."""
276        if len(data) <= 8:
277            # Zero-length datagrams still go on the wire as a Complete
278            # frame — the protocol permits them even if no real
279            # sub-protocol uses them.
280            frame = OlcbFrame(
281                header=make_datagram_header(
282                    DATAGRAM_COMPLETE, dest_alias, source_alias,
283                ),
284                data=data,
285            )
286            await self._tc.send_frame(frame)
287            return
288
289        # Multi-frame: First, then Middle...Middle, then Final.
290        chunks: list[bytes] = [
291            data[i:i + 8] for i in range(0, len(data), 8)
292        ]
293        for i, chunk in enumerate(chunks):
294            if i == 0:
295                kind = DATAGRAM_FIRST
296            elif i == len(chunks) - 1:
297                kind = DATAGRAM_FINAL
298            else:
299                kind = DATAGRAM_MIDDLE
300            frame = OlcbFrame(
301                header=make_datagram_header(kind, dest_alias, source_alias),
302                data=chunk,
303            )
304            await self._tc.send_frame(frame)
305
306    # ---- receive --------------------------------------------------------
307
308    async def _on_frame(self, frame: OlcbFrame) -> None:
309        kind = _extract_frame_kind(frame.header)
310        if kind in (
311            DATAGRAM_COMPLETE, DATAGRAM_FIRST,
312            DATAGRAM_MIDDLE, DATAGRAM_FINAL,
313        ):
314            await self._handle_datagram_frame(kind, frame)
315            return
316        # Not a datagram frame — is it a reply to something WE sent?
317        if kind == 0x1:
318            mti = _extract_mti(frame.header)
319            if mti in (MTI_DATAGRAM_RECEIVED_OK, MTI_DATAGRAM_REJECTED):
320                await self._handle_reply(mti, frame)
321
322    async def _handle_datagram_frame(
323        self, kind: int, frame: OlcbFrame,
324    ) -> None:
325        dest_alias = _extract_dest_alias(frame.header)
326        source_alias = _extract_source_alias(frame.header)
327
328        # We accept datagrams only when they're addressed to our own
329        # alias — the same rule real hardware follows. Everything else
330        # is on the bus but not for us.
331        if dest_alias != self._tc.source_alias:
332            return
333
334        if kind == DATAGRAM_COMPLETE:
335            body = bytes(frame.data)
336            await self._deliver(source_alias, dest_alias, body)
337            return
338
339        buf = self._partial.setdefault(source_alias, bytearray())
340        if kind == DATAGRAM_FIRST:
341            # Reset any dangling partial (peer aborted a previous send).
342            buf.clear()
343            buf.extend(frame.data)
344            return
345        if kind == DATAGRAM_MIDDLE:
346            buf.extend(frame.data)
347            return
348        if kind == DATAGRAM_FINAL:
349            buf.extend(frame.data)
350            body = bytes(buf)
351            self._partial.pop(source_alias, None)
352            await self._deliver(source_alias, dest_alias, body)
353
354    async def _deliver(
355        self, source_alias: int, dest_alias: int, body: bytes,
356    ) -> None:
357        # Send Received OK before firing our own signal so the reply
358        # is on the wire even if a subscriber raises.
359        await self._send_received_ok(source_alias)
360        dg = Datagram(
361            source_alias=source_alias,
362            dest_alias=dest_alias,
363            data=body,
364        )
365        await self.inbound.emit(dg)
366
367    async def _send_received_ok(self, dest_alias: int) -> None:
368        frame = OlcbFrame(
369            header=make_reply_header(
370                MTI_DATAGRAM_RECEIVED_OK,
371                self._tc.source_alias,
372            ),
373            # Addressed messages carry the destination alias in the
374            # first two body bytes, followed by protocol-specific
375            # fields (here: a single 0x00 = "no reply pending").
376            data=_encode_dest_prefix(dest_alias) + b"\x00",
377        )
378        await self._tc.send_frame(frame)
379
380    async def _handle_reply(self, mti: int, frame: OlcbFrame) -> None:
381        # The reply's *source* is the peer we sent to. Match on that.
382        peer_alias = _extract_source_alias(frame.header)
383        # For addressed messages, the destination alias travels in
384        # the first 2 body bytes, not the header.
385        addressed_dest = _decode_dest_alias_from_body(frame.data)
386        if addressed_dest != self._tc.source_alias:
387            return
388        pending = self._pending.get(peer_alias)
389        if pending is None or pending.fut.done():
390            return
391        # The status/error bytes follow the 2-byte destination
392        # prefix. Received OK carries one byte (0x00 or reply-pending
393        # flag); Rejected carries the 2-byte error code.
394        body = frame.data[2:]
395        if mti == MTI_DATAGRAM_RECEIVED_OK:
396            pending.fut.set_result(None)
397        else:  # MTI_DATAGRAM_REJECTED
398            code = 0
399            if len(body) >= 2:
400                code = (body[0] << 8) | body[1]
401            pending.fut.set_exception(DatagramRejected(error_code=code))

154@dataclass(frozen=True, slots=True)
155class DatagramRejected(Exception):
156    """Raised when the peer replies with ``Datagram Rejected`` to a
157    :meth:`DatagramEngine.send` call. The 2-byte error code is
158    exposed so callers can distinguish "permanent" (bit 15 set)
159    from "temporary" and log accordingly."""
160
161    error_code: int
162
163    def __str__(self) -> str:  # pragma: no cover — trivial
164        return f"datagram rejected (code {self.error_code:#06x})"

19@dataclass(frozen=True, slots=True)
20class EventId:
21    """64-bit OpenLCB event identifier."""
22
23    value: int  # 0..2**64-1
24
25    def __post_init__(self) -> None:
26        if not 0 <= self.value < (1 << 64):
27            raise ValueError(f"EventId must be 0..2**64-1, got {self.value}")
28
29    @classmethod
30    def parse(cls, text: str) -> EventId:
31        """Parse the dotted-hex or plain-hex text form into an :class:`EventId`."""
32        cleaned = text.replace(".", "").replace(" ", "").strip()
33        if not cleaned:
34            raise ValueError(f"empty EventId text {text!r}")
35        try:
36            return cls(int(cleaned, 16))
37        except ValueError as e:
38            raise ValueError(f"invalid EventId text {text!r}: {e}") from None
39
40    @property
41    def bytes(self) -> bytes:
42        return self.value.to_bytes(8, "big")
43
44    def __str__(self) -> str:
45        b = self.bytes
46        return ".".join(f"{x:02X}" for x in b)

49@dataclass(frozen=True, slots=True)
50class EventReport:
51    """Produced/Consumer Event Report — the workhorse message.
52
53    Sent by producers to announce "this event happened"; observed by
54    consumers to drive their state. The 64-bit ``event`` is enough; the
55    16-bit source ``alias`` identifies *which* node produced it (useful
56    for diagnostics; we don't validate aliases here).
57    """
58
59    event: EventId
60    source_alias: int  # 12-bit node alias, observed in the CAN header

 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

125class MemoryClient:
126    """Convenience API over the memory-config protocol.
127
128    Wraps a :class:`~jmri_openlcb.datagrams.DatagramEngine` so callers
129    talk to remote nodes in terms of ``read(space, address, count)`` /
130    ``write(space, address, data)`` and get bytes / None back
131    respectively. Failures raise :class:`MemoryError_`.
132
133    Each client instance registers ONE inbound listener on the
134    datagram engine and dispatches incoming replies to whatever
135    request is waiting. That means callers can await multiple
136    read/write operations against the same peer serially — a single
137    outstanding operation per peer is the assumption (matches the
138    protocol's implicit half-duplex nature at the memory layer).
139    """
140
141    def __init__(self, engine: DatagramEngine) -> None:
142        self._engine = engine
143        self._pending: dict[int, asyncio.Future[bytes]] = {}
144        self._unsubscribe = engine.inbound.connect(self._on_inbound)
145
146    def close(self) -> None:
147        self._unsubscribe()
148        for fut in self._pending.values():
149            if not fut.done():
150                fut.set_exception(RuntimeError("memory client closed"))
151        self._pending.clear()
152
153    async def read(
154        self,
155        peer_alias: int,
156        space: int,
157        address: int,
158        count: int,
159        *,
160        timeout: float = 3.0,
161    ) -> bytes:
162        """Read *count* bytes starting at *address* in *space* from
163        the peer. Automatically splits reads over
164        :data:`MAX_MEMORY_CHUNK` into multiple requests.
165        """
166        if count < 0:
167            raise ValueError(f"count must be non-negative: {count}")
168        chunks: list[bytes] = []
169        remaining = count
170        cursor = address
171        while remaining > 0:
172            batch = min(remaining, MAX_MEMORY_CHUNK)
173            chunk = await self._read_one(
174                peer_alias, space, cursor, batch, timeout=timeout,
175            )
176            chunks.append(chunk)
177            # A short reply signals end-of-space; stop early so we
178            # don't spin at the peer's expense.
179            if len(chunk) < batch:
180                break
181            cursor += batch
182            remaining -= batch
183        return b"".join(chunks)
184
185    async def write(
186        self,
187        peer_alias: int,
188        space: int,
189        address: int,
190        data: bytes,
191        *,
192        timeout: float = 3.0,
193    ) -> None:
194        """Write *data* starting at *address* in *space* on the peer.
195        Split into :data:`MAX_MEMORY_CHUNK`-byte writes."""
196        cursor = address
197        pos = 0
198        while pos < len(data):
199            batch = data[pos:pos + MAX_MEMORY_CHUNK]
200            await self._write_one(
201                peer_alias, space, cursor, batch, timeout=timeout,
202            )
203            cursor += len(batch)
204            pos += len(batch)
205
206    # ---- one-shot request path ----------------------------------------
207
208    async def _read_one(
209        self, peer_alias: int, space: int, address: int, count: int,
210        *, timeout: float,
211    ) -> bytes:
212        cmd, explicit = _encode_read_command(space)
213        body = self._build_request(cmd, address, explicit, space)
214        # count byte last; single byte, 1..64.
215        body += bytes([count & 0xFF])
216        return await self._request(peer_alias, body, timeout=timeout)
217
218    async def _write_one(
219        self, peer_alias: int, space: int, address: int, data: bytes,
220        *, timeout: float,
221    ) -> None:
222        cmd, explicit = _encode_write_command(space)
223        body = self._build_request(cmd, address, explicit, space) + data
224        await self._request(peer_alias, body, timeout=timeout)
225
226    def _build_request(
227        self, command: int, address: int, explicit_space: bool, space: int,
228    ) -> bytes:
229        addr_bytes = address.to_bytes(4, "big")
230        body = bytes([MEMORY_CONFIG, command]) + addr_bytes
231        if explicit_space:
232            body += bytes([space & 0xFF])
233        return body
234
235    async def _request(
236        self, peer_alias: int, body: bytes, *, timeout: float,
237    ) -> bytes:
238        """Send *body* as a datagram and await the peer's memory-config
239        reply (which arrives as an inbound datagram after the peer's
240        Received-OK)."""
241        if peer_alias in self._pending:
242            raise RuntimeError(
243                f"a memory op to alias {peer_alias:#x} is already in flight",
244            )
245        loop = asyncio.get_event_loop()
246        fut: asyncio.Future[bytes] = loop.create_future()
247        self._pending[peer_alias] = fut
248        try:
249            # The datagram engine's send() awaits Received OK before
250            # returning; the peer's memory-config *reply* arrives as
251            # a subsequent inbound datagram.
252            await self._engine.send(peer_alias, body, timeout=timeout)
253            data = await asyncio.wait_for(fut, timeout=timeout)
254        finally:
255            self._pending.pop(peer_alias, None)
256        return data
257
258    # ---- inbound reply handler ----------------------------------------
259
260    async def _on_inbound(self, dg: Datagram) -> None:
261        if len(dg.data) < 6:
262            return
263        if dg.data[0] != MEMORY_CONFIG:
264            return
265        command = dg.data[1]
266        # We match on peer alias (source of the datagram).
267        fut = self._pending.get(dg.source_alias)
268        if fut is None or fut.done():
269            return
270        if _is_failure_reply(command):
271            code = 0
272            # Optional 2-byte error code follows the address (6 bytes in).
273            if len(dg.data) >= 8:
274                code = (dg.data[6] << 8) | dg.data[7]
275            fut.set_exception(MemoryError_(error_code=code))
276            return
277        if _is_read_reply(command):
278            # Layout: proto, cmd, addr(4), [space], data...
279            # Compact spaces skip the extra space byte.
280            compact = _compact_space_bits_from_command(command)
281            data_start = 6 if compact is not None else 7
282            fut.set_result(bytes(dg.data[data_start:]))
283            return
284        if _is_write_reply(command):
285            # Write success carries no body; empty bytes signals done.
286            fut.set_result(b"")

109@dataclass(frozen=True, slots=True)
110class MemoryError_(Exception):
111    """Raised when a node returns a failure reply for a memory op.
112
113    Name has an underscore so it doesn't shadow Python's built-in.
114    """
115
116    error_code: int
117
118    def __str__(self) -> str:
119        return f"memory operation failed (code {self.error_code:#06x})"

23class OlcbConnection(SystemConnection):
24    name: ClassVar[str] = "OpenLCB"
25    prefix: ClassVar[str] = "X"
26
27    def __init__(self) -> None:
28        super().__init__()
29        self._port: BytePort | None = None
30        self._tc: OlcbTrafficController | None = None
31
32    async def open(self, settings: ConnectionSettings | None = None) -> None:
33        s = settings or {}
34        self._port = await _build_port(s)
35        self._tc = OlcbTrafficController(
36            self._port, source_alias=int(s.get("source_alias", 0x123))
37        )
38        await self._tc.start()
39
40        self.turnouts = OlcbTurnoutManager(self.prefix, tc=self._tc)  # type: ignore[assignment]
41        self.sensors = OlcbSensorManager(self.prefix, tc=self._tc)    # type: ignore[assignment]
42        # Well-known event power manager — produces + consumes the
43        # reserved 01.00.00.00.00.00.FF.FE / .FF event IDs so a
44        # single set_power() call converges every OpenLCB node
45        # listening to the bus. See :mod:`jmri_openlcb.power`.
46        self.power = OlcbPowerManager(self._tc)
47
48        await super().open(settings)
49
50    async def close(self) -> None:
51        # Drop the power manager's listener before we tear down the
52        # traffic controller so the manager doesn't fire against a
53        # closed signal.
54        if isinstance(self.power, OlcbPowerManager):
55            self.power.close()
56        if self._tc is not None:
57            await self._tc.aclose()
58            self._tc = None
59        self._port = None
60        await super().close()
61
62    @property
63    def traffic(self) -> OlcbTrafficController:
64        if self._tc is None:
65            raise RuntimeError("connection not open")
66        return self._tc

36@dataclass(frozen=True, slots=True)
37class OlcbFrame:
38    """One GridConnect frame, before MTI-level decoding."""
39
40    header: int       # 29-bit extended CAN ID
41    data: bytes       # 0..8 bytes

35class OlcbPowerManager(PowerManager):
36    """Power manager that produces + consumes the well-known events."""
37
38    def __init__(self, tc: OlcbTrafficController) -> None:
39        super().__init__()
40        self._tc = tc
41        self._unsubscribe = tc.events.connect(self._on_event)
42
43    async def set_power(self, state: PowerState) -> None:
44        # Emit the well-known event so every other node on the bus
45        # observes the transition. Local state is updated by the same
46        # inbound-event handler when our own broadcast comes back
47        # around, keeping the "we just heard about it from the bus"
48        # semantics uniform whether the change came from us or a peer.
49        if state is PowerState.ON:
50            await self._tc.produce_event(POWER_ON)
51        elif state is PowerState.OFF:
52            await self._tc.produce_event(POWER_OFF)
53        else:
54            # UNKNOWN is a "we don't know" state, not a command; no
55            # matching event exists.
56            return
57        # Some loopback / test transports don't echo our own frames
58        # back, so update the internal state explicitly. If the bus
59        # DOES echo, the inbound handler will see it and no-op on
60        # the already-current state (see ``_set_state`` in the base).
61        await self._set_state(state)
62
63    def close(self) -> None:
64        """Detach the inbound listener. Idempotent."""
65        if self._unsubscribe is not None:
66            self._unsubscribe()
67            self._unsubscribe = None
68
69    async def _on_event(self, report: EventReport) -> None:
70        if report.event == POWER_ON:
71            await self._set_state(PowerState.ON)
72        elif report.event == POWER_OFF:
73            await self._set_state(PowerState.OFF)

108class OlcbSensor(Sensor):
109    """A sensor bound to two event IDs (active/inactive)."""
110
111    __slots__ = ("_active_event", "_inactive_event")
112
113    def __init__(
114        self,
115        system_name: str,
116        active_event: EventId,
117        inactive_event: EventId,
118    ) -> None:
119        super().__init__(system_name)
120        self._active_event = active_event
121        self._inactive_event = inactive_event
122
123    @property
124    def active_event(self) -> EventId:
125        return self._active_event
126
127    @property
128    def inactive_event(self) -> EventId:
129        return self._inactive_event