jmri_nce.connection

  1"""``NceConnection`` — the SystemConnection subclass for NCE.
  2
  3Construct → ``open(settings)`` → bean managers and the background poller
  4tasks are wired and start running. ``close()`` cancels the pollers and
  5shuts down the traffic controller.
  6
  7Settings:
  8    {
  9        "kind": "loopback" | "serial",
 10        "command_set": "PH-Pro 2006+" | "PowerCab" | "Smart Booster SB5",
 11        "aiu_addresses": [1, 2, 3],     # AIUs to poll (default: empty)
 12        # for serial:
 13        "port": "COM7",
 14        "baud": 9600,
 15        # for loopback:
 16        "loopback_port": LoopbackPort,
 17    }
 18"""
 19from __future__ import annotations
 20
 21import asyncio
 22import logging
 23import time
 24from typing import Any, ClassVar
 25
 26from jmri_core.beans.fast_clock import HardwareClockState
 27from jmri_core.hal.connection import ConnectionSettings, SystemConnection
 28
 29from jmri_nce import messages as nce_messages
 30from jmri_nce.beans import NcePowerManager, NceSensorManager, NceTurnoutManager
 31from jmri_nce.command_set import COMMAND_SETS, NCE_PH_PRO_2006, NceCommandSet
 32from jmri_nce.detect import detect_command_set
 33from jmri_nce.monitors import NceAiuPoller, NceTurnoutMonitor
 34from jmri_nce.programmer import NceProgrammer, NceProgrammerManager
 35from jmri_nce.throttle import NceThrottleManager
 36from jmri_nce.traffic import NceTrafficController
 37from jmri_nce.transport import BytePort, LoopbackPort, SerialPort
 38
 39_log = logging.getLogger(__name__)
 40
 41
 42class NceConnection(SystemConnection):
 43    """NCE hardware connection. Prefix ``N``, name ``NCE``."""
 44
 45    name: ClassVar[str] = "NCE"
 46    prefix: ClassVar[str] = "N"
 47
 48    def __init__(self) -> None:
 49        super().__init__()
 50        self._port: BytePort | None = None
 51        self._tc: NceTrafficController | None = None
 52        self._cmd_set: NceCommandSet | None = None
 53        self._monitor: NceTurnoutMonitor | None = None
 54        self._aiu_poller: NceAiuPoller | None = None
 55        self._programmer: NceProgrammer | None = None
 56        self._tasks: list[asyncio.Task[None]] = []
 57        # Fast-clock running state — inferred from tick advancement across
 58        # consecutive reads (JMRI does the same). See ``read_hardware_clock``.
 59        self._last_clock_seconds: int | None = None
 60        self._last_clock_running: bool = False
 61
 62    # ----- lifecycle --------------------------------------------------------
 63
 64    async def open(self, settings: ConnectionSettings | None = None) -> None:
 65        s = settings or {}
 66        explicit = s.get("command_set")
 67        # Convenience: set ``JMRI_NCE_DEBUG=1`` to turn on wire-level
 68        # DEBUG for the ``jmri_nce`` logger without having to write a
 69        # separate uvicorn --log-config file. Uvicorn's own --log-level
 70        # only affects uvicorn's loggers; our module loggers inherit
 71        # from root, so we need to boost the module logger explicitly.
 72        # Go through :mod:`jmri_core.env` so the truthy/whitespace
 73        # semantics match every other JMRI_* var.
 74        from jmri_core.env import nce_debug  # noqa: PLC0415
 75        if nce_debug():
 76            _nce_root = logging.getLogger("jmri_nce")
 77            _nce_root.setLevel(logging.DEBUG)
 78            # Make sure there's a handler that will actually print the
 79            # DEBUG records. Uvicorn installs a root-level handler at
 80            # startup that emits INFO+; adding a StreamHandler here
 81            # ensures our DEBUG lines aren't swallowed.
 82            if not any(
 83                isinstance(h, logging.StreamHandler)
 84                for h in _nce_root.handlers
 85            ):
 86                h = logging.StreamHandler()
 87                h.setLevel(logging.DEBUG)
 88                h.setFormatter(logging.Formatter(
 89                    "%(levelname)s:%(name)s:%(message)s",
 90                ))
 91                _nce_root.addHandler(h)
 92                _nce_root.propagate = False
 93            _log.warning("jmri_nce DEBUG logging enabled via JMRI_NCE_DEBUG=1")
 94        _log.warning(
 95            "NCE connection opening: kind=%s port=%s command_set=%s",
 96            s.get("kind"), s.get("port"), explicit or "(default)",
 97        )
 98        self._port = await _build_port(s)
 99        self._tc = NceTrafficController(self._port)
100
101        # Command-set selection:
102        #
103        # 1. Explicit ``command_set`` setting wins — production users should
104        #    just specify which CS they have.
105        # 2. ``auto_detect`` is OPT-IN (default False). The version probe
106        #    opcode isn't verified across all firmwares; on some
107        #    PowerCab/PowerPro variants it leaves the CS waiting for
108        #    additional argument bytes, wedging the serial bus until the CS
109        #    is power-cycled. Don't enable auto-detect on production hardware
110        #    until the opcode is verified against your specific firmware.
111        # 3. Otherwise default to PH-Pro 2006+ (works on the largest set of
112        #    deployed NCE command stations).
113        if explicit and explicit in COMMAND_SETS:
114            self._cmd_set = COMMAND_SETS[explicit]
115        elif s.get("auto_detect", False):
116            self._cmd_set = await detect_command_set(self._tc)
117        else:
118            self._cmd_set = NCE_PH_PRO_2006
119
120        # Replace the base in-memory managers with NCE-aware ones.
121        nce_turnouts = NceTurnoutManager(self.prefix, tc=self._tc)
122        nce_sensors = NceSensorManager(self.prefix)
123        self.turnouts = nce_turnouts                      # type: ignore[assignment]
124        self.sensors = nce_sensors                        # type: ignore[assignment]
125        self.throttles = NceThrottleManager(self.prefix, tc=self._tc)
126        self.power = NcePowerManager(self._tc)
127        self._programmer = NceProgrammer(self._tc)
128        self.programmers = NceProgrammerManager(self._programmer)
129        # Replace the base placeholder ``ConsistManager`` (built in
130        # ``SystemConnection.__init__`` before drivers had a chance to
131        # install hardware-aware managers) with the NCE-aware one. This
132        # brings CS-managed (``command_station``) consists to life via
133        # 0xA2 sub-commands; software and advanced/CV19 consists still
134        # work because the subclass falls through to the base for those
135        # kinds. The rebind hooks used elsewhere aren't enough here —
136        # the fan-out overrides only live on ``NceConsistManager``.
137        from jmri_nce.consist import NceConsistManager  # noqa: PLC0415
138        self.consists = NceConsistManager(
139            self.throttles, self.prefix, self.programmers, tc=self._tc,
140        )
141
142        # Start polling tasks (turnout monitor + optional AIU poller).
143        # The turnout monitor can be opted out via ``disable_turnout_monitor``
144        # — useful when the memory base for a CS firmware variant isn't
145        # known yet, or when the poll is failing for some other reason and
146        # the user just wants the rest of the driver to function.
147        if not s.get("disable_turnout_monitor", False):
148            self._monitor = NceTurnoutMonitor(self._tc, nce_turnouts, self._cmd_set)
149            self._tasks.append(
150                asyncio.create_task(self._monitor.run(), name="nce-turnout-monitor"),
151            )
152
153        aius: list[int] = list(s.get("aiu_addresses", []) or [])
154        if aius and self._cmd_set.aiu_polling_supported:
155            self._aiu_poller = NceAiuPoller(self._tc, nce_sensors, aius)
156            self._tasks.append(asyncio.create_task(self._aiu_poller.run(), name="nce-aiu-poller"))
157
158        # CS reachability monitor. Polls the traffic controller's
159        # ``last_success_at`` timestamp once a second; if we haven't
160        # heard a completed round-trip in more than 3 seconds, mark
161        # the CS unreachable — the turnout monitor (1.5s cadence) or
162        # AIU poller (250ms) would have brought that number forward if
163        # the CS were still answering. Any fresh success flips it
164        # back on immediately.
165        #
166        # Start pessimistic: the base class defaults ``_cs_reachable``
167        # to True (matches ``is_open`` for drivers with no real
168        # detection), but because we DO have detection, we don't yet
169        # know if the CS is answering. Home-page renders in the first
170        # ~3 s would otherwise show "On" even against a dark CS; the
171        # monitor's first True-transition flips the badge live via
172        # the existing WebSocket push once we hear from the CS.
173        self._cs_reachable = False
174        self._tasks.append(asyncio.create_task(
175            self._cs_reachability_loop(), name="nce-cs-reachability",
176        ))
177
178        await super().open(settings)
179
180    async def _cs_reachability_loop(self) -> None:
181        """Flip ``_cs_reachable`` on the 3-second silence rule.
182
183        Runs until cancelled by :meth:`close`. Uses ``time.monotonic()``
184        to avoid wall-clock jumps confusing the window.
185        """
186        import contextlib  # noqa: PLC0415
187        try:
188            # Give the CS 3 seconds of grace on open — the first
189            # turnout-monitor poll or AIU read is what generates our
190            # first ``last_success_at`` value, and it can take a
191            # couple of seconds on a slow serial adapter.
192            await asyncio.sleep(3.0)
193            while True:
194                await asyncio.sleep(1.0)
195                if self._tc is None:
196                    return
197                last = self._tc.last_success_at
198                if last is None:
199                    # No successful round-trip has happened yet. Treat
200                    # as "not reachable" so the UI doesn't lie during
201                    # a stuck-open serial hang.
202                    await self._set_cs_reachable(False)
203                    continue
204                age = time.monotonic() - last
205                await self._set_cs_reachable(age <= 3.0)
206        except asyncio.CancelledError:
207            with contextlib.suppress(Exception):
208                return
209
210    async def close(self) -> None:
211        if self._monitor is not None:
212            self._monitor.stop()
213        if self._aiu_poller is not None:
214            self._aiu_poller.stop()
215        for t in self._tasks:
216            t.cancel()
217        import contextlib  # noqa: PLC0415
218        for t in self._tasks:
219            with contextlib.suppress(asyncio.CancelledError, Exception):
220                await t
221        self._tasks.clear()
222        if self._tc is not None:
223            await self._tc.aclose()
224            self._tc = None
225        self._port = None
226        await super().close()
227
228    # ----- accessors --------------------------------------------------------
229
230    @property
231    def command_set(self) -> NceCommandSet:
232        if self._cmd_set is None:
233            raise RuntimeError("connection not open")
234        return self._cmd_set
235
236    @property
237    def traffic(self) -> NceTrafficController:
238        if self._tc is None:
239            raise RuntimeError("connection not open")
240        return self._tc
241
242    @property
243    def programmer(self) -> NceProgrammer:
244        if self._programmer is None:
245            raise RuntimeError("connection not open")
246        return self._programmer
247
248    # ----- hardware clock ---------------------------------------------------
249
250    async def read_hardware_clock(self) -> HardwareClockState:
251        """Read the CS's fast-clock state.
252
253        Talks to command-station memory at ``CS_CLOCK_ADDR`` and decodes
254        the 16-byte reply into a :class:`HardwareClockState`. In 12h mode
255        the CS reports hours 1..12 with an AM/PM flag; we normalise to
256        24-hour before returning so the JSON/web layer never has to care.
257        """
258        cmd = nce_messages.read_clock()
259        reply = await self.traffic.send_command(cmd)
260        data = reply.data
261        if _log.isEnabledFor(logging.DEBUG):
262            _log.debug(
263                "NCE clock read bytes: %s",
264                " ".join(f"{b:02X}" for b in data),
265            )
266        scale = data[0]
267        seconds = data[2] & 0xFF
268        minutes = data[3] & 0xFF
269        hours_raw = data[4] & 0xFF
270        ampm_flag = data[5]              # ASCII 'A' (65) = AM, else PM
271        twenty_four = (data[6] & 0xFF) == 1
272        # In 12h mode, hours cycle 12,1..11,12,1..11 with AM flipping to PM
273        # at each 12. Normalise: AM 12 → 0, PM 12 → 12, PM 1..11 → 13..23.
274        if twenty_four:
275            hours = hours_raw % 24
276        else:
277            is_am = ampm_flag == ord("A")
278            h = hours_raw % 12               # 12 → 0
279            hours = h if is_am else h + 12
280        # Scale of 0 would divide-by-zero; treat as "unknown, don't touch".
281        rate = float(250 // scale) if scale else 1.0
282        # The CS's 16-byte poll has no clean "running" bit — byte [1] is a
283        # scale-prescaler tick that stays nonzero even when stopped, so
284        # decoding it as a run flag reports running=True for a stopped
285        # clock. JMRI's ``NceClockControl`` infers running by watching
286        # tick advancement across reads instead; we do the same:
287        # if this read's seconds byte differs from the last one, the CS
288        # advanced → running. If identical, → stopped. On the very first
289        # read we have no history and fall back to whatever the last
290        # explicit start/stop button intent was.
291        if self._last_clock_seconds is None:
292            running = self._last_clock_running
293        else:
294            running = seconds != self._last_clock_seconds
295        self._last_clock_seconds = seconds
296        self._last_clock_running = running
297        return HardwareClockState(
298            hours=hours,
299            minutes=minutes,
300            seconds=seconds,
301            rate=rate,
302            running=running,
303            twenty_four_hour=twenty_four,
304        )
305
306    async def set_hardware_clock_time(
307        self, hours: int, minutes: int, seconds: int = 0,
308    ) -> None:
309        """Write hh:mm:ss to the CS clock."""
310        await self.traffic.send_command(
311            nce_messages.write_clock_time(hours, minutes, seconds),
312        )
313
314    async def set_hardware_clock_rate(self, rate: float) -> None:
315        """Set the CS fast-clock ratio. NCE only stores integer ratios 1..15;
316        we round and clamp to that range so the user's float input doesn't
317        drop a command."""
318        r = max(1, min(15, int(round(rate))))
319        await self.traffic.send_command(nce_messages.set_clock_ratio(r))
320
321    async def start_hardware_clock(self) -> None:
322        await self.traffic.send_command(nce_messages.start_clock())
323        # Mark the run intent so the *next* read (which may sample the
324        # seconds byte before the CS has ticked) doesn't wrongly report
325        # stopped. Clear the last-seconds cache so tick-advancement
326        # inference restarts on the second post-start read.
327        self._last_clock_running = True
328        self._last_clock_seconds = None
329
330    async def stop_hardware_clock(self) -> None:
331        await self.traffic.send_command(nce_messages.stop_clock())
332        # Explicit stop overrides tick-advancement inference for the next
333        # read; the last-seconds cache stays valid because a stopped clock
334        # will read the same seconds byte again.
335        self._last_clock_running = False
336
337
338# ----- transport construction ----------------------------------------------
339
340
341async def _build_port(settings: dict[str, Any]) -> BytePort:
342    kind = settings.get("kind", "loopback")
343    if kind == "serial":
344        port = settings.get("port")
345        if not port:
346            raise ValueError("serial transport requires 'port'")
347        baud = int(settings.get("baud", 9600))
348        return await SerialPort.open(port=port, baud=baud)
349    if kind == "loopback":
350        loop_port = settings.get("loopback_port")
351        if not isinstance(loop_port, LoopbackPort):
352            raise ValueError(
353                "loopback transport requires 'loopback_port' to be a LoopbackPort"
354            )
355        return loop_port
356    raise ValueError(f"unknown NCE transport kind: {kind!r}")