Skip to content

pelinker.model_selection.artifacts

Artifact I/O for model-selection runs (grid CSV, fine metadata, screener eval).

merge_new_frames_into_per_sample_grid_csv(detail_path, new_frames)

Append grid rows to results_grid_per_sample.csv (merge + dedupe, atomic replace).

Source code in pelinker/model_selection/artifacts.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def merge_new_frames_into_per_sample_grid_csv(
    detail_path: pathlib.Path,
    new_frames: list[pd.DataFrame],
) -> None:
    """Append grid rows to ``results_grid_per_sample.csv`` (merge + dedupe, atomic replace)."""
    if not new_frames:
        return
    new_df = pd.concat(new_frames, ignore_index=True)
    if new_df.empty:
        return
    prior = read_optional_csv(detail_path)
    if prior is not None and not prior.empty:
        merged = pd.concat([prior, new_df], ignore_index=True)
    else:
        merged = new_df
    merged = dedupe_per_sample_grid(merged)
    tmp = detail_path.with_suffix(detail_path.suffix + ".tmp")
    merged.to_csv(tmp, index=False)
    tmp.replace(detail_path)

read_optional_jsonl_gzip(path)

Load gzipped JSON Lines (pandas); return None if missing or unreadable.

Source code in pelinker/model_selection/artifacts.py 
86
87
88
89
90
91
92
93
94
95
96
def read_optional_jsonl_gzip(path: pathlib.Path) -> pd.DataFrame | None:
    """Load gzipped JSON Lines (pandas); return None if missing or unreadable."""
    if not path.exists():
        return None
    try:
        df = pd.read_json(path, lines=True, compression="gzip")
        if df.empty:
            return None
        return df
    except Exception:
        return None

singleton_score_by_model_layer_from_checkpoint(ckpt)

Mean DBCV per (model, layer) for fusion proxy (singletons only).

Source code in pelinker/model_selection/artifacts.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def singleton_score_by_model_layer_from_checkpoint(
    ckpt: ModelSelectionCheckpoint,
) -> dict[tuple[str, str], float]:
    """Mean DBCV per (model, layer) for fusion proxy (singletons only)."""
    out = score_by_model_layer_from_checkpoint(ckpt.singleton_scores_by_key)
    if out:
        return out
    for key, row in ckpt.summaries_by_key.items():
        if not key.startswith("1:"):
            continue
        ml = model_layer_from_singleton_key(key)
        score = row.get("best_score")
        if score is not None:
            out[ml] = float(score)
    return out