pelinker.grid_export

Per-sample clustering grid CSV schema and selection at the consensus hyperparameter.

apply_chosen_min_cluster_size_to_grid(df_grid, chosen_by_combo)

Overwrite chosen_min_cluster_size per (model, layer) (e.g. after re-solving the grid).

Source code in pelinker/grid_export.py

def apply_chosen_min_cluster_size_to_grid(
    df_grid: pd.DataFrame,
    chosen_by_combo: dict[tuple[str, str], int],
) -> pd.DataFrame:
    """Overwrite ``chosen_min_cluster_size`` per (model, layer) (e.g. after re-solving the grid)."""
    if GRID_COL_CHOSEN_MIN_CLUSTER_SIZE not in df_grid.columns:
        raise ValueError(f"grid frame missing {GRID_COL_CHOSEN_MIN_CLUSTER_SIZE!r}")
    out = df_grid.copy()
    for (model, layer), mcs in chosen_by_combo.items():
        mask = (out["model"].astype(str) == model) & (out["layer"].astype(str) == layer)
        out.loc[mask, GRID_COL_CHOSEN_MIN_CLUSTER_SIZE] = int(mcs)
    return out

grid_chosen_hyperparameters_to_jsonable(solved_by_combo, optimization_config)

Build a JSON-serializable document of pooled grid solver results per (model, layer).

Source code in pelinker/grid_export.py

def grid_chosen_hyperparameters_to_jsonable(
    solved_by_combo: dict[tuple[str, str], SmoothedGridOptimumResult],
    optimization_config: ClusteringOptimizationConfig,
) -> dict[str, Any]:
    """Build a JSON-serializable document of pooled grid solver results per (model, layer)."""
    config = optimization_config
    combos = [
        _solved_combo_to_jsonable(model, layer, solved, config)
        for (model, layer), solved in sorted(solved_by_combo.items())
    ]
    return {
        "schema": "pelinker.grid_chosen_hyperparameters.v1",
        "solver_config": _solver_config_snapshot(config),
        "combinations": combos,
    }

grid_export_column_order()

Canonical column order for results_grid_per_sample.csv.

Source code in pelinker/grid_export.py

def grid_export_column_order() -> list[str]:
    """Canonical column order for ``results_grid_per_sample.csv``."""
    return [
        "model",
        "layer",
        "sample_idx",
        GRID_COL_CHOSEN_MIN_CLUSTER_SIZE,
        "min_cluster_size",
        "icm",
        "n_clusters",
        "dbcv",
        "ari",
    ]

grid_export_rows_from_report(report, *, model, layer, sample_idx, chosen_min_cluster_size)

Expand one sample's grid metrics_df into rows for results_grid_per_sample.csv.

chosen_min_cluster_size is the pooled consensus hyperparameter for the (model, layer) combination; it is duplicated on every grid row for that sample.

Source code in pelinker/grid_export.py

def grid_export_rows_from_report(
    report: ModelSelectionReport,
    *,
    model: str,
    layer: str,
    sample_idx: int,
    chosen_min_cluster_size: int,
) -> pd.DataFrame:
    """
    Expand one sample's grid ``metrics_df`` into rows for ``results_grid_per_sample.csv``.

    ``chosen_min_cluster_size`` is the pooled consensus hyperparameter for the
    (model, layer) combination; it is duplicated on every grid row for that sample.
    """
    return report.metrics_df.assign(
        model=model,
        layer=layer,
        sample_idx=sample_idx,
        **{GRID_COL_CHOSEN_MIN_CLUSTER_SIZE: int(chosen_min_cluster_size)},
    )

has_grid_points_for_dbcv_ari_scatter(df_grid)

True if df_grid has the columns needed for the DBCV vs ARI scatter.

Source code in pelinker/grid_export.py

def has_grid_points_for_dbcv_ari_scatter(df_grid: pd.DataFrame) -> bool:
    """True if ``df_grid`` has the columns needed for the DBCV vs ARI scatter."""
    return _GRID_POINT_COLUMNS.issubset(df_grid.columns)

per_combo_metrics_from_grid(df_grid)

Per (model, layer), list of per-sample grid metric tables for re-solving min_cluster_size.

Each DataFrame has columns among min_cluster_size, dbcv, ari, n_clusters, icm.

Source code in pelinker/grid_export.py

def per_combo_metrics_from_grid(
    df_grid: pd.DataFrame,
) -> dict[tuple[str, str], list[pd.DataFrame]]:
    """
    Per (model, layer), list of per-sample grid metric tables for re-solving ``min_cluster_size``.

    Each DataFrame has columns among ``min_cluster_size``, ``dbcv``, ``ari``, ``n_clusters``, ``icm``.
    """
    metric_cols = [
        c
        for c in ("min_cluster_size", "dbcv", "ari", "n_clusters", "icm")
        if c in df_grid.columns
    ]
    if "min_cluster_size" not in metric_cols or "dbcv" not in metric_cols:
        return {}

    out: dict[tuple[str, str], list[pd.DataFrame]] = {}
    for (model, layer), combo_df in df_grid.groupby(["model", "layer"], sort=False):
        metrics_list: list[pd.DataFrame] = []
        if "sample_idx" in combo_df.columns:
            groups = combo_df.groupby("sample_idx", sort=True)
        else:
            groups = [(0, combo_df)]
        for _sidx, sample_df in groups:
            metrics_list.append(sample_df[metric_cols].reset_index(drop=True))
        if metrics_list:
            out[(str(model), str(layer))] = metrics_list
    return out

select_grid_points_at_chosen_min_cluster_size(df_grid, *, require_n_clusters_gt_one=True)

One row per (model, layer, sample_idx): (dbcv, ari) at chosen_min_cluster_size.

Rows are taken from the grid sweep where min_cluster_size equals the consensus chosen_min_cluster_size for that (model, layer). This matches the vertical marker on per-combination error-bar plots.

Source code in pelinker/grid_export.py

def select_grid_points_at_chosen_min_cluster_size(
    df_grid: pd.DataFrame,
    *,
    require_n_clusters_gt_one: bool = True,
) -> pd.DataFrame:
    """
      One row per (model, layer, sample_idx): ``(dbcv, ari)`` at ``chosen_min_cluster_size``.

      Rows are taken from the grid sweep where ``min_cluster_size`` equals the consensus
    ``chosen_min_cluster_size`` for that (model, layer). This matches the vertical marker
      on per-combination error-bar plots.
    """
    if not _GRID_POINT_COLUMNS.issubset(df_grid.columns):
        missing = sorted(_GRID_POINT_COLUMNS - set(df_grid.columns))
        raise ValueError(f"grid frame missing required columns: {missing}")

    parts: list[pd.DataFrame] = []
    for (model, layer), g in df_grid.groupby(["model", "layer"], sort=False):
        chosen = _resolved_chosen_min_cluster_size(g[GRID_COL_CHOSEN_MIN_CLUSTER_SIZE])
        if chosen is None:
            continue
        mcs = g["min_cluster_size"].astype(np.float64)
        at = g.loc[mcs == chosen].copy()
        if require_n_clusters_gt_one and "n_clusters" in at.columns:
            at = at.loc[at["n_clusters"] > 1]
        if at.empty:
            continue
        at = at.drop_duplicates(
            subset=["model", "layer", "sample_idx"],
            keep="last",
        )
        at = at.loc[at["ari"].notna()]
        if at.empty:
            continue
        parts.append(at)

    if not parts:
        return df_grid.iloc[0:0].copy()
    return pd.concat(parts, ignore_index=True)

write_grid_chosen_hyperparameters(path, solved_by_combo, optimization_config)

Write grid_chosen_hyperparameters.json for downstream final fit.

Source code in pelinker/grid_export.py

def write_grid_chosen_hyperparameters(
    path: pathlib.Path,
    solved_by_combo: dict[tuple[str, str], SmoothedGridOptimumResult],
    optimization_config: ClusteringOptimizationConfig,
) -> None:
    """Write ``grid_chosen_hyperparameters.json`` for downstream final fit."""
    payload = grid_chosen_hyperparameters_to_jsonable(
        solved_by_combo, optimization_config
    )
    path = path.expanduser()
    path.parent.mkdir(parents=True, exist_ok=True)
    normalized = _json_normalize(payload)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(normalized, indent=2) + "\n", encoding="utf-8")
    tmp.replace(path)