#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json
import os
import time
from pathlib import Path, PureWindowsPath, PurePosixPath
from typing import Any, Dict, List, Literal, Optional
from packaging.version import InvalidVersion, Version
from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator, model_validator
from autoemx._version import __version__ as AUTOEMX_VERSION
import autoemx.utils.constants as cnst
from autoemx.config.runtime_configs import (
BulkMeasurementConfig,
ExpStandardsConfig,
MeasurementConfig,
MicroscopeConfig,
PlotConfig,
PowderMeasurementConfig,
SampleConfig,
SampleSubstrateConfig,
)
from .acquisition import ParticleInfo, SpectrumEntry
from .frames import FrameRecord
from .legacy_size_stats import migrate_legacy_size_statistics
from .particle_stats import ParticleSizeStatsResult, SizeMeasurement
from .segmentation import FrameSegmentation, SegmentationConfig
from .quantification import QuantificationConfig, QuantificationResult
[docs]
class LedgerConfigs(BaseModel):
"""Typed config bundle persisted in the ledger."""
microscope_cfg: MicroscopeConfig
sample_cfg: SampleConfig
measurement_cfg: MeasurementConfig
sample_substrate_cfg: SampleSubstrateConfig
plot_cfg: PlotConfig
model_config = ConfigDict(extra="forbid")
[docs]
@model_validator(mode="before")
@classmethod
def drop_legacy_quant_cfg(cls, data: Any) -> Any:
if isinstance(data, dict):
cleaned = dict(data)
cleaned.pop("quant_cfg", None)
return cleaned
return data
[docs]
@model_validator(mode="before")
@classmethod
def move_legacy_exp_std_w_frs(cls, data: Any) -> Any:
"""Legacy compatibility: move the standard's mass fractions from exp_stds_cfg to sample_cfg.
Old ledgers stored them in measurement_cfg.exp_stds_cfg.w_frs.
"""
if not isinstance(data, dict):
return data
measurement_cfg = data.get("measurement_cfg")
sample_cfg = data.get("sample_cfg")
if not isinstance(measurement_cfg, dict) or not isinstance(sample_cfg, dict):
return data
exp_stds_cfg = measurement_cfg.get("exp_stds_cfg")
if not isinstance(exp_stds_cfg, dict):
return data
exp_stds_cfg = dict(exp_stds_cfg)
legacy_w_frs = exp_stds_cfg.pop("w_frs", None)
if legacy_w_frs and exp_stds_cfg.get("is_exp_std_measurement") and not sample_cfg.get("w_frs"):
sample_cfg = {**sample_cfg, "w_frs": legacy_w_frs}
measurement_cfg = {**measurement_cfg, "exp_stds_cfg": exp_stds_cfg}
return {**data, "measurement_cfg": measurement_cfg, "sample_cfg": sample_cfg}
[docs]
@model_validator(mode="after")
def set_exp_std_w_frs(self) -> "LedgerConfigs":
"""Set the composition of an experimental standard from its formula, when not stored yet."""
exp_stds_cfg = self.measurement_cfg.exp_stds_cfg
if exp_stds_cfg is not None and exp_stds_cfg.is_exp_std_measurement and not self.sample_cfg.w_frs:
self.sample_cfg.w_frs = exp_stds_cfg.formula_w_frs()
return self
@property
def powder_meas_cfg(self) -> Optional[PowderMeasurementConfig]:
return self.measurement_cfg.powder_meas_cfg
@property
def bulk_meas_cfg(self) -> Optional[BulkMeasurementConfig]:
return self.measurement_cfg.bulk_meas_cfg
@property
def exp_stds_cfg(self) -> Optional[ExpStandardsConfig]:
return self.measurement_cfg.exp_stds_cfg
[docs]
class LedgerVersionError(ValueError):
"""Raised when a ledger written by a newer AutoEMX version cannot be read by the installed one."""
def _newer_ledger_version_message(payload: Any) -> Optional[str]:
"""Explain which AutoEMX version to use, if the ledger was written by a newer one than installed."""
ledger_version = payload.get("autoemx_version") if isinstance(payload, dict) else None
if not ledger_version:
return None
try:
is_newer = Version(str(ledger_version)) > Version(AUTOEMX_VERSION)
except InvalidVersion:
return None
if not is_newer:
return None
return (
f"This ledger was written by AutoEMX {ledger_version}, but AutoEMX {AUTOEMX_VERSION} is installed. "
f"Update AutoEMX, or read it in a separate environment: pip install autoemx=={ledger_version}"
)
[docs]
class SampleLedger(BaseModel):
"""Single-sample ledger with per-spectrum pointers and configuration bundle."""
# Version of AutoEMX that last saved the ledger. Ledgers saved before this field existed get the current
# version, since the current version reads them (through the legacy compatibility code).
autoemx_version: str = AUTOEMX_VERSION
sample_id: str
sample_path: str
configs: LedgerConfigs
spectra: List[SpectrumEntry]
particles: List[ParticleInfo] = Field(default_factory=list)
# Particle size statistics: frames acquired (with their segmentation results), segmentations, measurements, and
# statistics computed from frames under a segmentation (the active one shown and used by default)
frames: List[FrameRecord] = Field(default_factory=list)
segmentations: List[SegmentationConfig] = Field(default_factory=list)
size_measurements: List[SizeMeasurement] = Field(default_factory=list)
size_results: List[ParticleSizeStatsResult] = Field(default_factory=list)
active_size_result: Optional[int] = None
active_quant: Optional[int] = None
quantifications: List[QuantificationConfig] = Field(default_factory=list)
model_config = ConfigDict(extra="forbid")
[docs]
@model_validator(mode="before")
@classmethod
def move_legacy_size_statistics(cls, data: Any) -> Any:
"""Legacy compatibility: particle size statistics saved before frames and segmentations were recorded."""
return migrate_legacy_size_statistics(data)
[docs]
@model_validator(mode="before")
@classmethod
def move_legacy_known_phase_mixture(cls, data: Any) -> Any:
"""Legacy compatibility: is_known_powder_mixture_meas (powder settings) is now the quantification setting
is_known_phase_mixture, stored in the options of each quantification run."""
# Only saved ledgers (plain JSON data) can contain the legacy setting
if not isinstance(data, dict) or not isinstance(data.get("configs"), dict):
return data
measurement_cfg = data["configs"].get("measurement_cfg")
powder_meas_cfg = measurement_cfg.get("powder_meas_cfg") if isinstance(measurement_cfg, dict) else None
if not isinstance(powder_meas_cfg, dict) or "is_known_powder_mixture_meas" not in powder_meas_cfg:
return data
is_known_phase_mixture = bool(powder_meas_cfg["is_known_powder_mixture_meas"])
quantifications = []
for quant_config in data.get("quantifications") or []:
if isinstance(quant_config, dict) and isinstance(quant_config.get("options"), dict):
options = quant_config["options"]
if "is_known_phase_mixture" not in options:
quant_config = {**quant_config, "options": {**options, "is_known_phase_mixture": is_known_phase_mixture}}
quantifications.append(quant_config)
return {**data, "quantifications": quantifications}
[docs]
@field_validator("sample_path")
@classmethod
def validate_sample_path(cls, v: str) -> str:
# Check if it is a valid absolute path on EITHER Windows or POSIX
is_windows_abs = PureWindowsPath(v).is_absolute()
is_posix_abs = PurePosixPath(v).is_absolute()
if not (is_windows_abs or is_posix_abs):
raise ValueError("sample_path must be an absolute path")
return v
[docs]
@field_validator("active_quant")
@classmethod
def validate_active_quant(cls, v: Optional[int]) -> Optional[int]:
if v is not None and v < 0:
raise ValueError("active_quant must be a non-negative integer or None")
return v
[docs]
@model_validator(mode="after")
def validate_ledger_integrity(self) -> "SampleLedger":
particle_ids = [particle.id for particle in self.particles]
if len(particle_ids) != len(set(particle_ids)):
raise ValueError("ParticleInfo.id values must be unique within a ledger")
self._validate_size_statistics()
config_ids = [cfg.quantification_id for cfg in self.quantifications]
if len(config_ids) != len(set(config_ids)):
raise ValueError("quantification_id values must be unique within a ledger")
if config_ids:
if self.active_quant is None:
self.active_quant = config_ids[-1]
if self.active_quant not in set(config_ids):
raise ValueError("active_quant must reference an existing quantification_id")
elif self.active_quant is not None:
raise ValueError("active_quant must be None when no quantification configs are available")
config_id_set = set(config_ids)
for spectrum_index, spectrum in enumerate(self.spectra):
if spectrum.spectrum_relpath is not None:
resolved = (Path(self.sample_path) / spectrum.spectrum_relpath).resolve()
sample_root = Path(self.sample_path).resolve()
try:
resolved.relative_to(sample_root)
except ValueError as exc:
raise ValueError(
f"Spectrum at index {spectrum_index} has spectrum_relpath outside sample_path"
) from exc
if spectrum.instrument_background_relpath is not None:
resolved = (Path(self.sample_path) / spectrum.instrument_background_relpath).resolve()
sample_root = Path(self.sample_path).resolve()
try:
resolved.relative_to(sample_root)
except ValueError as exc:
raise ValueError(
f"Spectrum at index {spectrum_index} has instrument_background_relpath outside sample_path"
) from exc
result_ids = set()
for result in spectrum.quantification_results:
if result.quantification_id not in config_id_set:
raise ValueError(
f"Spectrum at index {spectrum_index} contains a quantification result "
f"with unknown quantification_id '{result.quantification_id}'"
)
if result.quantification_id in result_ids:
raise ValueError(
f"Spectrum at index {spectrum_index} contains duplicate "
f"quantification_id '{result.quantification_id}'"
)
result_ids.add(result.quantification_id)
return self
@staticmethod
def _load_counts_from_pointer_file(file_path: Path) -> List[float]:
"""Load counts from an external spectrum pointer file."""
def _read_once() -> List[float]:
suffix = file_path.suffix.lower()
if suffix in cnst.EMSA_SPECTRUM_EXTENSIONS:
counts: List[float] = []
in_data_section = False
with file_path.open("r", encoding="utf-8") as f:
for raw_line in f:
line = raw_line.strip()
if not line:
continue
# Accept both '#SPECTRUM' and '##SPECTRUM' as start of data
if line.startswith("#SPECTRUM") or line.startswith("##SPECTRUM"):
in_data_section = True
continue
if line.startswith("#"):
continue
if not in_data_section:
continue
# Accept only counts (no index). Reject lines with commas.
token = line.rstrip(",")
if "," in token:
raise ValueError(f"Pointer file '{file_path}' contains invalid spectrum data line with comma: '{line}'")
try:
counts.append(float(token))
except ValueError:
continue
if not counts:
raise ValueError(f"No spectrum counts found in pointer file '{file_path}'")
return counts
if suffix == ".json":
with file_path.open("r", encoding="utf-8") as f:
payload = json.load(f)
values = payload.get("spectrum_vals")
if not isinstance(values, list) or not values:
raise ValueError(f"JSON pointer file '{file_path}' does not contain non-empty 'spectrum_vals'")
return [float(v) for v in values]
raise ValueError(f"Unsupported spectrum pointer extension '{suffix}' for file '{file_path}'")
for attempt in range(2):
try:
return _read_once()
except KeyboardInterrupt:
if attempt == 0:
time.sleep(0.05)
continue
raise
raise RuntimeError(f"Failed to load spectrum counts from '{file_path}' after retries")
# ------------------------------------------------------------------ particle size statistics
def _validate_size_statistics(self) -> None:
"""Unique ids, and references to frames, segmentations, measurements and results that exist."""
for name, ids in (
("frame_id", [f.frame_id for f in self.frames]),
("segmentation_id", [s.segmentation_id for s in self.segmentations]),
("measurement_id", [m.measurement_id for m in self.size_measurements]),
("result_id", [r.result_id for r in self.size_results]),
):
if len(ids) != len(set(ids)):
raise ValueError(f"{name} values must be unique within a ledger")
segmentation_ids = {s.segmentation_id for s in self.segmentations}
frame_ids = {f.frame_id for f in self.frames}
measurement_ids = {m.measurement_id for m in self.size_measurements}
for frame in self.frames:
unknown = {s.segmentation_id for s in frame.segmentations} - segmentation_ids
if unknown:
raise ValueError(f"Frame {frame.frame_id} has results of unknown segmentations {sorted(unknown)}")
if frame.measurement_id is not None and frame.measurement_id not in measurement_ids:
raise ValueError(f"Frame {frame.frame_id} references unknown measurement {frame.measurement_id}")
for measurement in self.size_measurements:
if measurement.segmentation_id not in segmentation_ids:
raise ValueError(f"Size measurement {measurement.measurement_id} references an unknown segmentation")
unknown = set(measurement.frame_ids) - frame_ids
if unknown:
raise ValueError(f"Size measurement {measurement.measurement_id} references unknown frames "
f"{sorted(unknown)}")
for result in self.size_results:
for seg in (result.segmentation_id, result.eds_segmentation_id):
if seg is not None and seg not in segmentation_ids:
raise ValueError(f"Size result {result.result_id} references unknown segmentation {seg}")
if result.measurement_id is not None and result.measurement_id not in measurement_ids:
raise ValueError(f"Size result {result.result_id} references unknown measurement")
result_ids = [r.result_id for r in self.size_results]
if result_ids:
if self.active_size_result is None:
self.active_size_result = result_ids[-1]
if self.active_size_result not in result_ids:
raise ValueError("active_size_result must reference an existing result_id")
elif self.active_size_result is not None:
raise ValueError("active_size_result must be None without size results")
@property
def size_result(self) -> Optional[ParticleSizeStatsResult]:
"""The active particle size statistics result (None if there is none)."""
return next((r for r in self.size_results if r.result_id == self.active_size_result), None)
[docs]
def frame(self, frame_id: str) -> Optional[FrameRecord]:
return next((f for f in self.frames if f.frame_id == frame_id), None)
[docs]
def measurement(self, measurement_id: Optional[int]) -> Optional[SizeMeasurement]:
return next((m for m in self.size_measurements if m.measurement_id == measurement_id), None)
[docs]
def segmentation(self, segmentation_id: Optional[int]) -> Optional[SegmentationConfig]:
return next((s for s in self.segmentations if s.segmentation_id == segmentation_id), None)
[docs]
def frames_of(self, purpose: Optional[str] = None, measurement_id: Optional[int] = None) -> List[FrameRecord]:
"""Frames acquired for a purpose ('eds_search' or 'size_stats'), and/or by a measurement."""
return [f for f in self.frames if (purpose is None or f.purpose == purpose)
and (measurement_id is None or f.measurement_id == measurement_id)]
[docs]
def unique_frame_id(self, frame_id: str) -> str:
"""``frame_id``, made unique in the ledger with a ``#2``, ``#3``, ... suffix (frame acquired again)."""
used = {f.frame_id for f in self.frames}
new_id, n = frame_id, 2
while new_id in used:
new_id, n = f"{frame_id}#{n}", n + 1
return new_id
[docs]
def add_frame(self, frame: FrameRecord) -> FrameRecord:
"""Add an acquired frame (its id made unique if needed), returned as recorded."""
frame = frame.model_copy(update={"frame_id": self.unique_frame_id(frame.frame_id)})
self.frames.append(frame)
return frame
[docs]
def append_segmentation(self, config: SegmentationConfig) -> int:
"""Id of the segmentation with these settings: an existing one, or ``config`` added with a new id."""
for existing in self.segmentations:
if existing.fingerprint() == config.fingerprint():
return existing.segmentation_id
new_id = max((s.segmentation_id for s in self.segmentations), default=-1) + 1
self.segmentations.append(config.model_copy(update={"segmentation_id": new_id}))
return new_id
[docs]
def add_frame_segmentation(self, frame_id: str, result: FrameSegmentation) -> None:
"""Add the result of a segmentation of a frame (never replacing one)."""
if self.segmentation(result.segmentation_id) is None:
raise ValueError(f"Unknown segmentation {result.segmentation_id}")
frame = self.frame(frame_id)
if frame is None:
raise ValueError(f"Unknown frame {frame_id}")
frame.add_segmentation(result)
[docs]
def next_measurement_id(self) -> int:
return max((m.measurement_id for m in self.size_measurements), default=-1) + 1
[docs]
def append_size_result(self, result: ParticleSizeStatsResult) -> ParticleSizeStatsResult:
"""
Add a statistics result and make it the active one: it replaces a result computed from the same inputs
(measurement, segmentations, frames and particles left out), e.g. recalculated after more EDS frames;
otherwise it gets a new id. Returns the result as recorded.
"""
same = next((r for r in self.size_results if r.inputs_key() == result.inputs_key()), None)
if same is not None:
result = result.model_copy(update={"result_id": same.result_id})
self.size_results[self.size_results.index(same)] = result
else:
result = result.model_copy(update={
"result_id": max((r.result_id for r in self.size_results), default=-1) + 1})
self.size_results.append(result)
self.active_size_result = result.result_id
return result
[docs]
def append_quantification_config(self, quant_config: QuantificationConfig) -> None:
"""Append a quantification config, enforcing unique config identifiers."""
if any(existing.quantification_id == quant_config.quantification_id for existing in self.quantifications):
raise ValueError(
f"quantification_id '{quant_config.quantification_id}' already exists in this ledger"
)
self.quantifications.append(quant_config)
self.active_quant = quant_config.quantification_id
[docs]
def append_quantification_result(self, spectrum_index: int, result: QuantificationResult) -> None:
"""Append a quantification result to the requested spectrum entry."""
if spectrum_index < 0 or spectrum_index >= len(self.spectra):
raise IndexError(f"spectrum_index {spectrum_index} is out of range")
if result.quantification_id not in {cfg.quantification_id for cfg in self.quantifications}:
raise ValueError(
f"Quantification result references unknown quantification_id '{result.quantification_id}'"
)
if any(
existing.quantification_id == result.quantification_id
for existing in self.spectra[spectrum_index].quantification_results
):
raise ValueError(
f"Spectrum at index {spectrum_index} already contains quantification_id "
f"'{result.quantification_id}'"
)
self.spectra[spectrum_index].quantification_results.append(result)
[docs]
@classmethod
def from_json_file(cls, file_path: str | Path) -> "SampleLedger":
"""Load and validate a ledger from JSON in one call.
Raises
------
LedgerVersionError
If the ledger cannot be read because it was written by a newer AutoEMX version.
"""
path = Path(file_path)
with path.open("r", encoding="utf-8") as f:
payload = json.load(f)
try:
return cls.model_validate(payload)
except ValidationError as e:
message = _newer_ledger_version_message(payload)
if message is not None:
raise LedgerVersionError(f"{message}\n(ledger: {path})") from e
raise
[docs]
def to_dict(
self,
config_key_style: Literal["cfg", "config"] = "cfg",
*,
include_configs: bool = True,
) -> Dict[str, Any]:
"""Serialize ledger to a dict and choose config key suffix style."""
payload = self.model_dump(mode="json")
if not include_configs:
payload.pop("configs", None)
return payload
if config_key_style == "cfg":
return payload
if config_key_style != "config":
raise ValueError("config_key_style must be either 'cfg' or 'config'")
cfg_block = payload.get("configs", {})
renamed_cfg_block: Dict[str, Any] = {}
for key, value in cfg_block.items():
if key.endswith("_cfg"):
renamed_cfg_block[key[: -len("_cfg")] + "_config"] = value
else:
renamed_cfg_block[key] = value
payload["configs"] = renamed_cfg_block
return payload
[docs]
def to_json_file(
self,
file_path: str | Path,
*,
config_key_style: Literal["cfg", "config"] = "cfg",
include_configs: bool = True,
indent: int = 2,
) -> None:
"""Write ledger to JSON in one call, recording the AutoEMX version that saved it."""
self.autoemx_version = AUTOEMX_VERSION
path = Path(file_path)
payload = self.to_dict(config_key_style=config_key_style, include_configs=include_configs)
path.parent.mkdir(parents=True, exist_ok=True)
# Per-process temporary file, so concurrent writers never share it. Atomic replace avoids
# partial reads if another process opens the ledger mid-write.
tmp_path = path.with_name(f"{path.name}.{os.getpid()}.tmp")
try:
with tmp_path.open("w", encoding="utf-8") as f:
json.dump(payload, f, indent=indent, allow_nan=False)
f.write("\n")
os.replace(tmp_path, path)
except Exception:
if tmp_path.exists():
tmp_path.unlink()
raise