Skip to content

nidaqlib.channels

nidaqlib.channels

Channel specifications — :class:ChannelSpec and concrete subclasses.

AnalogInputVoltage dataclass

AnalogInputVoltage(
    *,
    physical_channel,
    name=None,
    unit=None,
    metadata=_empty_metadata(),
    min_val=-10.0,
    max_val=10.0,
    terminal_config=None,
    custom_scale_name=None,
)

Bases: ChannelSpec

Voltage analog-input channel.

Maps to Task.ai_channels.add_ai_voltage_chan on the NI side.

Attributes:

Name Type Description
min_val float

Lower limit of the expected input range, in volts.

max_val float

Upper limit of the expected input range, in volts. The NI driver uses the (min, max) range to select the most appropriate on-board gain.

terminal_config TerminalConfiguration | None

Terminal configuration (RSE / NRSE / DIFF / PSEUDO_DIFF). None lets NI pick the device default.

custom_scale_name str | None

Optional name of a pre-configured custom scale registered in MAX. When set, min_val/max_val are scaled engineering units, not volts.

__post_init__

__post_init__()

Validate the voltage range.

Source code in src/nidaqlib/channels/analog_input.py
def __post_init__(self) -> None:
    """Validate the voltage range."""
    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}"
        )

AnalogOutputVoltage dataclass

AnalogOutputVoltage(
    *,
    physical_channel,
    name=None,
    unit=None,
    metadata=_empty_metadata(),
    min_val=-10.0,
    max_val=10.0,
    safe_min=None,
    safe_max=None,
    requires_confirm=True,
    terminal_config=None,
    custom_scale_name=None,
)

Bases: ChannelSpec

Voltage analog-output channel.

Maps to Task.ao_channels.add_ao_voltage_chan on the NI side. Writes are gated through :meth:DaqSession.write, which rejects out-of-range values against safe_min / safe_max and requires confirm=True whenever any target channel sets requires_confirm (design doc §17.1).

Attributes:

Name Type Description
min_val float

Lower bound of the device output range, in volts.

max_val float

Upper bound of the device output range, in volts. NI uses (min, max) to select the output gain.

safe_min float | None

Optional lower-end safety clamp for application writes. None means "use min_val as the clamp." Out-of-range writes raise :class:NIDaqValidationError — never silently clamped.

safe_max float | None

Optional upper-end safety clamp. None means "use max_val."

requires_confirm bool

When True, every :meth:DaqSession.write targeting this channel must pass confirm=True. Defaults to True — outputs default to safe.

terminal_config TerminalConfiguration | None

Terminal configuration (RSE / DIFF / ...). None lets NI pick the device default.

custom_scale_name str | None

Optional name of a pre-configured custom scale registered in MAX. When set, min_val / max_val are engineering units, not volts.

effective_safe_max property

effective_safe_max

Resolved upper clamp — falls back to :attr:max_val.

effective_safe_min property

effective_safe_min

Resolved lower clamp — falls back to :attr:min_val.

__post_init__

__post_init__()

Validate the output and safety ranges.

Source code in src/nidaqlib/channels/analog_output.py
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}"
        )

ChannelSpec dataclass

ChannelSpec(
    *,
    physical_channel,
    name=None,
    unit=None,
    metadata=_empty_metadata(),
)

Application-facing description of one DAQ channel.

Attributes:

Name Type Description
physical_channel str

NI physical channel identifier, e.g. "Dev1/ai0".

name str | None

Optional friendly name; defaults to the physical channel.

unit str | None

Optional engineering unit string ("V", "degC", ...). Used by sinks for column headers; not interpreted by the backend.

metadata Mapping[str, str | int | float | bool]

Free-form scalar metadata propagated into emitted records.

display_name property

display_name

Return name if set, otherwise the physical channel.

kind class-attribute

kind = ''

Discriminator used by :meth:from_dict. Concrete subclasses override.

__post_init__

__post_init__()

Validate and freeze common channel metadata.

Source code in src/nidaqlib/channels/base.py
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)))

from_dict classmethod

from_dict(data)

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.

Parameters:

Name Type Description Default
data Mapping[str, Any]

Mapping carrying the kind discriminator and field values.

required

Raises:

Type Description
NIDaqValidationError

kind is missing, unknown, or does not match the concrete class.

Source code in src/nidaqlib/channels/base.py
@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)

to_dict

to_dict()

Serialise to a JSON/TOML-friendly dict, including kind.

Returns:

Type Description
dict[str, Any]

A dict carrying kind plus every dataclass field. Mappings are

dict[str, Any]

copied to plain dict so the result is JSON-encodable.

Source code in src/nidaqlib/channels/base.py
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

CounterEdgeCountInput dataclass

CounterEdgeCountInput(
    *,
    physical_channel,
    name=None,
    unit=None,
    metadata=_empty_metadata(),
    edge=Edge.RISING,
    initial_count=0,
    count_up=True,
)

Bases: ChannelSpec

Edge-count counter-input channel.

Maps to Task.ci_channels.add_ci_count_edges_chan on the NI side. Useful for encoders, totalisers, or anything that needs raw edge accumulation.

Attributes:

Name Type Description
edge Edge

Edge that increments / decrements the counter. Rising by default.

initial_count int

Starting value of the counter. Defaults to 0.

count_up bool

When True (default), every active edge increments the counter; when False, decrements. Mirrors NI's CountDirection.COUNT_UP / COUNT_DOWN.

__post_init__

__post_init__()

Validate common channel metadata.

Source code in src/nidaqlib/channels/counter_input.py
def __post_init__(self) -> None:
    """Validate common channel metadata."""
    ChannelSpec.__post_init__(self)

from_dict classmethod

from_dict(data)

Deserialise, restoring :class:Edge from its string value.

Source code in src/nidaqlib/channels/counter_input.py
@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)

to_dict

to_dict()

Serialise; encode :class:Edge to its string value.

Source code in src/nidaqlib/channels/counter_input.py
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

CounterFrequencyInput dataclass

CounterFrequencyInput(
    *,
    physical_channel,
    name=None,
    unit=None,
    metadata=_empty_metadata(),
    min_val,
    max_val,
    edge=Edge.RISING,
)

Bases: ChannelSpec

Frequency-measurement counter-input channel.

Maps to Task.ci_channels.add_ci_freq_chan on the NI side. NI uses (min_val, max_val) to choose timebases that resolve frequencies in the expected range.

Attributes:

Name Type Description
min_val float

Lower bound of the expected frequency, in Hz.

max_val float

Upper bound of the expected frequency, in Hz.

edge Edge

Edge of the input signal that increments the counter. Rising by default.

__post_init__

__post_init__()

Validate the expected frequency range.

Source code in src/nidaqlib/channels/counter_input.py
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}"
        )

from_dict classmethod

from_dict(data)

Deserialise, restoring :class:Edge from its string value.

Source code in src/nidaqlib/channels/counter_input.py
@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)

to_dict

to_dict()

Serialise; encode :class:Edge to its string value.

Source code in src/nidaqlib/channels/counter_input.py
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

CounterPeriodInput dataclass

CounterPeriodInput(
    *,
    physical_channel,
    name=None,
    unit=None,
    metadata=_empty_metadata(),
    min_val,
    max_val,
    edge=Edge.RISING,
)

Bases: ChannelSpec

Period-measurement counter-input channel.

Maps to Task.ci_channels.add_ci_period_chan on the NI side.

Attributes:

Name Type Description
min_val float

Lower bound of the expected period, in seconds.

max_val float

Upper bound of the expected period, in seconds.

edge Edge

Starting edge of the period measurement. Rising by default.

__post_init__

__post_init__()

Validate the expected period range.

Source code in src/nidaqlib/channels/counter_input.py
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}"
        )

from_dict classmethod

from_dict(data)

Deserialise, restoring :class:Edge from its string value.

Source code in src/nidaqlib/channels/counter_input.py
@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)

to_dict

to_dict()

Serialise; encode :class:Edge to its string value.

Source code in src/nidaqlib/channels/counter_input.py
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

CounterPulseFrequency dataclass

CounterPulseFrequency(
    *,
    physical_channel,
    name=None,
    unit=None,
    metadata=_empty_metadata(),
    frequency,
    duty_cycle=0.5,
    initial_delay=0.0,
    idle_high=False,
    requires_confirm=True,
)

Bases: ChannelSpec

Pulse-train counter output specified by frequency + duty cycle.

Maps to Task.co_channels.add_co_pulse_chan_freq on the NI side.

Attributes:

Name Type Description
frequency float

Pulse-train frequency, in Hz.

duty_cycle float

Fractional duty cycle in (0.0, 1.0). 0.5 = square wave.

initial_delay float

Optional delay before the first pulse, in seconds. Defaults to 0.

idle_high bool

When True, the line idles high (active-low pulses); otherwise idles low (active-high pulses).

requires_confirm bool

When True, every :meth:DaqSession.write targeting this channel must pass confirm=True. Defaults to True — counter outputs default to safe.

__post_init__

__post_init__()

Validate pulse-train parameters.

Source code in src/nidaqlib/channels/counter_output.py
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}")

CounterPulseTicks dataclass

CounterPulseTicks(
    *,
    physical_channel,
    name=None,
    unit=None,
    metadata=_empty_metadata(),
    source_terminal,
    high_ticks,
    low_ticks,
    initial_delay=0,
    idle_high=False,
    requires_confirm=True,
)

Bases: ChannelSpec

Pulse-train counter output specified by high / low tick counts.

Maps to Task.co_channels.add_co_pulse_chan_ticks on the NI side. The tick reference is given by source_terminal.

Attributes:

Name Type Description
source_terminal str

NI terminal supplying the tick clock (e.g. "/Dev1/20MHzTimebase").

high_ticks int

Number of source ticks in the high state.

low_ticks int

Number of source ticks in the low state.

initial_delay int

Optional initial-delay tick count.

idle_high bool

When True, the line idles high.

requires_confirm bool

Defaults to True.

__post_init__

__post_init__()

Validate pulse tick parameters.

Source code in src/nidaqlib/channels/counter_output.py
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}")

CounterPulseTime dataclass

CounterPulseTime(
    *,
    physical_channel,
    name=None,
    unit=None,
    metadata=_empty_metadata(),
    high_time,
    low_time,
    initial_delay=0.0,
    idle_high=False,
    requires_confirm=True,
)

Bases: ChannelSpec

Pulse-train counter output specified by high / low durations in seconds.

Maps to Task.co_channels.add_co_pulse_chan_time on the NI side.

Attributes:

Name Type Description
high_time float

High-state duration, in seconds.

low_time float

Low-state duration, in seconds.

initial_delay float

Optional delay before the first pulse, in seconds.

idle_high bool

When True, the line idles high (active-low pulses).

requires_confirm bool

Defaults to True.

__post_init__

__post_init__()

Validate pulse timing parameters.

Source code in src/nidaqlib/channels/counter_output.py
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}")

DigitalInput dataclass

DigitalInput(
    *,
    physical_channel,
    name=None,
    unit=None,
    metadata=_empty_metadata(),
    line_grouping_per_line=True,
)

Bases: ChannelSpec

Digital-input line or port.

Maps to Task.di_channels.add_di_chan on the NI side. physical_channel accepts NI's line / port grammar (Dev1/port0/line0, Dev1/port0:7, ...).

Attributes:

Name Type Description
line_grouping_per_line bool

When True, the backend treats each line as its own channel (NI LineGrouping.CHAN_PER_LINE). Defaults to True so multi-line specs round-trip cleanly into per-line reads. Set to False for one-channel-for-all-lines (CHAN_FOR_ALL_LINES).

DigitalOutput dataclass

DigitalOutput(
    *,
    physical_channel,
    name=None,
    unit=None,
    metadata=_empty_metadata(),
    requires_confirm=True,
    line_grouping_per_line=True,
)

Bases: ChannelSpec

Digital-output line or port.

Maps to Task.do_channels.add_do_chan on the NI side. Writes are gated through :meth:DaqSession.write, which requires confirm=True whenever any target channel sets requires_confirm (design doc §17.1).

Attributes:

Name Type Description
requires_confirm bool

When True, every :meth:DaqSession.write targeting this channel must pass confirm=True. Defaults to True — digital outputs are assumed to drive a real-world actuator unless the spec explicitly opts out.

line_grouping_per_line bool

When True, the backend treats each line as its own channel. Same semantics as :class:DigitalInput.line_grouping_per_line.

ThermocoupleInput dataclass

ThermocoupleInput(
    *,
    physical_channel,
    name=None,
    unit=None,
    metadata=_empty_metadata(),
    thermocouple_type,
    min_val,
    max_val,
    cjc_source=None,
    cjc_val=None,
    units=_default_temperature_units(),
)

Bases: ChannelSpec

Thermocouple analog-input channel.

Maps to Task.ai_channels.add_ai_thrmcpl_chan on the NI side. The enum-typed fields are stored as int values matching nidaqmx.constants so that to_dict/from_dict round-trips through JSON without dragging NI's enum machinery into the serialisation layer.

Attributes:

Name Type Description
thermocouple_type ThermocoupleType

One of nidaqmx.constants.ThermocoupleType (J, K, T, ...). Required; no sane default.

min_val float

Lower limit of the expected temperature, in units.

max_val float

Upper limit of the expected temperature, in units.

cjc_source CJCSource | None

Cold-junction compensation source. None lets NI pick the device default (typically built-in).

cjc_val float | None

Cold-junction reference temperature, in units. Only relevant for CJCSource.CONSTANT_USER_VALUE.

units TemperatureUnits

Temperature units for min_val / max_val and the returned data. Defaults to degrees Celsius.

__post_init__

__post_init__()

Validate the temperature range.

Source code in src/nidaqlib/channels/analog_input.py
def __post_init__(self) -> None:
    """Validate the temperature range."""
    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}"
        )

from_dict classmethod

from_dict(data)

Reconstruct, restoring enum members from their .value ints.

Source code in src/nidaqlib/channels/analog_input.py
@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> Self:
    """Reconstruct, restoring enum members from their ``.value`` ints."""
    kind = data.get("kind")
    if kind != cls.kind:
        raise NIDaqValidationError(f"kind mismatch: expected {cls.kind!r}, got {kind!r}")
    from nidaqmx.constants import (  # noqa: PLC0415
        CJCSource,
        TemperatureUnits,
        ThermocoupleType,
    )

    payload = {k: v for k, v in data.items() if k != "kind"}
    try:
        payload["thermocouple_type"] = ThermocoupleType(payload["thermocouple_type"])
    except (KeyError, ValueError) as exc:
        raise NIDaqValidationError(
            f"unknown ThermocoupleType {payload.get('thermocouple_type')!r}"
        ) from exc
    if payload.get("cjc_source") is not None:
        try:
            payload["cjc_source"] = CJCSource(payload["cjc_source"])
        except ValueError as exc:
            raise NIDaqValidationError(
                f"unknown CJCSource {payload.get('cjc_source')!r}"
            ) from exc
    units_raw = payload.get("units", TemperatureUnits.DEG_C.value)
    try:
        payload["units"] = TemperatureUnits(units_raw)
    except ValueError as exc:
        raise NIDaqValidationError(f"unknown TemperatureUnits {units_raw!r}") from exc
    return cls(**payload)

to_dict

to_dict()

Serialise via the enums' .value so the payload is JSON-encodable.

nidaqmx.constants enums do not inherit from :class:int, so we record .value (an int) and reconstruct the enum on :meth:from_dict.

Source code in src/nidaqlib/channels/analog_input.py
def to_dict(self) -> dict[str, Any]:
    """Serialise via the enums' ``.value`` so the payload is JSON-encodable.

    ``nidaqmx.constants`` enums do not inherit from :class:`int`, so we
    record ``.value`` (an int) and reconstruct the enum on
    :meth:`from_dict`.
    """
    # 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.
    payload = ChannelSpec.to_dict(self)
    payload["thermocouple_type"] = self.thermocouple_type.value
    payload["cjc_source"] = self.cjc_source.value if self.cjc_source is not None else None
    payload["units"] = self.units.value
    return payload

register_channel_kind

register_channel_kind(cls)

Register a concrete :class:ChannelSpec subclass for dispatch.

Used as a decorator on the subclass definition. Idempotent — re-registering the same kind is a no-op (helps reload-friendliness in notebooks).

Parameters:

Name Type Description Default
cls type[C]

The subclass to register. Must declare a non-empty kind.

required

Returns:

Type Description
type[C]

cls (so the decorator is transparent).

Raises:

Type Description
NIDaqValidationError

cls.kind is empty or already bound to a different class.

Source code in src/nidaqlib/channels/base.py
def register_channel_kind[C: ChannelSpec](cls: type[C]) -> type[C]:
    """Register a concrete :class:`ChannelSpec` subclass for dispatch.

    Used as a decorator on the subclass definition. Idempotent — re-registering
    the same ``kind`` is a no-op (helps reload-friendliness in notebooks).

    Args:
        cls: The subclass to register. Must declare a non-empty ``kind``.

    Returns:
        ``cls`` (so the decorator is transparent).

    Raises:
        NIDaqValidationError: ``cls.kind`` is empty or already bound to a
            different class.
    """
    if not cls.kind:
        raise NIDaqValidationError(f"{cls.__name__} must declare a non-empty 'kind' ClassVar")
    existing = _CHANNEL_REGISTRY.get(cls.kind)
    if existing is None:
        _CHANNEL_REGISTRY[cls.kind] = cls
    elif existing is not cls:
        raise NIDaqValidationError(
            f"channel kind {cls.kind!r} already bound to {existing.__name__}"
        )
    return cls