"""Dependency-independent models used by the qharbor tools framework."""
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from types import MappingProxyType
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from collections.abc import Mapping
from .errors import ConfigurationError
[docs]
def require_text(value: str, field_name: str) -> str:
"""Return a stripped non-empty string or raise a configuration error."""
if not isinstance(value, str) or not value.strip():
raise ConfigurationError(f"{field_name} must be a non-empty string.", operation="validate")
return value.strip()
[docs]
def frozen_mapping(value: Mapping[str, Any] | None = None) -> Mapping[str, Any]:
"""Copy a mapping so receipts and references cannot change unexpectedly."""
return MappingProxyType(dict(value or {}))
[docs]
@dataclass(frozen=True, slots=True)
class UserConfig:
"""User identity recorded in QHarbor metadata."""
initials: str
name: str
def __post_init__(self) -> None:
object.__setattr__(self, "initials", require_text(self.initials, "user.initials"))
object.__setattr__(self, "name", require_text(self.name, "user.name"))
[docs]
@dataclass(frozen=True, slots=True)
class DatasetConfig:
"""Metadata used to announce a QHarbor dataset."""
name: str
description: str
scope: str
subject_id: str
platform: str
project_id: str
component: str
step: str
sub_step: str
protocol_name: str
equipment: str
team: str = "Spin Qubit"
extra_attributes: Mapping[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
required = (
"name",
"description",
"scope",
"subject_id",
"platform",
"project_id",
"component",
"step",
"sub_step",
"protocol_name",
"equipment",
"team",
)
for field_name in required:
object.__setattr__(self, field_name, require_text(getattr(self, field_name), f"dataset.{field_name}"))
object.__setattr__(self, "extra_attributes", frozen_mapping(self.extra_attributes))
[docs]
def attributes(self, user: UserConfig) -> dict[str, Any]:
"""Return QHarbor attribute names expected by current data workflows."""
attributes = {
"UserInitials": user.initials,
"UserName": user.name,
"Team": self.team,
"SubjectID": self.subject_id,
"Platform": self.platform,
"ProjectID": self.project_id,
"Component": self.component,
"Step": self.step,
"SubStep": self.sub_step,
"ProtocolName": self.protocol_name,
"Equipment": self.equipment,
}
attributes.update(self.extra_attributes)
return attributes
[docs]
@dataclass(frozen=True, slots=True)
class ExperimentConfig:
"""Configuration for a new remote experiment."""
user: UserConfig
dataset: DatasetConfig
[docs]
@dataclass(frozen=True, slots=True)
class AnnouncementRequest:
"""Dependency-independent request passed to a remote backend."""
name: str
description: str
scope: str
attributes: Mapping[str, Any]
def __post_init__(self) -> None:
object.__setattr__(self, "name", require_text(self.name, "announcement.name"))
object.__setattr__(self, "description", require_text(self.description, "announcement.description"))
object.__setattr__(self, "scope", require_text(self.scope, "announcement.scope"))
object.__setattr__(self, "attributes", frozen_mapping(self.attributes))
[docs]
@dataclass(frozen=True, slots=True)
class RemoteDataset:
"""Stable public representation of a remote dataset."""
uuid: str
name: str
description: str = ""
attributes: Mapping[str, Any] = field(default_factory=dict)
files: tuple[str, ...] = ()
def __post_init__(self) -> None:
object.__setattr__(self, "uuid", require_text(self.uuid, "dataset.uuid"))
object.__setattr__(self, "name", require_text(self.name, "dataset.name"))
object.__setattr__(self, "attributes", frozen_mapping(self.attributes))
object.__setattr__(self, "files", tuple(self.files))
[docs]
@dataclass(frozen=True, slots=True)
class Artifact:
"""Portable data ready for publication by a remote backend."""
name: str
data: Any
media_type: str = "application/octet-stream"
metadata: Mapping[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
name = require_text(self.name, "artifact.name")
if Path(name).name != name or "/" in name or "\\" in name:
raise ConfigurationError("artifact.name must be a filename without directories.", operation="validate")
object.__setattr__(self, "name", name)
object.__setattr__(self, "media_type", require_text(self.media_type, "artifact.media_type"))
object.__setattr__(self, "metadata", frozen_mapping(self.metadata))
[docs]
@dataclass(frozen=True, slots=True)
class RunReference:
"""Serializable identity of a run stored by a measurement backend."""
backend: str
run_id: str
locator: str | None = None
metadata: Mapping[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
object.__setattr__(self, "backend", require_text(self.backend, "run.backend"))
object.__setattr__(self, "run_id", require_text(str(self.run_id), "run.run_id"))
object.__setattr__(self, "metadata", frozen_mapping(self.metadata))
[docs]
@dataclass(slots=True)
class StoredRun:
"""Backend run plus an optional native handle kept out of public receipts."""
reference: RunReference
native_handle: Any = None
[docs]
@dataclass(frozen=True, slots=True)
class BackendContext:
"""Experiment information supplied when preparing a local backend."""
experiment_name: str
subject_id: str
dataset_uuid: str
[docs]
@dataclass(frozen=True, slots=True)
class SaveRequest:
"""Validated results supplied to a measurement backend."""
experiment_name: str
subject_id: str
schema: Any
results: Mapping[str, Any]
metadata: Mapping[str, Any]
def __post_init__(self) -> None:
object.__setattr__(self, "results", frozen_mapping(self.results))
object.__setattr__(self, "metadata", frozen_mapping(self.metadata))