# SPDX-License-Identifier: AGPL-3.0-or-later
# Commercial license available
# © Concepts 1996–2026 Miroslav Šotek. All rights reserved.
# © Code 2020–2026 Miroslav Šotek. All rights reserved.
# ORCID: 0009-0009-3560-0851
# Contact: www.anulum.li | protoscience@anulum.li
# SCPN Fusion Core — complete FAIR-MAST magnetic archive capture
"""Capture every object and decoded array in a FAIR-MAST ``magnetics`` group."""
from __future__ import annotations
import hashlib
import json
from collections.abc import Mapping
from importlib import import_module
from pathlib import Path, PurePosixPath
from typing import Any, cast
import numpy as np
from numpy.typing import NDArray
from .mast_magnetic_archive_codec import (
JsonObject,
JsonValue,
MastCompleteMagneticArchiveEnvelope,
MastMagneticArchiveValidationError,
canonical_json_bytes,
decode_mast_complete_magnetic_archive_envelope,
encode_mast_complete_magnetic_archive_envelope,
mast_complete_magnetic_archive_sha256,
)
_PROVENANCE_SCHEMA = "scpn-fusion-open-disruption-data-provenance.v1"
_PRODUCER_PROJECT = "SCPN-FUSION-CORE"
_FAIR_MAST_ROOT = "https://s3.echo.stfc.ac.uk/mast/level2/shots"
[docs]
class MastMagneticArchiveDependencyError(ImportError):
"""Raised when the safe Zarr-v3 MAST dependency profile is unavailable."""
[docs]
def build_mast_complete_magnetic_archive_envelope(
provenance_path: Path,
shot_archive_root: Path,
) -> MastCompleteMagneticArchiveEnvelope:
"""Verify, decode and describe one complete authentic magnetic archive group.
Parameters
----------
provenance_path:
Exact tracked parent provenance manifest containing every source object.
shot_archive_root:
Local root of the fully materialised ``<shot>.zarr`` archive.
Returns
-------
MastCompleteMagneticArchiveEnvelope
Canonical review-only envelope binding all archive objects and arrays.
Raises
------
MastMagneticArchiveValidationError
If any object, array, clock, source identity or completeness invariant fails.
MastMagneticArchiveDependencyError
If Python or the installed optional dependency profile cannot decode Zarr v3.
"""
provenance_bytes = _read_regular_file(provenance_path, "provenance manifest")
provenance = _parse_json_object(provenance_bytes, "provenance manifest")
_require_equal(provenance.get("schema"), _PROVENANCE_SCHEMA, "provenance schema")
dataset = _as_object(provenance.get("dataset"), "provenance.dataset")
shot_id = _as_positive_integer(dataset.get("shot_id"), "provenance.dataset.shot_id")
_require_equal(dataset.get("device"), "MAST", "provenance.dataset.device")
groups = _as_string_list(dataset.get("downloaded_groups"), "downloaded_groups")
if "magnetics" not in groups:
raise MastMagneticArchiveValidationError("parent provenance omits magnetics group")
all_objects = _validate_parent_object_manifest(dataset, shot_id)
magnetic_prefix = f"raw/{shot_id}.zarr/magnetics/"
magnetic_objects = {
path: record for path, record in all_objects.items() if path.startswith(magnetic_prefix)
}
if not magnetic_objects:
raise MastMagneticArchiveValidationError("parent provenance contains no magnetic objects")
try:
archive_root = shot_archive_root.resolve(strict=True)
except OSError as exc:
raise MastMagneticArchiveValidationError("shot archive root is missing") from exc
if not archive_root.is_dir():
raise MastMagneticArchiveValidationError("shot archive root is not a directory")
magnetic_root = archive_root / "magnetics"
if not magnetic_root.is_dir() or magnetic_root.is_symlink():
raise MastMagneticArchiveValidationError("magnetics group is missing or symlinked")
_verify_complete_local_object_set(archive_root, magnetic_objects, shot_id)
verified_bytes: dict[str, bytes] = {}
for path, record in magnetic_objects.items():
local_path = archive_root / path.removeprefix(f"raw/{shot_id}.zarr/")
data = _read_regular_file(local_path, path)
_require_equal(len(data), record["size_bytes"], f"size for {path}")
_require_equal(
mast_complete_magnetic_archive_sha256(data), record["sha256"], f"digest for {path}"
)
verified_bytes[path] = data
group_metadata_path = f"{magnetic_prefix}zarr.json"
if group_metadata_path not in verified_bytes:
raise MastMagneticArchiveValidationError("magnetics group metadata object is missing")
group_metadata = _parse_json_object(
verified_bytes[group_metadata_path], "magnetics group metadata"
)
_require_equal(group_metadata.get("zarr_format"), 3, "group zarr format")
_require_equal(group_metadata.get("node_type"), "group", "group node type")
group_attributes = _as_object(group_metadata.get("attributes"), "group attributes")
_validate_group_identity_and_license(group_attributes)
source_ingestion_revision, source_ingestion_tree_state = _source_ingestion_identity(
group_attributes
)
zarr_group = _open_zarr_v3_group(magnetic_root)
array_names = tuple(sorted(cast(Any, zarr_group).array_keys()))
metadata_names = tuple(
sorted(
path.removeprefix(magnetic_prefix).removesuffix("/zarr.json")
for path in magnetic_objects
if path != group_metadata_path and path.endswith("/zarr.json")
)
)
if array_names != metadata_names:
raise MastMagneticArchiveValidationError(
f"decoded arrays differ from source metadata: decoded={array_names}, metadata={metadata_names}"
)
metadata_by_name = {
name: _parse_json_object(
verified_bytes[f"{magnetic_prefix}{name}/zarr.json"], f"array metadata {name}"
)
for name in array_names
}
clock_names = _discover_native_clocks(metadata_by_name)
arrays = [
_build_array_record(
name,
metadata_by_name[name],
cast(Any, zarr_group)[name],
magnetic_prefix,
magnetic_objects,
clock_names,
)
for name in array_names
]
clocks = [
_build_clock_record(name, cast(Any, zarr_group)[name], metadata_by_name[name])
for name in clock_names
]
magnetic_manifest_digest = _object_manifest_digest(magnetic_objects.values())
parent_files = _as_list(dataset["files"], "provenance.dataset.files")
parent_total_bytes = sum(
_as_nonnegative_integer(
_as_object(item, "provenance.dataset.files entry").get("size_bytes"),
"parent size_bytes",
)
for item in parent_files
)
root_url = f"{_FAIR_MAST_ROOT}/{shot_id}.zarr/magnetics/"
payload: JsonObject = {
"archive": {
"group": "magnetics",
"group_metadata": group_metadata,
"group_metadata_sha256": cast(str, magnetic_objects[group_metadata_path]["sha256"]),
"metadata_path": group_metadata_path,
"root_url": root_url,
"zarr_format": 3,
},
"arrays": cast(JsonValue, arrays),
"authority": {
"actionable": False,
"classification_performed": False,
"direct_actuation": False,
"execution_permitted": False,
"review_only": True,
},
"clocks": cast(JsonValue, clocks),
"completeness": {
"array_count": len(arrays),
"arrays_complete": True,
"clock_count": len(clocks),
"objects_complete": True,
"source_decoded": True,
},
"event_id": None,
"event_identity_state": "unresolved_facility_mapping",
"facility": "MAST",
"observation_id": f"mast-{shot_id}-complete-magnetics-{magnetic_manifest_digest[:16]}",
"producer_artifacts": cast(JsonValue, _producer_artifact_records()),
"provenance": {
"license": provenance.get("license"),
"limitations": provenance.get("limitations"),
"magnetic_group_manifest_sha256": magnetic_manifest_digest,
"object_count": len(magnetic_objects),
"objects": cast(JsonValue, list(magnetic_objects.values())),
"parent_download_manifest_sha256": dataset["download_manifest_sha256"],
"parent_object_count": len(parent_files),
"parent_total_bytes": parent_total_bytes,
"retrieved_at_utc": provenance.get("retrieved_at_utc"),
"schema": provenance.get("schema"),
"sha256": mast_complete_magnetic_archive_sha256(provenance_bytes),
"total_bytes": sum(
cast(int, record["size_bytes"]) for record in magnetic_objects.values()
),
},
"qualification": {
"calibration_state": "unresolved",
"channel_geometry_mapping_state": "unresolved",
"classification_eligible": False,
"diagnostic_semantics_state": "source_preserved_unqualified",
"event_clock_state": "unresolved",
"observation_operator_state": "not_supplied",
"phase_eligible": False,
"quality_state": "unknown",
"raw_samples_present": True,
"semantic_ingress_eligible": False,
"source_kind": "physical_archive",
"synthetic": False,
"source_clock_relationship_state": "unresolved",
"uncertainty_state": "unknown",
"validity_state": "unknown",
},
"reactor_configuration": "spherical_tokamak",
"shot_id": shot_id,
"producer_project": _PRODUCER_PROJECT,
"source_archive": "FAIR-MAST",
"source_ingestion_revision": source_ingestion_revision,
"source_ingestion_tree_state": source_ingestion_tree_state,
}
return encode_mast_complete_magnetic_archive_envelope(payload)
[docs]
def verify_mast_complete_magnetic_archive_source(
envelope: MastCompleteMagneticArchiveEnvelope | bytes,
shot_archive_root: Path,
) -> None:
"""Reverify every declared object and reject undeclared local archive objects."""
validated = (
decode_mast_complete_magnetic_archive_envelope(envelope)
if isinstance(envelope, bytes)
else decode_mast_complete_magnetic_archive_envelope(envelope.to_bytes())
)
payload = validated.payload
shot_id = _as_positive_integer(payload["shot_id"], "shot_id")
provenance = _as_object(payload["provenance"], "provenance")
objects = {
cast(str, record["path"]): record
for record in (
_as_object(item, "provenance object")
for item in _as_list(provenance["objects"], "provenance.objects")
)
}
try:
archive_root = shot_archive_root.resolve(strict=True)
except OSError as exc:
raise MastMagneticArchiveValidationError("shot archive root is missing") from exc
_verify_complete_local_object_set(archive_root, objects, shot_id)
for path, record in objects.items():
local_path = archive_root / path.removeprefix(f"raw/{shot_id}.zarr/")
data = _read_regular_file(local_path, path)
_require_equal(len(data), record["size_bytes"], f"size for {path}")
_require_equal(
mast_complete_magnetic_archive_sha256(data), record["sha256"], f"digest for {path}"
)
def _validate_parent_object_manifest(dataset: JsonObject, shot_id: int) -> dict[str, JsonObject]:
raw_files = _as_list(dataset.get("files"), "provenance.dataset.files")
if not raw_files:
raise MastMagneticArchiveValidationError("parent provenance file inventory is empty")
objects: dict[str, JsonObject] = {}
for index, raw_file in enumerate(raw_files):
record = _as_object(raw_file, f"provenance.dataset.files[{index}]")
if set(record) != {"path", "sha256", "size_bytes", "source_url"}:
raise MastMagneticArchiveValidationError("parent provenance object keys differ")
path = _as_nonempty_string(record.get("path"), f"files[{index}].path")
expected_prefix = f"raw/{shot_id}.zarr/"
pure_path = PurePosixPath(path)
if (
not path.startswith(expected_prefix)
or path != pure_path.as_posix()
or ".." in pure_path.parts
or path in objects
):
raise MastMagneticArchiveValidationError(
"parent object path is duplicate or cross-shot"
)
digest = _as_nonempty_string(record.get("sha256"), f"files[{index}].sha256")
if len(digest) != 64 or any(character not in "0123456789abcdef" for character in digest):
raise MastMagneticArchiveValidationError(
"parent object digest is not lowercase SHA-256"
)
size = _as_nonnegative_integer(record.get("size_bytes"), f"files[{index}].size_bytes")
source_url = _as_nonempty_string(record.get("source_url"), f"files[{index}].source_url")
expected_url = f"{_FAIR_MAST_ROOT}/" + path.removeprefix("raw/")
if source_url != expected_url:
raise MastMagneticArchiveValidationError(f"source URL does not bind {path}")
objects[path] = {
"path": path,
"sha256": digest,
"size_bytes": size,
"source_url": source_url,
}
ordered = dict(sorted(objects.items()))
digest = _object_manifest_digest(ordered.values())
_require_equal(
dataset.get("download_manifest_sha256"), digest, "parent download manifest digest"
)
return ordered
def _verify_complete_local_object_set(
archive_root: Path, objects: Mapping[str, JsonObject], shot_id: int
) -> None:
magnetic_root = archive_root / "magnetics"
actual_paths: set[str] = set()
for candidate in magnetic_root.rglob("*"):
if candidate.is_symlink():
raise MastMagneticArchiveValidationError(f"archive contains symlink: {candidate}")
if candidate.is_file():
relative = candidate.relative_to(archive_root).as_posix()
actual_paths.add(f"raw/{shot_id}.zarr/{relative}")
elif not candidate.is_dir():
raise MastMagneticArchiveValidationError(
f"archive contains non-regular object: {candidate}"
)
expected_paths = set(objects)
if actual_paths != expected_paths:
raise MastMagneticArchiveValidationError(
"local archive inventory differs: "
f"missing={sorted(expected_paths - actual_paths)}, "
f"undeclared={sorted(actual_paths - expected_paths)}"
)
def _build_array_record(
name: str,
metadata: JsonObject,
zarr_array: Any,
magnetic_prefix: str,
objects: Mapping[str, JsonObject],
clock_names: tuple[str, ...],
) -> JsonObject:
_require_equal(metadata.get("zarr_format"), 3, f"{name} zarr format")
_require_equal(metadata.get("node_type"), "array", f"{name} node type")
shape = _as_nonnegative_integer_list(metadata.get("shape"), f"{name} shape")
dimension_names = _dimension_names(metadata, shape, name)
if len(shape) != len(dimension_names):
raise MastMagneticArchiveValidationError(f"{name} shape/dimension rank differs")
decoded = np.asarray(zarr_array[...])
if list(decoded.shape) != shape:
raise MastMagneticArchiveValidationError(f"{name} decoded shape differs from metadata")
value_count = int(decoded.size) if decoded.ndim else 1
decoded_digest = _decoded_array_digest(decoded)
nonfinite_count = _nonfinite_count(decoded)
metadata_path = f"{magnetic_prefix}{name}/zarr.json"
data_prefix = f"{magnetic_prefix}{name}/"
data_paths = sorted(
path for path in objects if path.startswith(data_prefix) and path != metadata_path
)
attributes_value = metadata.get("attributes", {})
attributes = _as_object(attributes_value, f"{name} attributes")
return {
"archive_path": f"magnetics/{name}",
"attributes": attributes,
"clock_dimensions": cast(
JsonValue,
[dimension for dimension in dimension_names if dimension in clock_names],
),
"data_object_paths": cast(JsonValue, data_paths),
"data_type": metadata.get("data_type"),
"decoded_content_sha256": decoded_digest,
"decoded_nonfinite_count": nonfinite_count,
"decoded_value_count": value_count,
"dimension_names": cast(JsonValue, dimension_names),
"metadata_object_path": metadata_path,
"metadata_object_sha256": cast(str, objects[metadata_path]["sha256"]),
"name": name,
"shape": cast(JsonValue, shape),
"zarr_metadata": metadata,
}
def _discover_native_clocks(metadata_by_name: Mapping[str, JsonObject]) -> tuple[str, ...]:
clocks: list[str] = []
for name, metadata in metadata_by_name.items():
shape = _as_nonnegative_integer_list(metadata.get("shape"), f"{name} shape")
dimensions = _dimension_names(metadata, shape, name)
attributes = _as_object(metadata.get("attributes", {}), f"{name} attributes")
if len(shape) == 1 and dimensions == [name] and attributes.get("units") == "s":
clocks.append(name)
if not clocks:
raise MastMagneticArchiveValidationError("magnetics group declares no native clocks")
return tuple(sorted(clocks))
def _build_clock_record(name: str, zarr_array: Any, metadata: JsonObject) -> JsonObject:
values = np.asarray(zarr_array[...], dtype=np.float64)
if values.ndim != 1 or values.size < 2:
raise MastMagneticArchiveValidationError(f"clock {name} is not a nontrivial vector")
finite = bool(np.all(np.isfinite(values)))
intervals = np.diff(values)
increasing = bool(np.all(intervals > 0.0))
if not finite or not increasing:
raise MastMagneticArchiveValidationError(f"clock {name} is nonfinite or not increasing")
attributes = _as_object(metadata.get("attributes", {}), f"clock {name} attributes")
_require_equal(attributes.get("units"), "s", f"clock {name} units")
return {
"clock_kind_candidate": "shot_relative",
"clock_qualification": "unresolved",
"finite": True,
"first_value_s": float(values[0]),
"last_value_s": float(values[-1]),
"mapping_evidence_claimed": False,
"maximum_interval_s": float(np.max(intervals)),
"mean_interval_s": float(np.mean(intervals)),
"minimum_interval_s": float(np.min(intervals)),
"name": name,
"sample_count": int(values.size),
"strictly_increasing": True,
"units": "s",
}
def _dimension_names(metadata: JsonObject, shape: list[int], name: str) -> list[str]:
raw_dimensions = metadata.get("dimension_names")
if raw_dimensions is None and not shape:
return []
return _as_string_list(raw_dimensions, f"{name} dimensions")
def _decoded_array_digest(values: NDArray[Any]) -> str:
shape = list(values.shape)
if values.dtype.kind in {"O", "S", "U", "T"}:
content: JsonObject = {
"dtype": str(values.dtype),
"shape": cast(JsonValue, shape),
"values": cast(JsonValue, values.tolist()),
}
return mast_complete_magnetic_archive_sha256(canonical_json_bytes(content))
canonical_dtype = values.dtype.newbyteorder("<")
contiguous = np.ascontiguousarray(values.astype(canonical_dtype, copy=False))
header: JsonObject = {"dtype": contiguous.dtype.str, "shape": cast(JsonValue, shape)}
digest = hashlib.sha256(canonical_json_bytes(header))
digest.update(contiguous.tobytes(order="C"))
return digest.hexdigest()
def _nonfinite_count(values: NDArray[Any]) -> int:
if values.dtype.kind not in {"f", "c"}:
return 0
return int(values.size - np.count_nonzero(np.isfinite(values)))
def _open_zarr_v3_group(magnetic_root: Path) -> Any:
try:
zarr = import_module("zarr")
except ImportError as exc:
raise MastMagneticArchiveDependencyError(
"complete FAIR-MAST ingestion requires Python >=3.11 and scpn-fusion[mast]"
) from exc
version_parts = tuple(int(part) for part in zarr.__version__.split(".")[:2])
if version_parts < (3, 1):
raise MastMagneticArchiveDependencyError(
f"complete FAIR-MAST ingestion requires Zarr >=3.1,<3.2; found {zarr.__version__}"
)
if version_parts >= (3, 2):
raise MastMagneticArchiveDependencyError(
f"unreviewed Zarr version {zarr.__version__}; require the hash-locked MAST profile"
)
try:
return zarr.open_group(magnetic_root, mode="r")
except Exception as exc:
raise MastMagneticArchiveValidationError(
f"cannot open authentic Zarr v3 group: {exc}"
) from exc
def _validate_group_identity_and_license(attributes: JsonObject) -> None:
_require_equal(attributes.get("name"), "magnetics", "group name")
_require_equal(attributes.get("imas"), "magnetics", "group IMAS identity")
_require_equal(
attributes.get("license_url"),
"https://creativecommons.org/licenses/by-sa/4.0/",
"group license URL",
)
_require_equal(
attributes.get("license_name"),
"Creative Commons 4.0 BY-SA",
"group license name",
)
for key in ("description", "ingested_at", "commit_url"):
_as_nonempty_string(attributes.get(key), f"group {key}")
def _source_ingestion_identity(attributes: JsonObject) -> tuple[str, str]:
commit_url = _as_nonempty_string(attributes.get("commit_url"), "group commit_url")
expected_prefix = "https://github.com/ukaea/fair-mast-ingestion/tree/"
if not commit_url.startswith(expected_prefix):
raise MastMagneticArchiveValidationError("group commit URL is not FAIR-MAST ingestion")
identity = commit_url.removeprefix(expected_prefix)
tree_state = "dirty" if identity.endswith(" (dirty)") else "clean"
revision = identity.removesuffix(" (dirty)")
if len(revision) != 40 or any(character not in "0123456789abcdef" for character in revision):
raise MastMagneticArchiveValidationError("group ingestion revision is not a Git SHA-1")
return revision, tree_state
def _producer_artifact_records() -> list[JsonObject]:
module_dir = Path(__file__).resolve().parent
paths = [
module_dir / "mast_magnetic_archive.py",
module_dir / "mast_magnetic_archive_acquisition.py",
module_dir / "mast_magnetic_archive_codec.py",
]
return [
{
"path": f"src/scpn_fusion/io/{path.name}",
"sha256": mast_complete_magnetic_archive_sha256(_read_regular_file(path, path.name)),
}
for path in sorted(paths)
]
def _object_manifest_digest(records: Any) -> str:
lines: list[str] = []
for record in sorted(records, key=lambda item: cast(str, item["path"])):
lines.append(f"{record['sha256']}:{record['size_bytes']}:{record['path']}\n")
return mast_complete_magnetic_archive_sha256("".join(lines).encode("utf-8"))
def _parse_json_object(data: bytes, path: str) -> JsonObject:
def reject_duplicates(pairs: list[tuple[str, JsonValue]]) -> JsonObject:
result: JsonObject = {}
for key, value in pairs:
if key in result:
raise MastMagneticArchiveValidationError(f"{path} has duplicate key {key}")
result[key] = value
return result
try:
value = json.loads(data.decode("utf-8"), object_pairs_hook=reject_duplicates)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise MastMagneticArchiveValidationError(f"{path} is not valid UTF-8 JSON") from exc
return _as_object(cast(JsonValue, value), path)
def _read_regular_file(path: Path, label: str) -> bytes:
if path.is_symlink() or not path.is_file():
raise MastMagneticArchiveValidationError(f"{label} is missing, symlinked or non-regular")
try:
return path.read_bytes()
except OSError as exc:
raise MastMagneticArchiveValidationError(f"cannot read {label}: {exc}") from exc
def _as_object(value: object, path: str) -> JsonObject:
if not isinstance(value, dict) or any(not isinstance(key, str) for key in value):
raise MastMagneticArchiveValidationError(f"{path} must be an object")
return cast(JsonObject, value)
def _as_list(value: object, path: str) -> list[JsonValue]:
if not isinstance(value, list):
raise MastMagneticArchiveValidationError(f"{path} must be an array")
return cast(list[JsonValue], value)
def _as_string_list(value: object, path: str) -> list[str]:
result = _as_list(value, path)
if any(not isinstance(item, str) or not item for item in result):
raise MastMagneticArchiveValidationError(f"{path} must contain non-empty strings")
return cast(list[str], result)
def _as_nonnegative_integer_list(value: object, path: str) -> list[int]:
return [
_as_nonnegative_integer(item, f"{path}[{index}]")
for index, item in enumerate(_as_list(value, path))
]
def _as_nonnegative_integer(value: object, path: str) -> int:
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
raise MastMagneticArchiveValidationError(f"{path} must be a non-negative integer")
return value
def _as_positive_integer(value: object, path: str) -> int:
result = _as_nonnegative_integer(value, path)
if result == 0:
raise MastMagneticArchiveValidationError(f"{path} must be positive")
return result
def _as_nonempty_string(value: object, path: str) -> str:
if not isinstance(value, str) or not value.strip():
raise MastMagneticArchiveValidationError(f"{path} must be a non-empty string")
return value
def _require_equal(actual: object, expected: object, path: str) -> None:
if actual != expected or type(actual) is not type(expected):
raise MastMagneticArchiveValidationError(f"{path} must equal {expected!r}")
__all__ = [
"MastMagneticArchiveDependencyError",
"build_mast_complete_magnetic_archive_envelope",
"verify_mast_complete_magnetic_archive_source",
]