jmri_dccex.codec

 1"""DCC-EX text protocol codec.
 2
 3DCC-EX (formerly DCC++) is a bracket-delimited ASCII protocol:
 4
 5    <command arg1 arg2 ...>
 6
 7Commands and responses both use the same bracket convention. Multiple
 8commands can be sent back-to-back; the codec reads one complete bracketed
 9message at a time and strips the brackets.
10
11Common commands JMRI uses:
12
13* ``<1>`` / ``<0>`` — power on / off
14* ``<T id 1|0>`` — set turnout *id* to closed (1) or thrown (0)
15* ``<S id pin pullup>`` — define sensor (server-side state only)
16* ``<t cab loco speed dir>`` — throttle command (legacy form)
17* ``<R cv>`` — read CV; ``<W cv val>`` — write CV
18* ``<Q id>`` — sensor active report (server → client)
19* ``<q id>`` — sensor inactive report
20
21Responses follow the same shape; unknown messages decode to
22:class:`DccexUnknown` so listeners don't lose visibility.
23"""
24from __future__ import annotations
25
26from collections.abc import Awaitable, Callable
27from dataclasses import dataclass
28
29
30@dataclass(frozen=True, slots=True)
31class DccexMessage:
32    """A parsed message — command character + space-separated args."""
33
34    command: str          # first token (e.g. "1", "T", "Q")
35    args: tuple[str, ...] # remaining tokens
36
37    @property
38    def raw(self) -> str:
39        if not self.args:
40            return self.command
41        return f"{self.command} {' '.join(self.args)}"
42
43
44@dataclass(frozen=True, slots=True)
45class DccexUnknown(DccexMessage):
46    """Marker subclass: a well-formed bracket message we don't recognize."""
47
48
49def encode(msg: DccexMessage) -> bytes:
50    """Wrap a message in brackets and return as UTF-8 bytes."""
51    return f"<{msg.raw}>".encode()
52
53
54def decode(text: str) -> DccexMessage:
55    """Parse the content *between* the brackets into a :class:`DccexMessage`."""
56    text = text.strip()
57    if not text:
58        raise ValueError("empty DCC-EX message body")
59    tokens = text.split()
60    return DccexMessage(command=tokens[0], args=tuple(tokens[1:]))
61
62
63async def read_message(
64    read_byte: Callable[[], Awaitable[int]],
65) -> str:
66    """Read one bracketed message from the byte stream.
67
68    Bytes outside a ``<...>`` pair are silently discarded (matches DCC-EX
69    convention: newlines and other status text between messages are noise).
70    Returns the *unwrapped* contents (no brackets).
71    """
72    # Wait for opening bracket.
73    while True:
74        b = await read_byte()
75        if b == ord('<'):
76            break
77
78    out = bytearray()
79    while True:
80        b = await read_byte()
81        if b == ord('>'):
82            return out.decode("utf-8", errors="replace")
83        out.append(b)