jmri_ncemonitor.replay

  1"""Capture observed Cab Bus traffic to a file and replay it back.
  2
  3Capture format is JSON Lines — one JSON object per observed message:
  4
  5    {"t": 12345.678, "frame_hex": "806c..."}
  6
  7* ``t`` — monotonic timestamp in seconds (relative to capture start).
  8* ``frame_hex`` — the raw frame bytes (including the address byte) as
  9  hex; replay reconstructs the bytes verbatim.
 10
 11The format is line-oriented + plain-text so it's easy to diff, grep,
 12and edit by hand for test fixtures. :class:`BusReplay` reads such a file
 13and feeds the bytes into a :class:`LoopbackPort` so the same
 14:class:`CabBusMonitor` decode pipeline runs unchanged.
 15"""
 16from __future__ import annotations
 17
 18import asyncio
 19import json
 20import logging
 21import pathlib
 22import time
 23from collections.abc import Iterable
 24from dataclasses import dataclass, field
 25
 26from jmri_ncemonitor.codec import is_address_byte
 27from jmri_ncemonitor.transport import LoopbackPort, loopback_pair
 28
 29_log = logging.getLogger(__name__)
 30
 31
 32@dataclass
 33class CapturedFrame:
 34    """One captured frame: timestamp + raw bytes."""
 35
 36    t: float
 37    frame: bytes
 38
 39    def to_jsonl(self) -> str:
 40        return json.dumps({"t": round(self.t, 6), "frame_hex": self.frame.hex()})
 41
 42    @classmethod
 43    def from_jsonl(cls, line: str) -> CapturedFrame:
 44        obj = json.loads(line)
 45        return cls(t=float(obj["t"]), frame=bytes.fromhex(obj["frame_hex"]))
 46
 47
 48@dataclass
 49class BusCapture:
 50    """Accumulate observed frames; write them out as JSONL on demand."""
 51
 52    frames: list[CapturedFrame] = field(default_factory=list)
 53    _started_at: float = field(default_factory=time.monotonic)
 54
 55    def record(self, frame: bytes) -> None:
 56        """Record one decoded frame (raw bytes including the address byte)."""
 57        self.frames.append(CapturedFrame(
 58            t=time.monotonic() - self._started_at,
 59            frame=frame,
 60        ))
 61
 62    def write(self, path: str | pathlib.Path) -> None:
 63        """Persist the capture as JSONL."""
 64        p = pathlib.Path(path)
 65        with p.open("w", encoding="utf-8") as f:
 66            for cf in self.frames:
 67                f.write(cf.to_jsonl() + "\n")
 68
 69    @classmethod
 70    def read(cls, path: str | pathlib.Path) -> BusCapture:
 71        """Load a previously-written capture."""
 72        p = pathlib.Path(path)
 73        out = cls()
 74        out._started_at = 0.0  # consumed; only frame timestamps matter
 75        with p.open("r", encoding="utf-8") as f:
 76            for raw_line in f:
 77                line = raw_line.strip()
 78                if not line:
 79                    continue
 80                out.frames.append(CapturedFrame.from_jsonl(line))
 81        return out
 82
 83
 84class BusReplay:
 85    """Feed a captured stream into a :class:`LoopbackPort` for testing.
 86
 87    Construct with ``BusReplay(frames)`` (or via :meth:`from_file`); call
 88    :meth:`run` to start a background task that writes frames to the
 89    *write* side of a loopback pair. The *read* side is returned so callers
 90    can hand it to a :class:`CabBusMonitor`.
 91
 92    Timing modes:
 93
 94    * ``preserve_timing=False`` (default) — emit all frames immediately,
 95      one after the other. Good for unit tests.
 96    * ``preserve_timing=True`` — sleep between frames according to the
 97      captured timestamps. Useful when validating real-time behaviour.
 98
 99    The replay always appends a trailing address byte after the last
100    frame so the monitor's "read until next address" loop completes.
101    """
102
103    def __init__(
104        self,
105        frames: Iterable[CapturedFrame],
106        *,
107        preserve_timing: bool = False,
108    ) -> None:
109        self._frames = list(frames)
110        self._preserve_timing = preserve_timing
111        self._task: asyncio.Task[None] | None = None
112
113    @classmethod
114    def from_file(
115        cls, path: str | pathlib.Path, *, preserve_timing: bool = False
116    ) -> BusReplay:
117        capture = BusCapture.read(path)
118        return cls(capture.frames, preserve_timing=preserve_timing)
119
120    async def play_into(self, write_port: LoopbackPort) -> None:
121        """Write every captured frame into *write_port*."""
122        last_t: float | None = None
123        for cf in self._frames:
124            if self._preserve_timing and last_t is not None:
125                delay = cf.t - last_t
126                if delay > 0:
127                    await asyncio.sleep(delay)
128            last_t = cf.t
129            if not cf.frame or not is_address_byte(cf.frame[0]):
130                _log.warning("dropping malformed captured frame: %s", cf.frame.hex())
131                continue
132            await write_port.write(cf.frame)
133        # Tail boundary byte so the monitor's "read until next addr" returns.
134        await write_port.write(bytes([0x80]))
135
136    def start(self, write_port: LoopbackPort) -> asyncio.Task[None]:
137        """Kick off the replay as a background task. Returns the task."""
138        self._task = asyncio.create_task(self.play_into(write_port), name="cabbus-replay")
139        return self._task
140
141
142def make_replay_pair(
143    capture: BusCapture, *, preserve_timing: bool = False
144) -> tuple[LoopbackPort, BusReplay]:
145    """Convenience helper: return ``(monitor_port, replay)``.
146
147    The caller passes ``monitor_port`` to :class:`CabBusMonitor` and calls
148    ``replay.start(write_port)`` where ``write_port`` is the other half of
149    a fresh ``loopback_pair()``.
150    """
151    monitor_port, write_port = loopback_pair()
152    replay = BusReplay(capture.frames, preserve_timing=preserve_timing)
153    # Stash write_port on the replay so the test can drive it.
154    replay.__dict__["_write_port"] = write_port
155    return monitor_port, replay