nidaqlib.tasks¶

def __post_init__(self) -> None:
    """Validate the rectangular-shape invariant.

    Raises:
        NIDaqValidationError: ``data.shape`` does not equal
            ``(len(channels), samples_per_channel)``.
    """
    from nidaqlib.errors import NIDaqValidationError  # noqa: PLC0415

    n_channels = len(self.channels)
    expected = (n_channels, self.samples_per_channel)
    actual = tuple(self.data.shape)
    if actual != expected:
        raise NIDaqValidationError(
            f"DaqBlock data shape {actual} does not match (channels, "
            f"samples_per_channel) = {expected}"
        )

async def __aenter__(self) -> DaqSession:
    """Enter the async context — no-op; :func:`open_device` already configured/started."""
    return self

async def __aexit__(self, *exc_info: object) -> None:
    """Exit the async context — calls :meth:`close`."""
    del exc_info
    await self.close()

async def acquire(
    self,
    samples_per_channel: int,
    *,
    timeout: float | None = None,  # noqa: ASYNC109 — NI per-call timeout, not coroutine
) -> DaqBlock:
    """Run one finite acquisition and return its :class:`DaqBlock`.

    Convenience wrapper for the §12.3 finite-mode pattern: configure
    finite, start, read, stop. Requires a session whose
    :class:`Timing.mode` is :attr:`AcquisitionMode.FINITE`. After the
    read completes, the underlying NI task is stopped — call
    :meth:`start` again before another acquisition.

    Args:
        samples_per_channel: Number of samples per channel to read.
        timeout: Optional per-call timeout in seconds. Falls back to
            the session-wide default.

    Raises:
        NIDaqTaskStateError: The session is not started, is closed, or
            its timing mode is not :attr:`AcquisitionMode.FINITE`.
        NIDaqReadError / NIDaqTimeoutError: Surfaced from the backend.
    """
    self._require_started("acquire")
    timing = self._spec.timing
    if timing is None or timing.mode is not AcquisitionMode.FINITE:
        raise NIDaqTaskStateError(
            f"acquire() requires Timing.mode=FINITE; got {timing.mode if timing else None}",
            context=ErrorContext(task_name=self._spec.name, command_name="acquire"),
        )
    block = await self.read_block(samples_per_channel, timeout=timeout)
    await self.stop()
    return block

async def close(self) -> None:
    """Stop (if needed) and close the underlying task. Idempotent.

    ``__aexit__`` always calls this; explicit call is rare. Sessions that
    have opted into the every-N-samples callback bridge MUST instead use
    the recorder context manager — the bridge has its own ordered
    shutdown protocol (design doc §11.3.2) that this method does not
    implement.
    """
    if self._closed:
        return
    if self._callback_handle is not None:
        raise NIDaqTaskStateError(
            "cannot close a session while an every-N-samples callback bridge is active; "
            "exit the record(..., use_callback_bridge=True) context first",
            context=ErrorContext(task_name=self._spec.name, command_name="close"),
        )
    self._closed = True
    if self._task is None:
        return
    async with self._lock:
        if self._started:
            await run_sync(self._backend.stop_task, self._task)
            self._started = False
        await run_sync(self._backend.close_task, self._task)
        self._task = None
        self._configured = False

async def configure(self) -> None:
    """Create the underlying task and apply channels / timing / logging / trigger.

    After this method, ``raw_task`` is available and any pre-start hooks
    (notably the §11.3.2 buffer-event callback registration) may run.
    ``task.start()`` is **not** called — use :meth:`start` for that.

    On failure, the partial task is torn down so the session does not
    leak NI resources.

    Raises:
        NIDaqTaskStateError: Already configured, started, or closed.
    """
    if self._closed:
        raise NIDaqTaskStateError(
            f"session for task {self._spec.name!r} is closed",
            context=ErrorContext(task_name=self._spec.name, command_name="configure"),
        )
    if self._configured:
        raise NIDaqTaskStateError(
            f"session for task {self._spec.name!r} is already configured",
            context=ErrorContext(task_name=self._spec.name, command_name="configure"),
        )
    async with self._lock:
        await run_sync(self._configure_sync)
        self._configured = True

async def poll(
    self,
    *,
    timeout: float | None = None,  # noqa: ASYNC109 — NI per-call timeout, not coroutine
) -> DaqReading:
    """One-shot scalar read across all channels.

    Valid only for sessions that are not actively buffering a sample
    clock (``Timing.mode == ON_DEMAND`` or no ``Timing`` at all). For the
    live-scalar use case during a high-rate acquisition, use
    :func:`record` and read the most recent block's last column.

    Raises:
        NIDaqTaskStateError: The session is buffering a sample clock
            (continuous or finite mode and started).
    """
    self._require_started("poll")
    self._require_analog_input_task("poll")
    timing = self._spec.timing
    if timing is not None and timing.mode in (
        AcquisitionMode.CONTINUOUS,
        AcquisitionMode.FINITE,
    ):
        raise NIDaqTaskStateError(
            f"poll() is invalid for {timing.mode.value} tasks; use record() and "
            "inspect the most recent DaqBlock instead",
            context=ErrorContext(task_name=self._spec.name, command_name="poll"),
        )
    eff_timeout = timeout if timeout is not None else self._timeout
    async with self._lock:
        requested_at = datetime.now(UTC)
        monotonic_ns_start = time.monotonic_ns()
        data = await run_sync(
            self._backend.read_block,
            self._task,
            1,
            eff_timeout,
        )
        received_at = datetime.now(UTC)
        monotonic_ns_end = time.monotonic_ns()
    t_utc = requested_at + (received_at - requested_at) / 2
    t_mono_ns = (monotonic_ns_start + monotonic_ns_end) // 2
    # data shape is (n_channels, 1) — squeeze to per-channel scalars.
    names = self._channel_names()
    units = self._channel_units()
    values: dict[str, float | int | bool] = {
        name: float(data[i, 0]) for i, name in enumerate(names)
    }
    return DaqReading(
        device=self._spec.name,
        task=self._spec.name,
        values=values,
        units=units,
        t_mono_ns=t_mono_ns,
        t_utc=t_utc,
        t_midpoint_mono_ns=None,
        requested_at=requested_at,
        received_at=received_at,
        latency_s=(received_at - requested_at).total_seconds(),
        metadata=dict(self._spec.metadata),
        error=None,
    )

async def read_block(
    self,
    samples_per_channel: int,
    *,
    timeout: float | None = None,  # noqa: ASYNC109 — NI per-call timeout, not coroutine
) -> DaqBlock:
    """Read one rectangular :class:`DaqBlock`.

    Wraps the backend read in an ``run_sync`` so the
    event loop stays responsive during the blocking NI call. Increments
    the per-session ``first_sample_index`` cursor.

    Args:
        samples_per_channel: Samples per channel for this block.
        timeout: Optional per-call timeout in seconds; falls back to the
            session-wide default.

    Raises:
        NIDaqTaskStateError: The session is not started or is closed.
        NIDaqReadError / NIDaqTimeoutError: Surfaced from the backend.
    """
    self._require_started("read_block")
    self._require_analog_input_task("read_block")
    eff_timeout = timeout if timeout is not None else self._timeout
    async with self._lock:
        read_started_at = datetime.now(UTC)
        monotonic_ns = time.monotonic_ns()
        data = await run_sync(
            self._backend.read_block,
            self._task,
            samples_per_channel,
            eff_timeout,
        )
        read_finished_at = datetime.now(UTC)
        block = self._build_block(
            data=data,
            samples_per_channel=samples_per_channel,
            read_started_at=read_started_at,
            read_finished_at=read_finished_at,
            monotonic_ns=monotonic_ns,
        )
    return block

async def snapshot(self) -> NIDaqSnapshot:
    """Return an :class:`NIDaqSnapshot` of this session's current state.

    No I/O — built from the cached :class:`DeviceInfo` (one backend call
    at configure time) and the session's own lifecycle flags. Safe to
    call from any thread / event loop / callback context.
    """
    info = self._device_info
    timing = self._spec.timing
    return NIDaqSnapshot(
        name=self._spec.name,
        model=info.product_type if info is not None else None,
        firmware=None,
        serial=info.serial_number if info is not None else None,
        connected=self._configured and not self._closed,
        last_error=self._last_error_context,
        recoverable_error_count=self._recoverable_error_count,
        captured_at=datetime.now(UTC),
        task_name=self._spec.name,
        task_state=self.task_state(),
        channel_count=len(self._spec.channels),
        timing_mode=timing.mode.value if timing is not None else None,
        rate_hz=timing.rate_hz if timing is not None else None,
        physical_channels=tuple(ch.physical_channel for ch in self._spec.channels),
        product_type=info.product_type if info is not None else None,
        chassis=None,
        physical_module=info.name if info is not None else None,
    )

async def start(self, *, confirm: bool = False) -> None:
    """Start the configured task.

    :meth:`configure` must have run first. This method calls NI's
    ``task.start()`` and records the wall-clock anchor used for §8.7
    sample-time reconstruction. Calling :meth:`start` again after
    :meth:`stop` reuses the configured task and resets the
    block/sample counters for a new run.

    ``confirm=True`` is required for task kinds whose ``start`` call
    can actuate hardware immediately (currently counter-output pulse
    trains).

    Raises:
        NIDaqTaskStateError: Not configured, already started, or closed.
        NIDaqValidationError: Starting would actuate hardware without
            explicit confirmation.
    """
    if self._closed:
        raise NIDaqTaskStateError(
            f"session for task {self._spec.name!r} is closed",
            context=ErrorContext(task_name=self._spec.name, command_name="start"),
        )
    if not self._configured:
        raise NIDaqTaskStateError(
            f"session for task {self._spec.name!r} must be configured before start",
            context=ErrorContext(task_name=self._spec.name, command_name="start"),
        )
    if self._started:
        raise NIDaqTaskStateError(
            f"session for task {self._spec.name!r} is already started",
            context=ErrorContext(task_name=self._spec.name, command_name="start"),
        )
    self._validate_start_safety(confirm=confirm)
    async with self._lock:
        # Capture the wall-clock anchor as close to the start as possible
        # — `start_task` returns once NI has armed the clock, so the
        # first sample's wall-clock is approximately this timestamp + a
        # bounded device latency.
        anchor = datetime.now(UTC)
        try:
            await run_sync(self._backend.start_task, self._task)
        except BaseException:
            await run_sync(self._backend.close_task, self._task)
            self._task = None
            self._configured = False
            raise
        self._task_started_at = anchor
        self._first_sample_index = 0
        self._block_index = 0
        self._started = True

async def stop(self) -> None:
    """Stop the underlying task. Idempotent for not-yet-started sessions.

    Does NOT close the task. Use :meth:`close` to release NI resources.
    """
    if not self._started or self._closed or self._task is None:
        return
    async with self._lock:
        await run_sync(self._backend.stop_task, self._task)
        self._started = False

def task_state(self) -> TaskState:
    """Return the coarse :class:`TaskState` projection of this session.

    I/O-free — derived from internal lifecycle flags.
    """
    if self._closed:
        return TaskState.CLOSED
    if self._started:
        return TaskState.RUNNING
    if self._configured:
        # `configured-but-not-started` covers both "never started" and
        # "started then stopped". We distinguish by whether the start
        # anchor was ever set.
        return TaskState.STOPPED if self._task_started_at is not None else TaskState.CONFIGURED
    return TaskState.CREATED

async def write(
    self,
    values: Mapping[str, float | bool],
    *,
    confirm: bool = False,
    timeout: float | None = None,  # noqa: ASYNC109 — NI per-call timeout, not coroutine
) -> None:
    """Write one sample-per-channel to the task's output channels.

    Safety gate (design doc §17):

    - Keys of ``values`` must match the display names of the task's
      output channels (AO and/or DO). Unknown or missing keys raise
      :class:`NIDaqValidationError` before any I/O.
    - For analog-output channels with ``safe_min`` / ``safe_max`` set,
      values outside the resolved clamp window raise
      :class:`NIDaqValidationError`. **Never silently clamped.**
    - If any target channel has ``requires_confirm=True`` and
      ``confirm`` is ``False``, the call raises
      :class:`NIDaqValidationError`.

    Args:
        values: One value per output channel keyed by display name.
        confirm: Operator confirmation. Required (must be ``True``)
            whenever any target channel sets ``requires_confirm``.
        timeout: Per-call timeout in seconds. Falls back to the
            session-wide default.

    Raises:
        NIDaqTaskStateError: The session is not started or is closed.
        NIDaqValidationError: Safety-gate or shape rejection (see above).
        NIDaqWriteError / NIDaqTimeoutError: Surfaced from the backend.
    """
    # Late import — keeps the channel modules out of the session-import
    # graph for sessions that never write.
    from nidaqlib.channels.analog_output import AnalogOutputVoltage  # noqa: PLC0415
    from nidaqlib.channels.counter_output import (  # noqa: PLC0415
        CounterPulseFrequency,
        CounterPulseTicks,
        CounterPulseTime,
    )
    from nidaqlib.channels.digital_output import DigitalOutput  # noqa: PLC0415

    self._require_started("write")

    output_channels = [
        ch for ch in self._spec.channels if isinstance(ch, (AnalogOutputVoltage, DigitalOutput))
    ]
    if not output_channels:
        if any(
            isinstance(ch, (CounterPulseFrequency, CounterPulseTime, CounterPulseTicks))
            for ch in self._spec.channels
        ):
            raise NIDaqValidationError(
                "counter-output pulse trains are controlled by start()/stop(), not write(); "
                "start them with confirm=True",
                context=ErrorContext(task_name=self._spec.name, command_name="write"),
            )
        raise NIDaqValidationError(
            f"task {self._spec.name!r} has no output channels to write",
            context=ErrorContext(task_name=self._spec.name, command_name="write"),
        )
    has_ao = any(isinstance(ch, AnalogOutputVoltage) for ch in output_channels)
    has_do = any(isinstance(ch, DigitalOutput) for ch in output_channels)
    if has_ao and has_do:
        raise NIDaqValidationError(
            "write() does not support mixing analog-output and digital-output "
            "channels in one task",
            context=ErrorContext(task_name=self._spec.name, command_name="write"),
        )

    target_names = {ch.display_name for ch in output_channels}
    provided_names = set(values.keys())
    unknown = provided_names - target_names
    missing = target_names - provided_names
    if unknown or missing:
        raise NIDaqValidationError(
            f"write keys do not match task outputs (unknown={sorted(unknown)!r}, "
            f"missing={sorted(missing)!r})",
            context=ErrorContext(task_name=self._spec.name, command_name="write"),
        )

    needs_confirm = any(getattr(ch, "requires_confirm", False) for ch in output_channels)
    if needs_confirm and not confirm:
        raise NIDaqConfirmationRequiredError(
            f"task {self._spec.name!r}: write requires confirm=True (one or more "
            "channels are marked requires_confirm)",
            context=ErrorContext(task_name=self._spec.name, command_name="write"),
        )

    for ch in output_channels:
        value = values[ch.display_name]
        if isinstance(ch, AnalogOutputVoltage):
            lo = ch.effective_safe_min
            hi = ch.effective_safe_max
            fvalue = float(value)
            if fvalue < lo or fvalue > hi:
                raise NIDaqValidationError(
                    f"value {fvalue!r} for AO channel {ch.display_name!r} is outside "
                    f"safe range [{lo}, {hi}]",
                    context=ErrorContext(
                        task_name=self._spec.name,
                        channel_name=ch.display_name,
                        physical_channel=ch.physical_channel,
                        command_name="write",
                    ),
                )

    eff_timeout = timeout if timeout is not None else self._timeout
    async with self._lock:
        await run_sync(
            self._backend.write,
            self._task,
            dict(values),
            eff_timeout,
        )

def add_channel(self, channel: ChannelSpec) -> Self:
    """Append a channel to the task. Returns self for chaining."""
    self._channels.append(channel)
    return self

def build(self) -> TaskSpec:
    """Construct the immutable :class:`TaskSpec`."""
    return TaskSpec(
        name=self._name,
        channels=tuple(self._channels),
        timing=self._timing,
        metadata=dict(self._metadata),
    )

def with_metadata(self, metadata: Mapping[str, str | int | float | bool]) -> Self:
    """Merge ``metadata`` into the builder's metadata dict.

    Returns self for chaining. Later calls overwrite earlier keys.
    """
    self._metadata.update(metadata)
    return self

def with_timing(self, timing: Timing) -> Self:
    """Set the task's :class:`Timing`. Returns self for chaining."""
    self._timing = timing
    return self

def __post_init__(self) -> None:
    """Validate the channel list shape (the cheap, always-true invariants).

    Raises:
        NIDaqValidationError: ``channels`` is empty or contains a
            non-:class:`ChannelSpec` element.
    """
    if len(self.channels) == 0:
        raise NIDaqValidationError(f"TaskSpec {self.name!r}: at least one channel is required")
    if not self.name:
        raise NIDaqValidationError("TaskSpec.name must be a non-empty string")
    channels = tuple(self.channels)
    object.__setattr__(self, "channels", channels)
    for ch in self.channels:
        if not isinstance(ch, ChannelSpec):  # pyright: ignore[reportUnnecessaryIsInstance]
            raise NIDaqValidationError(
                f"TaskSpec {self.name!r}: channels must be ChannelSpec instances, "
                f"got {type(ch).__name__}"
            )
    names = [ch.display_name for ch in self.channels]
    duplicates = sorted({name for name in names if names.count(name) > 1})
    if duplicates:
        raise NIDaqValidationError(
            f"TaskSpec {self.name!r}: duplicate channel display names {duplicates!r}"
        )
    object.__setattr__(self, "metadata", MappingProxyType(dict(self.metadata)))

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> Self:
    """Deserialise from a dict produced by :meth:`to_dict`.

    Args:
        data: Mapping carrying the task-spec fields.

    Raises:
        NIDaqValidationError: A channel or trigger entry has an unknown
            ``kind``, or required structural fields are malformed.
    """
    from nidaqlib.tasks.triggers import TriggerSpec  # noqa: PLC0415

    timing_payload = data.get("timing")
    timing = Timing.from_dict(timing_payload) if timing_payload is not None else None
    trigger_payload = data.get("trigger")
    if trigger_payload is None:
        trigger = None
    elif isinstance(trigger_payload, Mapping):
        trigger = TriggerSpec.from_dict(trigger_payload)  # pyright: ignore[reportUnknownArgumentType]
    else:
        raise NIDaqValidationError(
            f"TaskSpec.trigger must be a mapping or null, got {type(trigger_payload).__name__}"
        )
    logging_payload = data.get("logging")
    logging = TdmsLogging.from_dict(logging_payload) if logging_payload is not None else None
    raw_channels: object = data.get("channels", [])
    if not isinstance(raw_channels, list):
        raise NIDaqValidationError(
            f"TaskSpec.channels must be a list, got {type(raw_channels).__name__}"
        )
    channels: list[ChannelSpec] = []
    for ch in raw_channels:  # pyright: ignore[reportUnknownVariableType]
        if not isinstance(ch, Mapping):
            raise NIDaqValidationError(
                f"TaskSpec.channels[*] must be a mapping, got {type(ch).__name__}"  # pyright: ignore[reportUnknownArgumentType]
            )
        channels.append(ChannelSpec.from_dict(ch))  # pyright: ignore[reportUnknownArgumentType]
    metadata_raw: object = data.get("metadata", {})
    if not isinstance(metadata_raw, Mapping):
        raise NIDaqValidationError(
            f"TaskSpec.metadata must be a mapping, got {type(metadata_raw).__name__}"
        )
    return cls(
        name=str(data["name"]),
        channels=channels,
        timing=timing,
        trigger=trigger,
        logging=logging,
        metadata=dict(metadata_raw),  # pyright: ignore[reportUnknownArgumentType]
    )

def replace(self, **updates: Any) -> Self:
    """Return a copy of this spec with ``updates`` applied.

    Mirrors ``dataclasses.replace`` but is exposed as a method for
    consistency with the rest of the API.
    """
    return dataclasses.replace(self, **updates)

def to_dict(self) -> dict[str, Any]:
    """Serialise to a JSON-friendly dict, dispatching channels by ``kind``."""
    return {
        "name": self.name,
        "channels": [ch.to_dict() for ch in self.channels],
        "timing": self.timing.to_dict() if self.timing is not None else None,
        "trigger": self.trigger.to_dict() if self.trigger is not None else None,
        "logging": self.logging.to_dict() if self.logging is not None else None,
        "metadata": dict(self.metadata),
    }

def __post_init__(self) -> None:
    """Validate timing parameters before they reach NI."""
    if self.rate_hz <= 0.0:
        raise NIDaqValidationError(f"rate_hz must be > 0, got {self.rate_hz!r}")
    if self.samples_per_channel is not None and self.samples_per_channel <= 0:
        raise NIDaqValidationError(
            f"samples_per_channel must be > 0 when set, got {self.samples_per_channel!r}"
        )

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> Self:
    """Deserialise from a dict produced by :meth:`to_dict`.

    Args:
        data: Mapping carrying the timing fields.

    Raises:
        NIDaqValidationError: An enum field carries an unknown value.
    """
    try:
        mode = AcquisitionMode(data.get("mode", AcquisitionMode.CONTINUOUS.value))
    except ValueError as exc:
        raise NIDaqValidationError(f"unknown AcquisitionMode {data.get('mode')!r}") from exc
    try:
        edge = Edge(data.get("active_edge", Edge.RISING.value))
    except ValueError as exc:
        raise NIDaqValidationError(f"unknown Edge {data.get('active_edge')!r}") from exc
    return cls(
        rate_hz=float(data["rate_hz"]),
        mode=mode,
        samples_per_channel=(
            int(data["samples_per_channel"])
            if data.get("samples_per_channel") is not None
            else None
        ),
        source=data.get("source"),
        active_edge=edge,
    )

def to_dict(self) -> dict[str, Any]:
    """Serialise to a JSON-friendly dict.

    Enum members serialise to their string values so the result is
    JSON-encodable without a custom encoder.
    """
    return {
        "rate_hz": self.rate_hz,
        "mode": self.mode.value,
        "samples_per_channel": self.samples_per_channel,
        "source": self.source,
        "active_edge": self.active_edge.value,
    }