pelinker.dim_selection.checkpoint

Structured checkpoint I/O for pelinker.dim_selection runs.

Shared machinery (version, fingerprint, atomic gzip-aware I/O, the fingerprint fields common to every search) lives in :mod:pelinker.checkpoint; only the cell-keyed schema is here.

DimSelectionCheckpoint dataclass

On-disk checkpoint for resumable PCA/UMAP dimension selection runs.

Source code in pelinker/dim_selection/checkpoint.py

@dataclass
class DimSelectionCheckpoint:
    """On-disk checkpoint for resumable PCA/UMAP dimension selection runs."""

    version: int = CHECKPOINT_VERSION
    run_fingerprint: str = ""
    created_at: str = ""
    updated_at: str = ""
    completed_cells: list[str] = field(default_factory=list)
    summaries_by_key: dict[str, dict[str, str | float | int | None]] = field(
        default_factory=dict
    )
    stages: dict[str, str] = field(default_factory=dict)
    failures: list[FailureRecord] = field(default_factory=list)

    def to_json_dict(self) -> dict[str, Any]:
        return {
            "version": self.version,
            "run_fingerprint": self.run_fingerprint,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
            "completed_cells": sorted(self.completed_cells),
            "summaries_by_key": {
                k: dict(v) for k, v in sorted(self.summaries_by_key.items())
            },
            "stages": dict(sorted(self.stages.items())),
            "failures": [
                {"cell_key": f.cell_key, "error": f.error, "at": f.at}
                for f in self.failures
            ],
        }

    @staticmethod
    def from_json_dict(data: dict[str, Any]) -> DimSelectionCheckpoint:
        require_checkpoint_version(data)
        failures_raw = data.get("failures") or []
        failures = [
            FailureRecord(
                cell_key=str(fr["cell_key"]),
                error=str(fr["error"]),
                at=str(fr["at"]),
            )
            for fr in failures_raw
        ]
        return DimSelectionCheckpoint(
            version=int(data["version"]),
            run_fingerprint=str(data["run_fingerprint"]),
            created_at=str(data.get("created_at", "")),
            updated_at=str(data.get("updated_at", "")),
            completed_cells=list(data.get("completed_cells") or []),
            summaries_by_key=dict(data.get("summaries_by_key") or {}),
            stages=dict(data.get("stages") or {}),
            failures=failures,
        )

FailureRecord dataclass

One recorded failure for a (pca, umap) cell.

Source code in pelinker/dim_selection/checkpoint.py

@dataclass
class FailureRecord:
    """One recorded failure for a (pca, umap) cell."""

    cell_key: str
    error: str
    at: str

compute_run_fingerprint(config)

sha256 of the canonicalized run config; a change invalidates the checkpoint.

Source code in pelinker/checkpoint.py

def compute_run_fingerprint(config: dict[str, Any]) -> str:
    """sha256 of the canonicalized run config; a change invalidates the checkpoint."""
    blob = json.dumps(config, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return sha256(blob).hexdigest()