jmri_nce.monitors

  1"""Polling tasks that turn NCE's request/response wire into push-style
  2state-change signals.
  3
  4NCE doesn't push state changes from the layout — clients must poll. This
  5module owns two background tasks:
  6
  7* :class:`NceTurnoutMonitor` reads the command-station's turnout-status
  8  memory in 16-byte chunks every ~1.5s, diffs against the previous
  9  snapshot, and emits state changes through any registered turnout.
 10* :class:`NceAiuPoller` reads each known AIU module's input word at a
 11  faster cadence (~250ms) to catch occupancy changes promptly.
 12
 13Both tasks run inside the connection's lifetime — started in
 14:meth:`NceConnection.open` and cancelled in :meth:`NceConnection.close`.
 15"""
 16from __future__ import annotations
 17
 18import asyncio
 19import logging
 20from typing import TYPE_CHECKING
 21
 22from jmri_core.state import SensorState, TurnoutState
 23
 24from jmri_nce.messages import read_accessory_memory, read_aiu
 25
 26if TYPE_CHECKING:
 27    from jmri_nce.beans import NceSensorManager, NceTurnoutManager
 28    from jmri_nce.command_set import NceCommandSet
 29    from jmri_nce.traffic import NceTrafficController
 30
 31_log = logging.getLogger(__name__)
 32
 33
 34class NceTurnoutMonitor:
 35    """Periodically poll the command-station turnout-status memory."""
 36
 37    POLL_INTERVAL_S = 1.5
 38    CHUNK_BYTES = 16          # 16 bytes × 8 turnouts/byte = 128 per chunk
 39
 40    def __init__(
 41        self,
 42        tc: NceTrafficController,
 43        mgr: NceTurnoutManager,
 44        cmd_set: NceCommandSet,
 45    ) -> None:
 46        self._tc = tc
 47        self._mgr = mgr
 48        self._cs = cmd_set
 49        # Pre-allocated snapshot buffer: the bytes we most recently read.
 50        self._snapshot = bytearray(cmd_set.turnout_table_bytes)
 51        self._stop = asyncio.Event()
 52
 53    async def run(self) -> None:
 54        """Run forever until stopped via :meth:`stop`."""
 55        import contextlib  # noqa: PLC0415
 56        try:
 57            while not self._stop.is_set():
 58                await self._poll_once()
 59                with contextlib.suppress(TimeoutError):
 60                    await asyncio.wait_for(
 61                        self._stop.wait(), timeout=self.POLL_INTERVAL_S
 62                    )
 63        except asyncio.CancelledError:
 64            return
 65
 66    def stop(self) -> None:
 67        self._stop.set()
 68
 69    async def _poll_once(self) -> None:
 70        for offset in range(0, self._cs.turnout_table_bytes, self.CHUNK_BYTES):
 71            addr = self._cs.turnout_memory_base + offset
 72            try:
 73                reply = await self._tc.send_command(read_accessory_memory(addr))
 74            except (TimeoutError, Exception) as e:  # noqa: BLE001
 75                _log.warning("NCE turnout-monitor poll failed at %#06x: %s", addr, e)
 76                return
 77            await self._diff_and_emit(offset, reply.data)
 78
 79    async def _diff_and_emit(self, offset: int, fresh: bytes) -> None:
 80        """Compare *fresh* against the snapshot at *offset*; emit per flipped bit."""
 81        old_chunk = self._snapshot[offset : offset + len(fresh)]
 82        for i, (old, new) in enumerate(zip(old_chunk, fresh, strict=False)):
 83            if old == new:
 84                continue
 85            changed = old ^ new
 86            for bit in range(8):
 87                if not (changed & (1 << bit)):
 88                    continue
 89                address = (offset + i) * 8 + bit + 1
 90                turnout = self._mgr.get_by_address(address)
 91                if turnout is None:
 92                    continue
 93                state = TurnoutState.CLOSED if (new & (1 << bit)) else TurnoutState.THROWN
 94                await turnout._set_known_state(state)
 95        self._snapshot[offset : offset + len(fresh)] = fresh
 96
 97
 98class NceAiuPoller:
 99    """Periodically poll the configured AIU modules and emit sensor state."""
100
101    POLL_INTERVAL_S = 0.25
102
103    def __init__(
104        self,
105        tc: NceTrafficController,
106        mgr: NceSensorManager,
107        aiu_addresses: list[int],
108    ) -> None:
109        self._tc = tc
110        self._mgr = mgr
111        self._aius = list(aiu_addresses)
112        self._snapshots: dict[int, int] = {}  # AIU address → last raw 16-bit word
113        self._stop = asyncio.Event()
114
115    async def run(self) -> None:
116        import contextlib  # noqa: PLC0415
117        try:
118            while not self._stop.is_set():
119                for aiu in self._aius:
120                    await self._poll_one(aiu)
121                with contextlib.suppress(TimeoutError):
122                    await asyncio.wait_for(
123                        self._stop.wait(), timeout=self.POLL_INTERVAL_S
124                    )
125        except asyncio.CancelledError:
126            return
127
128    def stop(self) -> None:
129        self._stop.set()
130
131    async def _poll_one(self, aiu: int) -> None:
132        try:
133            reply = await self._tc.send_command(read_aiu(aiu))
134        except (TimeoutError, Exception) as e:  # noqa: BLE001
135            _log.warning("NCE AIU-poll failed for %d: %s", aiu, e)
136            return
137        if len(reply.data) < 2:
138            return
139        word = (reply.data[0] << 8) | reply.data[1]
140        prev = self._snapshots.get(aiu, ~word & 0xFFFF)  # force diff first time
141        changed = prev ^ word
142        for bit in range(14):                 # 14 inputs per AIU
143            if not (changed & (1 << bit)):
144                continue
145            sensor_address = (aiu - 1) * 14 + bit + 1
146            sensor = self._mgr.get_by_address(sensor_address)
147            if sensor is None:
148                continue
149            state = SensorState.ACTIVE if (word & (1 << bit)) else SensorState.INACTIVE
150            await sensor._set_known_state(state)
151        self._snapshots[aiu] = word