nidaqlib.streaming¶

@asynccontextmanager
async def record(
    source: DaqSession,
    *,
    chunk_size: int,
    timeout: float = 10.0,  # noqa: ASYNC109 — per-NI-read timeout, not coroutine
    buffer_size: int = 16,
    error_policy: ErrorPolicy = ErrorPolicy.RAISE,
    overflow: OverflowPolicy = OverflowPolicy.DROP_OLDEST,
    use_callback_bridge: bool = False,
) -> AsyncGenerator[Recording[DaqBlock]]:
    """Hardware-clocked block acquisition.

    Yields a :class:`Recording[DaqBlock]`. The stream is closed when this
    context manager exits; ``summary`` is mutated in place during the run
    and is safe to read after exit.

    Args:
        source: A configured :class:`DaqSession`. Required state depends on
            ``use_callback_bridge``:

            * ``use_callback_bridge=False`` (Option A) — ``source`` must be
              **started**; wrap with :func:`~nidaqlib.tasks.open_device` (the
              default ``autostart=True``).
            * ``use_callback_bridge=True`` (Option B / §11.3.2) — ``source``
              must be **configured but not yet started**; pass
              ``autostart=False`` to ``open_device`` and let the recorder own
              the start. NI rejects buffer-event registration on a running
              task with -200960 ("Register all your DAQmx software events
              prior to starting the task").
        chunk_size: Samples per channel per emitted :class:`DaqBlock`.
        timeout: Per-read timeout in seconds (Option A only — Option B reads
            from the NI buffer with timeout 0).
        buffer_size: AnyIO memory-object stream buffer, in :class:`DaqBlock`
            slots. Older blocks may be dropped per ``overflow`` when full.
        error_policy: :attr:`ErrorPolicy.RAISE` (default) cancels the task
            group on error; :attr:`ErrorPolicy.RETURN` emits an error-tagged
            :class:`DaqBlock` and continues. Option B (callback bridge)
            currently honours only :attr:`RAISE` — :attr:`RETURN` is wired
            for the Option A producer.
        overflow: Backpressure policy. ``DROP_OLDEST`` is the
            hardware-clocked default — see design doc §13.3.
        use_callback_bridge: Opt into the §11.3.2 every-N-samples callback
            path. Default ``False`` selects Option A (blocking read in a
            worker thread).

    Raises:
        NIDaqTaskStateError: ``source`` is in the wrong lifecycle state for
            the selected mode (see ``source`` argument above).
        ValueError: ``chunk_size < 1`` or ``buffer_size < 1``.
    """
    if chunk_size < 1:
        raise ValueError(f"chunk_size must be >= 1, got {chunk_size!r}")
    if buffer_size < 1:
        raise ValueError(f"buffer_size must be >= 1, got {buffer_size!r}")
    _validate_record_state(
        source,
        error_policy=error_policy,
        use_callback_bridge=use_callback_bridge,
    )

    summary = AcquisitionSummary()
    timing = source.spec.timing
    rate_hz: float | None = (
        timing.rate_hz
        if timing is not None and timing.mode is not AcquisitionMode.ON_DEMAND
        else None
    )
    tx, rx = anyio.create_memory_object_stream[DaqBlock](max_buffer_size=buffer_size)
    drop_rx = rx.clone()

    # TDMS LoggingMode.LOG bypasses the application read path. If we tried
    # to drive the producer, ``read_block`` would block forever waiting on
    # samples that never arrive. Detect at entry and emit an empty stream
    # so consumers see ``blocks_emitted == 0``.
    if _is_log_only(source):
        async with rx, drop_rx:
            await tx.aclose()
            try:
                yield Recording(stream=rx, summary=summary, rate_hz=rate_hz)
            finally:
                summary.finished_at = datetime.now(UTC)
        return

    async with anyio.create_task_group() as tg, rx, drop_rx:
        if use_callback_bridge:
            await _start_bridge_producer(tg, source, tx, drop_rx, summary, chunk_size, overflow)
        else:
            _ = tg.start_soon(
                _blocking_producer,
                source,
                tx,
                drop_rx,
                summary,
                chunk_size,
                timeout,
                overflow,
                error_policy,
            )
        try:
            yield Recording(stream=rx, summary=summary, rate_hz=rate_hz)
        finally:
            await tx.aclose()
            tg.cancel()
    summary.finished_at = datetime.now(UTC)

@asynccontextmanager
async def record_polled(
    source: DaqSession | DaqManager | PollSource,
    *,
    rate_hz: float,
    error_policy: ErrorPolicy = ErrorPolicy.RAISE,
    overflow: OverflowPolicy = OverflowPolicy.BLOCK,
    buffer_size: int = 64,
) -> AsyncGenerator[Recording[_PolledItem]]:
    """Software-timed scalar polling at ``rate_hz``.

    Yields a :class:`Recording[T]`. The per-tick payload type ``T`` depends
    on ``source``:

    - :class:`DaqSession` → one :class:`DaqReading` per tick.
    - :class:`DaqManager` → ``Mapping[str, DeviceResult[DaqReading]]`` per
      tick (matches :meth:`DaqManager.poll`).
    - Any :class:`PollSource` (including :class:`PollSourceAdapter`) →
      ``Mapping[str, DeviceResult[DaqReading]]`` returned by its ``poll()``.

    ``summary`` is updated in place during the run; a final snapshot is
    frozen on exit.

    Args:
        source: A started :class:`DaqSession` (whose timing is ``None`` or
            :attr:`AcquisitionMode.ON_DEMAND`), a :class:`DaqManager`, or
            any :class:`PollSource`.
        rate_hz: Target poll rate, in Hz. Must be > 0.
        error_policy: :attr:`RAISE` cancels the recorder on a poll error;
            :attr:`RETURN` emits a :class:`DaqReading` (or per-task
            :class:`DeviceResult` row) with the error attached and continues.
        overflow: Backpressure policy. Defaults to :attr:`BLOCK` —
            software-timed pollers can pause safely (design §13.3).
        buffer_size: AnyIO send-stream capacity in payload slots.

    Raises:
        NIDaqTaskStateError: A session ``source`` is not started, or a
            manager ``source`` is closed / has no managed tasks.
        ValueError: ``rate_hz <= 0`` or ``buffer_size < 1``.
    """
    if rate_hz <= 0:
        raise ValueError(f"rate_hz must be > 0, got {rate_hz!r}")
    if buffer_size < 1:
        raise ValueError(f"buffer_size must be >= 1, got {buffer_size!r}")

    # Late imports — manager imports streaming.block, session pulls in
    # backend modules; both are heavy enough to keep off the recorder
    # module's import path.
    from nidaqlib.manager import DaqManager  # noqa: PLC0415
    from nidaqlib.tasks.session import DaqSession  # noqa: PLC0415

    if isinstance(source, DaqManager):
        if source.is_closed:
            raise NIDaqTaskStateError(
                "record_polled() requires an open DaqManager; got a closed manager",
                context=ErrorContext(command_name="record_polled"),
            )
        if not source.names:
            raise NIDaqTaskStateError(
                "record_polled() requires a DaqManager with at least one task",
                context=ErrorContext(command_name="record_polled"),
            )
    elif isinstance(source, DaqSession) and not source.is_started:
        raise NIDaqTaskStateError(
            f"record_polled() requires a started session; task {source.spec.name!r} is not running",
            context=ErrorContext(task_name=source.spec.name, command_name="record_polled"),
        )

    summary = AcquisitionSummary()
    period = 1.0 / rate_hz
    tx, rx = anyio.create_memory_object_stream[_PolledItem](max_buffer_size=buffer_size)
    drop_rx = rx.clone()

    async with anyio.create_task_group() as tg, rx, drop_rx:
        if isinstance(source, DaqManager):
            _ = tg.start_soon(
                _polled_manager_producer,
                source,
                tx,
                drop_rx,
                summary,
                period,
                error_policy,
                overflow,
            )
        elif isinstance(source, DaqSession):
            _ = tg.start_soon(
                _polled_producer,
                source,
                tx,
                drop_rx,
                summary,
                period,
                error_policy,
                overflow,
            )
        else:
            # Any other PollSource — drive its poll() directly. Emits the
            # same Mapping[str, DeviceResult[DaqReading]] shape as the
            # manager path.
            _ = tg.start_soon(
                _polled_source_producer,
                source,
                tx,
                drop_rx,
                summary,
                period,
                error_policy,
                overflow,
            )
        try:
            yield Recording(stream=rx, summary=summary, rate_hz=rate_hz)
        finally:
            await tx.aclose()
            tg.cancel()
    summary.finished_at = datetime.now(UTC)