jmri_ncemonitor

84@dataclass(frozen=True, slots=True)
85class AccessoryObserved(CabBusMessage):
86    """An accessory-decoder command seen on the bus.
87
88    The monitor captures the address that was thrown/closed and which way.
89    """
90    address: int  # always ADDR_ACCESSORY
91    accessory_address: int
92    closed: bool

 95@dataclass(frozen=True, slots=True)
 96class AiuUpdate(CabBusMessage):
 97    """An AIU input update observed on the bus."""
 98    address: int  # always ADDR_AIU_UPDATE
 99    aiu_module: int
100    input_word: int  # 16-bit raw input word

63class BodyOverflowError(RuntimeError):
64    """Raised when a single frame's body exceeds :data:`MAX_BODY_BYTES`.
65
66    The monitor's read loop logs this at WARNING and resyncs to the
67    next address byte on the wire. The raised instance carries the
68    partial buffer so operators can inspect what showed up on the
69    bus when the resync fired.
70    """
71
72    def __init__(self, partial: bytes) -> None:
73        super().__init__(
74            f"cab bus frame exceeded {MAX_BODY_BYTES}-byte cap "
75            f"(no address byte for {len(partial) - 1} data bytes)"
76        )
77        self.partial = partial

49@dataclass
50class BusCapture:
51    """Accumulate observed frames; write them out as JSONL on demand."""
52
53    frames: list[CapturedFrame] = field(default_factory=list)
54    _started_at: float = field(default_factory=time.monotonic)
55
56    def record(self, frame: bytes) -> None:
57        """Record one decoded frame (raw bytes including the address byte)."""
58        self.frames.append(CapturedFrame(
59            t=time.monotonic() - self._started_at,
60            frame=frame,
61        ))
62
63    def write(self, path: str | pathlib.Path) -> None:
64        """Persist the capture as JSONL."""
65        p = pathlib.Path(path)
66        with p.open("w", encoding="utf-8") as f:
67            for cf in self.frames:
68                f.write(cf.to_jsonl() + "\n")
69
70    @classmethod
71    def read(cls, path: str | pathlib.Path) -> BusCapture:
72        """Load a previously-written capture."""
73        p = pathlib.Path(path)
74        out = cls()
75        out._started_at = 0.0  # consumed; only frame timestamps matter
76        with p.open("r", encoding="utf-8") as f:
77            for raw_line in f:
78                line = raw_line.strip()
79                if not line:
80                    continue
81                out.frames.append(CapturedFrame.from_jsonl(line))
82        return out

 85class BusReplay:
 86    """Feed a captured stream into a :class:`LoopbackPort` for testing.
 87
 88    Construct with ``BusReplay(frames)`` (or via :meth:`from_file`); call
 89    :meth:`run` to start a background task that writes frames to the
 90    *write* side of a loopback pair. The *read* side is returned so callers
 91    can hand it to a :class:`CabBusMonitor`.
 92
 93    Timing modes:
 94
 95    * ``preserve_timing=False`` (default) — emit all frames immediately,
 96      one after the other. Good for unit tests.
 97    * ``preserve_timing=True`` — sleep between frames according to the
 98      captured timestamps. Useful when validating real-time behaviour.
 99
100    The replay always appends a trailing address byte after the last
101    frame so the monitor's "read until next address" loop completes.
102    """
103
104    def __init__(
105        self,
106        frames: Iterable[CapturedFrame],
107        *,
108        preserve_timing: bool = False,
109    ) -> None:
110        self._frames = list(frames)
111        self._preserve_timing = preserve_timing
112        self._task: asyncio.Task[None] | None = None
113
114    @classmethod
115    def from_file(
116        cls, path: str | pathlib.Path, *, preserve_timing: bool = False
117    ) -> BusReplay:
118        capture = BusCapture.read(path)
119        return cls(capture.frames, preserve_timing=preserve_timing)
120
121    async def play_into(self, write_port: LoopbackPort) -> None:
122        """Write every captured frame into *write_port*."""
123        last_t: float | None = None
124        for cf in self._frames:
125            if self._preserve_timing and last_t is not None:
126                delay = cf.t - last_t
127                if delay > 0:
128                    await asyncio.sleep(delay)
129            last_t = cf.t
130            if not cf.frame or not is_address_byte(cf.frame[0]):
131                _log.warning("dropping malformed captured frame: %s", cf.frame.hex())
132                continue
133            await write_port.write(cf.frame)
134        # Tail boundary byte so the monitor's "read until next addr" returns.
135        await write_port.write(bytes([0x80]))
136
137    def start(self, write_port: LoopbackPort) -> asyncio.Task[None]:
138        """Kick off the replay as a background task. Returns the task."""
139        self._task = asyncio.create_task(self.play_into(write_port), name="cabbus-replay")
140        return self._task

14class BytePort(ABC):
15    @abstractmethod
16    async def write(self, data: bytes) -> None: ...
17
18    @abstractmethod
19    async def read_byte(self) -> int: ...
20
21    @abstractmethod
22    async def aclose(self) -> None: ...
23
24    @property
25    @abstractmethod
26    def is_open(self) -> bool: ...

22class CabBusMessage:
23    """Marker base for every decoded message."""
24    address: int

27class CabBusMonitor:
28    """Owns one :class:`BytePort` and a background read task.
29
30    Listeners connect to :attr:`inbound` and receive every decoded message
31    in the order it appeared on the bus.
32    """
33
34    def __init__(self, port: BytePort) -> None:
35        self._port = port
36        self._reader = PeekableReader(port.read_byte)
37        self._read_task: asyncio.Task[None] | None = None
38        self._closed = False
39        self.inbound: Signal[CabBusMessage] = Signal()
40
41    async def start(self) -> None:
42        if self._read_task is not None:
43            return
44        self._read_task = asyncio.create_task(self._read_loop(), name="cabbus-monitor")
45
46    async def aclose(self) -> None:
47        self._closed = True
48        if self._read_task is not None:
49            self._read_task.cancel()
50            import contextlib  # noqa: PLC0415
51            with contextlib.suppress(asyncio.CancelledError, Exception):
52                await self._read_task
53            self._read_task = None
54        await self._port.aclose()
55
56    async def _read_loop(self) -> None:
57        try:
58            while not self._closed:
59                try:
60                    frame = await read_message_peekable(self._reader)
61                except asyncio.CancelledError:
62                    return
63                except BodyOverflowError as e:
64                    # A stuck bus buffered past the cap. Log the
65                    # partial for diagnostics and drop it — the next
66                    # read_message call resyncs on the wire.
67                    _log.warning(
68                        "Cab Bus body overflow (%d bytes buffered): %s",
69                        len(e.partial), e.partial.hex(),
70                    )
71                    continue
72                except Exception as e:  # noqa: BLE001
73                    _log.warning("Cab Bus read error: %s", e)
74                    continue
75                try:
76                    msg = decode(frame)
77                except ValueError as e:
78                    _log.warning("Cab Bus bad frame %s: %s", frame.hex(), e)
79                    continue
80                await self.inbound.emit(msg)
81        except asyncio.CancelledError:
82            return

27@dataclass(frozen=True, slots=True)
28class CabBusUnknown(CabBusMessage):
29    """A well-formed frame whose body we don't have a typed decoder for."""
30    address: int
31    data: bytes  # raw bytes (excluding the address marker byte itself)

37@dataclass(frozen=True, slots=True)
38class CabPoll(CabBusMessage):
39    """Master polling cab *address* for its current state.
40
41    Data is the 1-byte poll command (typically 0x00 = "send state").
42    """
43    address: int
44    command: int

47@dataclass(frozen=True, slots=True)
48class CabResponse(CabBusMessage):
49    """Cab *address* responding with its keyboard / encoder state.
50
51    Fields are decoded from the standard 8-byte response payload:
52
53    * ``key`` — last key code pressed (0 if none)
54    * ``speed`` — encoder speed value 0..127
55    * ``direction`` — 0 = reverse, 1 = forward
56    * ``function_low`` — bit-packed F0..F4
57    * ``function_high`` — bit-packed F5..F8
58    """
59    address: int
60    key: int
61    speed: int
62    direction: int
63    function_low: int
64    function_high: int

33@dataclass
34class CapturedFrame:
35    """One captured frame: timestamp + raw bytes."""
36
37    t: float
38    frame: bytes
39
40    def to_jsonl(self) -> str:
41        return json.dumps({"t": round(self.t, 6), "frame_hex": self.frame.hex()})
42
43    @classmethod
44    def from_jsonl(cls, line: str) -> CapturedFrame:
45        obj = json.loads(line)
46        return cls(t=float(obj["t"]), frame=bytes.fromhex(obj["frame_hex"]))

145@dataclass(frozen=True, slots=True)
146class ConsistObserved(CabBusMessage):
147    """A consist add / remove observed on the bus.
148
149    ``operation`` is 0 for add-to-consist, 1 for remove-from-consist,
150    2 for kill-consist. Address bytes carry the consist number
151    (1..127; matches the CV19 wire range).
152    """
153    address: int   # always ADDR_CONSIST_CMD
154    consist_number: int
155    loco_address: int
156    operation: int

76@dataclass(frozen=True, slots=True)
77class FastClockUpdate(CabBusMessage):
78    """Fast-clock packet observed on the bus."""
79    address: int  # always ADDR_FAST_CLOCK
80    hour: int
81    minute: int

10class LoopbackPort(BytePort):
11    def __init__(self, read_q: asyncio.Queue[int], write_q: asyncio.Queue[int]) -> None:
12        self._r = read_q
13        self._w = write_q
14        self._open = True
15
16    async def write(self, data: bytes) -> None:
17        if not self._open:
18            raise RuntimeError("port is closed")
19        for b in data:
20            await self._w.put(b)
21
22    async def read_byte(self) -> int:
23        return await self._r.get()
24
25    async def aclose(self) -> None:
26        self._open = False
27
28    @property
29    def is_open(self) -> bool:
30        return self._open

159@dataclass(frozen=True, slots=True)
160class MacroObserved(CabBusMessage):
161    """A macro trigger observed on the bus.
162
163    Macros are pre-programmed sequences of accessory operations the
164    command station holds in memory. This message captures which
165    macro number was fired.
166    """
167    address: int   # always ADDR_MACRO_CMD
168    macro_number: int

171@dataclass(frozen=True, slots=True)
172class MasterAck(CabBusMessage):
173    """An acknowledgement byte from the master to a cab.
174
175    The master emits these one-byte ACKs after processing a cab
176    command. ``code`` is the raw ack code — non-zero values indicate
177    an error condition specific to whatever the cab last sent.
178    """
179    address: int   # always ADDR_MASTER_ACK
180    code: int

36class NceMonitorConnection(SystemConnection):
37    """Passive NCE Cab Bus monitor. Prefix ``M``, name ``NCE Cab Bus monitor``."""
38
39    name: ClassVar[str] = "NCE Cab Bus monitor"
40    prefix: ClassVar[str] = "M"
41
42    def __init__(self) -> None:
43        super().__init__()
44        self._port: BytePort | None = None
45        self._monitor: CabBusMonitor | None = None
46
47    async def open(self, settings: ConnectionSettings | None = None) -> None:
48        s = settings or {}
49        self._port = await _build_port(s)
50        self._monitor = CabBusMonitor(self._port)
51        await self._monitor.start()
52
53        self.turnouts = PassiveTurnoutManager(self.prefix, self._monitor)   # type: ignore[assignment]
54        self.sensors = PassiveSensorManager(self.prefix, self._monitor)     # type: ignore[assignment]
55        self.power = PassivePowerManager()
56
57        await super().open(settings)
58
59    async def close(self) -> None:
60        if self._monitor is not None:
61            await self._monitor.aclose()
62            self._monitor = None
63        self._port = None
64        await super().close()
65
66    @property
67    def monitor(self) -> CabBusMonitor:
68        if self._monitor is None:
69            raise RuntimeError("connection not open")
70        return self._monitor

35class Parity9BitDecoder(BytePort):
36    """Translate a parity-tagged byte stream into the high-bit-marker form.
37
38    Constructor takes:
39
40    * ``read_tagged`` — awaitable returning ``(byte_value, is_address)``.
41      ``byte_value`` is the 7-bit payload; ``is_address`` indicates
42      whether this byte was received with mark parity (or however the
43      operator's driver signals address-ness).
44    * ``base_port`` — the underlying byte port. ``aclose`` is
45      delegated so lifecycle stays sane; ``read_byte`` is overridden
46      by this class.
47
48    The returned bytes carry the high bit set when ``is_address`` is
49    true, matching :func:`jmri_ncemonitor.codec.is_address_byte`.
50    """
51
52    def __init__(
53        self,
54        read_tagged: Callable[[], Awaitable[tuple[int, bool]]],
55        base_port: BytePort,
56    ) -> None:
57        self._read_tagged = read_tagged
58        self._base = base_port
59
60    async def read_byte(self) -> int:
61        value, is_address = await self._read_tagged()
62        # Force 7-bit payload; some drivers pass through a full byte
63        # that already has bit 7 clear/set — we mask so the semantic
64        # comes only from ``is_address``.
65        value &= 0x7F
66        return (value | 0x80) if is_address else value
67
68    async def write(self, data: bytes) -> None:  # pragma: no cover - passive
69        raise RuntimeError(
70            "NCE Cab Bus monitor is passive — writes are not supported"
71        )
72
73    async def aclose(self) -> None:
74        await self._base.aclose()
75
76    @property
77    def is_open(self) -> bool:
78        return self._base.is_open

144class PassivePowerManager(PowerManager):
145    """Power state is observed-only; ``set_power`` raises."""
146
147    async def set_power(self, state: PowerState) -> None:
148        raise RuntimeError(PASSIVE_REASON)

43class PassiveSensor(Sensor):
44    """Read-only sensor — its known state is observed off the bus."""
45
46    __slots__ = ("_address",)
47
48    def __init__(self, system_name: str, address: int) -> None:
49        super().__init__(system_name)
50        self._address = address
51
52    @property
53    def address(self) -> int:
54        return self._address

57class PassiveSensorManager(Manager[PassiveSensor]):
58    """Sensor manager driven by observed :class:`AiuUpdate` events."""
59
60    type_letter = "S"
61
62    def __init__(self, system_prefix: str, monitor: CabBusMonitor) -> None:
63        super().__init__(system_prefix)
64        self._by_address: dict[int, PassiveSensor] = {}
65        self._unsub = monitor.inbound.connect(self._on_message)
66
67    def _create(self, system_name: str) -> PassiveSensor:
68        address = _parse_address(system_name, self.full_prefix)
69        bean = PassiveSensor(system_name, address)
70        self._by_address[address] = bean
71        return bean
72
73    def get_by_address(self, address: int) -> PassiveSensor | None:
74        return self._by_address.get(address)
75
76    async def _on_message(self, msg: CabBusMessage) -> None:
77        if not isinstance(msg, AiuUpdate):
78            return
79        for bit in range(14):
80            sensor_addr = (msg.aiu_module - 1) * 14 + bit + 1
81            sensor = self._by_address.get(sensor_addr)
82            if sensor is None:
83                continue
84            state = (
85                SensorState.ACTIVE
86                if (msg.input_word & (1 << bit))
87                else SensorState.INACTIVE
88            )
89            await sensor._set_known_state(state)

 95class PassiveTurnout(Turnout):
 96    """Read-only turnout — commanded state cannot be set from this side."""
 97
 98    __slots__ = ("_address",)
 99
100    def __init__(self, system_name: str, address: int) -> None:
101        super().__init__(system_name)
102        self._address = address
103
104    @property
105    def address(self) -> int:
106        return self._address
107
108    async def set_commanded_state(self, state: TurnoutState) -> None:
109        raise RuntimeError(PASSIVE_REASON)

112class PassiveTurnoutManager(Manager[PassiveTurnout]):
113    """Turnout manager driven by observed :class:`AccessoryObserved` events."""
114
115    type_letter = "T"
116
117    def __init__(self, system_prefix: str, monitor: CabBusMonitor) -> None:
118        super().__init__(system_prefix)
119        self._by_address: dict[int, PassiveTurnout] = {}
120        self._unsub = monitor.inbound.connect(self._on_message)
121
122    def _create(self, system_name: str) -> PassiveTurnout:
123        address = _parse_address(system_name, self.full_prefix)
124        bean = PassiveTurnout(system_name, address)
125        self._by_address[address] = bean
126        return bean
127
128    def get_by_address(self, address: int) -> PassiveTurnout | None:
129        return self._by_address.get(address)
130
131    async def _on_message(self, msg: CabBusMessage) -> None:
132        if not isinstance(msg, AccessoryObserved):
133            return
134        bean = self._by_address.get(msg.accessory_address)
135        if bean is None:
136            return
137        state = TurnoutState.CLOSED if msg.closed else TurnoutState.THROWN
138        await bean._set_known_state(state)

131@dataclass(frozen=True, slots=True)
132class ProgrammingResponse(CabBusMessage):
133    """A response to a programming-track read seen on the bus.
134
135    Carries the CV number being read/written and the value byte. The
136    direction of the operation (read vs write) is encoded in the
137    ``operation`` byte the CS sends along.
138    """
139    address: int   # always ADDR_PROG_RESP
140    cv: int
141    value: int
142    operation: int   # 0 = read result, 1 = write ack, 2 = error