jmri_core.testing

Test-time utilities shared across driver packages.

Kept here rather than under each driver's tests/ because these formats — recorded-bus replay files being the current example — are one shared contract across drivers. A divergent per-driver shape would defeat the point of any protocol-agnostic capture tooling that reads them.

The modules here have no runtime hardware dependencies and are safe to import from production code (though there's no reason to).

 1"""Test-time utilities shared across driver packages.
 2
 3Kept here rather than under each driver's ``tests/`` because these
 4formats — recorded-bus replay files being the current example — are
 5one shared contract across drivers. A divergent per-driver shape
 6would defeat the point of any protocol-agnostic capture tooling
 7that reads them.
 8
 9The modules here have no runtime hardware dependencies and are safe
10to import from production code (though there's no reason to).
11"""
12from jmri_core.testing.bus_replay import (
13    BusFrame,
14    Direction,
15    iter_direction,
16    read_capture,
17    write_capture,
18)
19
20__all__ = [
21    "BusFrame",
22    "Direction",
23    "iter_direction",
24    "read_capture",
25    "write_capture",
26]
@dataclass(frozen=True, slots=True)
class BusFrame: 
47@dataclass(frozen=True, slots=True)
48class BusFrame:
49    """One captured frame."""
50
51    t: float
52    direction: Direction
53    frame: bytes
54
55    def to_jsonl(self) -> str:
56        return json.dumps({
57            "t": round(self.t, 6),
58            "dir": self.direction,
59            "frame_hex": self.frame.hex(),
60        })
61
62    @classmethod
63    def from_jsonl(cls, line: str) -> BusFrame:
64        obj = json.loads(line)
65        direction = obj.get("dir", "in")
66        if direction not in ("in", "out"):
67            raise ValueError(
68                f"invalid direction {direction!r}; must be 'in' or 'out'"
69            )
70        return cls(
71            t=float(obj["t"]),
72            direction=direction,   # type: ignore[arg-type]
73            frame=bytes.fromhex(obj["frame_hex"]),
74        )

One captured frame.

BusFrame(t: float, direction: Literal['in', 'out'], frame: bytes)
t: float
direction: Literal['in', 'out']
frame: bytes
def to_jsonl(self) -> str: 
55    def to_jsonl(self) -> str:
56        return json.dumps({
57            "t": round(self.t, 6),
58            "dir": self.direction,
59            "frame_hex": self.frame.hex(),
60        })
@classmethod
def from_jsonl(cls, line: str) -> BusFrame: 
62    @classmethod
63    def from_jsonl(cls, line: str) -> BusFrame:
64        obj = json.loads(line)
65        direction = obj.get("dir", "in")
66        if direction not in ("in", "out"):
67            raise ValueError(
68                f"invalid direction {direction!r}; must be 'in' or 'out'"
69            )
70        return cls(
71            t=float(obj["t"]),
72            direction=direction,   # type: ignore[arg-type]
73            frame=bytes.fromhex(obj["frame_hex"]),
74        )
Direction = typing.Literal['in', 'out']
def iter_direction( frames: Iterable[BusFrame], direction: Literal['in', 'out']) -> Iterator[BusFrame]: 
 99def iter_direction(
100    frames: Iterable[BusFrame], direction: Direction,
101) -> Iterator[BusFrame]:
102    """Filter capture frames by direction. Handy when a test only
103    cares about one side of the conversation (e.g. "assert every
104    ``in`` frame decodes cleanly")."""
105    for f in frames:
106        if f.direction == direction:
107            yield f

Filter capture frames by direction. Handy when a test only cares about one side of the conversation (e.g. "assert every in frame decodes cleanly").

def read_capture( path: str | pathlib._local.Path) -> list[BusFrame]: 
85def read_capture(path: str | pathlib.Path) -> list[BusFrame]:
86    """Load a capture. Blank lines and ``#``-prefixed comments are
87    silently skipped so hand-authored fixtures can carry notes."""
88    p = pathlib.Path(path)
89    out: list[BusFrame] = []
90    with p.open("r", encoding="utf-8") as f:
91        for raw_line in f:
92            line = raw_line.strip()
93            if not line or line.startswith("#"):
94                continue
95            out.append(BusFrame.from_jsonl(line))
96    return out

Load a capture. Blank lines and #-prefixed comments are silently skipped so hand-authored fixtures can carry notes.

def write_capture( path: str | pathlib._local.Path, frames: Iterable[BusFrame]) -> None: 
77def write_capture(path: str | pathlib.Path, frames: Iterable[BusFrame]) -> None:
78    """Serialize *frames* to *path* as JSONL. Overwrites."""
79    p = pathlib.Path(path)
80    with p.open("w", encoding="utf-8") as f:
81        for frame in frames:
82            f.write(frame.to_jsonl() + "\n")

Serialize frames to path as JSONL. Overwrites.