nidaqlib¶

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> Self:
    """Deserialise, restoring :class:`AnalogTriggerSlope` from its value."""
    kind = data.get("kind")
    if kind != cls.kind:
        raise NIDaqValidationError(f"kind mismatch: expected {cls.kind!r}, got {kind!r}")
    try:
        slope = AnalogTriggerSlope(data.get("slope", AnalogTriggerSlope.RISING.value))
    except ValueError as exc:
        raise NIDaqValidationError(f"unknown AnalogTriggerSlope {data.get('slope')!r}") from exc
    return cls(
        source=str(data["source"]),
        level=float(data["level"]),
        slope=slope,
    )

def to_dict(self) -> dict[str, Any]:
    """Serialise; encode :class:`AnalogTriggerSlope` to its string value."""
    return {
        "kind": self.kind,
        "source": self.source,
        "level": self.level,
        "slope": self.slope.value,
    }

def __post_init__(self) -> None:
    """Validate the ADC-timing pairing on top of the base channel checks."""
    # Direct call (not super()) — the @dataclass(slots=True) decorator
    # rewrites the class, which leaves super()'s __class__ cell pointing
    # at a now-unrelated class object.
    ChannelSpec.__post_init__(self)
    from nidaqmx.constants import ADCTimingMode  # noqa: PLC0415

    is_custom = self.adc_timing_mode is ADCTimingMode.CUSTOM
    if self.adc_custom_timing_mode is not None and not is_custom:
        raise NIDaqValidationError(
            f"adc_custom_timing_mode is only valid with "
            f"adc_timing_mode=ADCTimingMode.CUSTOM on {self.display_name!r}; "
            f"got adc_timing_mode={self.adc_timing_mode!r}"
        )
    if is_custom and self.adc_custom_timing_mode is None:
        raise NIDaqValidationError(
            f"adc_timing_mode=ADCTimingMode.CUSTOM requires adc_custom_timing_mode "
            f"on {self.display_name!r}"
        )

def to_dict(self) -> dict[str, Any]:
    """Serialise via each enum's ``.value`` so the payload is JSON-encodable."""
    # Direct call (not super()) — see __post_init__ note.
    payload = ChannelSpec.to_dict(self)
    payload["adc_timing_mode"] = (
        self.adc_timing_mode.value if self.adc_timing_mode is not None else None
    )
    payload["auto_zero_mode"] = (
        self.auto_zero_mode.value if self.auto_zero_mode is not None else None
    )
    return payload

def __post_init__(self) -> None:
    """Validate the voltage range."""
    AnalogInputBase.__post_init__(self)
    if self.min_val >= self.max_val:
        raise NIDaqValidationError(
            f"min_val must be < max_val for {self.display_name!r}; "
            f"got {self.min_val!r} >= {self.max_val!r}"
        )

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> Self:
    """Reconstruct, restoring enum members from their serialised ``.value`` ints."""
    kind = data.get("kind")
    if kind != cls.kind:
        raise NIDaqValidationError(f"kind mismatch: expected {cls.kind!r}, got {kind!r}")
    payload = {k: v for k, v in data.items() if k != "kind"}
    AnalogInputBase._restore_base_enums(payload)
    if payload.get("terminal_config") is not None:
        payload["terminal_config"] = _coerce_terminal_config(payload["terminal_config"])
    return cls(**payload)

def to_dict(self) -> dict[str, Any]:
    """Serialise, encoding ``terminal_config`` via its ``.value`` int."""
    payload = AnalogInputBase.to_dict(self)
    payload["terminal_config"] = (
        self.terminal_config.value if self.terminal_config is not None else None
    )
    return payload

def __post_init__(self) -> None:
    """Validate the output and safety ranges."""
    ChannelSpec.__post_init__(self)
    if self.min_val >= self.max_val:
        raise NIDaqValidationError(
            f"min_val must be < max_val for {self.display_name!r}; "
            f"got {self.min_val!r} >= {self.max_val!r}"
        )
    lo = self.effective_safe_min
    hi = self.effective_safe_max
    if lo > hi:
        raise NIDaqValidationError(
            f"safe_min must be <= safe_max for {self.display_name!r}; got {lo!r} > {hi!r}"
        )
    if lo < self.min_val or hi > self.max_val:
        raise NIDaqValidationError(
            f"safe range [{lo}, {hi}] must stay inside device range "
            f"[{self.min_val}, {self.max_val}] for {self.display_name!r}"
        )

def __post_init__(self) -> None:
    """Validate and freeze common channel metadata."""
    if not self.physical_channel:
        raise NIDaqValidationError("physical_channel must be a non-empty string")
    if self.name is not None and not self.name:
        raise NIDaqValidationError("name must be non-empty when provided")
    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`.

    On the base class, this dispatches to the registered subclass for the
    ``kind`` discriminator. On a concrete subclass, this validates that
    ``kind`` matches and constructs the dataclass directly.

    Args:
        data: Mapping carrying the ``kind`` discriminator and field values.

    Raises:
        NIDaqValidationError: ``kind`` is missing, unknown, or does not
            match the concrete class.
    """
    kind = data.get("kind")
    if cls is ChannelSpec:
        if not isinstance(kind, str):
            raise NIDaqValidationError(
                f"channel spec dict missing 'kind' discriminator (got {kind!r})"
            )
        target = _CHANNEL_REGISTRY.get(kind)
        if target is None:
            raise NIDaqValidationError(f"unknown channel kind {kind!r}")
        # Mypy can't see that the registered class returns Self here; the
        # dispatch is dynamic by design, so cast at the boundary.
        return target.from_dict(data)  # type: ignore[return-value]
    if kind != cls.kind:
        raise NIDaqValidationError(f"kind mismatch: expected {cls.kind!r}, got {kind!r}")
    payload = {k: v for k, v in data.items() if k != "kind"}
    return cls(**payload)

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

    Returns:
        A dict carrying ``kind`` plus every dataclass field. Mappings are
        copied to plain ``dict`` so the result is JSON-encodable.
    """
    payload: dict[str, Any] = {}
    for spec in dataclasses.fields(self):
        value = getattr(self, spec.name)
        if isinstance(value, Mapping):
            value = dict(cast("Mapping[str, Any]", value))
        payload[spec.name] = value
    payload["kind"] = self.kind
    return payload

def __post_init__(self) -> None:
    """Validate common channel metadata."""
    ChannelSpec.__post_init__(self)

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> Self:
    """Deserialise, restoring :class:`Edge` from its string value."""
    kind = data.get("kind")
    if kind != cls.kind:
        raise NIDaqValidationError(f"kind mismatch: expected {cls.kind!r}, got {kind!r}")
    payload = {k: v for k, v in data.items() if k != "kind"}
    try:
        payload["edge"] = Edge(payload.get("edge", Edge.RISING.value))
    except ValueError as exc:
        raise NIDaqValidationError(f"unknown Edge {payload.get('edge')!r}") from exc
    return cls(**payload)

def to_dict(self) -> dict[str, Any]:
    """Serialise; encode :class:`Edge` to its string value."""
    payload = ChannelSpec.to_dict(self)
    payload["edge"] = self.edge.value
    return payload

def __post_init__(self) -> None:
    """Validate the expected frequency range."""
    ChannelSpec.__post_init__(self)
    if self.min_val <= 0.0:
        raise NIDaqValidationError(f"min_val must be > 0 for {self.display_name!r}")
    if self.min_val >= self.max_val:
        raise NIDaqValidationError(
            f"min_val must be < max_val for {self.display_name!r}; "
            f"got {self.min_val!r} >= {self.max_val!r}"
        )

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> Self:
    """Deserialise, restoring :class:`Edge` from its string value."""
    kind = data.get("kind")
    if kind != cls.kind:
        raise NIDaqValidationError(f"kind mismatch: expected {cls.kind!r}, got {kind!r}")
    payload = {k: v for k, v in data.items() if k != "kind"}
    try:
        payload["edge"] = Edge(payload.get("edge", Edge.RISING.value))
    except ValueError as exc:
        raise NIDaqValidationError(f"unknown Edge {payload.get('edge')!r}") from exc
    return cls(**payload)

def to_dict(self) -> dict[str, Any]:
    """Serialise; encode :class:`Edge` to its string value."""
    payload = ChannelSpec.to_dict(self)
    payload["edge"] = self.edge.value
    return payload

def __post_init__(self) -> None:
    """Validate the expected period range."""
    ChannelSpec.__post_init__(self)
    if self.min_val <= 0.0:
        raise NIDaqValidationError(f"min_val must be > 0 for {self.display_name!r}")
    if self.min_val >= self.max_val:
        raise NIDaqValidationError(
            f"min_val must be < max_val for {self.display_name!r}; "
            f"got {self.min_val!r} >= {self.max_val!r}"
        )

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> Self:
    """Deserialise, restoring :class:`Edge` from its string value."""
    kind = data.get("kind")
    if kind != cls.kind:
        raise NIDaqValidationError(f"kind mismatch: expected {cls.kind!r}, got {kind!r}")
    payload = {k: v for k, v in data.items() if k != "kind"}
    try:
        payload["edge"] = Edge(payload.get("edge", Edge.RISING.value))
    except ValueError as exc:
        raise NIDaqValidationError(f"unknown Edge {payload.get('edge')!r}") from exc
    return cls(**payload)

def to_dict(self) -> dict[str, Any]:
    """Serialise; encode :class:`Edge` to its string value."""
    payload = ChannelSpec.to_dict(self)
    payload["edge"] = self.edge.value
    return payload

def __post_init__(self) -> None:
    """Validate pulse-train parameters."""
    ChannelSpec.__post_init__(self)
    if self.frequency <= 0.0:
        raise NIDaqValidationError(f"frequency must be > 0 for {self.display_name!r}")
    if not 0.0 < self.duty_cycle < 1.0:
        raise NIDaqValidationError(
            f"duty_cycle must be in (0.0, 1.0) for {self.display_name!r}; "
            f"got {self.duty_cycle!r}"
        )
    if self.initial_delay < 0.0:
        raise NIDaqValidationError(f"initial_delay must be >= 0 for {self.display_name!r}")

def __post_init__(self) -> None:
    """Validate pulse tick parameters."""
    ChannelSpec.__post_init__(self)
    if self.high_ticks <= 0 or self.low_ticks <= 0:
        raise NIDaqValidationError(
            f"high_ticks and low_ticks must be > 0 for {self.display_name!r}"
        )
    if self.initial_delay < 0:
        raise NIDaqValidationError(f"initial_delay must be >= 0 for {self.display_name!r}")

def __post_init__(self) -> None:
    """Validate pulse timing parameters."""
    ChannelSpec.__post_init__(self)
    if self.high_time <= 0.0 or self.low_time <= 0.0:
        raise NIDaqValidationError(
            f"high_time and low_time must be > 0 for {self.display_name!r}"
        )
    if self.initial_delay < 0.0:
        raise NIDaqValidationError(f"initial_delay must be >= 0 for {self.display_name!r}")

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 add(
    self,
    name: str,
    source: TaskSpec | DaqSession,
    *,
    backend: DaqBackend | None = None,
) -> DaqSession:
    """Register a task with this manager. Idempotent on duplicate ``name``.

    Performs a best-effort preflight conflict check against tasks
    already managed (design doc §15.3). NI is the final authority —
    the preflight only catches obvious overlaps.

    ``add`` does **not** allocate NI resources — it constructs a
    :class:`DaqSession` and records it. The session's
    :meth:`DaqSession.configure` (which creates the NI task and applies
    channels / timing / logging / triggers) runs lazily on the first
    :meth:`start` for the task. Any NI rejection of the ``spec`` (bad
    physical channel, unsupported channel kind, sample rate above
    device max, …) therefore surfaces at :meth:`start` time, not
    :meth:`add` time. The preflight catches operator-side overlap
    only; everything NI validates lives downstream.

    Args:
        name: Manager-side label for this task. Must be unique.
        source: Either a :class:`TaskSpec` (manager constructs a
            fresh :class:`DaqSession`) or a pre-built
            :class:`DaqSession` (manager registers it as-is). The
            pre-built form aligns with the ecosystem ``add(name,
            source)`` convention shared by :class:`watlowlib.WatlowManager`,
            :class:`alicatlib.AlicatManager`, and
            :class:`sartoriuslib.SartoriusManager`.
        backend: Optional :class:`DaqBackend`. Defaults to
            :class:`NidaqmxBackend` (lazy import). Ignored when
            ``source`` is a :class:`DaqSession` (which already
            carries its own backend).

    Returns:
        The :class:`DaqSession` registered under ``name``. Re-adding
        the same ``(name, source)`` returns the existing session and
        bumps a refcount.

    Raises:
        NIDaqTaskStateError: ``name`` already maps to a different
            spec, or the manager is closed.
        NIDaqResourceError: ``source`` overlaps physical channels
            with an already-managed task.
    """
    if self._closed:
        raise NIDaqTaskStateError(
            "DaqManager is closed",
            context=ErrorContext(task_name=name, command_name="manager.add"),
        )
    if isinstance(source, DaqSession):
        spec = source.spec
        prebuilt: DaqSession | None = source
    else:
        spec = source
        prebuilt = None
    async with self._global_lock:
        existing = self._sessions.get(name)
        if existing is not None:
            if self._specs.get(name) is not spec and self._specs.get(name) != spec:
                raise NIDaqTaskStateError(
                    f"task {name!r} already registered with a different spec",
                    context=ErrorContext(task_name=name, command_name="manager.add"),
                )
            self._refcounts[name] = self._refcounts.get(name, 1) + 1
            return existing

        if prebuilt is not None:
            session = prebuilt
        else:
            if backend is None:
                from nidaqlib.backend.nidaqmx_backend import NidaqmxBackend  # noqa: PLC0415

                backend = NidaqmxBackend()
            await self._preflight_conflicts(name, spec, backend)
            session = DaqSession(spec, backend)
        self._sessions[name] = session
        self._specs[name] = spec
        self._refcounts[name] = 1
        self._order.append(name)
        self._task_locks[name] = anyio.Lock()
        devices = tuple(sorted({_device_of(ch.physical_channel) for ch in spec.channels}))
        self._task_devices[name] = devices
        for ch in spec.channels:
            dev = _device_of(ch.physical_channel)
            self._device_locks.setdefault(dev, anyio.Lock())
        return session

async def close(self) -> None:
    """Tear down every managed session in LIFO order. Idempotent.

    Failures are collected into an :class:`ExceptionGroup`; one slow /
    broken close does not prevent others from running.
    """
    if self._closed:
        return
    self._closed = True
    # Snapshot under the global lock, then close outside it so unrelated
    # ops (e.g. a recorder still draining) are not blocked on a long NI
    # close call.
    async with self._global_lock:
        order = list(reversed(self._order))
        sessions = {name: self._sessions[name] for name in order if name in self._sessions}
        self._sessions.clear()
        self._specs.clear()
        self._refcounts.clear()
        self._order.clear()
        self._task_locks.clear()
        self._task_devices.clear()
        self._device_locks.clear()
    errors: list[BaseException] = []
    for name in order:
        session = sessions.get(name)
        if session is None:
            continue
        try:
            await session.close()
        except BaseException as exc:
            # Collected and re-grouped below — one slow / broken close
            # must not prevent the rest of the LIFO unwind from running.
            errors.append(exc)
    if errors:
        raise BaseExceptionGroup("DaqManager.close: one or more sessions failed", errors)

def get(self, name: str) -> DaqSession:
    """Return the session registered under ``name``.

    Raises:
        KeyError: ``name`` is unknown.
    """
    return self._sessions[name]

async def poll(
    self,
    names: Sequence[str] | None = None,
    *,
    timeout: float | None = None,  # noqa: ASYNC109 — NI per-call timeout, not coroutine
    error_policy: ErrorPolicy | None = None,
) -> Mapping[str, DeviceResult[DaqReading]]:
    """Poll one or more tasks once each. Returns one :class:`DaqReading` per task."""

    async def _do(session: DaqSession) -> DaqReading:
        return await session.poll(timeout=timeout)

    return await self._for_each(
        names,
        "poll",
        _do,
        error_policy=error_policy,
    )

async def read_block(
    self,
    samples_per_channel: int,
    names: Sequence[str] | None = None,
    *,
    timeout: float | None = None,  # noqa: ASYNC109 — NI per-call timeout, not coroutine
    error_policy: ErrorPolicy | None = None,
) -> Mapping[str, DeviceResult[DaqBlock]]:
    """Read one block per task in parallel."""

    async def _do(session: DaqSession) -> DaqBlock:
        return await session.read_block(samples_per_channel, timeout=timeout)

    return await self._for_each(
        names,
        "read_block",
        _do,
        error_policy=error_policy,
    )

async def remove(self, name: str) -> None:
    """Decrement refcount; tear down on the last :meth:`remove`.

    A no-op for unknown names — matches sibling parity.

    Raises:
        NIDaqError: Surfaced from session close (collected into a
            group when called from :meth:`close`).
    """
    async with self._global_lock:
        if name not in self._sessions:
            return
        self._refcounts[name] -= 1
        if self._refcounts[name] > 0:
            return
        session = self._sessions.pop(name)
        self._specs.pop(name, None)
        self._refcounts.pop(name, None)
        self._task_locks.pop(name, None)
        self._task_devices.pop(name, None)
        with contextlib.suppress(ValueError):
            self._order.remove(name)
    # Close outside the global lock so a slow NI close doesn't block
    # other manager ops on unrelated tasks.
    await session.close()

async def start(
    self,
    names: Sequence[str] | None = None,
    *,
    error_policy: ErrorPolicy | None = None,
    confirm: bool = False,
) -> Mapping[str, DeviceResult[None]]:
    """Start one or more managed tasks. Defaults to all in add-order."""

    async def _do(session: DaqSession) -> None:
        await _configure_then_start(session, confirm=confirm)

    return await self._for_each(
        names,
        "start",
        _do,
        error_policy=error_policy,
    )