pelinker.reporting

AllScreenerCvResult dataclass

Unified stratified CV: embedding screener (LDA vs SVM), manifold OOV, and stacked score.

Source code in pelinker/reporting.py

@dataclass(frozen=True)
class AllScreenerCvResult:
    """Unified stratified CV: embedding screener (LDA vs SVM), manifold OOV, and stacked score."""

    screener_lda: BinaryClassifierMetrics
    screener_svm: BinaryClassifierMetrics
    screener_best_kind: str
    screener_best: BinaryClassifierMetrics
    oov_winner_kind: str
    oov: BinaryClassifierMetrics
    combined: BinaryClassifierMetrics

BinaryClassifierMetrics dataclass

Precision / recall / F1 / AUC vs negative class (label 1); spread is fold-wise.

Source code in pelinker/reporting.py

@dataclass(frozen=True)
class BinaryClassifierMetrics:
    """Precision / recall / F1 / AUC vs negative class (label 1); spread is fold-wise."""

    precision: MetricMeanStd
    recall: MetricMeanStd
    f1: MetricMeanStd
    auc: MetricMeanStd

ClusteringFitMetrics dataclass

Fit-time clustering diagnostics at a fixed min_cluster_size.

Source code in pelinker/reporting.py

@dataclass(frozen=True)
class ClusteringFitMetrics:
    """Fit-time clustering diagnostics at a fixed ``min_cluster_size``."""

    min_cluster_size: int
    dbcv: float | None
    """HDBSCAN ``relative_validity_`` when available."""
    ari: float | None
    n_clusters_emergent: int
    noise_fraction: float
    n_samples: int

dbcv instance-attribute

HDBSCAN relative_validity_ when available.

ClusteringHyperparameters dataclass

HDBSCAN (and related) choices selected by the grid search / smoother.

Add fields here as more knobs participate in optimization; call sites then stay typed.

Source code in pelinker/reporting.py

@dataclass(frozen=True)
class ClusteringHyperparameters:
    """
    HDBSCAN (and related) choices selected by the grid search / smoother.

    Add fields here as more knobs participate in optimization; call sites then stay typed.
    """

    min_cluster_size: int

ClusteringSearchSummaryRow dataclass

One row of the model×layer clustering search table (singleton or fusion label).

Use :meth:to_flat_dict for CSV / pandas / heatmaps (legacy column names).

Source code in pelinker/reporting.py

@dataclass(frozen=True)
class ClusteringSearchSummaryRow:
    """
    One row of the model×layer clustering search table (singleton or fusion label).

    Use :meth:`to_flat_dict` for CSV / pandas / heatmaps (legacy column names).
    """

    model: str
    layer: str
    hyperparameters: HyperparameterSearchStats
    number_properties: MeanWithUncertainty
    n_clusters_emergent: MeanWithUncertainty
    dbcv: MeanWithUncertainty
    ari: MeanWithUncertainty | None
    all_screener_cv: AllScreenerCvResult | None = None
    n_rows_realized: MeanWithUncertainty | None = None
    """Mean (and std) realized mention-row count across this combination's draws.

    ``best_size`` is only interpretable against this scale; carrying it here is what lets
    :mod:`pelinker.scaling` fit ``min_cluster_size*(N)`` across runs.
    """

    def to_flat_dict(self) -> dict[str, str | float | None]:
        """Keys aligned with grid CSV / checkpoint / ``plot_heatmap`` expectations."""
        h = self.hyperparameters.min_cluster_size
        p = self.number_properties
        k = self.n_clusters_emergent
        d = self.dbcv
        row: dict[str, str | float | None] = {
            "model": self.model,
            "layer": self.layer,
            "best_size": h.mean,
            "best_size_std": h.std,
            "number_properties": p.mean,
            "number_properties_std": p.std,
            "n_clusters_emergent": k.mean,
            "n_clusters_emergent_std": k.std,
            "best_score": d.mean,
            "best_score_std": d.std,
        }
        nrr = self.n_rows_realized
        row["n_rows_realized"] = None if nrr is None else nrr.mean
        row["n_rows_realized_std"] = 0.0 if nrr is None else nrr.std
        if self.ari is None:
            row["ari"] = None
            row["ari_std"] = 0.0
        else:
            ari = self.ari
            row["ari"] = ari.mean
            row["ari_std"] = ari.std
        acv = self.all_screener_cv
        if acv is not None:
            row["screener_best_kind"] = acv.screener_best_kind
            _binary_metrics_into_row(row, acv.screener_best, "screener")
            row["oov_winner_kind"] = acv.oov_winner_kind
            _binary_metrics_into_row(row, acv.oov, "oov")
            _binary_metrics_into_row(row, acv.combined, "combined")
            _binary_metrics_into_row(row, acv.screener_lda, "screener_lda")
            _binary_metrics_into_row(row, acv.screener_svm, "screener_svm")
        return row

n_rows_realized = None class-attribute instance-attribute

Mean (and std) realized mention-row count across this combination's draws.

best_size is only interpretable against this scale; carrying it here is what lets :mod:pelinker.scaling fit min_cluster_size*(N) across runs.

to_flat_dict()

Keys aligned with grid CSV / checkpoint / plot_heatmap expectations.

Source code in pelinker/reporting.py

def to_flat_dict(self) -> dict[str, str | float | None]:
    """Keys aligned with grid CSV / checkpoint / ``plot_heatmap`` expectations."""
    h = self.hyperparameters.min_cluster_size
    p = self.number_properties
    k = self.n_clusters_emergent
    d = self.dbcv
    row: dict[str, str | float | None] = {
        "model": self.model,
        "layer": self.layer,
        "best_size": h.mean,
        "best_size_std": h.std,
        "number_properties": p.mean,
        "number_properties_std": p.std,
        "n_clusters_emergent": k.mean,
        "n_clusters_emergent_std": k.std,
        "best_score": d.mean,
        "best_score_std": d.std,
    }
    nrr = self.n_rows_realized
    row["n_rows_realized"] = None if nrr is None else nrr.mean
    row["n_rows_realized_std"] = 0.0 if nrr is None else nrr.std
    if self.ari is None:
        row["ari"] = None
        row["ari_std"] = 0.0
    else:
        ari = self.ari
        row["ari"] = ari.mean
        row["ari_std"] = ari.std
    acv = self.all_screener_cv
    if acv is not None:
        row["screener_best_kind"] = acv.screener_best_kind
        _binary_metrics_into_row(row, acv.screener_best, "screener")
        row["oov_winner_kind"] = acv.oov_winner_kind
        _binary_metrics_into_row(row, acv.oov, "oov")
        _binary_metrics_into_row(row, acv.combined, "combined")
        _binary_metrics_into_row(row, acv.screener_lda, "screener_lda")
        _binary_metrics_into_row(row, acv.screener_svm, "screener_svm")
    return row

HyperparameterSearchStats dataclass

Distribution of chosen hyperparameters across repeated clustering samples.

Source code in pelinker/reporting.py

@dataclass(frozen=True)
class HyperparameterSearchStats:
    """Distribution of chosen hyperparameters across repeated clustering samples."""

    min_cluster_size: MeanWithUncertainty

LinkerFitDiagnostics dataclass

Per-row training diagnostics for plotting (often stratified-subsampled).

Source code in pelinker/reporting.py

@dataclass(frozen=True)
class LinkerFitDiagnostics:
    """Per-row training diagnostics for plotting (often stratified-subsampled)."""

    pca_residual: np.ndarray
    pca_mahalanobis: np.ndarray
    pca_spectral_entropy: np.ndarray
    oov_label: np.ndarray
    """``1`` iff ``entity == negative_label`` (same convention as :class:`ModelSelectionReport`)."""
    screener_decision: np.ndarray
    projection_score: np.ndarray
    n_total: int
    """Original mention count before subsampling (same as ``len(prepared)`` at fit time)."""
    sample_random_state: int
    """RNG seed used for stratified subsampling (or configured seed when no subsample)."""

n_total instance-attribute

Original mention count before subsampling (same as len(prepared) at fit time).

oov_label instance-attribute

1 iff entity == negative_label (same convention as :class:ModelSelectionReport).

sample_random_state instance-attribute

RNG seed used for stratified subsampling (or configured seed when no subsample).

MeanWithUncertainty dataclass

Sample mean and standard deviation (ddof=1) over repeated runs; std=0 for a single run.

Source code in pelinker/reporting.py

@dataclass(frozen=True)
class MeanWithUncertainty:
    """Sample mean and standard deviation (ddof=1) over repeated runs; ``std=0`` for a single run."""

    mean: float
    std: float

MetricMeanStd dataclass

Mean and spread (sample std over CV folds) for one scalar metric.

Source code in pelinker/reporting.py

@dataclass(frozen=True)
class MetricMeanStd:
    """Mean and spread (sample std over CV folds) for one scalar metric."""

    mean: float
    std: float

ModelSelectionReport dataclass

Report containing clustering analysis results for one sample.

Source code in pelinker/reporting.py

@dataclass
class ModelSelectionReport:
    """Report containing clustering analysis results for one sample."""

    hyperparameters: ClusteringHyperparameters
    best_score: float
    """DBCV (``relative_validity_``) at the chosen ``min_cluster_size`` (mean when from aggregate)."""

    number_properties: int
    """Count of distinct KB ``entity`` labels in the frame used for PCA→UMAP (excludes ``pelinker.onto.NEGATIVE_LABEL`` when screening)."""

    n_clusters_emergent: int
    """HDBSCAN emergent cluster count at :attr:`hyperparameters` ``min_cluster_size``.

    In model-selection this is the grid-optimal ``best_size``, not an arbitrary MCS.
    To compare with ``Linker.fit`` at a fixed MCS, use
    :func:`n_clusters_at_min_cluster_size` on :attr:`metrics_df` instead.
    """

    metrics_df: pd.DataFrame
    assignments: pd.DataFrame
    pca_residuals: np.ndarray
    pca_mahalanobis: np.ndarray
    pca_spectral_entropy: np.ndarray
    oov_label: np.ndarray
    """Per-row OOV mask: ``1`` iff ``entity == negative_label`` (same length as ``pca_residuals``)."""
    umap_clustering: np.ndarray
    cluster_viz: np.ndarray
    cluster_viz_method: str
    pca_reduced: np.ndarray
    all_screener_cv: AllScreenerCvResult | None = None
    """Unified stratified CV for embedding screener, manifold OOV, and stacked score."""
    screener_oos_datapoints: PerDatapointScores | None = None
    """Per-datapoint OOS scores (not serialized in JSON clustering report)."""
    ari: float | None = None
    training_diagnostics: LinkerFitDiagnostics | None = None
    """Stratified-subsampled PCA quality + screener / manifold OOV scores (linker fit only)."""
    mention_quality: pd.DataFrame | None = None
    """All mentions (pos+neg) with PCA quality scores and oov_label; cluster=-1 for negatives."""
    distillation_fidelity: DistillationFidelityMetrics | None = None
    """Held-out agreement between the compact entity head and its HDBSCAN teacher.

    ``None`` for legacy fits (no student) and when the holdout was disabled or skipped.
    """
    min_cluster_size_provenance: MinClusterSizeProvenance | None = None
    """Where :attr:`hyperparameters` ``min_cluster_size`` came from (linker fit only).

    Explicit, extrapolated from a scale curve, or the bare default — recorded so a fitted
    model never carries an unexplained hyperparameter.
    """
    n_rows_realized: int | None = None
    """Mention rows this draw actually clustered on.

    The realized N, **not** the ``clustering_sample_rows`` cap: when the frame is smaller
    than the cap, or the cap is unset, the two differ. Sample-size-dependent
    hyperparameters (``min_cluster_size``, and through it HDBSCAN ``min_samples``) are
    only interpretable against this number — see :mod:`pelinker.scaling`.
    """

all_screener_cv = None class-attribute instance-attribute

Unified stratified CV for embedding screener, manifold OOV, and stacked score.

best_score instance-attribute

DBCV (relative_validity_) at the chosen min_cluster_size (mean when from aggregate).

distillation_fidelity = None class-attribute instance-attribute

Held-out agreement between the compact entity head and its HDBSCAN teacher.

None for legacy fits (no student) and when the holdout was disabled or skipped.

mention_quality = None class-attribute instance-attribute

All mentions (pos+neg) with PCA quality scores and oov_label; cluster=-1 for negatives.

min_cluster_size_provenance = None class-attribute instance-attribute

Where :attr:hyperparameters min_cluster_size came from (linker fit only).

Explicit, extrapolated from a scale curve, or the bare default — recorded so a fitted model never carries an unexplained hyperparameter.

n_clusters_emergent instance-attribute

HDBSCAN emergent cluster count at :attr:hyperparameters min_cluster_size.

In model-selection this is the grid-optimal best_size, not an arbitrary MCS. To compare with Linker.fit at a fixed MCS, use :func:n_clusters_at_min_cluster_size on :attr:metrics_df instead.

n_rows_realized = None class-attribute instance-attribute

Mention rows this draw actually clustered on.

The realized N, not the clustering_sample_rows cap: when the frame is smaller than the cap, or the cap is unset, the two differ. Sample-size-dependent hyperparameters (min_cluster_size, and through it HDBSCAN min_samples) are only interpretable against this number — see :mod:pelinker.scaling.

number_properties instance-attribute

Count of distinct KB entity labels in the frame used for PCA→UMAP (excludes pelinker.onto.NEGATIVE_LABEL when screening).

oov_label instance-attribute

Per-row OOV mask: 1 iff entity == negative_label (same length as pca_residuals).

screener_oos_datapoints = None class-attribute instance-attribute

Per-datapoint OOS scores (not serialized in JSON clustering report).

training_diagnostics = None class-attribute instance-attribute

Stratified-subsampled PCA quality + screener / manifold OOV scores (linker fit only).

ModelSelectionRunReport dataclass

Standardized aggregate report for one model-selection run.

Source code in pelinker/reporting.py

@dataclass(frozen=True)
class ModelSelectionRunReport:
    """Standardized aggregate report for one model-selection run."""

    schema: str
    generated_at: str
    run_fingerprint: str
    run_config: dict[str, Any]
    checkpoint: dict[str, Any]
    combinations: list[dict[str, Any]]
    failures: list[dict[str, Any]]
    best_overall: dict[str, Any] | None
    best_per_model: dict[str, float]

NegativeScreenerInSampleMetrics dataclass

Train-set precision / recall / F1 for detecting negative_label (binary label 1).

Source code in pelinker/reporting.py

@dataclass(frozen=True)
class NegativeScreenerInSampleMetrics:
    """Train-set precision / recall / F1 for detecting ``negative_label`` (binary label 1)."""

    precision: float
    recall: float
    f1: float
    n_kb_mentions: int
    """Rows whose ``entity`` is not the synthetic negative label (class 0)."""
    n_negative_label_mentions: int
    """Rows whose ``entity`` equals the synthetic negative label (class 1)."""
    kind: ScreenerKind

n_kb_mentions instance-attribute

Rows whose entity is not the synthetic negative label (class 0).

n_negative_label_mentions instance-attribute

Rows whose entity equals the synthetic negative label (class 1).

PerDatapointScores dataclass

Out-of-sample scores per stratified-fold test datapoint.

Source code in pelinker/reporting.py

@dataclass(frozen=True)
class PerDatapointScores:
    """Out-of-sample scores per stratified-fold test datapoint."""

    orig_idx: list[int]
    entity: list[str]
    y_true: list[int]
    screener_lda_score: list[float]
    screener_svm_score: list[float]
    screener_best_score: list[float]
    oov_score: list[float]
    combined_score: list[float]

clustering_report_to_jsonable_dict(report)

Flatten a :class:ClusteringReport into JSON-serializable built-ins (no DataFrames/ndarrays).

Intended for json.dumps or for pickling a stable, language-adjacent blob. Schema version is stored under "schema" for forward compatibility.

Source code in pelinker/reporting.py

def clustering_report_to_jsonable_dict(report: ModelSelectionReport) -> dict[str, Any]:
    """
    Flatten a :class:`ClusteringReport` into JSON-serializable built-ins (no DataFrames/ndarrays).

    Intended for ``json.dumps`` or for pickling a stable, language-adjacent blob. Schema version
    is stored under ``"schema"`` for forward compatibility.
    """
    ari_out: float | None
    if report.ari is None:
        ari_out = None
    else:
        ari_f = float(report.ari)
        ari_out = None if math.isnan(ari_f) or math.isinf(ari_f) else ari_f

    return {
        "schema": _JSON_CLUSTERING_REPORT_SCHEMA,
        "hyperparameters": {
            "min_cluster_size": int(report.hyperparameters.min_cluster_size),
        },
        "min_cluster_size_provenance": (
            None
            if report.min_cluster_size_provenance is None
            else report.min_cluster_size_provenance.to_jsonable()
        ),
        "distillation_fidelity": (
            None
            if report.distillation_fidelity is None
            else report.distillation_fidelity.to_jsonable()
        ),
        "n_rows_realized": (
            None if report.n_rows_realized is None else int(report.n_rows_realized)
        ),
        "best_score": _json_normalize(float(report.best_score)),
        "number_properties": int(report.number_properties),
        "n_clusters_emergent": int(report.n_clusters_emergent),
        "metrics_df": _dataframe_to_jsonable_records(report.metrics_df),
        "assignments": _dataframe_to_jsonable_records(report.assignments),
        "pca_residuals": _ndarray_to_jsonable_nested(report.pca_residuals),
        "pca_mahalanobis": _ndarray_to_jsonable_nested(report.pca_mahalanobis),
        "pca_spectral_entropy": _ndarray_to_jsonable_nested(
            report.pca_spectral_entropy
        ),
        "oov_label": _ndarray_to_jsonable_nested(report.oov_label),
        "umap_clustering": _ndarray_to_jsonable_nested(report.umap_clustering),
        "cluster_viz": _ndarray_to_jsonable_nested(report.cluster_viz),
        "cluster_viz_method": str(report.cluster_viz_method),
        "pca_reduced": _ndarray_to_jsonable_nested(report.pca_reduced),
        "ari": ari_out,
        "all_screener_cv": (
            None
            if report.all_screener_cv is None
            else _json_normalize(_all_screener_cv_to_jsonable(report.all_screener_cv))
        ),
        "training_diagnostics": (
            None
            if report.training_diagnostics is None
            else _json_normalize(
                _linker_fit_diagnostics_to_jsonable(report.training_diagnostics)
            )
        ),
    }

clustering_search_summary_row_from_flat_dict(row)

Reconstruct :class:ClusteringSearchSummaryRow from :meth:to_flat_dict output.

Source code in pelinker/reporting.py

def clustering_search_summary_row_from_flat_dict(
    row: dict[str, str | float | None],
) -> ClusteringSearchSummaryRow:
    """Reconstruct :class:`ClusteringSearchSummaryRow` from :meth:`to_flat_dict` output."""
    ari_raw = row.get("ari")
    ari_block: MeanWithUncertainty | None
    if ari_raw is None or (isinstance(ari_raw, float) and math.isnan(ari_raw)):
        ari_block = None
    else:
        ari_block = MeanWithUncertainty(
            mean=float(ari_raw),
            std=float(row.get("ari_std") or 0.0),
        )
    # Absent in checkpoints written before n_rows_realized existed; stays None there.
    nrr_raw = row.get("n_rows_realized")
    nrr_block: MeanWithUncertainty | None
    if nrr_raw is None or (isinstance(nrr_raw, float) and math.isnan(nrr_raw)):
        nrr_block = None
    else:
        nrr_block = MeanWithUncertainty(
            mean=float(nrr_raw),
            std=float(row.get("n_rows_realized_std") or 0.0),
        )
    return ClusteringSearchSummaryRow(
        model=str(row["model"]),
        layer=str(row["layer"]),
        hyperparameters=HyperparameterSearchStats(
            min_cluster_size=MeanWithUncertainty(
                mean=float(row["best_size"]),
                std=float(row["best_size_std"] or 0.0),
            ),
        ),
        number_properties=MeanWithUncertainty(
            mean=float(row["number_properties"]),
            std=float(row["number_properties_std"] or 0.0),
        ),
        n_clusters_emergent=MeanWithUncertainty(
            mean=float(row["n_clusters_emergent"]),
            std=float(row["n_clusters_emergent_std"] or 0.0),
        ),
        dbcv=MeanWithUncertainty(
            mean=float(row["best_score"]),
            std=float(row["best_score_std"] or 0.0),
        ),
        ari=ari_block,
        all_screener_cv=_all_screener_cv_from_flat_row(row),
        n_rows_realized=nrr_block,
    )

entity_negative_label_mask_01(entities, negative_label)

Per-row binary labels aligned with entities: 1 if the row's entity equals negative_label (same convention as the negative screener positive class), else 0.

Source code in pelinker/reporting.py

def entity_negative_label_mask_01(
    entities: pd.Series | np.ndarray,
    negative_label: str,
) -> np.ndarray:
    """
    Per-row binary labels aligned with ``entities``: ``1`` if the row's ``entity`` equals
    ``negative_label`` (same convention as the negative screener positive class), else ``0``.
    """
    if isinstance(entities, pd.Series):
        s = entities.astype(str).to_numpy()
    else:
        s = np.asarray(entities).astype(str)
    if s.size == 0:
        return np.zeros(0, dtype=np.int64)
    return (s == negative_label).astype(np.int64, copy=False)

linker_fit_cluster_composition_path(report_dir)

Filesystem path for the entity-weighted cluster composition artifact.

Source code in pelinker/reporting.py

def linker_fit_cluster_composition_path(
    report_dir: str | pathlib.Path,
) -> pathlib.Path:
    """Filesystem path for the entity-weighted cluster composition artifact."""
    return (
        pathlib.Path(report_dir).expanduser() / LINKER_FIT_CLUSTER_COMPOSITION_BASENAME
    )

linker_fit_clustering_report_path(report_dir)

Filesystem path for the fit-time :class:ClusteringReport JSON under report_dir.

Source code in pelinker/reporting.py

def linker_fit_clustering_report_path(report_dir: str | pathlib.Path) -> pathlib.Path:
    """Filesystem path for the fit-time :class:`ClusteringReport` JSON under ``report_dir``."""
    return pathlib.Path(report_dir).expanduser() / LINKER_FIT_CLUSTERING_REPORT_BASENAME

linker_fit_kb_out_path(report_dir)

Filesystem path for the KB-out catalog JSON under report_dir.

Source code in pelinker/reporting.py

def linker_fit_kb_out_path(report_dir: str | pathlib.Path) -> pathlib.Path:
    """Filesystem path for the KB-out catalog JSON under ``report_dir``."""
    return pathlib.Path(report_dir).expanduser() / LINKER_FIT_KB_OUT_BASENAME

mention_quality_frame(dfr, *, neg_mask, cluster_kb, pca_residuals, pca_mahalanobis, pca_spectral_entropy, negative_label)

Per-mention PCA quality and labels for all rows (KB clustered; negatives cluster=-1).

Source code in pelinker/reporting.py

def mention_quality_frame(
    dfr: pd.DataFrame,
    *,
    neg_mask: np.ndarray,
    cluster_kb: np.ndarray,
    pca_residuals: np.ndarray,
    pca_mahalanobis: np.ndarray,
    pca_spectral_entropy: np.ndarray,
    negative_label: str,
) -> pd.DataFrame:
    """Per-mention PCA quality and labels for all rows (KB clustered; negatives cluster=-1)."""
    optional = ["pmid", "mention"]
    optional_cols = [c for c in optional if c in dfr.columns]
    out = dfr[["entity", *optional_cols]].copy()
    cluster_full = np.full(len(dfr), -1, dtype=np.int64)
    cluster_full[~neg_mask] = np.asarray(cluster_kb, dtype=np.int64).ravel()
    out["cluster"] = cluster_full
    out["oov_label"] = entity_negative_label_mask_01(dfr["entity"], negative_label)
    out["pca_residual"] = np.asarray(pca_residuals, dtype=np.float64).ravel()
    out["pca_mahalanobis"] = np.asarray(pca_mahalanobis, dtype=np.float64).ravel()
    out["pca_spectral_entropy"] = np.asarray(
        pca_spectral_entropy, dtype=np.float64
    ).ravel()
    ordered = [
        "entity",
        *optional_cols,
        "cluster",
        "oov_label",
        "pca_residual",
        "pca_mahalanobis",
        "pca_spectral_entropy",
    ]
    return out[ordered]

model_selection_run_report_path(report_dir)

Filesystem path for the standardized model-selection aggregate report.

Source code in pelinker/reporting.py

def model_selection_run_report_path(report_dir: str | pathlib.Path) -> pathlib.Path:
    """Filesystem path for the standardized model-selection aggregate report."""
    return pathlib.Path(report_dir).expanduser() / MODEL_SELECTION_RUN_REPORT_BASENAME

model_selection_summary_json_path(report_dir)

Top-level replot summary (rankings, best combos) as plain JSON.

Source code in pelinker/reporting.py

def model_selection_summary_json_path(report_dir: str | pathlib.Path) -> pathlib.Path:
    """Top-level replot summary (rankings, best combos) as plain JSON."""
    return pathlib.Path(report_dir).expanduser() / MODEL_SELECTION_SUMMARY_JSON_BASENAME

n_clusters_at_min_cluster_size(metrics_df, min_cluster_size)

n_clusters from a grid metrics_df row at min_cluster_size (for fit parity).

Source code in pelinker/reporting.py

def n_clusters_at_min_cluster_size(
    metrics_df: pd.DataFrame,
    min_cluster_size: int,
) -> int | None:
    """``n_clusters`` from a grid ``metrics_df`` row at ``min_cluster_size`` (for fit parity)."""
    if (
        "min_cluster_size" not in metrics_df.columns
        or "n_clusters" not in metrics_df.columns
    ):
        return None
    hit = metrics_df.loc[
        metrics_df["min_cluster_size"] == min_cluster_size, "n_clusters"
    ]
    if hit.empty:
        return None
    return int(hit.iloc[0])

read_cluster_composition_json(path)

Load a composition table and metadata written by :func:write_cluster_composition_json.

Source code in pelinker/reporting.py

def read_cluster_composition_json(
    path: str | pathlib.Path,
) -> tuple[pd.DataFrame, dict[str, Any]]:
    """Load a composition table and metadata written by :func:`write_cluster_composition_json`."""
    p = pathlib.Path(path).expanduser()
    with gzip.open(p, mode="rt", encoding="utf-8") as f:
        raw: dict[str, Any] = json.load(f)
    schema = str(raw.get("schema", ""))
    if schema not in (
        "pelinker.fit_cluster_composition.v1",
        _FIT_CLUSTER_COMPOSITION_SCHEMA,
    ):
        raise ValueError(f"Unsupported cluster composition schema: {schema!r}")
    meta = {k: v for k, v in raw.items() if k != "rows"}
    return pd.DataFrame(raw["rows"]), meta

read_clustering_report_json(path)

Load a :class:ModelSelectionReport written by :func:write_clustering_report_json.

Supports schema pelinker.clustering_report.v10 (cluster-space viz coords).

Source code in pelinker/reporting.py

def read_clustering_report_json(path: str | pathlib.Path) -> ModelSelectionReport:
    """
    Load a :class:`ModelSelectionReport` written by :func:`write_clustering_report_json`.

    Supports schema ``pelinker.clustering_report.v10`` (cluster-space viz coords).
    """
    p = pathlib.Path(path).expanduser()
    with gzip.open(p, mode="rt", encoding="utf-8") as f:
        raw: dict[str, Any] = json.load(f)

    schema = str(raw.get("schema", ""))
    if schema != "pelinker.clustering_report.v10":
        raise ValueError(f"Unsupported clustering report schema: {schema!r}")

    hp = raw["hyperparameters"]
    h = ClusteringHyperparameters(min_cluster_size=int(hp["min_cluster_size"]))
    metrics_df = pd.DataFrame(raw["metrics_df"])
    assignments = pd.DataFrame(raw["assignments"])

    def _farr(key: str) -> np.ndarray:
        return np.asarray(raw[key], dtype=np.float64)

    def _iarr(key: str) -> np.ndarray:
        return np.asarray(raw[key], dtype=np.int64)

    ari_raw = raw.get("ari")
    ari: float | None
    if ari_raw is None:
        ari = None
    else:
        ari = float(ari_raw)

    # Nested CV summaries are not round-tripped here (linker fit reports use ``None``).
    all_cv: AllScreenerCvResult | None = None

    td_raw = raw.get("training_diagnostics")
    training_diagnostics = _linker_fit_diagnostics_from_jsonable(td_raw)

    return ModelSelectionReport(
        hyperparameters=h,
        best_score=float(raw["best_score"]),
        number_properties=int(raw["number_properties"]),
        n_clusters_emergent=int(raw["n_clusters_emergent"]),
        metrics_df=metrics_df,
        assignments=assignments,
        pca_residuals=_farr("pca_residuals"),
        pca_mahalanobis=_farr("pca_mahalanobis"),
        pca_spectral_entropy=_farr("pca_spectral_entropy"),
        oov_label=_iarr("oov_label"),
        umap_clustering=np.asarray(raw["umap_clustering"], dtype=np.float64),
        cluster_viz=np.asarray(raw["cluster_viz"], dtype=np.float64),
        cluster_viz_method=str(raw["cluster_viz_method"]),
        pca_reduced=np.asarray(raw["pca_reduced"], dtype=np.float64),
        all_screener_cv=all_cv,
        screener_oos_datapoints=None,
        ari=ari,
        training_diagnostics=training_diagnostics,
    )

read_kb_out_json(path)

Load KB-out catalog JSON.

Source code in pelinker/reporting.py

def read_kb_out_json(path: str | pathlib.Path) -> dict[str, Any]:
    """Load KB-out catalog JSON."""
    p = pathlib.Path(path).expanduser()
    with p.open(encoding="utf-8") as f:
        raw: dict[str, Any] = json.load(f)
    if str(raw.get("schema", "")) != _KB_OUT_SCHEMA:
        raise ValueError(f"Unsupported KB-out schema: {raw.get('schema')!r}")
    return raw

subsample_diagnostics_stratified(full, *, max_rows, random_state)

Stratified subsample by oov_label (preserve class proportions, at least one row per non-empty class when two classes exist). Mirrors logic used in pairgrid plotting.

Source code in pelinker/reporting.py

def subsample_diagnostics_stratified(
    full: LinkerFitDiagnostics,
    *,
    max_rows: int,
    random_state: int,
) -> LinkerFitDiagnostics:
    """
    Stratified subsample by ``oov_label`` (preserve class proportions, at least one row per
    non-empty class when two classes exist). Mirrors logic used in pairgrid plotting.
    """
    if max_rows < 1:
        raise ValueError("max_rows must be >= 1")
    n_total = full.n_total
    n = int(len(full.pca_residual))
    if n != n_total:
        raise ValueError(
            f"LinkerFitDiagnostics length mismatch: len(arrays)={n} vs n_total={n_total}"
        )

    def _copy_same() -> LinkerFitDiagnostics:
        return _copy_linker_fit_diagnostics(full, random_state)

    if n <= max_rows:
        return _copy_same()

    rng = np.random.default_rng(random_state)
    y = np.asarray(full.oov_label, dtype=np.int64).ravel()
    classes = np.unique(y)

    if len(classes) == 1:
        idx_all = np.flatnonzero(y == int(classes[0]))
        k = min(max_rows, len(idx_all))
        chosen = rng.choice(idx_all, size=k, replace=False)
    else:
        parts: list[np.ndarray] = []
        for cls in classes:
            idx = np.flatnonzero(y == int(cls))
            k_i = max(1, int(round(max_rows * len(idx) / n)))
            k_i = min(k_i, len(idx))
            parts.append(rng.choice(idx, size=k_i, replace=False))
        chosen = np.concatenate(parts)
        if len(chosen) > max_rows:
            chosen = rng.choice(chosen, size=max_rows, replace=False)

    chosen = np.sort(chosen.astype(np.int64, copy=False))
    return _slice_linker_fit_diagnostics(full, chosen, random_state)

summarize_clustering_reports_for_search(reports, *, model, layer, pooled_min_cluster_size=None)

Aggregate repeated :class:ClusteringReport runs into one search summary row.

When pooled_min_cluster_size is set (after aggregating grid curves across samples), best_size / best_size_std report that single consensus hyperparameter (std is 0) and dbcv is the mean (and std) of each sample's DBCV at that grid point. n_clusters_emergent remains the mean of per-sample fits at each sample's grid-optimal size; use :func:n_clusters_at_min_cluster_size on pooled grid CSV rows for a fixed MCS.

Otherwise (independent runs or legacy callers) best_size is the mean of per-report chosen sizes and dbcv is the mean of each report's best_score.

Raises:

Type	Description
ValueError	if reports is empty.

Source code in pelinker/reporting.py

def summarize_clustering_reports_for_search(
    reports: Sequence[ModelSelectionReport],
    *,
    model: str,
    layer: str,
    pooled_min_cluster_size: int | None = None,
) -> ClusteringSearchSummaryRow:
    """
    Aggregate repeated :class:`ClusteringReport` runs into one search summary row.

    When ``pooled_min_cluster_size`` is set (after aggregating grid curves across samples),
    ``best_size`` / ``best_size_std`` report that single consensus hyperparameter (std is 0)
    and ``dbcv`` is the mean (and std) of each sample's DBCV **at that grid point**.
    ``n_clusters_emergent`` remains the mean of per-sample fits at each sample's grid-optimal
    size; use :func:`n_clusters_at_min_cluster_size` on pooled grid CSV rows for a fixed MCS.

    Otherwise (independent runs or legacy callers) ``best_size`` is the mean of per-report
    chosen sizes and ``dbcv`` is the mean of each report's ``best_score``.

    Raises:
        ValueError: if ``reports`` is empty.
    """
    if not reports:
        raise ValueError("reports must be non-empty")

    sizes = np.array(
        [r.hyperparameters.min_cluster_size for r in reports], dtype=np.float64
    )
    scores = np.array([r.best_score for r in reports], dtype=np.float64)
    nprops = np.array([r.number_properties for r in reports], dtype=np.float64)
    n_clusters = np.array([r.n_clusters_emergent for r in reports], dtype=np.float64)
    ari_vals = [float(r.ari) for r in reports if r.ari is not None]

    n = len(reports)
    std_nprops = float(np.std(nprops)) if n > 1 else 0.0
    std_n_clusters = float(np.std(n_clusters)) if n > 1 else 0.0

    if pooled_min_cluster_size is not None:
        sizes_mean = float(pooled_min_cluster_size)
        std_sizes = 0.0
        dbcv_at: list[float] = []
        for r in reports:
            m = r.metrics_df
            hit = m.loc[m["min_cluster_size"] == pooled_min_cluster_size, "dbcv"]
            if len(hit) > 0:
                dbcv_at.append(float(hit.iloc[0]))
        if dbcv_at:
            arr_dbcv = np.array(dbcv_at, dtype=np.float64)
            dbcv_mean = float(np.mean(arr_dbcv))
            dbcv_std = float(np.std(arr_dbcv)) if len(arr_dbcv) > 1 else 0.0
        else:
            dbcv_mean = float(np.mean(scores))
            dbcv_std = float(np.std(scores)) if n > 1 else 0.0
    else:
        sizes_mean = float(np.mean(sizes))
        std_sizes = float(np.std(sizes)) if n > 1 else 0.0
        dbcv_mean = float(np.mean(scores))
        dbcv_std = float(np.std(scores)) if n > 1 else 0.0

    ari_block: MeanWithUncertainty | None
    if ari_vals:
        arr = np.array(ari_vals, dtype=np.float64)
        ari_block = MeanWithUncertainty(
            mean=float(np.mean(arr)),
            std=float(np.std(arr)) if len(arr) > 1 else 0.0,
        )
    else:
        ari_block = None

    acv_reports = [r.all_screener_cv for r in reports if r.all_screener_cv is not None]
    pooled_acv = _pool_all_screener_cv_results(acv_reports) if acv_reports else None

    rows_realized = [
        float(r.n_rows_realized) for r in reports if r.n_rows_realized is not None
    ]
    n_rows_block: MeanWithUncertainty | None = None
    if rows_realized:
        arr_rows = np.array(rows_realized, dtype=np.float64)
        n_rows_block = MeanWithUncertainty(
            mean=float(np.mean(arr_rows)),
            std=float(np.std(arr_rows)) if len(arr_rows) > 1 else 0.0,
        )

    return ClusteringSearchSummaryRow(
        model=model,
        layer=layer,
        hyperparameters=HyperparameterSearchStats(
            min_cluster_size=MeanWithUncertainty(
                mean=sizes_mean,
                std=std_sizes,
            ),
        ),
        number_properties=MeanWithUncertainty(
            mean=float(np.mean(nprops)),
            std=std_nprops,
        ),
        n_clusters_emergent=MeanWithUncertainty(
            mean=float(np.mean(n_clusters)),
            std=std_n_clusters,
        ),
        dbcv=MeanWithUncertainty(
            mean=dbcv_mean,
            std=dbcv_std,
        ),
        ari=ari_block,
        all_screener_cv=pooled_acv,
        n_rows_realized=n_rows_block,
    )

write_cluster_composition_json(path, composition_df, *, top_n=3, weighting='inv_sqrt_mention_count', exclude_noise=False, summary=None, max_clusters_in_rows=None, indent=2)

Serialize a processed cluster-composition table written at fit time.

Source code in pelinker/reporting.py

def write_cluster_composition_json(
    path: str | pathlib.Path,
    composition_df: pd.DataFrame,
    *,
    top_n: int = 3,
    weighting: str = "inv_sqrt_mention_count",
    exclude_noise: bool = False,
    summary: dict[str, Any] | None = None,
    max_clusters_in_rows: int | None = None,
    indent: int = 2,
) -> None:
    """Serialize a processed cluster-composition table written at fit time."""
    from pelinker.cluster_composition_viz import ENTITY_WEIGHTING_INV_SQRT

    p = pathlib.Path(path).expanduser()
    p.parent.mkdir(parents=True, exist_ok=True)
    payload: dict[str, Any] = {
        "schema": _FIT_CLUSTER_COMPOSITION_SCHEMA,
        "top_n": int(top_n),
        "weighting": weighting or ENTITY_WEIGHTING_INV_SQRT,
        "exclude_noise": bool(exclude_noise),
        "rows": _dataframe_to_jsonable_records(composition_df),
    }
    if summary is not None:
        payload["summary"] = _json_normalize(summary)
    if max_clusters_in_rows is not None:
        payload["max_clusters_in_rows"] = int(max_clusters_in_rows)
    with gzip.open(p, mode="wt", encoding="utf-8", compresslevel=9) as f:
        json.dump(payload, f, indent=indent)

write_clustering_report_json(path, report, *, indent=2)

Serialize report with :func:clustering_report_to_jsonable_dict to UTF-8 JSON.

Parent directories are created when missing.

Source code in pelinker/reporting.py

def write_clustering_report_json(
    path: str | pathlib.Path, report: ModelSelectionReport, *, indent: int = 2
) -> None:
    """
    Serialize ``report`` with :func:`clustering_report_to_jsonable_dict` to UTF-8 JSON.

    Parent directories are created when missing.
    """
    p = pathlib.Path(path).expanduser()
    p.parent.mkdir(parents=True, exist_ok=True)
    payload = clustering_report_to_jsonable_dict(report)

    with gzip.open(p, mode="wt", encoding="utf-8", compresslevel=9) as f:
        json.dump(payload, f, indent=indent)

write_kb_out_json(path, payload, *, indent=2)

Write :func:~pelinker.kb_out.build_kb_out_catalog output.

Source code in pelinker/reporting.py

def write_kb_out_json(
    path: str | pathlib.Path,
    payload: dict[str, Any],
    *,
    indent: int = 2,
) -> None:
    """Write :func:`~pelinker.kb_out.build_kb_out_catalog` output."""
    p = pathlib.Path(path).expanduser()
    p.parent.mkdir(parents=True, exist_ok=True)
    if str(payload.get("schema", "")) != _KB_OUT_SCHEMA:
        raise ValueError(
            f"Expected schema {_KB_OUT_SCHEMA!r}, got {payload.get('schema')!r}"
        )
    with p.open("w", encoding="utf-8") as f:
        json.dump(_json_normalize(payload), f, indent=indent, ensure_ascii=False)