capa.calibration¶

def load_artifact(path: Path) -> HeatFluxTuneArtifact:
    """Read and validate a single tune artifact file."""
    try:
        text = path.read_text(encoding="utf-8")
    except OSError as exc:
        raise TuneArtifactError(f"cannot read tune artifact {path}: {exc}") from exc
    return from_toml(text)

def load_latest(directory: Path) -> HeatFluxTuneArtifact | None:
    """Return the artifact pointed to by ``<directory>/latest.toml``.

    Returns ``None`` when the directory does not exist, when
    ``latest.toml`` is missing, or when its ``id`` pointer references a
    file that no longer exists. Any *parse* failure on the pointed-at
    file raises :class:`TuneArtifactError` — a corrupt artifact is
    different from an absent one.
    """
    pointer_path = directory / "latest.toml"
    if not pointer_path.is_file():
        return None
    try:
        pointer = tomllib.loads(pointer_path.read_text(encoding="utf-8"))
    except (OSError, tomllib.TOMLDecodeError) as exc:
        raise TuneArtifactError(f"latest pointer at {pointer_path} unreadable: {exc}") from exc
    target_id = pointer.get("id")
    if not isinstance(target_id, str) or not target_id:
        raise TuneArtifactError(f"latest pointer at {pointer_path} is missing a string 'id'")
    target_path = directory / f"{target_id}.toml"
    if not target_path.is_file():
        return None
    return load_artifact(target_path)

def save_artifact(artifact: HeatFluxTuneArtifact, directory: Path) -> Path:
    """Persist ``artifact`` under ``directory`` and update ``latest.toml``.

    Writes ``<directory>/<artifact.id>.toml``. Refuses to overwrite an
    existing file with that id — a clash means the caller picked a
    non-unique id and the operator deserves to see it rather than have
    yesterday's tune silently clobbered.

    Updates ``latest.toml`` to point at the new id. If ``latest.toml``
    previously pointed at a different artifact, its target file is
    copied to ``<old-id>.toml.bak-<YYYY-MM-DD>`` first; that file is
    *not* removed (the on-disk record stays available under its own
    name, the ``.bak-…`` copy is the audit ledger of "this was the
    latest on this date").
    """
    directory.mkdir(parents=True, exist_ok=True)

    target_path = directory / f"{artifact.id}.toml"
    if target_path.exists():
        raise TuneArtifactError(
            f"tune artifact already exists at {target_path}; refusing to overwrite "
            f"(pick a unique id or remove the existing file deliberately)"
        )

    pointer_path = directory / "latest.toml"
    previous_id: str | None = None
    if pointer_path.is_file():
        try:
            previous = tomllib.loads(pointer_path.read_text(encoding="utf-8"))
        except (OSError, tomllib.TOMLDecodeError) as exc:
            raise TuneArtifactError(
                f"existing latest pointer at {pointer_path} is unreadable; "
                f"resolve manually before saving a new tune: {exc}"
            ) from exc
        prev = previous.get("id")
        if isinstance(prev, str) and prev and prev != artifact.id:
            previous_id = prev

    if previous_id is not None:
        previous_target = directory / f"{previous_id}.toml"
        if previous_target.is_file():
            backup_name = f"{previous_id}.toml.bak-{date.today().isoformat()}"
            backup_path = directory / backup_name
            if not backup_path.exists():
                shutil.copy2(previous_target, backup_path)

    target_path.write_text(to_toml(artifact), encoding="utf-8")
    pointer_path.write_text(
        tomli_w.dumps({"id": artifact.id, "updated_at": datetime.now(UTC)}),
        encoding="utf-8",
    )
    return target_path

def to_toml(artifact: HeatFluxTuneArtifact) -> str:
    """Serialize an artifact to TOML text.

    ``datetime`` is preserved as a native TOML datetime via ``tomli_w``,
    so a round-trip through :func:`from_toml` reconstructs an equal
    object. ``None``-valued optional fields are emitted as missing keys
    rather than as empty strings.
    """
    payload: dict[str, Any] = {
        "id": artifact.id,
        "rig": artifact.rig,
        "heater_device": artifact.heater_device,
        "heater_setpoint_channel": artifact.heater_setpoint_channel,
        "heater_pv_channel": artifact.heater_pv_channel,
        "flux_channel": artifact.flux_channel,
        "geometry": artifact.geometry,
        "accepted_at": artifact.accepted_at,
        "procedure_id": artifact.procedure_id,
        "procedure_version": artifact.procedure_version,
    }
    if artifact.gauge_calibration_ref is not None:
        payload["gauge_calibration_ref"] = artifact.gauge_calibration_ref
    if artifact.operator_id is not None:
        payload["operator_id"] = artifact.operator_id
    if artifact.capa_git_sha is not None:
        payload["capa_git_sha"] = artifact.capa_git_sha
    payload["points"] = [
        {
            "target_flux_kw_m2": p.target_flux_kw_m2,
            "heater_setpoint_c": p.heater_setpoint_c,
            "measured_flux_mean_kw_m2": p.measured_flux_mean_kw_m2,
            "measured_flux_std_kw_m2": p.measured_flux_std_kw_m2,
            "measured_flux_slope_kw_m2_per_min": p.measured_flux_slope_kw_m2_per_min,
            "heater_pv_mean_c": p.heater_pv_mean_c,
            "soak_s": p.soak_s,
            "accepted": p.accepted,
            "accept_reason": p.accept_reason,
        }
        for p in artifact.points
    ]
    return tomli_w.dumps(payload)