servomexlib.streaming¶

@asynccontextmanager
async def record(
    source: PollSource,
    *,
    rate_hz: float,
    duration: float | None = None,
    names: Sequence[str] | None = None,
    timeout: float | None = None,
    overflow: OverflowPolicy = OverflowPolicy.BLOCK,
    buffer_size: int = 64,
    auto_reconnect: bool = False,
    reconnect_factory: Callable[[], Awaitable[PollSource]] | None = None,
) -> AsyncGenerator[Recording[Sequence[Sample]]]:
    """Record polled samples into a receive stream at an absolute cadence.

    Usage::

        async with record(analyzer, rate_hz=2, duration=10) as rec:
            async for batch in rec.stream:
                for sample in batch:
                    print(sample.channel, sample.value)

    Args:
        source: Any :class:`PollSource` (an :class:`Analyzer` or a
            :class:`ServomexManager`).
        rate_hz: Target cadence; ``target[n] = start + n / rate_hz``. Must be > 0.
        duration: Total acquisition seconds, or ``None`` for "until the caller
            exits the CM".
        names: Subset of device names to poll (manager only); ``None`` polls all.
        timeout: Per-poll I/O ceiling passed to ``source.poll_samples``.
        overflow: Backpressure policy when the buffer is full.
        buffer_size: Receive-stream capacity, in per-tick batches.
        auto_reconnect: Treat :class:`ServomexConnectionError` from the source as
            a transient drop: log, back off, optionally rebuild via
            ``reconnect_factory``, and keep going (missed ticks count late).
        reconnect_factory: Rebuilds the source after a disconnect when supplied.

    Yields:
        A :class:`Recording[Sequence[Sample]]`.

    Raises:
        ValueError: ``rate_hz <= 0``, ``duration <= 0``, or ``buffer_size < 1``.
    """
    if rate_hz <= 0:
        raise ValueError(f"rate_hz must be > 0, got {rate_hz!r}")
    if duration is not None and duration <= 0:
        raise ValueError(f"duration must be > 0 or None, got {duration!r}")
    if buffer_size < 1:
        raise ValueError(f"buffer_size must be >= 1, got {buffer_size!r}")

    period = 1.0 / rate_hz
    total_ticks = None if duration is None else max(1, round(duration * rate_hz))

    send_stream, receive_stream = anyio.create_memory_object_stream[Sequence[Sample]](
        max_buffer_size=buffer_size,
    )
    # Producer-side clone for DROP_OLDEST eviction, off the consumer's iterator.
    drop_rx = receive_stream.clone()

    started_at = datetime.now(UTC)
    summary = AcquisitionSummary(started_at=started_at)
    tick_durations_ms: list[float] = []
    _logger.info(
        "recorder.start rate_hz=%s duration_s=%s overflow=%s buffer_size=%s names=%s",
        rate_hz,
        duration,
        overflow.value,
        buffer_size,
        list(names) if names is not None else None,
    )

    try:
        async with anyio.create_task_group() as tg, receive_stream:

            async def _producer_entrypoint() -> None:
                await _run_producer(
                    source,
                    send_stream,
                    drop_rx,
                    names,
                    timeout,
                    period,
                    total_ticks,
                    overflow,
                    summary,
                    tick_durations_ms,
                    auto_reconnect=auto_reconnect,
                    reconnect_factory=reconnect_factory,
                )

            tg.start_soon(_producer_entrypoint)
            try:
                yield Recording(stream=receive_stream, summary=summary, rate_hz=rate_hz)
            finally:
                tg.cancel_scope.cancel()
    except BaseExceptionGroup as eg:
        # A lone producer failure (e.g. ServomexConnectionError without
        # auto_reconnect) surfaces as a single-member group from the task group;
        # collapse it so callers catch the concrete error, not the wrapper.
        raise _collapse(eg) from None
    finally:
        summary.finished_at = datetime.now(UTC)
        summary.tick_duration_ms_p50, summary.tick_duration_ms_p99 = _tick_percentiles(
            tick_durations_ms
        )
        _logger.info(
            "recorder.stop emitted=%s late=%s max_drift_ms=%.3f tick_p50_ms=%.3f tick_p99_ms=%.3f",
            summary.samples_emitted,
            summary.samples_late,
            summary.max_drift_ms,
            summary.tick_duration_ms_p50,
            summary.tick_duration_ms_p99,
        )