File size: 1,597 Bytes
700e896 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 | """Per-phase Parquet checkpoints (K13) — persist each phase's output frame for inspection.
Configurable via ``settings.debug.checkpoints`` (default OFF → zero writes, zero overhead). When on, each
phase writes ``<checkpoint_dir>/<NN>_<name>.parquet`` atomically. Any failure is logged and swallowed — a
checkpoint must NEVER break a run. This is what makes each phase independently inspectable/testable.
"""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING
import polars as pl
from .io import atomic_write_parquet
from .logging import get_logger
from .paths import REPO_ROOT
if TYPE_CHECKING: # runtime-free import — avoids any coupling/cycle with the config module
from config import Settings
def checkpoint(index: int, name: str, frame: pl.DataFrame, settings: Settings) -> None:
"""Persist ``frame`` as ``<NN>_<name>.parquet`` when checkpoints are enabled; otherwise a no-op."""
if not settings.debug.checkpoints:
return
base = Path(settings.debug.checkpoint_dir)
if not base.is_absolute():
base = REPO_ROOT / base # resolve relative dirs against the repo root, not the cwd
target = base / f"{index:02d}_{name}.parquet"
try:
atomic_write_parquet(frame, target)
get_logger("checkpoint").info(
"checkpoint.write", phase=name, rows=frame.height, cols=frame.width, path=str(target)
)
except Exception as failure: # observational only — never break the rank
get_logger("checkpoint").warning("checkpoint.fail", phase=name, error=str(failure))
|