jmri_nce.traffic

  1"""``NceTrafficController`` — sync request/response over a byte stream.
  2
  3NCE is single-master: the PC sends a command, the command station replies
  4with ``reply_len`` data bytes followed by one ACK byte. Multiple commands
  5are serialized through one :class:`asyncio.Lock` because the command station
  6processes them one at a time.
  7
  8This is the structural opposite of :mod:`jmri_loconet.traffic`, which has
  9no concept of "in flight" — every LocoNet frame is broadcast and any
 10listener may decode it. The :class:`SystemConnection` ABC abstracts both
 11shapes without forcing either onto the other; that this driver dropped in
 12without changes to ``jmri-core`` is the validation that the abstraction
 13holds.
 14"""
 15from __future__ import annotations
 16
 17import asyncio
 18import logging
 19import time
 20
 21from jmri_nce.messages import NceCommand, NceCommandError, NceReply
 22from jmri_nce.opcodes import ACK_ERROR
 23from jmri_nce.transport.port import BytePort
 24
 25_log = logging.getLogger(__name__)
 26
 27
 28class NceTrafficController:
 29    """One command in flight at a time. All sends serialize through ``_lock``."""
 30
 31    # Fast-fail behaviour: once the CS has been silent for more than
 32    # ``SILENCE_THRESHOLD_S`` (matching the reachability monitor's
 33    # window plus a small margin so the two don't fight for control),
 34    # ``send_command`` refuses to touch the serial port and raises
 35    # ``TimeoutError`` immediately. Otherwise the browser's 1-Hz
 36    # fast-clock polls + the turnout monitor's 1.5 s poll cadence
 37    # queue up behind the ``_lock`` and each pays the full 500 ms
 38    # timeout — a home-page render or profile-load can stall for
 39    # 5–10 seconds even though we already know the CS is off.
 40    #
 41    # Every ``PROBE_INTERVAL_S`` one command IS let through to probe
 42    # whether the CS has come back. On success the timestamp gets
 43    # updated, silence drops below threshold, and normal operation
 44    # resumes.
 45    SILENCE_THRESHOLD_S = 4.0
 46    PROBE_INTERVAL_S = 5.0
 47
 48    def __init__(self, port: BytePort) -> None:
 49        self._port = port
 50        self._lock = asyncio.Lock()
 51        self._closed = False
 52        # Timestamp (``time.monotonic()`` scale) of the most recent
 53        # completed round-trip with the CS. ``None`` before any
 54        # command has been sent. Read by :class:`NceConnection`'s
 55        # reachability monitor to flip the CS-power state; write-only
 56        # here — updates on every success in :meth:`send_command` and
 57        # in :meth:`consume_trailer` when the trailer arrives.
 58        self.last_success_at: float | None = None
 59        # Fallback reference for the fast-fail silence calculation.
 60        # If the CS has been off since server start, ``last_success_at``
 61        # is always ``None`` and would defeat the short-circuit —
 62        # every command would still pay its own serial timeout. Using
 63        # ``max(start_time, last_success_at)`` means an always-off CS
 64        # is detected the moment startup grace elapses, and a CS that
 65        # went off after some traffic is measured from its last reply.
 66        self._start_time: float = time.monotonic()
 67        # Next moment at which a probe attempt is allowed while the CS
 68        # is known silent. See the class-level fast-fail comment.
 69        self._next_probe_at: float | None = None
 70
 71    async def aclose(self) -> None:
 72        self._closed = True
 73        await self._port.aclose()
 74
 75    async def send_command(self, cmd: NceCommand) -> NceReply:
 76        """Write *cmd*, await its reply with timeout, return the reply data.
 77
 78        Raises:
 79          - :class:`asyncio.TimeoutError` if the command station doesn't
 80            answer within ``cmd.timeout_s``;
 81          - :class:`NceCommandError` if the ACK byte indicates failure.
 82        """
 83        if self._closed:
 84            raise RuntimeError("traffic controller is closed")
 85        # Fast-fail if the CS has been silent long enough that the
 86        # reachability monitor has already flipped it to unreachable.
 87        # Without this, a browser page-load can queue ~10 commands (fast
 88        # clock + turnout monitor + user actions) that each pay 500 ms
 89        # of serial timeout in series — 5–9 s of hang on GET / and
 90        # POST /profiles/load. Let one probe through every
 91        # ``PROBE_INTERVAL_S`` so recovery is detected.
 92        #
 93        # ``reference`` is the later of first-open and last successful
 94        # round-trip. If the CS has been off since startup,
 95        # ``last_success_at`` is ``None`` — measuring silence from
 96        # ``_start_time`` still trips the short-circuit once startup
 97        # grace elapses.
 98        now = time.monotonic()
 99        reference = (
100            self.last_success_at
101            if self.last_success_at is not None
102            else self._start_time
103        )
104        silence = now - reference
105        if silence > self.SILENCE_THRESHOLD_S:
106            if self._next_probe_at is not None and now < self._next_probe_at:
107                raise TimeoutError(
108                    f"NCE CS silent for {silence:.1f}s; "
109                    f"{cmd.name} short-circuited"
110                )
111            # Let this one through as a probe; re-arm for the next
112            # window so only one attempt hits serial per interval.
113            self._next_probe_at = now + self.PROBE_INTERVAL_S
114        async with self._lock:
115            # Drain any bytes the CS left in the read buffer before the
116            # request. Without this, leftover bytes — CS startup chatter,
117            # or extras the CS sends after a documented reply — are read
118            # as the *current* command's ACK on the next send. (Action
119            # commands then fail with random non-``!`` ACKs.)
120            await self._drain_pending()
121            _log.debug("NCE send %s: %s", cmd.name, cmd.request.hex())
122            await self._port.write(cmd.request)
123            # Read commands return exactly ``reply_len`` data bytes (no ACK).
124            # Action commands return ``reply_len`` data bytes + 1 ACK byte.
125            n = cmd.reply_len if cmd.is_read else cmd.reply_len + 1
126            try:
127                full = await asyncio.wait_for(
128                    self._port.read_exact(n),
129                    timeout=cmd.timeout_s,
130                )
131            except TimeoutError:
132                _log.warning("NCE command %s timed out", cmd.name)
133                raise
134            _log.debug("NCE recv %s: %s", cmd.name, full.hex())
135            # Any completed round-trip counts as evidence the CS is on.
136            self.last_success_at = time.monotonic()
137            if cmd.is_read:
138                return NceReply(command=cmd, data=full)
139            data = full[: cmd.reply_len]
140            ack = full[cmd.reply_len : cmd.reply_len + 1]
141            # NCE's reply byte semantics vary by command and firmware: some
142            # commands return ``!`` (ACK_OK), others echo the opcode or
143            # respond with a status byte that's command-specific. JMRI Java
144            # treats action commands as fire-and-forget (null listener,
145            # no value check). The one signal we *do* care about is the
146            # explicit-NACK byte ``b'0'`` (ACK_ERROR) — anything else is
147            # accepted as success.
148            if ack == ACK_ERROR:
149                raise NceCommandError(cmd, ack, data)
150            return NceReply(command=cmd, data=data)
151
152    async def _drain_pending(self) -> None:
153        """Discard any bytes already sitting in the read buffer.
154
155        Non-blocking: returns immediately if the buffer is empty. Avoids
156        the ~5ms-per-command timeout penalty of a polling loop.
157        """
158        stray = await self._port.read_pending()
159        if stray:
160            _log.debug("NCE drained %d stale byte(s): %s", len(stray), stray.hex())
161
162    async def consume_trailer(self, max_wait_s: float = 0.1) -> bytes:
163        """Wait briefly for a trailing byte and consume it if it arrives.
164
165        NCE PowerPro firmware quirk: some commands (notably
166        RD_CV_DIRECT on successful decoder ack) send TWO bytes back —
167        the payload byte plus a trailing ``!`` ACK. Our command spec
168        for RD_CV_DIRECT expects ``reply_len=1``, so the second byte
169        lands in the serial buffer. Usually the next command's drain
170        catches it, but occasionally it arrives *after* the drain
171        window closes but *before* the next command's expected reply
172        — corrupting the reply as though we'd shifted by one byte.
173
174        Callers who know their command has an optional trailer should
175        call this immediately after :meth:`send_command`. Holds no
176        lock (the caller's ``send_command`` call already released it),
177        which is fine because we're only READING from the port, not
178        writing — a concurrent write can't interleave.
179
180        Returns the trailer byte if seen, empty bytes otherwise.
181        """
182        try:
183            trailer = await asyncio.wait_for(
184                self._port.read_exact(1),
185                timeout=max_wait_s,
186            )
187        except TimeoutError:
188            return b""
189        _log.debug("NCE consumed trailer byte: %02x", trailer[0])
190        return trailer