nidaqlib.testing¶

def add_channel(self, task: _FakeTask, spec: ChannelSpec) -> None:
    """Append ``spec`` to ``task.channels``."""
    task.channels.append(spec)
    self.operations.append(_Operation("add_channel", task.name, spec.physical_channel))

def close_task(self, task: _FakeTask) -> None:
    """Mark ``task`` closed. Idempotent."""
    if task.closed:
        return
    task.closed = True
    self.operations.append(_Operation("close_task", task.name))

def configure_logging(self, task: _FakeTask, logging: TdmsLogging) -> None:
    """Record ``logging`` on ``task``."""
    task.logging = logging
    self.operations.append(_Operation("configure_logging", task.name, f"path={logging.path!s}"))

def configure_timing(self, task: _FakeTask, timing: Timing) -> None:
    """Record ``timing`` on ``task``."""
    task.timing = timing
    self.operations.append(
        _Operation("configure_timing", task.name, f"rate_hz={timing.rate_hz}")
    )

def configure_trigger(self, task: _FakeTask, trigger: TriggerSpec) -> None:
    """Record ``trigger`` on ``task`` for test inspection."""
    task.trigger = trigger
    detail = f"kind={trigger.kind};source={trigger.source}"
    self.operations.append(_Operation("configure_trigger", task.name, detail))

def create_task(self, name: str) -> _FakeTask:
    """Create and return a new :class:`_FakeTask`.

    Raises:
        NIDaqBackendError: A task with ``name`` already exists.
    """
    if name in self._tasks:
        raise NIDaqBackendError(
            f"task {name!r} already exists",
            context=ErrorContext(task_name=name, operation="create_task"),
        )
    task = _FakeTask(name=name)
    self._tasks[name] = task
    self.operations.append(_Operation("create_task", name))
    return task

def device_info(self, device: str) -> DeviceInfo | None:
    """Return scripted ``DeviceInfo`` for ``device`` if registered, else ``None``.

    Tests register product types via :meth:`register_device_info` so the
    manager's module-level preflight can be exercised against the fake.
    Default behaviour (no registration) returns ``None``, matching the
    Protocol's "unknown device" semantics.
    """
    return self._device_info.get(device)

def read_block(
    self,
    task: _FakeTask,
    samples_per_channel: int,
    timeout: float,
) -> np.ndarray:
    """Pop the next scripted block, fall back to a deterministic ramp.

    Raises:
        NIDaqTimeoutError: A scripted timeout exception was queued.
        NIDaqReadError: A scripted read exception was queued, or the
            queue is empty and no ``read_block_default_shape`` is set.
    """
    del timeout  # The fake never blocks — timeout is metadata for tests.
    errs = self._read_errors.get(task.name)
    if errs:
        err = errs.pop(0)
        self.operations.append(_Operation("read_block_error", task.name, str(err)))
        if isinstance(err, (NIDaqReadError, NIDaqTimeoutError)):
            raise err
        raise NIDaqReadError(
            f"scripted read error: {err!r}",
            context=ErrorContext(task_name=task.name, operation="read_block"),
        ) from err
    scripted = self._blocks.get(task.name)
    if scripted:
        block = scripted.pop(0)
    elif self._default_shape is not None:
        n_channels, _ = self._default_shape
        i = self._read_counter[task.name]
        block = np.full(
            (n_channels, samples_per_channel),
            fill_value=float(i),
            dtype=np.float64,
        )
        self._read_counter[task.name] = i + 1
    else:
        raise NIDaqReadError(
            f"no scripted blocks remain for task {task.name!r}",
            context=ErrorContext(task_name=task.name, operation="read_block"),
        )
    if block.shape[1] != samples_per_channel:
        # Reshape on the fly for tests that pre-build a long stream and
        # let the recorder choose the chunk size.
        block = block[:, :samples_per_channel]
    self.operations.append(_Operation("read_block", task.name, str(block.shape)))
    return block

def register_device_info(self, device: str, *, product_type: str) -> None:
    """Scripted DeviceInfo for tests of the manager's module-level preflight."""
    from nidaqlib.system.models import DeviceInfo as _DeviceInfo  # noqa: PLC0415

    self._device_info[device] = _DeviceInfo(
        name=device,
        product_type=product_type,
        serial_number=None,
        ai_physical_channels=(),
        ao_physical_channels=(),
        di_lines=(),
        do_lines=(),
        ci_physical_channels=(),
        co_physical_channels=(),
    )

def register_every_n_samples(
    self,
    task: _FakeTask,
    n: int,
    callback: Callable[[int], None],
) -> _FakeTask:
    """Stash ``callback`` on the task. Returns ``task`` as the handle.

    Mirrors NI's ordering invariant: registration must precede
    ``task.start()``. Real NI rejects post-start registration with
    -200960 ("Register all your DAQmx software events prior to starting
    the task"); the fake raises an analogous
    :class:`NIDaqBackendError` so the unit suite catches violations
    that the hardware would otherwise surface only at integration.

    Raises:
        NIDaqBackendError: A callback is already registered, or the task
            has already been started.
    """
    if task.started:
        raise NIDaqBackendError(
            f"task {task.name!r} is already started; "
            "register_every_n_samples must run before start_task "
            "(NI rejects post-start registration with -200960)",
            context=ErrorContext(task_name=task.name, operation="register_every_n_samples"),
        )
    if task.callback is not None:
        raise NIDaqBackendError(
            f"task {task.name!r} already has a buffer-event callback",
            context=ErrorContext(task_name=task.name, operation="register_every_n_samples"),
        )
    task.callback = callback
    task.callback_n = n
    self.operations.append(_Operation("register_every_n_samples", task.name, f"n={n}"))
    return task

def simulate_callbacks(
    self,
    task: _FakeTask,
    *,
    firings: int,
    cadence_s: float = 0.0,
) -> threading.Thread:
    """Fire the registered callback ``firings`` times from a worker thread.

    Models the behaviour of NI's DAQmx driver thread so the §11.3.2
    bridge can be exercised end-to-end in unit tests. The callback runs
    on a fresh ``threading.Thread`` (NOT the asyncio event loop).

    The simulator stops early if the callback is unregistered between
    firings — this models NI's "any pending events are discarded" note
    on ``stop_task``.

    Args:
        task: The fake task on which the callback was registered.
        firings: Number of times to invoke the callback.
        cadence_s: Optional sleep between firings, in seconds. Use 0 for
            a tight burst, > 0 to mimic a finite sample-clock cadence.

    Returns:
        The :class:`threading.Thread` running the simulator. Tests
        usually do not need to ``.join()`` this — the bridge tests rely
        on the recorder's own shutdown to drain pending chunks. Joinable
        if asserting on thread liveness.
    """

    def _run() -> None:
        for _ in range(firings):
            cb = task.callback
            if cb is None:
                return
            cb(task.callback_n)
            if cadence_s > 0.0:
                threading.Event().wait(cadence_s)

    thread = threading.Thread(
        target=_run,
        name=f"FakeDaqBackend-cb-sim[{task.name}]",
        daemon=True,
    )
    thread.start()
    self._sim_threads.append(thread)
    return thread

def start_task(self, task: _FakeTask) -> None:
    """Mark ``task`` started."""
    task.started = True
    self.operations.append(_Operation("start_task", task.name))

def stop_task(self, task: _FakeTask) -> None:
    """Mark ``task`` stopped."""
    task.started = False
    self.operations.append(_Operation("stop_task", task.name))

def unregister_every_n_samples(self, task: _FakeTask, handle: Any) -> None:
    """Clear the buffer-event callback on ``task``.

    Mirrors NI's ordering invariant: unregistration requires the task
    to be stopped. Real NI rejects post-running unregister with -200986
    ("DAQmx software event cannot be unregistered because the task is
    running"); the fake raises an analogous
    :class:`NIDaqBackendError` so the unit suite catches violations
    that real hardware would otherwise surface only at integration.

    Raises:
        NIDaqBackendError: The task is still running.
    """
    del handle
    if task.started:
        raise NIDaqBackendError(
            f"task {task.name!r} is still running; "
            "unregister_every_n_samples must run after stop_task "
            "(NI rejects post-running unregister with -200986)",
            context=ErrorContext(task_name=task.name, operation="unregister_every_n_samples"),
        )
    task.callback = None
    task.callback_n = 0
    self.operations.append(_Operation("unregister_every_n_samples", task.name))

def write(
    self,
    task: _FakeTask,
    values: Mapping[str, float | bool],
    timeout: float,
) -> None:
    """Record one write — for tests asserting on outputs.

    Validation parity with the real backend: missing channel keys raise
    :class:`NIDaqConfigurationError`; scripted errors raise from the
    per-task ``write_errors`` queue.

    Raises:
        NIDaqConfigurationError: ``values`` is missing entries for one
            or more output channels.
        NIDaqWriteError / NIDaqTimeoutError: A scripted error was queued.
    """
    del timeout  # The fake never blocks — timeout is metadata for tests.
    from nidaqlib.channels.analog_output import AnalogOutputVoltage  # noqa: PLC0415
    from nidaqlib.channels.digital_output import DigitalOutput  # noqa: PLC0415

    output_channels = [
        ch for ch in task.channels if isinstance(ch, (AnalogOutputVoltage, DigitalOutput))
    ]
    if not output_channels:
        raise NIDaqConfigurationError(
            "task has no writable channels (AO or DO)",
            context=ErrorContext(task_name=task.name, operation="write"),
        )
    names = [ch.display_name for ch in output_channels]
    missing = [n for n in names if n not in values]
    if missing:
        raise NIDaqConfigurationError(
            f"write missing values for channel(s): {missing!r}",
            context=ErrorContext(task_name=task.name, operation="write"),
        )

    errs = self._write_errors.get(task.name)
    if errs:
        err = errs.pop(0)
        self.operations.append(_Operation("write_error", task.name, str(err)))
        if isinstance(err, (NIDaqWriteError, NIDaqTimeoutError)):
            raise err
        raise NIDaqWriteError(
            f"scripted write error: {err!r}",
            context=ErrorContext(task_name=task.name, operation="write"),
        ) from err

    snapshot: dict[str, float | bool] = {n: values[n] for n in names}
    task.writes.append(snapshot)
    task.last_write = snapshot
    self.operations.append(_Operation("write", task.name, repr(snapshot)))