def render_model_selection_summary_figures(
report_path: pathlib.Path,
*,
checkpoint_path: pathlib.Path | None = None,
grid_csv_path: pathlib.Path | None = None,
fine_screener_eval_path: pathlib.Path | None = None,
fine_metadata_path: pathlib.Path | None = None,
optimization_config: ClusteringOptimizationConfig | None = None,
all_pca_pairgrid_samples: bool = False,
) -> SummaryFigureRenderResult:
"""
Regenerate aggregate model-selection figures from on-disk artifacts (no parquet re-load).
Uses the checkpoint summaries for score heatmaps and AUC panels, the optional
per-sample grid CSV for the DBCV vs ARI scatter, gzipped JSON Lines
``fine_screener_eval.jsonl.gz`` for ROC comparison, and
``fine_metadata.jsonl.gz`` for the PCA quality pair grid.
Writes the same PNG filenames as ``model_selection`` end-of-run and corresponding
PDF siblings: ``model.perf.heatmap.png``, ``model.*.heatmap.png`` (scores, ARI,
screener LDA/SVM/best, OOV, combined), ``model.screener_oov_auc.png``,
``model.roc_curves.png``, ``model.roc_best.png`` (best combined-AUC combo),
``model.dbcv_vs_ari.png``, ``model_selection.summary.json``, and per-combination
``{model}_{layer}_sample{n}_pca_quality_pairgrid.png`` from fine metadata (lowest
sample index per combo unless ``all_pca_pairgrid_samples`` is true).
"""
report_path = report_path.expanduser()
ckpt_file = (
checkpoint_path.expanduser()
if checkpoint_path
else report_path / DEFAULT_CHECKPOINT_NAME
)
detail_path = (
grid_csv_path.expanduser()
if grid_csv_path
else report_path / CLUSTERING_SEARCH_GRID_PER_SAMPLE_CSV_BASENAME
)
screener_path = (
fine_screener_eval_path.expanduser()
if fine_screener_eval_path
else report_path / FINE_SCREENER_EVAL_BASENAME
)
fm_path = (
fine_metadata_path.expanduser()
if fine_metadata_path
else report_path / CLUSTERING_SEARCH_FINE_METADATA_BASENAME
)
written: list[pathlib.Path] = []
skipped: list[str] = []
chosen_hyperparameters_path: pathlib.Path | None = None
chosen_by_combo_flat: tuple[tuple[str, str, int], ...] = ()
summary_json_path: pathlib.Path | None = None
df_grid_detail = read_optional_csv(detail_path)
if df_grid_detail is not None and not df_grid_detail.empty:
df_grid_detail = dedupe_per_sample_grid(df_grid_detail)
combo_metrics = _per_combo_metrics_from_grid(df_grid_detail)
solver_config = optimization_config or ClusteringOptimizationConfig()
solved_by_combo = solve_pooled_grid_by_combo_from_grid(
df_grid_detail, solver_config
)
chosen_by_combo = {
combo: result.chosen_min_cluster_size
for combo, result in solved_by_combo.items()
}
if chosen_by_combo:
df_grid_detail = apply_chosen_min_cluster_size_to_grid(
df_grid_detail, chosen_by_combo
)
tmp_grid = detail_path.with_suffix(detail_path.suffix + ".tmp")
df_grid_detail.to_csv(tmp_grid, index=False)
tmp_grid.replace(detail_path)
written.append(detail_path)
chosen_hyperparameters_path = (
report_path / CLUSTERING_SEARCH_GRID_CHOSEN_JSON_BASENAME
)
write_grid_chosen_hyperparameters(
chosen_hyperparameters_path, solved_by_combo, solver_config
)
written.append(chosen_hyperparameters_path)
chosen_by_combo_flat = tuple(
(model, layer, int(mcs))
for (model, layer), mcs in sorted(chosen_by_combo.items())
)
scatter_path = _summary_plot_path(report_path, "model.dbcv_vs_ari")
if plot_dbcv_vs_ari_from_grid(df_grid_detail, scatter_path):
written.append(scatter_path)
else:
skipped.append(
"DBCV vs ARI scatter: insufficient grid columns or no ARI data"
)
# Per-combination metric plots (DBCV / ARI / n_clusters vs min_cluster_size).
for (model, layer), (metrics_list, _stored_mcs) in combo_metrics.items():
combo = (model, layer)
solve_result = solved_by_combo.get(combo)
chosen_mcs = float(chosen_by_combo.get(combo, _stored_mcs or float("nan")))
safe_layer = layer.replace("/", "_").replace("+", "__")
if len(metrics_list) > 1:
out_p = report_path / f"{model}_{safe_layer}_error_bars.png"
plot_metrics_with_error_bars(
metrics_list,
out_p,
chosen_min_cluster_size=chosen_mcs,
grid_solve=solve_result,
)
else:
out_p = report_path / f"{model}_{safe_layer}.png"
plot_metrics(metrics_list[0], out_p)
written.append(out_p)
else:
skipped.append(f"Grid CSV missing or empty: {detail_path}")
if not ckpt_file.exists():
skipped.append(f"Checkpoint missing (heatmaps/bar/ROC need it): {ckpt_file}")
return SummaryFigureRenderResult(
written_paths=tuple(written),
skipped_messages=tuple(skipped),
chosen_hyperparameters_path=chosen_hyperparameters_path,
chosen_by_combo=chosen_by_combo_flat,
summary_json_path=summary_json_path,
)
try:
ckpt = load_checkpoint(ckpt_file)
except (OSError, ValueError) as e:
skipped.append(f"Checkpoint unreadable ({ckpt_file}): {e}")
return SummaryFigureRenderResult(
written_paths=tuple(written),
skipped_messages=tuple(skipped),
chosen_hyperparameters_path=chosen_hyperparameters_path,
chosen_by_combo=chosen_by_combo_flat,
summary_json_path=summary_json_path,
)
results = results_from_checkpoint(ckpt)
if not results:
skipped.append("Checkpoint has no summaries_by_key rows")
return SummaryFigureRenderResult(
written_paths=tuple(written),
skipped_messages=tuple(skipped),
chosen_hyperparameters_path=chosen_hyperparameters_path,
chosen_by_combo=chosen_by_combo_flat,
summary_json_path=summary_json_path,
)
df_results = pd.DataFrame([r.to_flat_dict() for r in results])
df_results.insert(
0,
"combo_key",
[combo_key_for_results_row(row) for _, row in df_results.iterrows()],
)
df_results = df_results.sort_values(["model", "layer"])
try:
summary_json_path = write_model_selection_summary_from_results(
report_path,
df_results,
chosen_by_combo=chosen_by_combo_flat,
chosen_hyperparameters_path=chosen_hyperparameters_path,
)
written.append(summary_json_path)
except OSError as e:
skipped.append(f"Summary JSON: write failed: {e}")
df_heatmap = df_results[~df_results["model"].isin(["fusion2", "fusion3"])].copy()
if len(df_heatmap) > 0:
heatmap_path = _summary_plot_path(report_path, "model.perf.heatmap")
plot_heatmap(
df_heatmap, heatmap_path, metric="best_score", metric_label="Best Score"
)
written.append(heatmap_path)
else:
skipped.append("Score heatmap: no singleton (non-fusion2/3) rows")
if len(df_heatmap) > 0:
if "ari" in df_heatmap.columns and df_heatmap["ari"].notna().any():
ari_heatmap_path = _summary_plot_path(report_path, "model.ari.heatmap")
plot_heatmap(
df_heatmap,
ari_heatmap_path,
metric="ari",
metric_label="ARI",
)
written.append(ari_heatmap_path)
else:
skipped.append("ARI heatmap: missing or all-NaN ARI column")
auc_heat_specs: list[tuple[str, str, pathlib.Path]] = [
(
"screener_lda_auc_mean",
"Screener LDA AUC",
pathlib.Path("model.screener_lda_auc.heatmap.png"),
),
(
"screener_svm_auc_mean",
"Screener SVM AUC",
pathlib.Path("model.screener_svm_auc.heatmap.png"),
),
(
"screener_auc_mean",
"Screener AUC (best)",
pathlib.Path("model.screener_auc.heatmap.png"),
),
("oov_auc_mean", "OOV AUC", pathlib.Path("model.oov_auc.heatmap.png")),
(
"combined_auc_mean",
"Combined AUC",
pathlib.Path("model.combined_auc.heatmap.png"),
),
]
if len(df_heatmap) > 0:
for col_auc, label_auc, fn_auc in auc_heat_specs:
if col_auc not in df_heatmap.columns:
skipped.append(
f"{label_auc} heatmap: column {col_auc!r} not in summaries (older checkpoint?)"
)
continue
if not df_heatmap[col_auc].notna().any():
skipped.append(f"{label_auc} heatmap: all NaN")
continue
out_p = report_path / fn_auc
plot_heatmap(
df_heatmap,
out_p,
metric=col_auc,
metric_label=label_auc,
secondary_metric=None,
)
written.append(out_p)
else:
skipped.append(
"Screener LDA/SVM/best, OOV, combined AUC heatmaps: no singleton rows"
)
bar_path = _summary_plot_path(report_path, "model.screener_oov_auc")
if plot_screener_oov_bar(df_heatmap, bar_path):
written.append(bar_path)
else:
skipped.append(
"Screener/OOV/combined bar: missing metrics or empty after fusion filter"
)
if screener_path.exists():
roc_df = read_optional_jsonl_gzip(screener_path)
if (
roc_df is not None
and not roc_df.empty
and "combined_auc_mean" in df_results.columns
):
top_df = df_results.dropna(subset=["combined_auc_mean"]).nlargest(
3,
"combined_auc_mean",
)
top_keys_top = top_df["combo_key"].astype(str).drop_duplicates().tolist()
roc_out = _summary_plot_path(report_path, "model.roc_curves")
if plot_roc_comparison(roc_df, roc_out, combo_keys=top_keys_top):
written.append(roc_out)
else:
skipped.append(
"ROC curves: plot_roc_comparison returned False (columns or data)"
)
best_key = str(top_df["combo_key"].iloc[0])
roc_best_out = _summary_plot_path(report_path, "model.roc_best")
if plot_roc_comparison(roc_df, roc_best_out, combo_keys=[best_key]):
written.append(roc_best_out)
else:
skipped.append(
"Best-case ROC: plot_roc_comparison returned False (columns or data)"
)
else:
skipped.append(
"ROC curves: empty screener eval or no combined_auc_mean in summaries"
)
else:
skipped.append(f"ROC curves: file missing {screener_path}")
fm = read_optional_jsonl_gzip(fm_path)
if fm is not None and not fm.empty:
pca_written, pca_skipped = write_pca_quality_pairgrids_from_fine_metadata(
fm,
report_path,
source_name=f"summary figures ({fm_path})",
all_samples=all_pca_pairgrid_samples,
)
written.extend(pca_written)
skipped.extend(pca_skipped)
else:
skipped.append(
f"PCA quality pair grid: fine metadata missing or empty: {fm_path}"
)
return SummaryFigureRenderResult(
written_paths=tuple(written),
skipped_messages=tuple(skipped),
chosen_hyperparameters_path=chosen_hyperparameters_path,
chosen_by_combo=chosen_by_combo_flat,
summary_json_path=summary_json_path,
)