Source code for semi_cr.core.lab.utils.qharbor_tools.schemas

"""Measurement schemas shared by all local measurement backends."""

from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING, Any

import numpy as np

from .errors import ConfigurationError
from .models import require_text

if TYPE_CHECKING:
    from collections.abc import Mapping, Sequence




@dataclass(frozen=True, slots=True)
class ParameterSpec:
    """A backend-independent measurement parameter definition."""

    name: str
    label: str | None = None
    unit: str = ""
    setpoints: tuple[str, ...] = ()
    paramtype: str = "numeric"

    def __post_init__(self) -> None:
        object.__setattr__(self, "name", require_text(self.name, "parameter.name"))
        if self.label is not None:
            object.__setattr__(self, "label", require_text(self.label, "parameter.label"))
        object.__setattr__(self, "setpoints", tuple(self.setpoints))
        object.__setattr__(self, "paramtype", require_text(self.paramtype, "parameter.paramtype"))





@dataclass(frozen=True, slots=True)
class MeasurementSchema:
    """Backend-independent description of one saved measurement."""

    name: str
    parameters: tuple[ParameterSpec, ...] | Sequence[ParameterSpec]

    def __post_init__(self) -> None:
        object.__setattr__(self, "name", require_text(self.name, "schema.name"))
        parameters = tuple(self.parameters)
        if not parameters:
            raise ConfigurationError("schema.parameters must contain at least one parameter.", operation="validate")
        names = [parameter.name for parameter in parameters]
        if len(names) != len(set(names)):
            raise ConfigurationError("schema parameter names must be unique.", operation="validate")
        known = set(names)
        for parameter in parameters:
            missing = set(parameter.setpoints) - known
            if missing:
                raise ConfigurationError(
                    f"Parameter '{parameter.name}' references unknown setpoints: {sorted(missing)}.",
                    operation="validate",
                )
            if parameter.name in parameter.setpoints:
                raise ConfigurationError(
                    f"Parameter '{parameter.name}' cannot be its own setpoint.",
                    operation="validate",
                )
        object.__setattr__(self, "parameters", parameters)

    @property
    def parameter_names(self) -> tuple[str, ...]:
        """Return parameter names in registration order."""

        return tuple(parameter.name for parameter in self.parameters)



    def validate_results(self, results: Mapping[str, Any]) -> None:
        """Validate keys and array shapes before a backend has side effects."""

        expected = set(self.parameter_names)
        received = set(results)
        missing = expected - received
        extra = received - expected
        if missing or extra:
            details = []
            if missing:
                details.append(f"missing={sorted(missing)}")
            if extra:
                details.append(f"unexpected={sorted(extra)}")
            raise ConfigurationError(f"Results do not match the schema ({', '.join(details)}).", operation="validate")

        for parameter in self.parameters:
            result_shape = np.shape(results[parameter.name])
            for setpoint_name in parameter.setpoints:
                setpoint_shape = np.shape(results[setpoint_name])
                if setpoint_shape != result_shape:
                    raise ConfigurationError(
                        f"Parameter '{parameter.name}' has shape {result_shape}, but setpoint '{setpoint_name}' has shape {setpoint_shape}.",
                        operation="validate",
                    )