Task specifications¶

def __post_init__(self) -> None:
    """Enforce the hard minimum-3 floor."""
    if self.buffers < _MIN_BUFFERS:
        raise DtolValidationError(
            f"BufferPlan.buffers must be >= {_MIN_BUFFERS} (got {self.buffers}); "
            "see docs/design.md §8.7a for the minimum-3 rationale",
            context=ErrorContext(operation="BufferPlan.__post_init__"),
        )
    if self.samples_per_buffer <= 0:
        raise DtolValidationError(
            f"BufferPlan.samples_per_buffer must be positive (got {self.samples_per_buffer})",
            context=ErrorContext(operation="BufferPlan.__post_init__"),
        )

def __post_init__(self) -> None:
    """Validate mode-specific field requirements."""
    if self.multiscan_count < 1:
        raise DtolValidationError(
            f"RetriggerSpec.multiscan_count must be >= 1 (got {self.multiscan_count})",
            context=ErrorContext(operation="RetriggerSpec.__post_init__"),
        )
    if self.mode == RetriggerMode.INTERNAL and self.frequency_hz is None:
        raise DtolValidationError(
            "RetriggerSpec.frequency_hz is required when mode is INTERNAL",
            context=ErrorContext(operation="RetriggerSpec.__post_init__"),
        )
    if self.mode != RetriggerMode.INTERNAL and self.frequency_hz is not None:
        raise DtolValidationError(
            "RetriggerSpec.frequency_hz is only valid when mode is INTERNAL",
            context=ErrorContext(operation="RetriggerSpec.__post_init__"),
        )
    if self.mode == RetriggerMode.EXTRA and self.source is None:
        raise DtolValidationError(
            "RetriggerSpec.source is required when mode is EXTRA",
            context=ErrorContext(operation="RetriggerSpec.__post_init__"),
        )

def __post_init__(self) -> None:
    """Validate the spec against the §8.1 / §4.2 matrix."""
    if not self.name:
        raise DtolValidationError(
            "TaskSpec.name must be a non-empty string",
            context=ErrorContext(operation="TaskSpec.__post_init__"),
        )
    if not self.channels:
        raise DtolValidationError(
            f"TaskSpec(name={self.name!r}).channels is empty; at least one channel is required",
            context=ErrorContext(
                operation="TaskSpec.__post_init__",
                task_name=self.name,
            ),
        )

    # Channel-kind homogeneity: one HDASS = one subsystem type.
    inferred = self.infer_subsystem_type()
    if self.subsystem_type is not None and self.subsystem_type != inferred:
        raise DtolValidationError(
            f"TaskSpec.subsystem_type={self.subsystem_type.value} conflicts "
            f"with inferred type {inferred.value} from the channel kinds",
            context=ErrorContext(
                operation="TaskSpec.__post_init__",
                task_name=self.name,
            ),
        )

    # data_flow / timing / buffers matrix — see implementation-plan.md §4.2.
    self._validate_data_flow_matrix()

    # Wrap metadata to enforce immutability.
    if not isinstance(self.metadata, MappingProxyType):
        object.__setattr__(self, "metadata", MappingProxyType(dict(self.metadata)))

def infer_subsystem_type(self) -> SubsystemType:
    """Derive the subsystem kind from the channel kinds.

    Returns:
        The implied :class:`SubsystemType`.

    Raises:
        DtolValidationError: Channels span multiple subsystem kinds
            or include an unrecognised kind.  One HDASS is one
            subsystem of one type — mixing is illegal.
    """
    inferred: SubsystemType | None = None
    for channel in self.channels:
        kind_type = _channel_subsystem_type(channel)
        if inferred is None:
            inferred = kind_type
        elif inferred != kind_type:
            raise DtolValidationError(
                f"TaskSpec(name={self.name!r}).channels mixes subsystem "
                f"kinds ({inferred.value} vs {kind_type.value}); "
                "split into separate TaskSpecs (one HDASS per type).",
                context=ErrorContext(
                    operation="TaskSpec.infer_subsystem_type",
                    task_name=self.name,
                ),
            )
    if inferred is None:
        raise DtolValidationError(
            f"TaskSpec(name={self.name!r}).channels is empty; at least one channel is required",
            context=ErrorContext(
                operation="TaskSpec.infer_subsystem_type",
                task_name=self.name,
            ),
        )
    return inferred

def __post_init__(self) -> None:
    """Validate the external-divider / clock-source / samples-ceiling combos."""
    if self.rate_hz <= 0.0:
        raise DtolValidationError(
            f"Timing.rate_hz must be positive, got {self.rate_hz}",
            context=ErrorContext(operation="Timing.__post_init__"),
        )
    if self.clock_source == ClockSource.EXTERNAL and self.external_divider is None:
        raise DtolValidationError(
            "Timing.external_divider is required when clock_source is EXTERNAL",
            context=ErrorContext(operation="Timing.__post_init__"),
        )
    if self.clock_source == ClockSource.INTERNAL and self.external_divider is not None:
        raise DtolValidationError(
            "Timing.external_divider is forbidden when clock_source is INTERNAL",
            context=ErrorContext(operation="Timing.__post_init__"),
        )
    if self.samples_per_channel is not None and self.samples_per_channel <= 0:
        raise DtolValidationError(
            f"Timing.samples_per_channel must be positive (got {self.samples_per_channel})",
            context=ErrorContext(operation="Timing.__post_init__"),
        )

def __getstate__(self) -> dict[str, Any]:
    """Unwrap MappingProxyType views so the block can be pickled."""
    return _slots_to_picklable_state(self, _DAQ_BLOCK_MAPPING_FIELDS)

def __post_init__(self) -> None:
    """Validate array shapes and freeze mapping + ndarray mutability."""
    n_channels = len(self.channels)
    expected_shape = (n_channels, self.samples_per_channel)
    if self.data.shape != expected_shape:
        raise DtolValidationError(
            f"DaqBlock.data shape {self.data.shape} does not match "
            f"(len(channels)={n_channels}, samples_per_channel="
            f"{self.samples_per_channel})",
            context=ErrorContext(operation="DaqBlock.__post_init__"),
        )
    if self.raw_codes is not None and self.raw_codes.shape != expected_shape:
        raise DtolValidationError(
            f"DaqBlock.raw_codes shape {self.raw_codes.shape} does not "
            f"match data shape {expected_shape}",
            context=ErrorContext(operation="DaqBlock.__post_init__"),
        )
    if self.cjc_data is not None and self.cjc_data.shape != expected_shape:
        raise DtolValidationError(
            f"DaqBlock.cjc_data shape {self.cjc_data.shape} does not "
            f"match data shape {expected_shape}",
            context=ErrorContext(operation="DaqBlock.__post_init__"),
        )
    for ch_name, mask in self.sensor_status.items():
        if mask.shape != (self.samples_per_channel,):
            raise DtolValidationError(
                f"DaqBlock.sensor_status[{ch_name!r}] shape {mask.shape} "
                f"does not match (samples_per_channel="
                f"{self.samples_per_channel},)",
                context=ErrorContext(operation="DaqBlock.__post_init__"),
            )
    if self.block_index < 0:
        raise DtolValidationError(
            f"DaqBlock.block_index must be >= 0 (got {self.block_index})",
            context=ErrorContext(operation="DaqBlock.__post_init__"),
        )
    if self.first_sample_index < 0:
        raise DtolValidationError(
            f"DaqBlock.first_sample_index must be >= 0 (got {self.first_sample_index})",
            context=ErrorContext(operation="DaqBlock.__post_init__"),
        )

    # Lock arrays read-only so sinks cannot mutate the drainer's copy.
    self.data.setflags(write=False)
    if self.raw_codes is not None:
        self.raw_codes.setflags(write=False)
    if self.cjc_data is not None:
        self.cjc_data.setflags(write=False)
    for mask in self.sensor_status.values():
        mask.setflags(write=False)

    # Freeze mappings.
    if not isinstance(self.units, MappingProxyType):
        object.__setattr__(self, "units", MappingProxyType(dict(self.units)))
    if not isinstance(self.sensor_status, MappingProxyType):
        object.__setattr__(
            self,
            "sensor_status",
            MappingProxyType(dict(self.sensor_status)),
        )

def __setstate__(self, state: dict[str, Any]) -> None:
    """Restore slotted fields and re-run shape/freeze validation."""
    _restore_slots_state(self, state)
    self.__post_init__()

def __getstate__(self) -> dict[str, Any]:
    """Unwrap MappingProxyType views — they are not pickle-friendly."""
    return _slots_to_picklable_state(self, _DAQ_READING_MAPPING_FIELDS)

def __post_init__(self) -> None:
    """Freeze the mapping fields so callers cannot mutate them post-hoc."""
    for name in ("values", "units", "sensor_status", "metadata"):
        current = getattr(self, name)
        if not isinstance(current, MappingProxyType):
            object.__setattr__(self, name, MappingProxyType(dict(current)))

def __setstate__(self, state: dict[str, Any]) -> None:
    """Restore slotted fields and re-wrap mappings via :meth:`__post_init__`."""
    _restore_slots_state(self, state)
    self.__post_init__()

def to_dict(self) -> dict[str, Any]:
    """JSON-friendly mapping suitable for the row-oriented sinks."""
    return {
        "device": self.device,
        "task": self.task,
        "values": dict(self.values),
        "units": dict(self.units),
        "requested_at": self.requested_at.isoformat(),
        "received_at": self.received_at.isoformat(),
        "t_utc": self.t_utc.isoformat(),
        "t_mono_ns": self.t_mono_ns,
        "t_midpoint_mono_ns": self.t_midpoint_mono_ns,
        "latency_s": self.latency_s,
        "sensor_status": {k: v.value for k, v in self.sensor_status.items()},
        "metadata": dict(self.metadata),
        "error": None if self.error is None else str(self.error),
    }

def __getstate__(self) -> dict[str, Any]:
    """Unwrap MappingProxyType so the sample can be pickled."""
    return _slots_to_picklable_state(self, _DAQ_SAMPLE_MAPPING_FIELDS)

def __post_init__(self) -> None:
    """Freeze the metadata mapping."""
    if not isinstance(self.metadata, MappingProxyType):
        object.__setattr__(
            self,
            "metadata",
            MappingProxyType(dict(self.metadata)),
        )

def __setstate__(self, state: dict[str, Any]) -> None:
    """Restore slotted fields and re-wrap the metadata mapping."""
    _restore_slots_state(self, state)
    self.__post_init__()

def to_dict(self) -> dict[str, Any]:
    """JSON-friendly mapping suitable for row-oriented sinks."""
    return {
        "device": self.device,
        "task": self.task,
        "channel": self.channel,
        "value": self.value,
        "unit": self.unit,
        "sample_index": self.sample_index,
        "block_index": self.block_index,
        "t_mono_ns": self.t_mono_ns,
        "t_utc": self.t_utc.isoformat(),
        "sensor_status": self.sensor_status.value,
        "is_linearised": self.is_linearised,
        "metadata": dict(self.metadata),
    }

def configure_continuous(
    self,
    hdass: int,
    spec: TaskSpec,
    capabilities: CapabilitySet,
) -> None:
    """Run the continuous-mode pre-commit configuration sequence.

    Stops short of ``commit()`` — the §12.3.2 ordering requires
    notification registration and buffer queueing BEFORE
    ``olDaConfig``. The callback bridge / ``record()`` drives the
    commit step after wiring its bridge.

    Sequence (docs/implementation-plan.md §5.7):

    1. ``set_data_flow(OL_DF_CONTINUOUS)`` (or ``OL_DF_CONTINUOUS_*``).
    2. For each channel: MULTI_SENSOR retype if needed, then
       ``add_channel(...)``.
    3. ``set_channel_list([phys, ...])`` — drives the continuous-mode
       channel list separately from the gain-list entries.
    4. ``set_clock(...)``.
    5. ``set_trigger(...)``.
    6. ``set_wrap_mode(...)``.
    7. ``set_stop_on_error(...)``.

    Args:
        hdass: Subsystem handle.
        spec: Task spec with ``data_flow`` in {CONTINUOUS, FINITE}.
        capabilities: Subsystem capability snapshot — drives the
            MULTI_SENSOR dispatch.

    Raises:
        DtolTaskStateError: If ``spec`` is not in a continuous mode.
    """
    if spec.data_flow == DataFlow.SINGLE_VALUE:
        from dtollib.errors import (  # noqa: PLC0415
            DtolTaskStateError,
            ErrorContext,
        )

        raise DtolTaskStateError(
            "configure_continuous: data_flow=single_value; "
            "use configure_single_value() instead",
            context=ErrorContext(
                operation="TaskBuilder.configure_continuous",
                task_name=spec.name,
            ),
        )
    if spec.timing is None or spec.buffers is None:
        from dtollib.errors import (  # noqa: PLC0415
            DtolTaskStateError,
            ErrorContext,
        )

        raise DtolTaskStateError(
            "configure_continuous requires both Timing and BufferPlan",
            context=ErrorContext(
                operation="TaskBuilder.configure_continuous",
                task_name=spec.name,
            ),
        )

    self._backend.set_data_flow(hdass, DATA_FLOW_TO_OL[spec.data_flow])

    # Per-channel configuration (mirrors single-value path).
    for list_index, channel in enumerate(spec.channels):
        _require_io_type_supported(channel, capabilities)
        if capabilities.supports_multisensor:
            io_type = channel.kind_to_multi_sensor_type()
            self._backend.set_multi_sensor_type(hdass, channel.physical_channel, io_type)
        self._backend.add_channel(hdass, list_index, channel)

    # Continuous channel-list: flat list of physical channel indices.
    self._backend.set_channel_list(hdass, [c.physical_channel for c in spec.channels])

    # Clock.
    clock_source = (
        OL_CLK_INTERNAL if spec.timing.clock_source == ClockSource.INTERNAL else OL_CLK_EXTERNAL
    )
    self._backend.set_clock(
        hdass,
        rate_hz=spec.timing.rate_hz,
        clock_source=clock_source,
        external_divider=spec.timing.external_divider,
    )

    # Trigger.
    kind, threshold_channel, threshold_level = _trigger_to_sdk(spec.trigger)
    self._backend.set_trigger(
        hdass,
        kind=kind,
        threshold_channel=threshold_channel,
        threshold_level=threshold_level,
    )

    # Triggered-scan retrigger.  When the timing carries a
    # RetriggerSpec the SDK collects ``multiscan_count`` scans per trigger
    # at the configured retrigger source/rate.
    if spec.timing.retrigger is not None:
        self._configure_retrigger(hdass, spec.timing.retrigger)

    # Wrap mode (NONE for FINITE; MULTIPLE for continuous; SINGLE for DAC).
    wrap = _wrap_mode_to_sdk(spec.buffers.wrap_mode)
    self._backend.set_wrap_mode(hdass, wrap)

    # DMA usage. The SDK requires olDaSetDmaUsage(min(1, NUMDMACHANS))
    # for continuous mode even when the subsystem reports zero DMA
    # channels (the DT9805/06 report NUMDMACHANS==0 yet still need the
    # call — docs/decisions.md). Pass 1 when DMA is available, else 0,
    # and always make the call.
    self._backend.set_dma_usage(hdass, 1 if capabilities.supports_dma else 0)

    self._backend.set_stop_on_error(hdass, spec.stop_on_error)

def configure_counter(
    self,
    hdass: int,
    spec: TaskSpec,
    capabilities: CapabilitySet,
) -> None:
    """Configure a counter/timer, quadrature, or tachometer subsystem.

    Counter subsystems are read on demand after :meth:`start`; there is
    no channel/gain list or sample clock to set up.  The critical
    ordering invariant is **C/T mode first** — ``olDaSetCTMode`` re-types
    the counter and must precede gate / pulse / edge setters.  The fake
    backend rejects out-of-order calls.

    Args:
        hdass: Subsystem handle.
        spec: Task spec whose channels are counter/quadrature/tachometer.
        capabilities: Subsystem capability snapshot (unused today; kept
            for signature parity with the AI/continuous paths).
    """
    for channel in spec.channels:
        _require_counter_mode_supported(_counter_mode_for(channel), capabilities, channel)
        self._configure_counter_channel(hdass, channel)
    self._backend.set_stop_on_error(hdass, spec.stop_on_error)
    self._backend.commit(hdass)

def configure_single_value(
    self,
    hdass: int,
    spec: TaskSpec,
    capabilities: CapabilitySet,
) -> None:
    """Run the single-value configuration sequence.

    Sequence (docs/implementation-plan.md §4.3):

    1. ``set_data_flow(OL_DF_SINGLEVALUE)``.
    2. For each channel:
       a. If the channel is MULTI_SENSOR per the capability set,
          ``set_multi_sensor_type(...)`` FIRST.
       b. ``add_channel(...)`` — drives
          ``olDaSetChannelType`` + ``olDaSetChannelRange`` +
          ``olDaSetGainListEntry`` (+ ``olDaSetThermocoupleType``).
    3. ``set_stop_on_error(spec.stop_on_error)``.
    4. ``commit()`` — ``olDaConfig``.

    Args:
        hdass: Subsystem handle from
            :meth:`~dtollib.backend.DtolBackend.get_dass`.
        spec: Task specification.  Must have
            ``data_flow == DataFlow.SINGLE_VALUE`` — validated by
            ``TaskSpec.__post_init__`` plus an explicit assert
            here for the type narrower.
        capabilities: Live capability snapshot for the
            subsystem.  Drives the MULTI_SENSOR dispatch.
    """
    if spec.data_flow != DataFlow.SINGLE_VALUE:
        # TaskSpec validation should have caught this.  Belt + braces.
        from dtollib.errors import (  # noqa: PLC0415
            DtolTaskStateError,
            ErrorContext,
        )

        raise DtolTaskStateError(
            f"configure_single_value: data_flow={spec.data_flow.value}; "
            "use configure_continuous() for non-single-value tasks",
            context=ErrorContext(
                operation="TaskBuilder.configure_single_value",
                task_name=spec.name,
            ),
        )

    self._backend.set_data_flow(hdass, DATA_FLOW_TO_OL[spec.data_flow])

    for list_index, channel in enumerate(spec.channels):
        # MULTI_SENSOR ordering — docs/design.md §8.5a.  On
        # subsystems that report supports_multisensor, every
        # channel is re-typed at configure time; the spec's
        # ``kind_to_multi_sensor_type`` returns the SDK-facing
        # IOType discriminator.
        _require_io_type_supported(channel, capabilities)
        _validate_digital_port(channel, capabilities)
        if capabilities.supports_multisensor:
            io_type = channel.kind_to_multi_sensor_type()
            self._backend.set_multi_sensor_type(hdass, channel.physical_channel, io_type)

        self._backend.add_channel(hdass, list_index, channel)

    self._backend.set_stop_on_error(hdass, spec.stop_on_error)
    self._backend.commit(hdass)