jmri_openlcb.codec

  1"""GridConnect ASCII codec for CAN-over-text.
  2
  3OpenLCB messages travel as CAN 2.0B extended frames; many bridges convert
  4them to GridConnect ASCII for serial/TCP transport:
  5
  6    :X<11-hex-digit header><0..16 hex digit data>;
  7
  8The header encodes:
  9
 10* bit 17 — always 1 (extended frame)
 11* bits 12..16 — MTI variant
 12* bit 11 — always 1 (OpenLCB frame)
 13* bits 0..11 — source node alias
 14
 15The lower 12 bits of the header are the **source node alias**. The next
 164 bits identify the **MTI** (message type indicator). MTIs we care about:
 17
 18  0x5B4 = Producer/Consumer Event Report (PCER) — the all-important event-emit message.
 19
 20Other MTIs (verify node, identify producers/consumers, datagrams) are
 21preserved as :class:`OlcbUnknown` so the monitor sees them but the
 22driver doesn't act on them.
 23"""
 24from __future__ import annotations
 25
 26from collections.abc import Awaitable, Callable
 27from dataclasses import dataclass
 28
 29from jmri_openlcb.events import EventId, EventReport
 30
 31# MTI constant for Producer/Consumer Event Report.
 32MTI_PCER = 0x5B4
 33
 34
 35@dataclass(frozen=True, slots=True)
 36class OlcbFrame:
 37    """One GridConnect frame, before MTI-level decoding."""
 38
 39    header: int       # 29-bit extended CAN ID
 40    data: bytes       # 0..8 bytes
 41
 42
 43@dataclass(frozen=True, slots=True)
 44class OlcbUnknown:
 45    """A well-formed CAN frame whose MTI we don't have a typed wrapper for."""
 46
 47    header: int
 48    data: bytes
 49
 50
 51def encode_frame(frame: OlcbFrame) -> bytes:
 52    """Encode an :class:`OlcbFrame` to GridConnect ASCII (with trailing newline)."""
 53    body = f":X{frame.header:08X}N{frame.data.hex().upper()};"
 54    return body.encode("ascii")
 55
 56
 57def decode_frame(text: str) -> OlcbFrame:
 58    """Parse one GridConnect message into an :class:`OlcbFrame`.
 59
 60    Accepts either ``:X<hex>N<hex>;`` (newer LCC format) or ``:X<hex>;`` for
 61    data-less frames.
 62    """
 63    text = text.strip()
 64    if not text.startswith(":X") or not text.endswith(";"):
 65        raise ValueError(f"not a GridConnect frame: {text!r}")
 66    body = text[2:-1]
 67    if "N" in body:
 68        head_hex, _, data_hex = body.partition("N")
 69    else:
 70        head_hex, data_hex = body, ""
 71    try:
 72        header = int(head_hex, 16)
 73        data = bytes.fromhex(data_hex) if data_hex else b""
 74    except ValueError as e:
 75        raise ValueError(f"invalid GridConnect frame {text!r}: {e}") from None
 76    return OlcbFrame(header=header, data=data)
 77
 78
 79def extract_mti(header: int) -> int:
 80    """Pull the MTI bits out of the 29-bit header (bits 12..23)."""
 81    return (header >> 12) & 0xFFF
 82
 83
 84def extract_source_alias(header: int) -> int:
 85    """Pull the 12-bit source node alias from the header low bits."""
 86    return header & 0x0FFF
 87
 88
 89def make_pcer_header(source_alias: int) -> int:
 90    """Compose the 29-bit CAN header for a PCER frame from *source_alias*."""
 91    if not 0 <= source_alias <= 0xFFF:
 92        raise ValueError(f"source alias must be 12-bit, got {source_alias:#x}")
 93    # bit 28 set (extended frame), MTI in bits 12..23, alias in bits 0..11.
 94    return (1 << 28) | (MTI_PCER << 12) | source_alias
 95
 96
 97def encode_pcer(event: EventId, source_alias: int) -> bytes:
 98    """Encode an event report as a GridConnect frame."""
 99    return encode_frame(OlcbFrame(
100        header=make_pcer_header(source_alias),
101        data=event.bytes,
102    ))
103
104
105def decode_pcer_if_pcer(frame: OlcbFrame) -> EventReport | None:
106    """Return an :class:`EventReport` iff *frame* is a PCER, else None."""
107    if extract_mti(frame.header) != MTI_PCER:
108        return None
109    if len(frame.data) != 8:
110        return None
111    return EventReport(
112        event=EventId(int.from_bytes(frame.data, "big")),
113        source_alias=extract_source_alias(frame.header),
114    )
115
116
117async def read_frame(read_byte: Callable[[], Awaitable[int]]) -> str:
118    """Read one GridConnect frame (terminated by ``;``)."""
119    # Skip until ':X' — the frame start.
120    out = bytearray()
121    while True:
122        b = await read_byte()
123        if b == ord(":"):
124            out = bytearray([b])
125            break
126    while True:
127        b = await read_byte()
128        out.append(b)
129        if b == ord(";"):
130            return out.decode("ascii", errors="replace")