Testing helpers¶

def abort(self, hdass: int) -> None:
    """Transition RUNNING → ABORTING → INITIALIZED."""
    with self._lock:
        self._log("abort", hdass)
        self._consume_scripted("abort")
        self._require_open(hdass, op="abort")
        current = self.state_of(hdass)
        # Abort is idempotent / always valid; matches SDK behaviour.
        if current == SubsystemState.RUNNING:
            self._set_state(hdass, SubsystemState.INITIALIZED)

def add_channel(
    self,
    hdass: int,
    list_index: int,
    spec: ChannelSpec,
) -> None:
    """Add a channel to the channel/gain list, enforcing MULTI_SENSOR ordering."""
    from dtollib.channels.analog_input import (  # noqa: PLC0415
        ThermocoupleInput,
    )

    with self._lock:
        self._log("add_channel", (hdass, list_index, type(spec).__name__))
        self._consume_scripted("add_channel")
        sub = self._require_open(hdass, op="add_channel")

        # MULTI_SENSOR ordering invariant (docs/design.md §8.5a).
        if sub.capabilities.supports_multisensor:
            key = (hdass, spec.physical_channel)
            if key not in self._multi_sensor_typed:
                raise DtolTaskStateError(
                    f"fake backend: add_channel on MULTI_SENSOR channel "
                    f"{spec.physical_channel} before set_multi_sensor_type; "
                    f"see docs/design.md §8.5a (silent wrong-data bug)",
                    context=ErrorContext(
                        operation="add_channel",
                        channel=spec.physical_channel,
                        channel_name=spec.name,
                    ),
                )
            self.multi_sensor_specs[(hdass, spec.physical_channel)] = spec

        if isinstance(spec, ThermocoupleInput):
            self.thermocouple_types[(hdass, spec.physical_channel)] = (
                spec.thermocouple_type.value
            )

def alloc_buffer(
    self,
    n_samples: int,
    sample_dtype_bytes: int,
    *,
    zero_init: bool = True,
) -> int:
    """Allocate a synthetic HBUF backed by a numpy ndarray."""
    import numpy as np  # noqa: PLC0415

    with self._lock:
        self._log("alloc_buffer", (n_samples, sample_dtype_bytes, zero_init))
        self._consume_scripted("alloc_buffer")
        dtype = np.int16 if sample_dtype_bytes == _INT16_SAMPLE_BYTES else np.int32
        payload = np.zeros(n_samples, dtype=dtype)
        hbuf = self._next_hbuf
        self._next_hbuf += 1
        self._hbuf_state[hbuf] = BufferState.IDLE
        self._hbuf_payload[hbuf] = payload
        self._hbuf_valid_samples[hbuf] = 0
        return hbuf

def arm(self, hdass: int) -> None:
    """``olDaConfig`` #2 — continuous mode only; arm for ``start``.

    Enforces the bench-proven ordering: ``arm`` runs after the
    notification window is wired (:meth:`register_notification`) and
    the Ready queue is seeded. It raises if either precondition is
    unmet — register-BEFORE-arm and queue-BEFORE-arm — and refuses on a
    single-value subsystem (which has no second config).
    """
    with self._lock:
        self._log("arm", hdass)
        self._consume_scripted("arm")
        self._require_open(hdass, op="arm")
        if self.state_of(hdass) != SubsystemState.CONFIGURED_FOR_CONTINUOUS:
            raise DtolTaskStateError(
                "fake backend: arm requires a committed continuous task "
                f"(state is {self.state_of(hdass).value})",
                context=ErrorContext(operation="arm"),
            )
        if self._notify_callbacks.get(hdass) is None:
            raise DtolTaskStateError(
                "fake backend: arm BEFORE register_notification "
                "(violates register-before-arm ordering)",
                context=ErrorContext(operation="arm"),
            )
        if not self._ready_queue.get(hdass):
            raise DtolTaskStateError(
                "fake backend: arm BEFORE queueing buffers "
                "(violates queue-before-arm ordering)",
                context=ErrorContext(operation="arm"),
            )
        self._armed.add(hdass)

def code_to_volts(self, hdass: int, code: int, gain: float) -> float:
    """Offset-binary conversion mirroring the real DT9805/06 A/D.

    Matches :class:`~dtollib.backend.dataacq.DataAcqBackend.code_to_volts`
    on a ±10 V, 16-bit, offset-binary subsystem so the application-side
    thermocouple read path is exercised faithfully in unit tests: code
    ``32768`` → 0 V, ``0`` → −10 V, ``65535`` → +10 V, referred through
    ``gain``.
    """
    from dtollib.capi.conversion import code_to_input_volts  # noqa: PLC0415

    with self._lock:
        self._log("code_to_volts", (hdass, code, gain))
        self._consume_scripted("code_to_volts")
        self._require_open(hdass, op="code_to_volts")
        return code_to_input_volts(code, gain, vmin=-10.0, vmax=10.0, resolution_bits=16)

def commit(self, hdass: int) -> None:
    """``olDaConfig`` #1 — INITIALIZED → CONFIGURED_FOR_{SINGLE_VALUE|CONTINUOUS}.

    This is config #1: it runs after channel/clock/wrap setup and
    before the notification window + Ready queue are wired. It does
    NOT require a registered notification or queued buffers — those are
    preconditions of :meth:`arm` (config #2), matching the bench-proven
    ordering (docs/decisions.md).
    """
    from dtollib.capi.constants import OL_DF_SINGLEVALUE  # noqa: PLC0415

    with self._lock:
        self._log("commit", hdass)
        self._consume_scripted("commit")
        self._require_open(hdass, op="commit")
        current = self.state_of(hdass)
        if current not in {SubsystemState.INITIALIZED, SubsystemState.IO_COMPLETE}:
            raise DtolTaskStateError(
                f"fake backend: commit invalid from state {current.value}",
                context=ErrorContext(operation="commit"),
            )
        mode = self._data_flow_mode.get(hdass, OL_DF_SINGLEVALUE)
        if mode == OL_DF_SINGLEVALUE:
            self._set_state(hdass, SubsystemState.CONFIGURED_FOR_SINGLE_VALUE)
        else:
            self._set_state(hdass, SubsystemState.CONFIGURED_FOR_CONTINUOUS)
        self._armed.discard(hdass)
        self._configured.add(hdass)

def copy_buffer(self, hbuf: int, n_samples: int, sample_dtype_bytes: int) -> bytes:
    """Copy a synthetic HBUF's samples back out to a host byte payload."""
    with self._lock:
        self._log("copy_buffer", (hbuf, n_samples))
        self._consume_scripted("copy_buffer")
        payload = self._hbuf_payload.get(hbuf)
        if payload is None or self._hbuf_state.get(hbuf) == BufferState.RELEASED:
            raise DtolResourceError(
                f"fake backend: copy_buffer on invalid HBUF {hbuf}",
                context=ErrorContext(operation="copy_buffer"),
            )
        return bytes(payload[:n_samples].tobytes())

def copy_inprocess_buffer(
    self,
    hbuf: int,
    n_samples: int,
    sample_dtype_bytes: int,
) -> bytes:
    """Copy the in-process HBUF's valid samples without waiting.

    Models ``olDmCopyFromBuffer``: returns at most ``n_samples`` of the
    HBUF's recorded valid samples as raw bytes (empty when the buffer
    holds nothing yet). ``sample_dtype_bytes`` is accepted for signature
    parity; the synthetic payload already carries its own dtype.
    """
    import numpy as np  # noqa: PLC0415

    del sample_dtype_bytes  # payload dtype is intrinsic to the ndarray
    with self._lock:
        self._log("copy_inprocess_buffer", (hbuf, n_samples))
        self._consume_scripted("copy_inprocess_buffer")
        payload = self._hbuf_payload.get(hbuf)
        if payload is None:
            return b""
        valid = self._hbuf_valid_samples.get(hbuf, int(np.asarray(payload).size))
        take = max(0, min(int(valid), int(n_samples)))
        if take == 0:
            return b""
        flat = np.ascontiguousarray(np.asarray(payload).ravel()[:take])
        return bytes(flat.tobytes())

def copy_to_buffer(self, hbuf: int, data: bytes, n_samples: int) -> None:
    """Fill a synthetic HBUF from a host byte payload (waveform pre-fill)."""
    import numpy as np  # noqa: PLC0415

    with self._lock:
        self._log("copy_to_buffer", (hbuf, n_samples))
        self._consume_scripted("copy_to_buffer")
        payload = self._hbuf_payload.get(hbuf)
        if payload is None or self._hbuf_state.get(hbuf) == BufferState.RELEASED:
            raise DtolResourceError(
                f"fake backend: copy_to_buffer on invalid HBUF {hbuf}",
                context=ErrorContext(operation="copy_to_buffer"),
            )
        incoming = np.frombuffer(data, dtype=payload.dtype, count=n_samples)
        payload[:n_samples] = incoming
        self._hbuf_valid_samples[hbuf] = n_samples
        self._filled_hbufs.add(hbuf)

def ct_mode_of(self, hdass: int) -> int | None:
    """Test accessor — the last C/T mode set on ``hdass`` (or None)."""
    with self._lock:
        return self._ct_mode.get(hdass)

def enum_boards(self) -> list[BoardInfo]:
    """Satisfies :meth:`DtolBackend.enum_boards` — returns scripted boards."""
    with self._lock:
        self._log("enum_boards")
        self._consume_scripted("enum_boards")
        return [
            BoardInfo(
                name=b.name,
                model=b.model or b.name,
                driver_name=b.driver_name,
                instance=b.instance,
            )
            for b in self.boards
        ]

def enum_subsystems(self, board_name: str) -> list[SubsystemInfo]:
    """Satisfies :meth:`DtolBackend.enum_subsystems` — scripted subsystems."""
    with self._lock:
        self._log("enum_subsystems", board_name)
        self._consume_scripted("enum_subsystems")

        from dtollib.backend.dataacq import (  # noqa: PLC0415
            SUBSYS_TYPE_TO_ENUM,
        )

        board = self._find_board(board_name)
        return [
            SubsystemInfo(
                type=SUBSYS_TYPE_TO_ENUM[sub.type],
                element=sub.element,
                num_channels=sub.capabilities.num_channels,
                supports_singlevalue=sub.capabilities.supports_singlevalue,
                supports_continuous=sub.capabilities.supports_continuous,
                supports_simultaneous_sh=sub.capabilities.supports_simultaneous_sh,
                supports_multisensor=sub.capabilities.supports_multisensor,
                supports_dma=sub.capabilities.supports_dma,
                returns_floats=sub.capabilities.returns_floats,
                max_throughput_hz=sub.capabilities.max_throughput_hz,
                cgl_depth=sub.capabilities.cgl_depth,
            )
            for sub in board.subsystems
        ]

def fail_next(self, operation: str, *, code: int) -> None:
    """Cause the next call to ``operation`` to raise.

    Args:
        operation: Operation name as logged in
            :attr:`operations` (e.g. ``"initialize"``).
        code: SDK-style ECODE.  Used to construct an
            :class:`~dtollib.errors.ErrorContext`; the raised
            exception is selected via
            :func:`~dtollib.capi.errors.classify` so tests
            exercise the same routing as the real backend.
    """
    self.scripted_failures[operation] = code

def fire_buffer_done(
    self,
    hdass: int,
    *,
    fill: npt.NDArray[Any] | None = None,
) -> int | None:
    """Promote the next Ready HBUF to Done and fire ``BUFFER_DONE``.

    Returns the HBUF that was moved (None if Ready queue empty).
    Optionally writes a synthetic payload into the HBUF before firing.
    """
    with self._lock:
        ready = self._ready_queue.setdefault(hdass, deque())
        if not ready:
            return None
        hbuf = ready.popleft()
        if fill is not None:
            self._hbuf_payload[hbuf] = fill
            self._hbuf_valid_samples[hbuf] = int(fill.size)
        else:
            payload = self._hbuf_payload.get(hbuf)
            if payload is not None:
                self._hbuf_valid_samples[hbuf] = int(payload.size)
        self._hbuf_state[hbuf] = BufferState.COMPLETED
        self._done_queue.setdefault(hdass, deque()).append(hbuf)
    # Fire outside the lock — same reason as fire_event.
    self.fire_event(hdass, SdkEventKind.BUFFER_DONE)
    return hbuf

def fire_event(
    self,
    hdass: int,
    kind: SdkEventKind,
    *,
    wparam: int = 0,
    lparam: int = 0,
) -> None:
    """Synchronously fire an SDK notification event on the registered callback.

    Used by continuous callback-bridge tests to inject events without a
    real SDK driver thread. Raises :class:`DtolTaskStateError` if no
    callback is registered.
    """
    from dtollib.capi.constants import (  # noqa: PLC0415
        OLDA_WM_BUFFER_DONE,
        OLDA_WM_BUFFER_REUSED,
        OLDA_WM_EVENT_DONE,
        OLDA_WM_IO_COMPLETE,
        OLDA_WM_MEASURE_DONE,
        OLDA_WM_OVERRUN_ERROR,
        OLDA_WM_PRETRIGGER_BUFFER_DONE,
        OLDA_WM_QUEUE_DONE,
        OLDA_WM_QUEUE_STOPPED,
        OLDA_WM_TRIGGER_ERROR,
        OLDA_WM_UNDERRUN_ERROR,
    )

    kind_to_msg_id: dict[SdkEventKind, int] = {
        SdkEventKind.BUFFER_DONE: OLDA_WM_BUFFER_DONE,
        SdkEventKind.PRETRIGGER_BUFFER_DONE: OLDA_WM_PRETRIGGER_BUFFER_DONE,
        SdkEventKind.BUFFER_REUSED: OLDA_WM_BUFFER_REUSED,
        SdkEventKind.QUEUE_DONE: OLDA_WM_QUEUE_DONE,
        SdkEventKind.QUEUE_STOPPED: OLDA_WM_QUEUE_STOPPED,
        SdkEventKind.IO_COMPLETE: OLDA_WM_IO_COMPLETE,
        SdkEventKind.EVENT_DONE: OLDA_WM_EVENT_DONE,
        SdkEventKind.MEASURE_DONE: OLDA_WM_MEASURE_DONE,
        SdkEventKind.TRIGGER_ERROR: OLDA_WM_TRIGGER_ERROR,
        SdkEventKind.OVERRUN_ERROR: OLDA_WM_OVERRUN_ERROR,
        SdkEventKind.UNDERRUN_ERROR: OLDA_WM_UNDERRUN_ERROR,
    }
    with self._lock:
        callback = self._notify_callbacks.get(hdass)
        if callback is None:
            raise DtolTaskStateError(
                f"fake backend: fire_event with no notification registered on {hdass}",
                context=ErrorContext(operation="fire_event"),
            )
        msg_id = kind_to_msg_id[kind]
    # Run the callback OUTSIDE the lock so realistic event-handler
    # code that re-enters the backend does not deadlock.
    callback(msg_id, wparam, lparam)

def flush_buffers(self, hdass: int) -> None:
    """Empty the Ready and Done queues."""
    with self._lock:
        self._log("flush_buffers", hdass)
        self._consume_scripted("flush_buffers")
        self._require_open(hdass, op="flush_buffers")
        self._ready_queue[hdass] = deque()
        self._done_queue[hdass] = deque()

def force_hbuf_state(self, hbuf: int, state: BufferState) -> None:
    """Test-only state mutator for invariant coverage."""
    with self._lock:
        self._hbuf_state[hbuf] = state

def free_buffer(self, hbuf: int) -> None:
    """Release a synthetic HBUF; rejects free while INPROCESS."""
    with self._lock:
        self._log("free_buffer", hbuf)
        self._consume_scripted("free_buffer")
        state = self._hbuf_state.get(hbuf)
        if state is None:
            raise DtolResourceError(
                f"fake backend: free_buffer on unknown HBUF {hbuf}",
                context=ErrorContext(operation="free_buffer"),
            )
        if state == BufferState.INPROCESS:
            raise DtolTaskStateError(
                f"fake backend: free_buffer on INPROCESS HBUF {hbuf} "
                "(violates §8.14 invariant)",
                context=ErrorContext(operation="free_buffer"),
            )
        self._hbuf_state[hbuf] = BufferState.RELEASED
        self._hbuf_payload.pop(hbuf, None)

def get_buffer(self, hdass: int) -> int | None:
    """Pop the next HBUF from the Done queue (None if empty)."""
    with self._lock:
        self._log("get_buffer", hdass)
        self._consume_scripted("get_buffer")
        self._require_open(hdass, op="get_buffer")
        done = self._done_queue.setdefault(hdass, deque())
        if not done:
            return None
        hbuf = done.popleft()
        self._hbuf_state[hbuf] = BufferState.COMPLETED
        return hbuf

def get_buffer_valid_samples(self, hbuf: int) -> int:
    """Return the recorded valid-sample count for the HBUF."""
    with self._lock:
        self._log("get_buffer_valid_samples", hbuf)
        self._consume_scripted("get_buffer_valid_samples")
        return self._hbuf_valid_samples.get(hbuf, 0)

def get_cjc_temperature(self, hdass: int, channel: int) -> float:
    """Return a plausible CJC temperature; tests override via ``scalar_values``."""
    with self._lock:
        self._log("get_cjc_temperature", (hdass, channel))
        self._consume_scripted("get_cjc_temperature")
        self._require_open(hdass, op="get_cjc_temperature")
        return 25.0  # room temperature default; tests can override via the dict.

def get_clock_frequency(self, hdass: int) -> float:
    """Return the recorded clock rate (driver may quantise the request)."""
    with self._lock:
        self._log("get_clock_frequency", hdass)
        self._consume_scripted("get_clock_frequency")
        self._require_open(hdass, op="get_clock_frequency")
        return self._clock_rate_hz.get(hdass, 0.0)

def get_dass(self, hdrvr: int, subsystem_type: int, element: int) -> int:
    """Satisfies :meth:`DtolBackend.get_dass` — returns synthetic HDASS."""
    with self._lock:
        self._log("get_dass", (hdrvr, subsystem_type, element))
        self._consume_scripted("get_dass")

        name = self._hdrvr_to_name.get(hdrvr)
        if name is None:
            raise DtolResourceError(
                f"fake backend: get_dass on unknown HDRVR {hdrvr}",
                context=ErrorContext(operation="get_dass"),
            )
        board = self._find_board(name)
        for sub in board.subsystems:
            if sub.type == subsystem_type and sub.element == element:
                hdass = self._next_hdass
                self._next_hdass += 1
                self._open_subsystems[hdass] = sub
                return hdass
        raise DtolResourceError(
            f"fake backend: no subsystem (type={subsystem_type}, "
            f"element={element}) on {name!r}",
            context=ErrorContext(
                operation="get_dass",
                board=name,
                element=element,
                extra={"subsys_type": subsystem_type},
            ),
        )