jmri_core.testing.bus_replay

  1"""Recorded-bus replay format shared by driver test suites.
  2
  3The Stage-0 scoping doc called for "capture real LocoNet traffic
  4with wireshark/JMRI's monitor, replay against the Python TC, assert
  5state matches." This module ships the on-disk file shape both the
  6LocoNet and NCE test suites use.
  7
  8Wire shape (JSONL — one JSON object per line)::
  9
 10    {"t": 0.000000, "dir": "in",  "frame_hex": "e703bf..."}
 11    {"t": 0.043500, "dir": "out", "frame_hex": "8000..."}
 12
 13Fields:
 14
 15* ``t`` — monotonic timestamp in seconds, relative to the first frame.
 16* ``dir`` — ``"in"`` = from command station to us, ``"out"`` = from us
 17  to CS. Direction matters because a driver's assertion order depends
 18  on which side spoke first.
 19* ``frame_hex`` — the frame bytes as hex, including any framing bytes
 20  (length prefix, address byte). No checksum stripping here — the
 21  wire form is the canonical form.
 22
 23The format is line-oriented and plain-text on purpose: easy to diff,
 24grep, and hand-edit into synthetic fixtures when you don't have real
 25traffic to work with. Trailing whitespace and blank lines are
 26ignored on read; comments (``#`` at column 0) are permitted so
 27test-authors can annotate captures.
 28
 29Chose to live under ``jmri_core.testing`` rather than
 30``tests/_replay.py`` in each driver package because the format is
 31one shared contract — a divergent LocoNet-only shape would defeat
 32the point of the driver-agnostic capture tooling that would use
 33this.
 34"""
 35from __future__ import annotations
 36
 37import json
 38import pathlib
 39from collections.abc import Iterable, Iterator
 40from dataclasses import dataclass
 41from typing import Literal
 42
 43Direction = Literal["in", "out"]
 44
 45
 46@dataclass(frozen=True, slots=True)
 47class BusFrame:
 48    """One captured frame."""
 49
 50    t: float
 51    direction: Direction
 52    frame: bytes
 53
 54    def to_jsonl(self) -> str:
 55        return json.dumps({
 56            "t": round(self.t, 6),
 57            "dir": self.direction,
 58            "frame_hex": self.frame.hex(),
 59        })
 60
 61    @classmethod
 62    def from_jsonl(cls, line: str) -> BusFrame:
 63        obj = json.loads(line)
 64        direction = obj.get("dir", "in")
 65        if direction not in ("in", "out"):
 66            raise ValueError(
 67                f"invalid direction {direction!r}; must be 'in' or 'out'"
 68            )
 69        return cls(
 70            t=float(obj["t"]),
 71            direction=direction,   # type: ignore[arg-type]
 72            frame=bytes.fromhex(obj["frame_hex"]),
 73        )
 74
 75
 76def write_capture(path: str | pathlib.Path, frames: Iterable[BusFrame]) -> None:
 77    """Serialize *frames* to *path* as JSONL. Overwrites."""
 78    p = pathlib.Path(path)
 79    with p.open("w", encoding="utf-8") as f:
 80        for frame in frames:
 81            f.write(frame.to_jsonl() + "\n")
 82
 83
 84def read_capture(path: str | pathlib.Path) -> list[BusFrame]:
 85    """Load a capture. Blank lines and ``#``-prefixed comments are
 86    silently skipped so hand-authored fixtures can carry notes."""
 87    p = pathlib.Path(path)
 88    out: list[BusFrame] = []
 89    with p.open("r", encoding="utf-8") as f:
 90        for raw_line in f:
 91            line = raw_line.strip()
 92            if not line or line.startswith("#"):
 93                continue
 94            out.append(BusFrame.from_jsonl(line))
 95    return out
 96
 97
 98def iter_direction(
 99    frames: Iterable[BusFrame], direction: Direction,
100) -> Iterator[BusFrame]:
101    """Filter capture frames by direction. Handy when a test only
102    cares about one side of the conversation (e.g. "assert every
103    ``in`` frame decodes cleanly")."""
104    for f in frames:
105        if f.direction == direction:
106            yield f