jmri_nce

12class BytePort(ABC):
13    """Bidirectional byte-stream port for NCE."""
14
15    @abstractmethod
16    async def write(self, data: bytes) -> None: ...
17
18    @abstractmethod
19    async def read_exact(self, n: int) -> bytes: ...
20
21    @abstractmethod
22    async def read_pending(self) -> bytes:
23        """Read any bytes already buffered, without waiting for more.
24
25        Used by the traffic controller to drain stale bytes between
26        commands. Returns ``b""`` if nothing is currently buffered.
27        """
28
29    @abstractmethod
30    async def aclose(self) -> None: ...
31
32    @property
33    @abstractmethod
34    def is_open(self) -> bool: ...

16class LoopbackPort(BytePort):
17    """One half of a connected pair."""
18
19    def __init__(self, read_q: asyncio.Queue[int], write_q: asyncio.Queue[int]) -> None:
20        self._r = read_q
21        self._w = write_q
22        self._open = True
23
24    async def write(self, data: bytes) -> None:
25        if not self._open:
26            raise RuntimeError("port is closed")
27        for b in data:
28            await self._w.put(b)
29
30    async def read_exact(self, n: int) -> bytes:
31        out = bytearray()
32        for _ in range(n):
33            out.append(await self._r.get())
34        return bytes(out)
35
36    async def read_pending(self) -> bytes:
37        """Drain anything currently sitting in the receive queue."""
38        out = bytearray()
39        while not self._r.empty():
40            out.append(self._r.get_nowait())
41        return bytes(out)
42
43    async def aclose(self) -> None:
44        self._open = False
45
46    @property
47    def is_open(self) -> bool:
48        return self._open

 99class NceAiuPoller:
100    """Periodically poll the configured AIU modules and emit sensor state."""
101
102    POLL_INTERVAL_S = 0.25
103
104    def __init__(
105        self,
106        tc: NceTrafficController,
107        mgr: NceSensorManager,
108        aiu_addresses: list[int],
109    ) -> None:
110        self._tc = tc
111        self._mgr = mgr
112        self._aius = list(aiu_addresses)
113        self._snapshots: dict[int, int] = {}  # AIU address → last raw 16-bit word
114        self._stop = asyncio.Event()
115
116    async def run(self) -> None:
117        import contextlib  # noqa: PLC0415
118        try:
119            while not self._stop.is_set():
120                for aiu in self._aius:
121                    await self._poll_one(aiu)
122                with contextlib.suppress(TimeoutError):
123                    await asyncio.wait_for(
124                        self._stop.wait(), timeout=self.POLL_INTERVAL_S
125                    )
126        except asyncio.CancelledError:
127            return
128
129    def stop(self) -> None:
130        self._stop.set()
131
132    async def _poll_one(self, aiu: int) -> None:
133        try:
134            reply = await self._tc.send_command(read_aiu(aiu))
135        except (TimeoutError, Exception) as e:  # noqa: BLE001
136            _log.warning("NCE AIU-poll failed for %d: %s", aiu, e)
137            return
138        if len(reply.data) < 2:
139            return
140        word = (reply.data[0] << 8) | reply.data[1]
141        prev = self._snapshots.get(aiu, ~word & 0xFFFF)  # force diff first time
142        changed = prev ^ word
143        for bit in range(14):                 # 14 inputs per AIU
144            if not (changed & (1 << bit)):
145                continue
146            sensor_address = (aiu - 1) * 14 + bit + 1
147            sensor = self._mgr.get_by_address(sensor_address)
148            if sensor is None:
149                continue
150            state = SensorState.ACTIVE if (word & (1 << bit)) else SensorState.INACTIVE
151            await sensor._set_known_state(state)
152        self._snapshots[aiu] = word

42@dataclass(frozen=True, slots=True)
43class NceCommand:
44    """One outbound request.
45
46    NCE has two distinct reply shapes:
47
48    * **Action commands** (power on/off, accessory throw, write CV, ...)
49      reply with a single ACK byte (``!`` on success). ``reply_len`` is
50      the number of *data* bytes that precede the ACK — usually 0.
51    * **Read commands** (read memory, read AIU, ...) reply with exactly
52      *reply_len* bytes of raw data, no trailing ACK. Set ``is_read=True``
53      and the traffic controller will not consume an ACK byte after.
54    """
55
56    name: str
57    request: bytes
58    reply_len: int = 0
59    timeout_s: float = 0.5
60    is_read: bool = False

71class NceCommandError(Exception):
72    """The command station replied with a non-OK ACK byte."""
73
74    def __init__(self, command: NceCommand, ack: bytes, data: bytes) -> None:
75        super().__init__(
76            f"NCE command {command.name!r} failed: ack={ack!r} data={data!r}"
77        )
78        self.command = command
79        self.ack = ack
80        self.data = data

23@dataclass(frozen=True, slots=True)
24class NceCommandSet:
25    """Per-firmware constants used by services."""
26
27    name: str
28    turnout_memory_base: int    # base of accessory/turnout status table
29    turnout_table_bytes: int    # 8 turnouts per byte (256 bytes → 2048 turnouts)
30    macro_memory_base: int      # base of macro table
31    consist_memory_base: int    # base of consist table
32    supports_ops_mode_prog: bool
33    aiu_polling_supported: bool

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

119class NcePowerManager(PowerManager):
120    """Send OP_POWER_ON / OP_POWER_OFF; state is set on successful reply."""
121
122    def __init__(self, tc: NceTrafficController) -> None:
123        super().__init__()
124        self._tc = tc
125
126    async def set_power(self, state: PowerState) -> None:
127        if state == PowerState.ON:
128            await self._tc.send_command(power_on())
129        elif state == PowerState.OFF:
130            await self._tc.send_command(power_off())
131        # NCE has no IDLE wire command; ignore other targets here.
132        await self._set_state(state)

 47class NceProgrammer:
 48    """CV access against an NCE command station."""
 49
 50    def __init__(self, tc: NceTrafficController) -> None:
 51        self._tc = tc
 52        self._programming_in_progress = asyncio.Event()
 53
 54    @property
 55    def is_programming(self) -> bool:
 56        """True while a direct-mode session is in flight."""
 57        return self._programming_in_progress.is_set()
 58
 59    async def wait_idle(self, timeout: float = 30.0) -> None:
 60        """Wait for any in-progress direct-mode session to finish."""
 61        if not self._programming_in_progress.is_set():
 62            return
 63        # ``wait`` returns True when set; we want to wait until it's clear.
 64        # asyncio.Event has no direct "wait clear" — poll briefly.
 65        deadline = asyncio.get_event_loop().time() + timeout
 66        while self._programming_in_progress.is_set():
 67            if asyncio.get_event_loop().time() > deadline:
 68                raise TimeoutError("programming did not complete in time")
 69            await asyncio.sleep(0.05)
 70
 71    # ----- ops mode (Programming On the Main) ----------------------------
 72    #
 73    # Rebuilt on the 2026-07-14 audit path: raw NMRA DCC packet wrapped
 74    # in NCE ``QUEUEn_TMP_CMD`` (opcodes 0x93..0x96). See
 75    # :func:`jmri_nce.messages.ops_mode_write_cv` for the packet layout.
 76    # NCE PowerPro has no ops-mode CV read (RailCom is out of scope), so
 77    # only ``write_cv_ops`` exists — reads must go via direct mode.
 78
 79    async def write_cv_ops(
 80        self, loco_address: int, cv: int, value: int,
 81        *, long_address: bool = False,
 82    ) -> None:
 83        """Ops-mode CV write (POM). Fires a single DCC packet on the main.
 84
 85        Unlike ``write_cv_direct``, this does *not* pause layout traffic:
 86        the CS just queues a service-mode packet in-between throttle
 87        packets. Result reaches the loco during normal operation.
 88
 89        There's no wire-level "did the decoder accept it?" — POM is
 90        inherently unverified without RailCom. The returned ACK confirms
 91        the CS queued the packet, not that the decoder wrote the CV.
 92        """
 93        await self._tc.send_command(
 94            ops_mode_write_cv(
 95                loco_address, cv, value, long_address=long_address,
 96            ),
 97        )
 98
 99    # ----- direct mode (programming track) -------------------------------
100    #
101    # NCE PowerPro's direct-mode CV opcodes (``0xA8`` / ``0xA9``) require
102    # the CS to be in *programming mode* — a persistent state entered
103    # via ``ENTER_PROG_CMD`` (0x9E) and exited via ``EXIT_PROG_CMD``
104    # (0x9F). Outside programming mode the CS immediately NoAcks the
105    # command with ``'0'`` (0x30) *without* running any DCC packets on
106    # the track. That looks like an "instant NoAck" on the wire (~ms,
107    # not the 2–6 s a real programming attempt takes).
108    #
109    # Java JMRI's ``NceProgrammer`` wraps every direct-mode read/write
110    # with enter/exit. We mirror that here in :meth:`_prog_op` — the
111    # write and read helpers both go through it so exit always happens
112    # even when the operation raises.
113
114    async def _prog_op(
115        self, op, *, consume_trailer: bool = False,
116    ) -> "NceReply":  # noqa: F821
117        """Run *op* (an NceCommand) wrapped in ENTER_PROG / EXIT_PROG.
118
119        The enter and exit commands are short (single-byte requests,
120        1-byte ACK reply) so the wrapper adds ~10 ms of latency, dwarfed
121        by the 2–6 s of any real direct-mode CV op. Best-effort exit:
122        if the wrapped op raises, we still send EXIT_PROG so the CS
123        returns to normal operation and layout traffic resumes. The
124        exit's own reply is swallowed on failure — we don't want it
125        masking the original error.
126
127        *consume_trailer*: for RD_CV_DIRECT (and any future command
128        where NCE PowerPro appends a trailing ACK byte after the
129        payload), set True so the trailer is eaten before EXIT_PROG
130        runs. Without this, the trailer can land inside EXIT_PROG's
131        expected-1-byte reply window and cascade into stream
132        desynchronization.
133        """
134        await self._tc.send_command(enter_programming_mode())
135        try:
136            reply = await self._tc.send_command(op)
137            if consume_trailer:
138                await self._tc.consume_trailer(max_wait_s=0.1)
139            return reply
140        finally:
141            try:
142                await self._tc.send_command(exit_programming_mode())
143            except Exception as e:  # noqa: BLE001
144                _log.warning(
145                    "NCE: EXIT_PROG failed after %s: %s", op.name, e,
146                )
147
148    async def write_cv_direct(self, cv: int, value: int) -> None:
149        """Service-mode write. Pauses layout traffic for ~2–6 s.
150
151        Wraps the write with :meth:`_prog_op` so the CS is in
152        programming mode when the ``WRITE_DIR_CV_CMD`` (0xA8) opcode
153        arrives — a hard requirement of NCE PowerPro firmware.
154        """
155        self._programming_in_progress.set()
156        try:
157            await self._prog_op(direct_mode_write_cv(cv, value))
158        finally:
159            self._programming_in_progress.clear()
160
161    async def read_cv_direct(self, cv: int) -> int:
162        """Service-mode read. Same caveats as :meth:`write_cv_direct`.
163
164        Wrapped in ENTER_PROG / EXIT_PROG for the same reason as writes.
165        Raises :class:`NceCommandError` when the decoder doesn't
166        acknowledge — NCE signals that by sending the single byte
167        ``0x30`` (ASCII ``'0'``). JMRI Java's ``NceProgrammer`` treats
168        byte ``'0'`` as ``NoAck`` regardless of whether the actual CV
169        happens to hold value 48; we mirror that here since the
170        alternative (a silent "value 48 always") is worse than the rare
171        false negative when a CV genuinely equals 48.
172
173        Firmware quirk on successful reads: NCE PowerPro sends *two*
174        bytes back — the CV value plus a trailing ``!`` ACK. Our
175        command spec asks for ``reply_len=1`` so the trailer lands in
176        the serial buffer. If the next command's drain catches it,
177        fine; if it doesn't (race), the trailer contaminates the next
178        reply. We proactively consume it here with a short blocking
179        wait, closing that race window.
180        """
181        self._programming_in_progress.set()
182        try:
183            reply = await self._prog_op(
184                direct_mode_read_cv(cv), consume_trailer=True,
185            )
186            byte = reply.data[0]
187            if byte == 0x30:   # '0' = NCE NoAck sentinel
188                # NoAck is a single byte on the wire; the trailer eater
189                # in ``_prog_op`` times out silently in that case,
190                # which is fine.
191                raise NceCommandError(
192                    reply.command, b"0",
193                    b"decoder did not ack read (no decoder on prog track?)",
194                )
195            return byte
196        except NceCommandError:
197            raise
198        finally:
199            self._programming_in_progress.clear()