Channels¶

def __post_init__(self) -> None:
    """Reject ``min_val >= max_val`` before the SDK does."""
    # Explicit two-arg super(): @dataclass(slots=True) recreates the class,
    # which breaks zero-arg super() on Python < 3.14 (CPython gh-90562).
    super(AnalogInputVoltage, self).__post_init__()
    if self.min_val >= self.max_val:
        raise DtolValidationError(
            f"AnalogInputVoltage: min_val={self.min_val} must be "
            f"strictly less than max_val={self.max_val}",
            context=ErrorContext(
                operation="AnalogInputVoltage.__post_init__",
                channel=self.physical_channel,
                channel_name=self.name,
            ),
        )

def kind_to_multi_sensor_type(self) -> IOType:
    """``AnalogInputVoltage`` → :attr:`IOType.VOLTAGE_IN`."""
    return IOType.VOLTAGE_IN

def __post_init__(self) -> None:
    """Reject inconsistent ranges before the SDK ever sees them."""
    # Explicit two-arg super(): @dataclass(slots=True) recreates the class,
    # which breaks zero-arg super() on Python < 3.14 (CPython gh-90562).
    super(AnalogOutputVoltage, self).__post_init__()
    if self.min_val >= self.max_val:
        raise self._reject(
            f"min_val={self.min_val} must be strictly less than max_val={self.max_val}"
        )
    if self.safe_min is not None and self.safe_min < self.min_val:
        raise self._reject(
            f"safe_min={self.safe_min} is below the device range min_val={self.min_val}"
        )
    if self.safe_max is not None and self.safe_max > self.max_val:
        raise self._reject(
            f"safe_max={self.safe_max} is above the device range max_val={self.max_val}"
        )
    if (
        self.safe_min is not None
        and self.safe_max is not None
        and self.safe_min >= self.safe_max
    ):
        raise self._reject(
            f"safe_min={self.safe_min} must be strictly less than safe_max={self.safe_max}"
        )

def in_device_range(self, value: float) -> bool:
    """True if ``value`` lies within the device electrical range."""
    return self.min_val <= value <= self.max_val

def in_safe_band(self, value: float) -> bool:
    """True if ``value`` lies within the configured safe band.

    An unset bound (``None``) does not constrain that side. With both
    bounds unset, any in-range value is considered "in band".
    """
    if self.safe_min is not None and value < self.safe_min:
        return False
    return not (self.safe_max is not None and value > self.safe_max)

def kind_to_multi_sensor_type(self) -> IOType:
    """``AnalogOutputVoltage`` → :attr:`IOType.VOLTAGE_OUT`."""
    return IOType.VOLTAGE_OUT

def __post_init__(self) -> None:
    """Reject non-physical resistance."""
    super(BridgeInput, self).__post_init__()
    if self.nominal_resistance_ohms <= 0.0:
        raise DtolValidationError(
            f"BridgeInput: nominal_resistance_ohms must be positive "
            f"(got {self.nominal_resistance_ohms})",
            context=ErrorContext(
                operation="BridgeInput.__post_init__",
                channel=self.physical_channel,
                channel_name=self.name,
            ),
        )

def kind_to_multi_sensor_type(self) -> IOType:
    """``BridgeInput`` → :attr:`IOType.BRIDGE`."""
    return IOType.BRIDGE

def __post_init__(self) -> None:
    """Wrap mutable ``metadata`` mappings to enforce immutability."""
    if not isinstance(self.metadata, MappingProxyType):
        object.__setattr__(self, "metadata", MappingProxyType(dict(self.metadata)))

def kind_to_multi_sensor_type(self) -> IOType:
    """Return the :class:`IOType` discriminator for a MULTI_SENSOR channel.

    Called by the :class:`~dtollib.tasks.TaskBuilder` immediately
    before any per-type setter on a ``MULTI_SENSOR`` channel
    (docs/design.md §8.5a). Subclasses override; the base raises.

    Raises:
        NotImplementedError: This subclass does not know how to
            re-type a multi-sensor channel.
    """
    raise NotImplementedError(
        f"{type(self).__name__} cannot configure a MULTI_SENSOR channel; "
        "override kind_to_multi_sensor_type()."
    )

def to_dict(self) -> dict[str, Any]:
    """JSON-friendly mapping with ``kind`` discriminator embedded.

    Built field-by-field (not via :func:`dataclasses.asdict`, which
    deep-copies and cannot pickle the ``MappingProxyType`` metadata).
    :func:`~dtollib.channels.channel_from_dict` reverses this.
    """
    from dataclasses import fields  # noqa: PLC0415

    data: dict[str, Any] = {f.name: getattr(self, f.name) for f in fields(self)}
    data["kind"] = type(self).kind
    # MappingProxyType is not directly JSON serialisable; cast.
    data["metadata"] = dict(self.metadata)
    return data

def __post_init__(self) -> None:
    """Reject a non-positive measurement window."""
    # Explicit two-arg super(): @dataclass(slots=True) recreates the class,
    # which breaks zero-arg super() on Python < 3.14 (CPython gh-90562).
    super(CounterFrequency, self).__post_init__()
    if self.gate_period_s is not None and self.gate_period_s <= 0.0:
        raise DtolValidationError(
            f"CounterFrequency[{self.display_name}]: gate_period_s must be "
            f"positive (got {self.gate_period_s})",
            context=ErrorContext(
                operation="CounterFrequency.__post_init__",
                channel=self.physical_channel,
                channel_name=self.name,
            ),
        )

def __post_init__(self) -> None:
    """Reject ``min_val >= max_val`` before the SDK does."""
    super(CurrentInput, self).__post_init__()
    if self.min_val >= self.max_val:
        raise DtolValidationError(
            f"CurrentInput: min_val={self.min_val} must be strictly less "
            f"than max_val={self.max_val}",
            context=ErrorContext(
                operation="CurrentInput.__post_init__",
                channel=self.physical_channel,
                channel_name=self.name,
            ),
        )

def kind_to_multi_sensor_type(self) -> IOType:
    """``CurrentInput`` → :attr:`IOType.CURRENT`."""
    return IOType.CURRENT

def kind_to_multi_sensor_type(self) -> IOType:
    """``DigitalInputPort`` → :attr:`IOType.DIGITAL_INPUT`."""
    return IOType.DIGITAL_INPUT

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> DigitalLine:
    """Reconstruct a :class:`DigitalLine` from :meth:`to_dict`."""
    return cls(
        bit=data["bit"],
        name=data.get("name"),
        safe_value=data.get("safe_value"),
        requires_confirm=data.get("requires_confirm"),
    )

def to_dict(self) -> dict[str, Any]:
    """JSON-friendly mapping (reversed by :meth:`from_dict`)."""
    return {
        "bit": self.bit,
        "name": self.name,
        "safe_value": self.safe_value,
        "requires_confirm": self.requires_confirm,
    }

def kind_to_multi_sensor_type(self) -> IOType:
    """``DigitalOutputPort`` → :attr:`IOType.DIGITAL_OUTPUT`."""
    return IOType.DIGITAL_OUTPUT

def __post_init__(self) -> None:
    """Reject DC coupling and disabled excitation — invalid for IEPE."""
    super(IepeInput, self).__post_init__()
    if self.coupling is CouplingType.DC:
        raise DtolValidationError(
            "IepeInput requires AC coupling (an IEPE sensor rides a DC bias "
            "that must be blocked); got coupling=DC",
            context=self._ctx("IepeInput.__post_init__"),
        )
    if self.excitation_source is ExcitationSource.DISABLED:
        raise DtolValidationError(
            "IepeInput requires a drive current; excitation_source=DISABLED is invalid",
            context=self._ctx("IepeInput.__post_init__"),
        )
    if self.excitation_current_a <= 0.0:
        raise DtolValidationError(
            f"IepeInput: excitation_current_a must be positive "
            f"(got {self.excitation_current_a})",
            context=self._ctx("IepeInput.__post_init__"),
        )

def kind_to_multi_sensor_type(self) -> IOType:
    """``IepeInput`` → :attr:`IOType.ACCELEROMETER`."""
    return IOType.ACCELEROMETER

def __post_init__(self) -> None:
    """Reject a non-positive pulse width."""
    # Explicit two-arg super(): @dataclass(slots=True) recreates the class,
    # which breaks zero-arg super() on Python < 3.14 (CPython gh-90562).
    # RepetitiveOneShotOutput inherits this method; ``self`` is a subtype
    # of the recreated OneShotOutput, so the two-arg form stays correct.
    super(OneShotOutput, self).__post_init__()
    if self.pulse_width_s <= 0.0:
        raise DtolValidationError(
            f"{type(self).__name__}[{self.display_name}]: pulse_width_s must be "
            f"positive (got {self.pulse_width_s})",
            context=ErrorContext(
                operation=f"{type(self).__name__}.__post_init__",
                channel=self.physical_channel,
                channel_name=self.name,
            ),
        )

def __post_init__(self) -> None:
    """Reject a non-positive frequency or an out-of-range duty cycle."""
    # Explicit two-arg super(): @dataclass(slots=True) recreates the class,
    # which breaks zero-arg super() on Python < 3.14 (CPython gh-90562).
    super(PulseTrainOutput, self).__post_init__()
    if self.frequency_hz <= 0.0:
        raise self._reject(f"frequency_hz must be positive (got {self.frequency_hz})")
    if not (0.0 < self.duty_cycle < 1.0):
        raise self._reject(f"duty_cycle must be in (0, 1) (got {self.duty_cycle})")

def kind_to_multi_sensor_type(self) -> IOType:
    """``ResistanceInput`` → :attr:`IOType.RESISTANCE`."""
    return IOType.RESISTANCE

def __post_init__(self) -> None:
    """Enforce the custom-vs-standard coefficient contract."""
    super(RtdInput, self).__post_init__()
    is_custom = self.rtd_type is RtdType.CUSTOM
    has_ab = self.a is not None and self.b is not None
    if is_custom and not has_ab:
        raise DtolValidationError(
            "RtdInput(rtd_type=CUSTOM) requires both 'a' and 'b' "
            "Callendar-Van Dusen coefficients",
            context=self._ctx("RtdInput.__post_init__"),
        )
    if not is_custom and (self.a is not None or self.b is not None or self.c is not None):
        raise DtolValidationError(
            f"RtdInput(rtd_type={self.rtd_type.value}): a/b/c coefficients "
            "are only valid with rtd_type=CUSTOM (the standard curve fixes them)",
            context=self._ctx("RtdInput.__post_init__"),
        )
    if self.r0 <= 0.0:
        raise DtolValidationError(
            f"RtdInput: r0 must be positive (got {self.r0})",
            context=self._ctx("RtdInput.__post_init__"),
        )

def kind_to_multi_sensor_type(self) -> IOType:
    """``RtdInput`` → :attr:`IOType.RTD`."""
    return IOType.RTD

def __post_init__(self) -> None:
    """Reject non-physical gage factor / resistance."""
    super(StrainInput, self).__post_init__()
    if self.gage_factor <= 0.0:
        raise DtolValidationError(
            f"StrainInput: gage_factor must be positive (got {self.gage_factor})",
            context=self._ctx("StrainInput.__post_init__"),
        )
    if self.gage_resistance_ohms <= 0.0:
        raise DtolValidationError(
            f"StrainInput: gage_resistance_ohms must be positive "
            f"(got {self.gage_resistance_ohms})",
            context=self._ctx("StrainInput.__post_init__"),
        )

def kind_to_multi_sensor_type(self) -> IOType:
    """``StrainInput`` → :attr:`IOType.STRAIN_GAGE`."""
    return IOType.STRAIN_GAGE

def kind_to_multi_sensor_type(self) -> IOType:
    """``ThermistorInput`` → :attr:`IOType.THERMISTOR`."""
    return IOType.THERMISTOR

def __post_init__(self) -> None:
    """Reject ranges outside NIST operating envelope, before SDK does."""
    super(ThermocoupleInput, self).__post_init__()
    if self.min_val_degc >= self.max_val_degc:
        raise DtolValidationError(
            f"ThermocoupleInput: min_val_degc={self.min_val_degc} must be "
            f"strictly less than max_val_degc={self.max_val_degc}",
            context=ErrorContext(
                operation="ThermocoupleInput.__post_init__",
                channel=self.physical_channel,
                channel_name=self.name,
            ),
        )
    envelope_lo, envelope_hi = get_thermocouple_range(self.thermocouple_type.value)
    if self.min_val_degc < envelope_lo or self.max_val_degc > envelope_hi:
        raise DtolValidationError(
            f"ThermocoupleInput Type {self.thermocouple_type.value}: "
            f"range [{self.min_val_degc}, {self.max_val_degc}] °C "
            f"falls outside the NIST envelope "
            f"[{envelope_lo}, {envelope_hi}] °C",
            context=ErrorContext(
                operation="ThermocoupleInput.__post_init__",
                channel=self.physical_channel,
                channel_name=self.name,
                extra={"tc_type": self.thermocouple_type.value},
            ),
        )

def kind_to_multi_sensor_type(self) -> IOType:
    """``ThermocoupleInput`` → :attr:`IOType.THERMOCOUPLE`."""
    return IOType.THERMOCOUPLE