Skip to content

spatial_graph_algorithms.compare

Helpers for comparative simulation, denoising, reconstruction, and benchmark-gym studies. Runners return a ComparisonResult with built-in summary, ranking, delta, plotting, and I/O helpers.

API Reference

spatial_graph_algorithms.compare.ComparisonResult dataclass

Results of a multi-method reconstruction comparison study.

Wraps the raw tidy DataFrame produced by :func:run_comparison and exposes convenience methods for summarising, ranking, and plotting without boilerplate pandas.

The raw DataFrame is always accessible via :attr:df.

Parameters:

Name Type Description Default
df DataFrame

One row per graph_spec × seed × reconstruction_spec.

 required

Examples:

>>> from spatial_graph_algorithms.compare import parameter_grid, run_comparison
>>> graphs = parameter_grid(cases=[{"n": 50, "mode": "knn", "k": 4}])
>>> recons = parameter_grid(cases=[{"method": "mds"}])
>>> result = run_comparison(graph_specs=graphs, reconstruction_specs=recons, seeds=[1])
>>> isinstance(result.df, pd.DataFrame)
True
Source code in src/spatial_graph_algorithms/compare/__init__.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
@dataclass
class ComparisonResult:
    """Results of a multi-method reconstruction comparison study.

    Wraps the raw tidy DataFrame produced by :func:`run_comparison` and
    exposes convenience methods for summarising, ranking, and plotting
    without boilerplate pandas.

    The raw DataFrame is always accessible via :attr:`df`.

    Parameters
    ----------
    df : pandas.DataFrame
        One row per ``graph_spec × seed × reconstruction_spec``.

    Examples
    --------
    >>> from spatial_graph_algorithms.compare import parameter_grid, run_comparison
    >>> graphs = parameter_grid(cases=[{"n": 50, "mode": "knn", "k": 4}])
    >>> recons = parameter_grid(cases=[{"method": "mds"}])
    >>> result = run_comparison(graph_specs=graphs, reconstruction_specs=recons, seeds=[1])
    >>> isinstance(result.df, pd.DataFrame)
    True
    """

    df: pd.DataFrame

    # ------------------------------------------------------------------
    # Analysis helpers
    # ------------------------------------------------------------------

    def summary(
        self,
        *,
        by: list[str] | None = None,
        metrics: list[str] | None = None,
    ) -> pd.DataFrame:
        """Return mean metrics grouped by method and graph condition.

        Only rows with ``status == "ok"`` are included.

        Parameters
        ----------
        by : list of str, optional
            Columns to group by.  Default is ``["graph_label", "method"]``.
        metrics : list of str, optional
            Metric columns to aggregate.  Default is ``["cpd", "knn"]``.

        Returns
        -------
        pandas.DataFrame
            Mean of each metric for each group.  Groups with no successful
            rows are absent.

        Examples
        --------
        >>> result.summary()  # doctest: +SKIP
                                         cpd    knn
        graph_label  method
        mode=knn__k=4  landmark_mds  0.8821  0.7341
                       mds           0.7512  0.6103
        """
        by = list(by) if by is not None else _DEFAULT_SUMMARY_BY
        metrics = list(metrics) if metrics is not None else _DEFAULT_METRICS
        ok = self.df[self.df["status"] == "ok"]
        present = [m for m in metrics if m in ok.columns]
        if not present:
            return pd.DataFrame()
        return ok.groupby(by)[present].mean().round(4)

    def best(
        self,
        *,
        metric: str = "cpd",
        by: list[str] | None = None,
        higher_is_better: bool = True,
    ) -> pd.DataFrame:
        """Return the best-performing method per group.

        Parameters
        ----------
        metric : str
            Metric column to rank by.  Default is ``"cpd"``.
        by : list of str, optional
            Grouping columns.  Default is ``["graph_label"]``.
        higher_is_better : bool
            If ``True`` (default), select the row with the highest metric
            value.  Set to ``False`` for error / loss metrics.

        Returns
        -------
        pandas.DataFrame
            One row per unique ``by`` group, showing the best method and its
            mean metric value (averaged over seeds).

        Raises
        ------
        ValueError
            If ``metric`` is not a column in the results DataFrame.

        Examples
        --------
        >>> result.best(metric="cpd")  # doctest: +SKIP
              graph_label        method     cpd
        0  mode=knn__k=4  landmark_mds  0.8821
        """
        by_cols = list(by) if by is not None else _DEFAULT_BEST_BY
        ok = self.df[self.df["status"] == "ok"]
        if metric not in ok.columns:
            available = [c for c in ok.columns if ok[c].dtype.kind == "f"]
            raise ValueError(
                f"Metric {metric!r} not found in results.  "
                f"Available numeric columns: {available}"
            )
        grp_cols = by_cols + ["method"]
        agg = ok.groupby(grp_cols)[metric].mean().reset_index()
        if higher_is_better:
            idx = agg.groupby(by_cols)[metric].idxmax()
        else:
            idx = agg.groupby(by_cols)[metric].idxmin()
        return agg.loc[idx.values].reset_index(drop=True)

    def delta(
        self,
        *,
        metrics: list[str] | None = None,
        baseline: str = "none",
        denoise_col: str = "denoise_label",
        by: list[str] | None = None,
    ) -> pd.DataFrame:
        """Return metric deltas relative to a baseline denoising condition.

        Parameters
        ----------
        metrics : list of str, optional
            Metric columns to compare. Default is ``["cpd", "knn"]``.
        baseline : str
            Baseline value in *denoise_col*. Default is ``"none"``.
        denoise_col : str
            Column identifying the denoising condition. Default is
            ``"denoise_label"``.
        by : list of str, optional
            Columns that identify matched rows. Default is
            ``["graph_label", "seed", "reconstruction_label"]`` when present.

        Returns
        -------
        pandas.DataFrame
            Successful non-baseline rows with ``delta_<metric>`` columns.

        Raises
        ------
        ValueError
            If required columns are missing.

        Examples
        --------
        >>> result.delta()  # doctest: +SKIP
        """
        metrics = list(metrics) if metrics is not None else ["cpd", "knn"]
        if denoise_col not in self.df.columns:
            raise ValueError(f"Column {denoise_col!r} not found in results.")
        default_by = ["graph_label", "graph_id", "seed", "reconstruction_label"]
        by_cols = list(by) if by is not None else [c for c in default_by if c in self.df.columns]
        missing = [c for c in by_cols + metrics if c not in self.df.columns]
        if missing:
            raise ValueError(f"Required columns missing from results: {missing}")

        ok = self.df[self.df["status"] == "ok"].copy()
        base = ok[ok[denoise_col] == baseline][by_cols + metrics]
        if base.empty:
            return pd.DataFrame()
        base = base.rename(columns={metric: f"{metric}_baseline" for metric in metrics})
        merged = ok[ok[denoise_col] != baseline].merge(base, on=by_cols, how="inner")
        for metric in metrics:
            merged[f"delta_{metric}"] = merged[metric] - merged[f"{metric}_baseline"]
        return merged

    # ------------------------------------------------------------------
    # Plotting
    # ------------------------------------------------------------------

    def plot(
        self,
        *,
        metric: str = "cpd",
        by: str = "method",
        hue: str | None = "graph_label",
        ax: plt.Axes | None = None,
    ) -> plt.Figure:
        """Bar chart of a quality metric grouped by method and condition.

        Means are computed over all successful rows (``status == "ok"``).
        Error bars show one standard deviation.

        Parameters
        ----------
        metric : str
            Metric column to plot.  Default is ``"cpd"``.
        by : str
            Column that defines the x-axis categories.  Default is
            ``"method"``.
        hue : str, optional
            Column that defines the colour grouping.  Default is
            ``"graph_label"``.  Pass ``None`` for a single-colour chart.
        ax : matplotlib.axes.Axes, optional
            Axes to draw on.  A new figure is created when omitted.

        Returns
        -------
        matplotlib.figure.Figure

        Raises
        ------
        ValueError
            If ``metric`` is not a column in the results DataFrame.

        Examples
        --------
        >>> fig = result.plot(metric="cpd", by="method")  # doctest: +SKIP
        """
        import matplotlib.pyplot as plt

        ok = self.df[self.df["status"] == "ok"]
        if metric not in ok.columns:
            raise ValueError(f"Metric {metric!r} not found in results.")

        group_cols = ([hue, by] if hue is not None and hue in ok.columns else [by])
        effective_hue = hue if hue in group_cols else None

        agg = ok.groupby(group_cols)[metric].agg(["mean", "std"]).reset_index()

        if ax is None:
            fig, ax = plt.subplots(figsize=(max(5, len(agg) * 0.9), 4))
        else:
            fig = ax.get_figure()

        by_vals = sorted(agg[by].unique())
        colors = plt.cm.tab10.colors  # type: ignore[attr-defined]

        if effective_hue is not None:
            hue_vals = sorted(agg[effective_hue].unique())
            n_hue = len(hue_vals)
            width = 0.75 / n_hue
            for i, hue_val in enumerate(hue_vals):
                sub = agg[agg[effective_hue] == hue_val]
                x_map = {v: j for j, v in enumerate(by_vals)}
                x_pos = [x_map[v] + (i - n_hue / 2 + 0.5) * width for v in sub[by]]
                ax.bar(
                    x_pos,
                    sub["mean"],
                    width=width,
                    label=str(hue_val),
                    yerr=sub["std"].fillna(0),
                    capsize=3,
                    color=colors[i % len(colors)],
                    alpha=0.85,
                )
            ax.legend(title=effective_hue, bbox_to_anchor=(1.01, 1), loc="upper left")
        else:
            means = [agg.loc[agg[by] == v, "mean"].mean() for v in by_vals]
            stds = [agg.loc[agg[by] == v, "std"].mean() for v in by_vals]
            ax.bar(
                range(len(by_vals)),
                means,
                yerr=[s if pd.notna(s) else 0 for s in stds],
                capsize=3,
                color=colors[0],
                alpha=0.85,
            )

        ax.set_xticks(range(len(by_vals)))
        ax.set_xticklabels(by_vals, rotation=15, ha="right")
        ax.set_xlabel(by)
        ax.set_ylabel(metric)
        title = f"{metric} by {by}"
        if effective_hue:
            title += f", coloured by {effective_hue}"
        ax.set_title(title)
        fig.tight_layout()
        return fig

    # ------------------------------------------------------------------
    # I/O
    # ------------------------------------------------------------------

    def save(self, path: str | Path) -> None:
        """Save results to a CSV or Parquet file.

        The format is inferred from the file extension (``.parquet`` → Parquet,
        anything else → CSV).

        Parameters
        ----------
        path : str or Path
            Destination path.  Parent directories are created automatically.

        Examples
        --------
        >>> result.save("results/comparison.csv")  # doctest: +SKIP
        """
        p = Path(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        if p.suffix == ".parquet":
            self.df.to_parquet(p, index=False)
        else:
            self.df.to_csv(p, index=False)

    @classmethod
    def load(cls, path: str | Path) -> ComparisonResult:
        """Load results previously saved with :meth:`save`.

        Parameters
        ----------
        path : str or Path
            Path to a CSV or Parquet file created by :meth:`save`.

        Returns
        -------
        ComparisonResult

        Examples
        --------
        >>> result = ComparisonResult.load("results/comparison.csv")  # doctest: +SKIP
        """
        p = Path(path)
        if p.suffix == ".parquet":
            return cls(df=pd.read_parquet(p))
        return cls(df=pd.read_csv(p))

    # ------------------------------------------------------------------
    # Display
    # ------------------------------------------------------------------

    def __repr__(self) -> str:
        n_total = len(self.df)
        n_ok = int((self.df["status"] == "ok").sum()) if "status" in self.df.columns else n_total
        n_err = n_total - n_ok
        metrics = [m for m in _DEFAULT_METRICS if m in self.df.columns]
        return (
            f"ComparisonResult(rows={n_total}, ok={n_ok}, errors={n_err}, "
            f"metrics={metrics})"
        )

    def _repr_html_(self) -> str:
        return self.df._repr_html_()  # type: ignore[return-value]

Methods:

summary(*, by=None, metrics=None)

Return mean metrics grouped by method and graph condition.

Only rows with status == "ok" are included.

Parameters:

Name Type Description Default
by list of str

Columns to group by. Default is ["graph_label", "method"].

 None
metrics list of str

Metric columns to aggregate. Default is ["cpd", "knn"].

 None

Returns:

Type Description
DataFrame

Mean of each metric for each group. Groups with no successful rows are absent.

Examples:

>>> result.summary()
                                 cpd    knn
graph_label  method
mode=knn__k=4  landmark_mds  0.8821  0.7341
               mds           0.7512  0.6103
Source code in src/spatial_graph_algorithms/compare/__init__.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def summary(
    self,
    *,
    by: list[str] | None = None,
    metrics: list[str] | None = None,
) -> pd.DataFrame:
    """Return mean metrics grouped by method and graph condition.

    Only rows with ``status == "ok"`` are included.

    Parameters
    ----------
    by : list of str, optional
        Columns to group by.  Default is ``["graph_label", "method"]``.
    metrics : list of str, optional
        Metric columns to aggregate.  Default is ``["cpd", "knn"]``.

    Returns
    -------
    pandas.DataFrame
        Mean of each metric for each group.  Groups with no successful
        rows are absent.

    Examples
    --------
    >>> result.summary()  # doctest: +SKIP
                                     cpd    knn
    graph_label  method
    mode=knn__k=4  landmark_mds  0.8821  0.7341
                   mds           0.7512  0.6103
    """
    by = list(by) if by is not None else _DEFAULT_SUMMARY_BY
    metrics = list(metrics) if metrics is not None else _DEFAULT_METRICS
    ok = self.df[self.df["status"] == "ok"]
    present = [m for m in metrics if m in ok.columns]
    if not present:
        return pd.DataFrame()
    return ok.groupby(by)[present].mean().round(4)

best(*, metric='cpd', by=None, higher_is_better=True)

Return the best-performing method per group.

Parameters:

Name Type Description Default
metric str

Metric column to rank by. Default is "cpd".

 'cpd'
by list of str

Grouping columns. Default is ["graph_label"].

 None
higher_is_better bool

If True (default), select the row with the highest metric value. Set to False for error / loss metrics.

 True

Returns:

Type Description
DataFrame

One row per unique by group, showing the best method and its mean metric value (averaged over seeds).

Raises:

Type Description
ValueError

If metric is not a column in the results DataFrame.

Examples:

>>> result.best(metric="cpd")
      graph_label        method     cpd
0  mode=knn__k=4  landmark_mds  0.8821
Source code in src/spatial_graph_algorithms/compare/__init__.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def best(
    self,
    *,
    metric: str = "cpd",
    by: list[str] | None = None,
    higher_is_better: bool = True,
) -> pd.DataFrame:
    """Return the best-performing method per group.

    Parameters
    ----------
    metric : str
        Metric column to rank by.  Default is ``"cpd"``.
    by : list of str, optional
        Grouping columns.  Default is ``["graph_label"]``.
    higher_is_better : bool
        If ``True`` (default), select the row with the highest metric
        value.  Set to ``False`` for error / loss metrics.

    Returns
    -------
    pandas.DataFrame
        One row per unique ``by`` group, showing the best method and its
        mean metric value (averaged over seeds).

    Raises
    ------
    ValueError
        If ``metric`` is not a column in the results DataFrame.

    Examples
    --------
    >>> result.best(metric="cpd")  # doctest: +SKIP
          graph_label        method     cpd
    0  mode=knn__k=4  landmark_mds  0.8821
    """
    by_cols = list(by) if by is not None else _DEFAULT_BEST_BY
    ok = self.df[self.df["status"] == "ok"]
    if metric not in ok.columns:
        available = [c for c in ok.columns if ok[c].dtype.kind == "f"]
        raise ValueError(
            f"Metric {metric!r} not found in results.  "
            f"Available numeric columns: {available}"
        )
    grp_cols = by_cols + ["method"]
    agg = ok.groupby(grp_cols)[metric].mean().reset_index()
    if higher_is_better:
        idx = agg.groupby(by_cols)[metric].idxmax()
    else:
        idx = agg.groupby(by_cols)[metric].idxmin()
    return agg.loc[idx.values].reset_index(drop=True)

delta(*, metrics=None, baseline='none', denoise_col='denoise_label', by=None)

Return metric deltas relative to a baseline denoising condition.

Parameters:

Name Type Description Default
metrics list of str

Metric columns to compare. Default is ["cpd", "knn"].

 None
baseline str

Baseline value in denoise_col. Default is "none".

 'none'
denoise_col str

Column identifying the denoising condition. Default is "denoise_label".

 'denoise_label'
by list of str

Columns that identify matched rows. Default is ["graph_label", "seed", "reconstruction_label"] when present.

 None

Returns:

Type Description
DataFrame

Successful non-baseline rows with delta_<metric> columns.

Raises:

Type Description
ValueError

If required columns are missing.

Examples:

>>> result.delta()
Source code in src/spatial_graph_algorithms/compare/__init__.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def delta(
    self,
    *,
    metrics: list[str] | None = None,
    baseline: str = "none",
    denoise_col: str = "denoise_label",
    by: list[str] | None = None,
) -> pd.DataFrame:
    """Return metric deltas relative to a baseline denoising condition.

    Parameters
    ----------
    metrics : list of str, optional
        Metric columns to compare. Default is ``["cpd", "knn"]``.
    baseline : str
        Baseline value in *denoise_col*. Default is ``"none"``.
    denoise_col : str
        Column identifying the denoising condition. Default is
        ``"denoise_label"``.
    by : list of str, optional
        Columns that identify matched rows. Default is
        ``["graph_label", "seed", "reconstruction_label"]`` when present.

    Returns
    -------
    pandas.DataFrame
        Successful non-baseline rows with ``delta_<metric>`` columns.

    Raises
    ------
    ValueError
        If required columns are missing.

    Examples
    --------
    >>> result.delta()  # doctest: +SKIP
    """
    metrics = list(metrics) if metrics is not None else ["cpd", "knn"]
    if denoise_col not in self.df.columns:
        raise ValueError(f"Column {denoise_col!r} not found in results.")
    default_by = ["graph_label", "graph_id", "seed", "reconstruction_label"]
    by_cols = list(by) if by is not None else [c for c in default_by if c in self.df.columns]
    missing = [c for c in by_cols + metrics if c not in self.df.columns]
    if missing:
        raise ValueError(f"Required columns missing from results: {missing}")

    ok = self.df[self.df["status"] == "ok"].copy()
    base = ok[ok[denoise_col] == baseline][by_cols + metrics]
    if base.empty:
        return pd.DataFrame()
    base = base.rename(columns={metric: f"{metric}_baseline" for metric in metrics})
    merged = ok[ok[denoise_col] != baseline].merge(base, on=by_cols, how="inner")
    for metric in metrics:
        merged[f"delta_{metric}"] = merged[metric] - merged[f"{metric}_baseline"]
    return merged

plot(*, metric='cpd', by='method', hue='graph_label', ax=None)

Bar chart of a quality metric grouped by method and condition.

Means are computed over all successful rows (status == "ok"). Error bars show one standard deviation.

Parameters:

Name Type Description Default
metric str

Metric column to plot. Default is "cpd".

 'cpd'
by str

Column that defines the x-axis categories. Default is "method".

 'method'
hue str

Column that defines the colour grouping. Default is "graph_label". Pass None for a single-colour chart.

 'graph_label'
ax Axes

Axes to draw on. A new figure is created when omitted.

 None

Returns:

Type Description
Figure

Raises:

Type Description
ValueError

If metric is not a column in the results DataFrame.

Examples:

>>> fig = result.plot(metric="cpd", by="method")
Source code in src/spatial_graph_algorithms/compare/__init__.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
def plot(
    self,
    *,
    metric: str = "cpd",
    by: str = "method",
    hue: str | None = "graph_label",
    ax: plt.Axes | None = None,
) -> plt.Figure:
    """Bar chart of a quality metric grouped by method and condition.

    Means are computed over all successful rows (``status == "ok"``).
    Error bars show one standard deviation.

    Parameters
    ----------
    metric : str
        Metric column to plot.  Default is ``"cpd"``.
    by : str
        Column that defines the x-axis categories.  Default is
        ``"method"``.
    hue : str, optional
        Column that defines the colour grouping.  Default is
        ``"graph_label"``.  Pass ``None`` for a single-colour chart.
    ax : matplotlib.axes.Axes, optional
        Axes to draw on.  A new figure is created when omitted.

    Returns
    -------
    matplotlib.figure.Figure

    Raises
    ------
    ValueError
        If ``metric`` is not a column in the results DataFrame.

    Examples
    --------
    >>> fig = result.plot(metric="cpd", by="method")  # doctest: +SKIP
    """
    import matplotlib.pyplot as plt

    ok = self.df[self.df["status"] == "ok"]
    if metric not in ok.columns:
        raise ValueError(f"Metric {metric!r} not found in results.")

    group_cols = ([hue, by] if hue is not None and hue in ok.columns else [by])
    effective_hue = hue if hue in group_cols else None

    agg = ok.groupby(group_cols)[metric].agg(["mean", "std"]).reset_index()

    if ax is None:
        fig, ax = plt.subplots(figsize=(max(5, len(agg) * 0.9), 4))
    else:
        fig = ax.get_figure()

    by_vals = sorted(agg[by].unique())
    colors = plt.cm.tab10.colors  # type: ignore[attr-defined]

    if effective_hue is not None:
        hue_vals = sorted(agg[effective_hue].unique())
        n_hue = len(hue_vals)
        width = 0.75 / n_hue
        for i, hue_val in enumerate(hue_vals):
            sub = agg[agg[effective_hue] == hue_val]
            x_map = {v: j for j, v in enumerate(by_vals)}
            x_pos = [x_map[v] + (i - n_hue / 2 + 0.5) * width for v in sub[by]]
            ax.bar(
                x_pos,
                sub["mean"],
                width=width,
                label=str(hue_val),
                yerr=sub["std"].fillna(0),
                capsize=3,
                color=colors[i % len(colors)],
                alpha=0.85,
            )
        ax.legend(title=effective_hue, bbox_to_anchor=(1.01, 1), loc="upper left")
    else:
        means = [agg.loc[agg[by] == v, "mean"].mean() for v in by_vals]
        stds = [agg.loc[agg[by] == v, "std"].mean() for v in by_vals]
        ax.bar(
            range(len(by_vals)),
            means,
            yerr=[s if pd.notna(s) else 0 for s in stds],
            capsize=3,
            color=colors[0],
            alpha=0.85,
        )

    ax.set_xticks(range(len(by_vals)))
    ax.set_xticklabels(by_vals, rotation=15, ha="right")
    ax.set_xlabel(by)
    ax.set_ylabel(metric)
    title = f"{metric} by {by}"
    if effective_hue:
        title += f", coloured by {effective_hue}"
    ax.set_title(title)
    fig.tight_layout()
    return fig

save(path)

Save results to a CSV or Parquet file.

The format is inferred from the file extension (.parquet → Parquet, anything else → CSV).

Parameters:

Name Type Description Default
path str or Path

Destination path. Parent directories are created automatically.

 required

Examples:

>>> result.save("results/comparison.csv")
Source code in src/spatial_graph_algorithms/compare/__init__.py 
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
def save(self, path: str | Path) -> None:
    """Save results to a CSV or Parquet file.

    The format is inferred from the file extension (``.parquet`` → Parquet,
    anything else → CSV).

    Parameters
    ----------
    path : str or Path
        Destination path.  Parent directories are created automatically.

    Examples
    --------
    >>> result.save("results/comparison.csv")  # doctest: +SKIP
    """
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    if p.suffix == ".parquet":
        self.df.to_parquet(p, index=False)
    else:
        self.df.to_csv(p, index=False)

load(path) classmethod

Load results previously saved with :meth:save.

Parameters:

Name Type Description Default
path str or Path

Path to a CSV or Parquet file created by :meth:save.

 required

Returns:

Type Description
ComparisonResult

Examples:

>>> result = ComparisonResult.load("results/comparison.csv")
Source code in src/spatial_graph_algorithms/compare/__init__.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
@classmethod
def load(cls, path: str | Path) -> ComparisonResult:
    """Load results previously saved with :meth:`save`.

    Parameters
    ----------
    path : str or Path
        Path to a CSV or Parquet file created by :meth:`save`.

    Returns
    -------
    ComparisonResult

    Examples
    --------
    >>> result = ComparisonResult.load("results/comparison.csv")  # doctest: +SKIP
    """
    p = Path(path)
    if p.suffix == ".parquet":
        return cls(df=pd.read_parquet(p))
    return cls(df=pd.read_csv(p))

spatial_graph_algorithms.compare.parameter_grid(*, base=None, vary=None, cases=None, groups=None, where=None, label_keys=None, drop_none=False)

Build a list of parameter dictionaries for comparison studies.

Parameters:

Name Type Description Default
base mapping

Parameters shared by every cartesian product case.

 None
vary mapping

Parameter values to expand using :func:itertools.product.

 None
cases iterable of mapping

Explicit hand-picked cases. Useful for non-cartesian comparisons.

 None
groups iterable of mapping

Multiple grid definitions. Each group can contain base, vary, cases, where, label_keys, and drop_none. Groups avoid creating invalid cartesian products for unrelated parameters.

 None
where callable

Predicate used to keep or discard expanded specs.

 None
label_keys iterable of str

Keys used to auto-generate "_label". Defaults to all public keys.

 None
drop_none bool

If True, remove keys with value None after filtering.

 False

Returns:

Type Description
list of dict

Parameter specs. Each spec has a readable "_label" unless one was provided explicitly.

Examples:

>>> from spatial_graph_algorithms.compare import parameter_grid
>>> parameter_grid(base={"n": 100}, vary={"mode": ["knn"], "k": [4, 8]})
[{'_label': 'mode=knn__k=4', 'n': 100, 'mode': 'knn', 'k': 4}, ...]
Source code in src/spatial_graph_algorithms/compare/__init__.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
def parameter_grid(
    *,
    base: Mapping[str, Any] | None = None,
    vary: Mapping[str, Iterable[Any]] | None = None,
    cases: Iterable[Mapping[str, Any]] | None = None,
    groups: Iterable[GridGroup] | None = None,
    where: Callable[[Spec], bool] | None = None,
    label_keys: Iterable[str] | None = None,
    drop_none: bool = False,
) -> list[Spec]:
    """Build a list of parameter dictionaries for comparison studies.

    Parameters
    ----------
    base : mapping, optional
        Parameters shared by every cartesian product case.
    vary : mapping, optional
        Parameter values to expand using :func:`itertools.product`.
    cases : iterable of mapping, optional
        Explicit hand-picked cases.  Useful for non-cartesian comparisons.
    groups : iterable of mapping, optional
        Multiple grid definitions.  Each group can contain ``base``, ``vary``,
        ``cases``, ``where``, ``label_keys``, and ``drop_none``.  Groups avoid
        creating invalid cartesian products for unrelated parameters.
    where : callable, optional
        Predicate used to keep or discard expanded specs.
    label_keys : iterable of str, optional
        Keys used to auto-generate ``"_label"``.  Defaults to all public keys.
    drop_none : bool
        If ``True``, remove keys with value ``None`` after filtering.

    Returns
    -------
    list of dict
        Parameter specs.  Each spec has a readable ``"_label"`` unless one was
        provided explicitly.

    Examples
    --------
    >>> from spatial_graph_algorithms.compare import parameter_grid
    >>> parameter_grid(base={"n": 100}, vary={"mode": ["knn"], "k": [4, 8]})
    [{'_label': 'mode=knn__k=4', 'n': 100, 'mode': 'knn', 'k': 4}, ...]
    """
    specs: list[Spec] = []

    if groups is not None:
        for group in groups:
            group_specs = parameter_grid(
                base=group.get("base"),
                vary=group.get("vary"),
                cases=group.get("cases"),
                where=group.get("where"),
                label_keys=group.get("label_keys", label_keys),
                drop_none=bool(group.get("drop_none", drop_none)),
            )
            specs.extend(group_specs)

    specs.extend(_expand_single_grid(base=base, vary=vary, cases=cases))

    filtered: list[Spec] = []
    for spec in specs:
        if where is not None and not where(dict(spec)):
            continue
        clean = {k: v for k, v in spec.items() if not (drop_none and v is None)}
        _ensure_label(clean, label_keys=label_keys)
        filtered.append(clean)
    return filtered

spatial_graph_algorithms.compare.dry_run_comparison(*, graph_specs, reconstruction_specs, seeds)

Preview comparison combinations without generating graphs.

Parameters:

Name Type Description Default
graph_specs iterable of mapping

Graph-generation specs, usually returned by :func:parameter_grid.

 required
reconstruction_specs iterable of mapping

Reconstruction specs, usually returned by :func:parameter_grid.

 required
seeds iterable of int

Top-level seeds to combine with each graph and reconstruction spec.

 required

Returns:

Type Description
DataFrame

One row per planned graph_spec × seed × reconstruction_spec with labels and parameter columns. No simulation, reconstruction, or metric computation is performed.
Source code in src/spatial_graph_algorithms/compare/__init__.py 
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
def dry_run_comparison(
    *,
    graph_specs: Iterable[Mapping[str, Any]],
    reconstruction_specs: Iterable[Mapping[str, Any]],
    seeds: Iterable[int],
) -> pd.DataFrame:
    """Preview comparison combinations without generating graphs.

    Parameters
    ----------
    graph_specs : iterable of mapping
        Graph-generation specs, usually returned by :func:`parameter_grid`.
    reconstruction_specs : iterable of mapping
        Reconstruction specs, usually returned by :func:`parameter_grid`.
    seeds : iterable of int
        Top-level seeds to combine with each graph and reconstruction spec.

    Returns
    -------
    pandas.DataFrame
        One row per planned ``graph_spec × seed × reconstruction_spec`` with
        labels and parameter columns.  No simulation, reconstruction, or metric
        computation is performed.
    """
    graph_specs_list = [_with_label(dict(spec)) for spec in graph_specs]
    reconstruction_specs_list = [_with_label(dict(spec)) for spec in reconstruction_specs]
    seeds_list = [int(seed) for seed in seeds]

    rows = [
        _base_row(
            graph_spec=graph_spec,
            recon_spec=recon_spec,
            seed=seed,
            status="planned",
            error=None,
            generation_seconds=None,
        )
        for graph_spec in graph_specs_list
        for seed in seeds_list
        for recon_spec in reconstruction_specs_list
    ]
    return pd.DataFrame(rows)

spatial_graph_algorithms.compare.run_comparison(*, graph_specs, reconstruction_specs, seeds, dim=None, k_neighbors=15, compute_distortion=False, verbose=True)

Run a simulation/reconstruction comparison and return a :class:ComparisonResult.

Each graph spec is generated once per seed. All reconstruction specs are then applied to that graph, so generation cost is not repeated.

Parameters:

Name Type Description Default
graph_specs iterable of mapping

Specs passed to :func:spatial_graph_algorithms.simulate.generate. Private keys starting with "_" are treated as metadata.

 required
reconstruction_specs iterable of mapping

Specs passed to :func:spatial_graph_algorithms.reconstruct.reconstruct. Each spec must include "method". Other public keys are passed as method-specific keyword arguments.

 required
seeds iterable of int

Top-level seeds. Each graph spec is generated once per seed, then all reconstruction specs are run against that graph.

 required
dim int

Reconstruction dimensionality. Defaults to the graph spec dim if present, otherwise 2.

 None
k_neighbors int

Number of neighbours for reconstruction quality KNN evaluation.

 15
compute_distortion bool

Whether to compute the O(n²) distortion metric.

 False
verbose bool

If True (default), print one progress line per completed row.

 True

Returns:

Type Description
ComparisonResult

One row per graph_spec × seed × reconstruction_spec.

Examples:

>>> from spatial_graph_algorithms.compare import parameter_grid, run_comparison
>>> graphs = parameter_grid(cases=[{"n": 50, "mode": "knn", "k": 4}])
>>> recons = parameter_grid(cases=[{"method": "mds"}])
>>> result = run_comparison(
...     graph_specs=graphs, reconstruction_specs=recons, seeds=[1], verbose=False
... )
>>> result.df["status"].iloc[0]
'ok'
Source code in src/spatial_graph_algorithms/compare/__init__.py 
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
def run_comparison(
    *,
    graph_specs: Iterable[Mapping[str, Any]],
    reconstruction_specs: Iterable[Mapping[str, Any]],
    seeds: Iterable[int],
    dim: int | None = None,
    k_neighbors: int = 15,
    compute_distortion: bool = False,
    verbose: bool = True,
) -> ComparisonResult:
    """Run a simulation/reconstruction comparison and return a :class:`ComparisonResult`.

    Each graph spec is generated once per seed.  All reconstruction specs
    are then applied to that graph, so generation cost is not repeated.

    Parameters
    ----------
    graph_specs : iterable of mapping
        Specs passed to :func:`spatial_graph_algorithms.simulate.generate`.
        Private keys starting with ``"_"`` are treated as metadata.
    reconstruction_specs : iterable of mapping
        Specs passed to :func:`spatial_graph_algorithms.reconstruct.reconstruct`.
        Each spec must include ``"method"``.  Other public keys are passed as
        method-specific keyword arguments.
    seeds : iterable of int
        Top-level seeds.  Each graph spec is generated once per seed, then all
        reconstruction specs are run against that graph.
    dim : int, optional
        Reconstruction dimensionality.  Defaults to the graph spec ``dim`` if
        present, otherwise ``2``.
    k_neighbors : int
        Number of neighbours for reconstruction quality KNN evaluation.
    compute_distortion : bool
        Whether to compute the O(n²) distortion metric.
    verbose : bool
        If ``True`` (default), print one progress line per completed row.

    Returns
    -------
    ComparisonResult
        One row per ``graph_spec × seed × reconstruction_spec``.

    Examples
    --------
    >>> from spatial_graph_algorithms.compare import parameter_grid, run_comparison
    >>> graphs = parameter_grid(cases=[{"n": 50, "mode": "knn", "k": 4}])
    >>> recons = parameter_grid(cases=[{"method": "mds"}])
    >>> result = run_comparison(
    ...     graph_specs=graphs, reconstruction_specs=recons, seeds=[1], verbose=False
    ... )
    >>> result.df["status"].iloc[0]
    'ok'
    """
    graph_specs_list = [_with_label(dict(spec)) for spec in graph_specs]
    reconstruction_specs_list = [_with_label(dict(spec)) for spec in reconstruction_specs]
    seeds_list = [int(seed) for seed in seeds]
    rows: list[Spec] = []

    total = len(graph_specs_list) * len(seeds_list) * len(reconstruction_specs_list)
    completed = 0

    for graph_spec in graph_specs_list:
        graph_params = _public_params(graph_spec)
        graph_params.pop("seed", None)
        for seed in seeds_list:
            graph_start = time.perf_counter()
            try:
                sg = generate(**graph_params, seed=seed)
                generation_seconds = time.perf_counter() - graph_start
                graph_metrics = evaluate(
                    sg,
                    k_neighbors=k_neighbors,
                    compute_distortion=compute_distortion,
                )
            except Exception as exc:  # noqa: BLE001 - experiments should continue.
                generation_seconds = time.perf_counter() - graph_start
                for recon_spec in reconstruction_specs_list:
                    completed += 1
                    row = _base_row(
                        graph_spec=graph_spec,
                        recon_spec=recon_spec,
                        seed=seed,
                        status="generation_error",
                        error=f"{type(exc).__name__}: {exc}",
                        generation_seconds=generation_seconds,
                    )
                    rows.append(row)
                    if verbose:
                        _print_row(completed, total, graph_spec, recon_spec, seed, row)
                continue

            for recon_spec in reconstruction_specs_list:
                row = _base_row(
                    graph_spec=graph_spec,
                    recon_spec=recon_spec,
                    seed=seed,
                    status="ok",
                    error=None,
                    generation_seconds=generation_seconds,
                )
                row.update(graph_metrics)

                recon_params = _public_params(recon_spec)
                method = recon_params.pop("method", None)
                recon_params.pop("seed", None)
                if method is None:
                    row.update(
                        {
                            "status": "reconstruction_error",
                            "error": "ValueError: reconstruction spec must include 'method'",
                            "reconstruction_seconds": None,
                        }
                    )
                    completed += 1
                    rows.append(row)
                    if verbose:
                        _print_row(completed, total, graph_spec, recon_spec, seed, row)
                    continue

                recon_dim = dim if dim is not None else int(graph_params.get("dim", 2))
                recon_start = time.perf_counter()
                try:
                    sg_rec = reconstruct(
                        sg,
                        method=str(method),
                        dim=recon_dim,
                        seed=seed,
                        **recon_params,
                    )
                    row["reconstruction_seconds"] = time.perf_counter() - recon_start
                    row.update(
                        evaluate(
                            sg_rec,
                            k_neighbors=k_neighbors,
                            compute_distortion=compute_distortion,
                        )
                    )
                except ImportError as exc:
                    row.update(
                        {
                            "status": "method_unavailable",
                            "error": f"{type(exc).__name__}: {exc}",
                            "reconstruction_seconds": time.perf_counter() - recon_start,
                            "cpd": None,
                            "knn": None,
                            "distortion": None,
                        }
                    )
                except Exception as exc:  # noqa: BLE001 - experiments should continue.
                    row.update(
                        {
                            "status": "reconstruction_error",
                            "error": f"{type(exc).__name__}: {exc}",
                            "reconstruction_seconds": time.perf_counter() - recon_start,
                            "cpd": None,
                            "knn": None,
                            "distortion": None,
                        }
                    )
                completed += 1
                rows.append(row)
                if verbose:
                    _print_row(completed, total, graph_spec, recon_spec, seed, row)

    return ComparisonResult(df=pd.DataFrame(rows))

spatial_graph_algorithms.compare.run_denoise_comparison(*, graph_specs, denoise_specs, seeds, k_neighbors=15, compute_distortion=False, verbose=True)

Run a simulated denoising benchmark and return tidy quality metrics.

Each graph spec is generated once per seed, then every denoising spec is scored and filtered on that graph. Ground-truth false-edge labels from simulation are evaluated with :func:spatial_graph_algorithms.metrics.evaluate_denoising.

Parameters:

Name Type Description Default
graph_specs iterable of mapping

Specs passed to :func:spatial_graph_algorithms.simulate.generate.

 required
denoise_specs iterable of mapping

Denoising specs. Each spec must include "method" and may include "fraction_to_remove" plus scorer-specific keyword arguments.

 required
seeds iterable of int

Top-level seeds for graph generation and stochastic scorers.

 required
k_neighbors int

Number of neighbours used when graph-level metrics are computed.

 15
compute_distortion bool

Whether to compute O(n²) distortion for generated graphs.

 False
verbose bool

If True (default), print one progress line per completed row.

 True

Returns:

Type Description
ComparisonResult

One row per graph_spec × seed × denoise_spec.

Examples:

>>> graphs = parameter_grid(cases=[{"n": 50, "false_edge_fraction": 0.1}])
>>> denoisers = parameter_grid(cases=[{"method": "jaccard", "fraction_to_remove": 0.05}])
>>> result = run_denoise_comparison(
...     graph_specs=graphs, denoise_specs=denoisers, seeds=[0], verbose=False
... )
>>> "f1" in result.df.columns
True
Source code in src/spatial_graph_algorithms/compare/__init__.py 
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
def run_denoise_comparison(
    *,
    graph_specs: Iterable[Mapping[str, Any]],
    denoise_specs: Iterable[Mapping[str, Any]],
    seeds: Iterable[int],
    k_neighbors: int = 15,
    compute_distortion: bool = False,
    verbose: bool = True,
) -> ComparisonResult:
    """Run a simulated denoising benchmark and return tidy quality metrics.

    Each graph spec is generated once per seed, then every denoising spec is
    scored and filtered on that graph. Ground-truth false-edge labels from
    simulation are evaluated with :func:`spatial_graph_algorithms.metrics.evaluate_denoising`.

    Parameters
    ----------
    graph_specs : iterable of mapping
        Specs passed to :func:`spatial_graph_algorithms.simulate.generate`.
    denoise_specs : iterable of mapping
        Denoising specs. Each spec must include ``"method"`` and may include
        ``"fraction_to_remove"`` plus scorer-specific keyword arguments.
    seeds : iterable of int
        Top-level seeds for graph generation and stochastic scorers.
    k_neighbors : int
        Number of neighbours used when graph-level metrics are computed.
    compute_distortion : bool
        Whether to compute O(n²) distortion for generated graphs.
    verbose : bool
        If ``True`` (default), print one progress line per completed row.

    Returns
    -------
    ComparisonResult
        One row per ``graph_spec × seed × denoise_spec``.

    Examples
    --------
    >>> graphs = parameter_grid(cases=[{"n": 50, "false_edge_fraction": 0.1}])
    >>> denoisers = parameter_grid(cases=[{"method": "jaccard", "fraction_to_remove": 0.05}])
    >>> result = run_denoise_comparison(
    ...     graph_specs=graphs, denoise_specs=denoisers, seeds=[0], verbose=False
    ... )
    >>> "f1" in result.df.columns
    True
    """
    graph_specs_list = [_with_label(dict(spec)) for spec in graph_specs]
    denoise_specs_list = [_with_label(dict(spec)) for spec in denoise_specs]
    seeds_list = [int(seed) for seed in seeds]
    rows: list[Spec] = []

    total = len(graph_specs_list) * len(seeds_list) * len(denoise_specs_list)
    completed = 0

    for graph_spec in graph_specs_list:
        graph_params = _public_params(graph_spec)
        graph_params.pop("seed", None)
        for seed in seeds_list:
            graph_start = time.perf_counter()
            try:
                sg = generate(**graph_params, seed=seed)
                generation_seconds = time.perf_counter() - graph_start
                graph_metrics = evaluate(
                    sg,
                    k_neighbors=k_neighbors,
                    compute_distortion=compute_distortion,
                )
            except Exception as exc:  # noqa: BLE001 - experiments should continue.
                generation_seconds = time.perf_counter() - graph_start
                for denoise_spec in denoise_specs_list:
                    completed += 1
                    row = _base_denoise_row(
                        graph_spec=graph_spec,
                        denoise_spec=denoise_spec,
                        seed=seed,
                        status="generation_error",
                        error=f"{type(exc).__name__}: {exc}",
                        generation_seconds=generation_seconds,
                    )
                    rows.append(row)
                    if verbose:
                        _print_denoise_row(completed, total, row)
                continue

            for denoise_spec in denoise_specs_list:
                completed += 1
                row = _base_denoise_row(
                    graph_spec=graph_spec,
                    denoise_spec=denoise_spec,
                    seed=seed,
                    status="ok",
                    error=None,
                    generation_seconds=generation_seconds,
                )
                row.update(graph_metrics)
                row.update(_run_denoise_step(sg, denoise_spec, seed=seed))
                rows.append(row)
                if verbose:
                    _print_denoise_row(completed, total, row)

    return ComparisonResult(df=pd.DataFrame(rows))

spatial_graph_algorithms.compare.run_pipeline_comparison(*, graph_specs, reconstruction_specs, denoise_specs, seeds, include_raw_baseline=True, dim=None, k_neighbors=15, compute_distortion=False, verbose=True)

Run denoise-to-reconstruction comparisons on simulated graphs.

Parameters:

Name Type Description Default
graph_specs iterable of mapping

Specs passed to :func:spatial_graph_algorithms.simulate.generate.

 required
reconstruction_specs iterable of mapping

Specs passed to :func:spatial_graph_algorithms.reconstruct.reconstruct.

 required
denoise_specs iterable of mapping

Denoising specs applied before reconstruction.

 required
seeds iterable of int

Top-level seeds for graph generation, denoising, and reconstruction.

 required
include_raw_baseline bool

If True (default), include rows with denoise_label == "none".

 True
dim int

Reconstruction dimensionality. Defaults to graph spec dim or 2.

 None
k_neighbors int

Number of neighbours for reconstruction quality KNN evaluation.

 15
compute_distortion bool

Whether to compute the O(n²) distortion metric.

 False
verbose bool

If True (default), print one progress line per completed row.

 True

Returns:

Type Description
ComparisonResult

One row per graph, denoising condition, seed, and reconstruction spec.

Examples:

>>> graphs = parameter_grid(cases=[{"n": 50, "false_edge_fraction": 0.1}])
>>> denoisers = parameter_grid(cases=[{"method": "jaccard", "fraction_to_remove": 0.05}])
>>> recons = parameter_grid(cases=[{"method": "mds"}])
>>> result = run_pipeline_comparison(
...     graph_specs=graphs, reconstruction_specs=recons,
...     denoise_specs=denoisers, seeds=[0], verbose=False
... )
>>> set(result.df["denoise_label"]) == {"none", denoisers[0]["_label"]}
True
Source code in src/spatial_graph_algorithms/compare/__init__.py 
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
def run_pipeline_comparison(
    *,
    graph_specs: Iterable[Mapping[str, Any]],
    reconstruction_specs: Iterable[Mapping[str, Any]],
    denoise_specs: Iterable[Mapping[str, Any]],
    seeds: Iterable[int],
    include_raw_baseline: bool = True,
    dim: int | None = None,
    k_neighbors: int = 15,
    compute_distortion: bool = False,
    verbose: bool = True,
) -> ComparisonResult:
    """Run denoise-to-reconstruction comparisons on simulated graphs.

    Parameters
    ----------
    graph_specs : iterable of mapping
        Specs passed to :func:`spatial_graph_algorithms.simulate.generate`.
    reconstruction_specs : iterable of mapping
        Specs passed to :func:`spatial_graph_algorithms.reconstruct.reconstruct`.
    denoise_specs : iterable of mapping
        Denoising specs applied before reconstruction.
    seeds : iterable of int
        Top-level seeds for graph generation, denoising, and reconstruction.
    include_raw_baseline : bool
        If ``True`` (default), include rows with ``denoise_label == "none"``.
    dim : int, optional
        Reconstruction dimensionality. Defaults to graph spec ``dim`` or ``2``.
    k_neighbors : int
        Number of neighbours for reconstruction quality KNN evaluation.
    compute_distortion : bool
        Whether to compute the O(n²) distortion metric.
    verbose : bool
        If ``True`` (default), print one progress line per completed row.

    Returns
    -------
    ComparisonResult
        One row per graph, denoising condition, seed, and reconstruction spec.

    Examples
    --------
    >>> graphs = parameter_grid(cases=[{"n": 50, "false_edge_fraction": 0.1}])
    >>> denoisers = parameter_grid(cases=[{"method": "jaccard", "fraction_to_remove": 0.05}])
    >>> recons = parameter_grid(cases=[{"method": "mds"}])
    >>> result = run_pipeline_comparison(
    ...     graph_specs=graphs, reconstruction_specs=recons,
    ...     denoise_specs=denoisers, seeds=[0], verbose=False
    ... )
    >>> set(result.df["denoise_label"]) == {"none", denoisers[0]["_label"]}
    True
    """
    graph_specs_list = [_with_label(dict(spec)) for spec in graph_specs]
    reconstruction_specs_list = [_with_label(dict(spec)) for spec in reconstruction_specs]
    denoise_specs_list = [_with_label(dict(spec)) for spec in denoise_specs]
    seeds_list = [int(seed) for seed in seeds]
    rows: list[Spec] = []

    n_denoise_conditions = len(denoise_specs_list) + int(include_raw_baseline)
    total = (
        len(graph_specs_list)
        * len(seeds_list)
        * n_denoise_conditions
        * len(reconstruction_specs_list)
    )
    completed = 0

    for graph_spec in graph_specs_list:
        graph_params = _public_params(graph_spec)
        graph_params.pop("seed", None)
        for seed in seeds_list:
            graph_start = time.perf_counter()
            try:
                sg = generate(**graph_params, seed=seed)
                generation_seconds = time.perf_counter() - graph_start
            except Exception as exc:  # noqa: BLE001
                generation_seconds = time.perf_counter() - graph_start
                for denoise_spec in _pipeline_denoise_specs(
                    denoise_specs_list, include_raw_baseline=include_raw_baseline
                ):
                    for recon_spec in reconstruction_specs_list:
                        completed += 1
                        row = _base_pipeline_row(
                            graph_spec=graph_spec,
                            denoise_spec=denoise_spec,
                            recon_spec=recon_spec,
                            seed=seed,
                            status="generation_error",
                            error=f"{type(exc).__name__}: {exc}",
                            generation_seconds=generation_seconds,
                        )
                        rows.append(row)
                        if verbose:
                            _print_row(completed, total, graph_spec, recon_spec, seed, row)
                continue

            graph_variants: list[tuple[Mapping[str, Any], Any, Spec]] = []
            if include_raw_baseline:
                graph_variants.append(
                    (
                        {"_label": "none", "method": None},
                        sg,
                        {"score_seconds": 0.0, "filter_seconds": 0.0},
                    )
                )
            for denoise_spec in denoise_specs_list:
                denoise_result = _run_denoise_step(sg, denoise_spec, seed=seed)
                if denoise_result["status"] == "ok":
                    graph_variants.append(
                        (denoise_spec, denoise_result.pop("_sg_clean"), denoise_result)
                    )
                else:
                    for recon_spec in reconstruction_specs_list:
                        completed += 1
                        row = _base_pipeline_row(
                            graph_spec=graph_spec,
                            denoise_spec=denoise_spec,
                            recon_spec=recon_spec,
                            seed=seed,
                            status=denoise_result["status"],
                            error=denoise_result["error"],
                            generation_seconds=generation_seconds,
                        )
                        row.update({k: v for k, v in denoise_result.items() if k != "_sg_clean"})
                        rows.append(row)
                        if verbose:
                            _print_row(completed, total, graph_spec, recon_spec, seed, row)

            for denoise_spec, sg_variant, denoise_result in graph_variants:
                for recon_spec in reconstruction_specs_list:
                    completed += 1
                    row = _base_pipeline_row(
                        graph_spec=graph_spec,
                        denoise_spec=denoise_spec,
                        recon_spec=recon_spec,
                        seed=seed,
                        status="ok",
                        error=None,
                        generation_seconds=generation_seconds,
                    )
                    row.update({k: v for k, v in denoise_result.items() if k != "_sg_clean"})
                    _run_reconstruction_step(
                        row,
                        sg_variant,
                        graph_params=graph_params,
                        recon_spec=recon_spec,
                        seed=seed,
                        dim=dim,
                        k_neighbors=k_neighbors,
                        compute_distortion=compute_distortion,
                    )
                    rows.append(row)
                    if verbose:
                        _print_row(completed, total, graph_spec, recon_spec, seed, row)

    return ComparisonResult(df=pd.DataFrame(rows))

spatial_graph_algorithms.compare.benchmark_io.save_benchmark_graph(sg, path, *, metadata=None)

Save a benchmark graph as inspectable array, table, and JSON files.

Parameters:

Name Type Description Default
sg SpatialGraph

Graph instance to save.

 required
path str or Path

Destination directory. It is created when missing.

 required
metadata dict

Extra graph-level metadata written to metadata.json.

 None

Returns:

Type Description
None

Examples:

>>> from tempfile import TemporaryDirectory
>>> from spatial_graph_algorithms.simulate import generate
>>> from spatial_graph_algorithms.compare.benchmark_io import save_benchmark_graph
>>> with TemporaryDirectory() as tmp:
...     save_benchmark_graph(generate(n=20, seed=0), tmp)
Source code in src/spatial_graph_algorithms/compare/benchmark_io.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def save_benchmark_graph(
    sg: SpatialGraph,
    path: str | Path,
    *,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Save a benchmark graph as inspectable array, table, and JSON files.

    Parameters
    ----------
    sg : SpatialGraph
        Graph instance to save.
    path : str or Path
        Destination directory. It is created when missing.
    metadata : dict, optional
        Extra graph-level metadata written to ``metadata.json``.

    Returns
    -------
    None

    Examples
    --------
    >>> from tempfile import TemporaryDirectory
    >>> from spatial_graph_algorithms.simulate import generate
    >>> from spatial_graph_algorithms.compare.benchmark_io import save_benchmark_graph
    >>> with TemporaryDirectory() as tmp:
    ...     save_benchmark_graph(generate(n=20, seed=0), tmp)
    """
    out = Path(path)
    out.mkdir(parents=True, exist_ok=True)

    scipy.sparse.save_npz(out / "adjacency.npz", sg.adjacency_matrix)
    _save_array(out / "positions.npy", sg.positions)
    _save_array(out / "reconstructed_positions.npy", sg.reconstructed_positions)
    _save_frame(out / "edge_metadata.csv", sg.edge_metadata)
    _save_frame(out / "node_metadata.csv", sg.node_metadata)

    payload = dict(metadata or {})
    payload["n_nodes"] = int(sg.n_nodes)
    payload["n_edges"] = int(sg.n_edges)
    payload["node_id_map"] = _jsonable_mapping(sg.node_id_map)
    payload["source_indices"] = (
        None if sg.source_indices is None else sg.source_indices.astype(int).tolist()
    )
    (out / "metadata.json").write_text(json.dumps(payload, indent=2, sort_keys=True))

spatial_graph_algorithms.compare.benchmark_io.load_benchmark_graph(path)

Load a benchmark graph saved by :func:save_benchmark_graph.

Parameters:

Name Type Description Default
path str or Path

Benchmark graph directory.

 required

Returns:

Type Description
SpatialGraph

Materialized graph with adjacency, positions, and metadata restored.

Raises:

Type Description
FileNotFoundError

If the directory or required adjacency file does not exist.

Examples:

>>> from tempfile import TemporaryDirectory
>>> from spatial_graph_algorithms.simulate import generate
>>> from spatial_graph_algorithms.compare.benchmark_io import (
...     load_benchmark_graph, save_benchmark_graph,
... )
>>> with TemporaryDirectory() as tmp:
...     save_benchmark_graph(generate(n=20, seed=0), tmp)
...     sg = load_benchmark_graph(tmp)
>>> sg.n_nodes
20
Source code in src/spatial_graph_algorithms/compare/benchmark_io.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def load_benchmark_graph(path: str | Path) -> SpatialGraph:
    """Load a benchmark graph saved by :func:`save_benchmark_graph`.

    Parameters
    ----------
    path : str or Path
        Benchmark graph directory.

    Returns
    -------
    SpatialGraph
        Materialized graph with adjacency, positions, and metadata restored.

    Raises
    ------
    FileNotFoundError
        If the directory or required adjacency file does not exist.

    Examples
    --------
    >>> from tempfile import TemporaryDirectory
    >>> from spatial_graph_algorithms.simulate import generate
    >>> from spatial_graph_algorithms.compare.benchmark_io import (
    ...     load_benchmark_graph, save_benchmark_graph,
    ... )
    >>> with TemporaryDirectory() as tmp:
    ...     save_benchmark_graph(generate(n=20, seed=0), tmp)
    ...     sg = load_benchmark_graph(tmp)
    >>> sg.n_nodes
    20
    """
    root = Path(path)
    if not root.exists():
        raise FileNotFoundError(f"Benchmark graph directory not found: {root}")
    adj_path = root / "adjacency.npz"
    if not adj_path.exists():
        raise FileNotFoundError(f"Benchmark adjacency file not found: {adj_path}")

    metadata = _read_json(root / "metadata.json")
    return SpatialGraph(
        adjacency_matrix=scipy.sparse.load_npz(adj_path).tocsr(),
        positions=_load_array(root / "positions.npy"),
        reconstructed_positions=_load_array(root / "reconstructed_positions.npy"),
        edge_metadata=_load_frame(root / "edge_metadata.csv"),
        node_metadata=_load_frame(root / "node_metadata.csv"),
        node_id_map=_restore_mapping(metadata.get("node_id_map")),
        source_indices=_restore_array(metadata.get("source_indices")),
        keep_lcc=False,
    )

spatial_graph_algorithms.compare.benchmark_io.graph_difficulty_metadata(sg)

Compute false-edge burden metadata for a benchmark graph.

Parameters:

Name Type Description Default
sg SpatialGraph

Graph with optional edge_metadata["is_false"] labels.

 required

Returns:

Type Description
dict

Edge counts and false-edge ratios. Values are None when false-edge labels are unavailable.

Examples:

>>> from spatial_graph_algorithms.simulate import generate
>>> from spatial_graph_algorithms.compare.benchmark_io import graph_difficulty_metadata
>>> meta = graph_difficulty_metadata(generate(n=50, false_edge_fraction=0.1, seed=0))
>>> "n_false_edges" in meta
True
Source code in src/spatial_graph_algorithms/compare/benchmark_io.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def graph_difficulty_metadata(sg: SpatialGraph) -> dict[str, float | int | None]:
    """Compute false-edge burden metadata for a benchmark graph.

    Parameters
    ----------
    sg : SpatialGraph
        Graph with optional ``edge_metadata["is_false"]`` labels.

    Returns
    -------
    dict
        Edge counts and false-edge ratios. Values are ``None`` when false-edge
        labels are unavailable.

    Examples
    --------
    >>> from spatial_graph_algorithms.simulate import generate
    >>> from spatial_graph_algorithms.compare.benchmark_io import graph_difficulty_metadata
    >>> meta = graph_difficulty_metadata(generate(n=50, false_edge_fraction=0.1, seed=0))
    >>> "n_false_edges" in meta
    True
    """
    if sg.edge_metadata is None or "is_false" not in sg.edge_metadata.columns:
        return {
            "n_true_edges": None,
            "n_false_edges": None,
            "true_false_edge_ratio": None,
            "false_edge_ratio": None,
            "false_edge_fraction": None,
            "false_edge_rate_of_total": None,
        }

    is_false = sg.edge_metadata["is_false"].astype(bool)
    n_false = int(is_false.sum())
    n_total = int(len(is_false))
    n_true = n_total - n_false
    return {
        "n_true_edges": n_true,
        "n_false_edges": n_false,
        "true_false_edge_ratio": None if n_false == 0 else float(n_true / n_false),
        "false_edge_ratio": None if n_true == 0 else float(n_false / n_true),
        "false_edge_fraction": None if n_total == 0 else float(n_false / n_total),
        "false_edge_rate_of_total": None if n_total == 0 else float(n_false / n_total),
    }

spatial_graph_algorithms.compare.gym.BenchmarkSuite dataclass

Describe a materialized benchmark suite.

Parameters:

Name Type Description Default
name str

Stable suite name.

 required
root Path

Directory containing the manifest and graph files.

 required
manifest dict

Parsed manifest payload.

 required

Examples:

>>> suite = BenchmarkSuite(name="demo", root=Path("."), manifest={"name": "demo"})
>>> suite.name
'demo'
Source code in src/spatial_graph_algorithms/compare/gym.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@dataclass
class BenchmarkSuite:
    """Describe a materialized benchmark suite.

    Parameters
    ----------
    name : str
        Stable suite name.
    root : pathlib.Path
        Directory containing the manifest and graph files.
    manifest : dict
        Parsed manifest payload.

    Examples
    --------
    >>> suite = BenchmarkSuite(name="demo", root=Path("."), manifest={"name": "demo"})
    >>> suite.name
    'demo'
    """

    name: str
    root: Path
    manifest: dict[str, Any]

    @property
    def description(self) -> str:
        """Return the suite description from the manifest."""
        return str(self.manifest.get("description", ""))

    @property
    def rationale(self) -> str:
        """Return the suite rationale from the manifest."""
        return str(self.manifest.get("rationale", ""))

    @property
    def graph_paths(self) -> list[Path]:
        """Return materialized graph directories declared by the manifest."""
        raw_paths = self.manifest.get("graph_paths", [])
        return [(self.root / path).resolve() for path in raw_paths]

    @property
    def tasks(self) -> dict[str, Any]:
        """Return task definitions keyed by task name."""
        return dict(self.manifest.get("tasks", {}))

Attributes

description property

Return the suite description from the manifest.

rationale property

Return the suite rationale from the manifest.

graph_paths property

Return materialized graph directories declared by the manifest.

tasks property

Return task definitions keyed by task name.

spatial_graph_algorithms.compare.gym.list_suites(*, benchmark_root='data/benchmark')

List benchmark suites with manifests under benchmark_root.

Parameters:

Name Type Description Default
benchmark_root str or Path

Directory containing suite subdirectories. Default is "data/benchmark".

 'data/benchmark'

Returns:

Type Description
list of str

Suite names sorted alphabetically.

Examples:

>>> isinstance(list_suites(), list)
True
Source code in src/spatial_graph_algorithms/compare/gym.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def list_suites(*, benchmark_root: str | Path = "data/benchmark") -> list[str]:
    """List benchmark suites with manifests under *benchmark_root*.

    Parameters
    ----------
    benchmark_root : str or Path
        Directory containing suite subdirectories. Default is
        ``"data/benchmark"``.

    Returns
    -------
    list of str
        Suite names sorted alphabetically.

    Examples
    --------
    >>> isinstance(list_suites(), list)
    True
    """
    root = Path(benchmark_root)
    if not root.exists():
        return []
    suites = []
    for child in root.iterdir():
        if child.is_dir() and _manifest_path(child) is not None:
            suites.append(child.name)
    return sorted(suites)

spatial_graph_algorithms.compare.gym.load_suite(suite, *, benchmark_root='data/benchmark')

Load a benchmark suite manifest by name or path.

Parameters:

Name Type Description Default
suite str or Path

Suite name under benchmark_root, suite directory, or manifest file.

 required
benchmark_root str or Path

Directory searched when suite is a name. Default is "data/benchmark".

 'data/benchmark'

Returns:

Type Description
BenchmarkSuite

Parsed suite definition.

Raises:

Type Description
FileNotFoundError

If no manifest can be found.
ValueError

If the manifest does not contain a name.

Examples:

>>> from tempfile import TemporaryDirectory
>>> with TemporaryDirectory() as tmp:
...     root = Path(tmp) / "demo"
...     root.mkdir()
...     _ = (root / "manifest.json").write_text('{"name": "demo"}')
...     suite = load_suite(root)
>>> suite.name
'demo'
Source code in src/spatial_graph_algorithms/compare/gym.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def load_suite(
    suite: str | Path,
    *,
    benchmark_root: str | Path = "data/benchmark",
) -> BenchmarkSuite:
    """Load a benchmark suite manifest by name or path.

    Parameters
    ----------
    suite : str or Path
        Suite name under *benchmark_root*, suite directory, or manifest file.
    benchmark_root : str or Path
        Directory searched when *suite* is a name. Default is
        ``"data/benchmark"``.

    Returns
    -------
    BenchmarkSuite
        Parsed suite definition.

    Raises
    ------
    FileNotFoundError
        If no manifest can be found.
    ValueError
        If the manifest does not contain a ``name``.

    Examples
    --------
    >>> from tempfile import TemporaryDirectory
    >>> with TemporaryDirectory() as tmp:
    ...     root = Path(tmp) / "demo"
    ...     root.mkdir()
    ...     _ = (root / "manifest.json").write_text('{"name": "demo"}')
    ...     suite = load_suite(root)
    >>> suite.name
    'demo'
    """
    path = Path(suite)
    if path.is_file():
        manifest_path = path
        root = path.parent
    else:
        root = path if path.exists() else Path(benchmark_root) / str(suite)
        manifest_path = _manifest_path(root)
        if manifest_path is None:
            raise FileNotFoundError(f"No manifest found for benchmark suite: {suite}")

    manifest = _read_manifest(manifest_path)
    name = manifest.get("name")
    if not name:
        raise ValueError(f"Benchmark manifest must contain a name: {manifest_path}")
    return BenchmarkSuite(name=str(name), root=root.resolve(), manifest=manifest)

spatial_graph_algorithms.compare.gym.run_suite(suite, *, task, benchmark_root='data/benchmark', k_neighbors=15, compute_distortion=False, checkpoint_dir=None, resume=True, verbose=True)

Run one benchmark task over a materialized suite.

Parameters:

Name Type Description Default
suite str, Path, or BenchmarkSuite

Suite name, manifest path, suite directory, or loaded suite.

 required
task ('denoise', 'reconstruction', 'pipeline')

Benchmark task to run.

 "denoise"
benchmark_root str or Path

Suite root used when suite is a name. Default is "data/benchmark".

 'data/benchmark'
k_neighbors int

Number of neighbours for KNN reconstruction quality.

 15
compute_distortion bool

Whether to compute O(n²) distortion during reconstruction tasks.

 False
checkpoint_dir str or Path

Directory for a crash-safe JSONL checkpoint. When given, every finished row is appended to <checkpoint_dir>/<suite>__<task>.jsonl as it completes so an interrupted run can be resumed without lost work.

 None
resume bool

When True (default) and checkpoint_dir is set, rows already present in the checkpoint are reused instead of recomputed. When False any existing checkpoint is truncated before the run.

 True
verbose bool

If True, print one progress line per row.

 True

Returns:

Type Description
ComparisonResult

Tidy benchmark results.

Raises:

Type Description
ValueError

If the task is unsupported or the suite has no materialized graphs.

Examples:

>>> run_suite("missing", task="denoise")
Traceback (most recent call last):
FileNotFoundError: ...
Source code in src/spatial_graph_algorithms/compare/gym.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def run_suite(
    suite: str | Path | BenchmarkSuite,
    *,
    task: BenchmarkTask,
    benchmark_root: str | Path = "data/benchmark",
    k_neighbors: int = 15,
    compute_distortion: bool = False,
    checkpoint_dir: str | Path | None = None,
    resume: bool = True,
    verbose: bool = True,
) -> ComparisonResult:
    """Run one benchmark task over a materialized suite.

    Parameters
    ----------
    suite : str, Path, or BenchmarkSuite
        Suite name, manifest path, suite directory, or loaded suite.
    task : {"denoise", "reconstruction", "pipeline"}
        Benchmark task to run.
    benchmark_root : str or Path
        Suite root used when *suite* is a name. Default is
        ``"data/benchmark"``.
    k_neighbors : int
        Number of neighbours for KNN reconstruction quality.
    compute_distortion : bool
        Whether to compute O(n²) distortion during reconstruction tasks.
    checkpoint_dir : str or Path, optional
        Directory for a crash-safe JSONL checkpoint. When given, every finished
        row is appended to ``<checkpoint_dir>/<suite>__<task>.jsonl`` as it
        completes so an interrupted run can be resumed without lost work.
    resume : bool
        When ``True`` (default) and *checkpoint_dir* is set, rows already
        present in the checkpoint are reused instead of recomputed. When
        ``False`` any existing checkpoint is truncated before the run.
    verbose : bool
        If ``True``, print one progress line per row.

    Returns
    -------
    ComparisonResult
        Tidy benchmark results.

    Raises
    ------
    ValueError
        If the task is unsupported or the suite has no materialized graphs.

    Examples
    --------
    >>> run_suite("missing", task="denoise")  # doctest: +IGNORE_EXCEPTION_DETAIL
    Traceback (most recent call last):
    FileNotFoundError: ...
    """
    loaded = suite if isinstance(suite, BenchmarkSuite) else load_suite(
        suite, benchmark_root=benchmark_root
    )
    graph_refs = _suite_graph_refs(loaded)
    if not graph_refs:
        raise ValueError(
            f"Benchmark suite {loaded.name!r} has no graph_paths or real_datasets."
        )

    writer = None
    if checkpoint_dir is not None:
        path = Path(checkpoint_dir) / f"{loaded.name}__{task}.jsonl"
        writer = CheckpointWriter(path, resume=resume)

    if task == "denoise":
        if loaded.manifest.get("kind") == "experimental":
            raise ValueError("Experimental suites do not support denoise-only F1/AUC tasks.")
        rows = _run_denoise_suite(loaded, graph_refs, writer=writer, verbose=verbose)
    elif task == "reconstruction":
        rows = _run_reconstruction_suite(
            loaded,
            graph_refs,
            k_neighbors=k_neighbors,
            compute_distortion=compute_distortion,
            max_eval_nodes=_task_option(loaded, task, "max_eval_nodes"),
            writer=writer,
            verbose=verbose,
        )
    elif task == "pipeline":
        rows = _run_pipeline_suite(
            loaded,
            graph_refs,
            k_neighbors=k_neighbors,
            compute_distortion=compute_distortion,
            max_eval_nodes=_task_option(loaded, task, "max_eval_nodes"),
            writer=writer,
            verbose=verbose,
        )
    else:
        raise ValueError(f"Unsupported benchmark task: {task!r}")

    return ComparisonResult(df=pd.DataFrame(rows))

spatial_graph_algorithms.compare.datasets.list_experimental_datasets()

List built-in experimental benchmark dataset names.

Returns:

Type Description
list of str

Dataset names sorted alphabetically.

Examples:

>>> "human_tonsil" in list_experimental_datasets()
True
Source code in src/spatial_graph_algorithms/compare/datasets.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def list_experimental_datasets() -> list[str]:
    """List built-in experimental benchmark dataset names.

    Returns
    -------
    list of str
        Dataset names sorted alphabetically.

    Examples
    --------
    >>> "human_tonsil" in list_experimental_datasets()
    True
    """
    return sorted(EXPERIMENTAL_DATASETS)

spatial_graph_algorithms.compare.datasets.load_experimental_dataset(name, *, data_root='data')

Load an experimental graph and attach available ground-truth positions.

Nodes absent from original_positions.csv receive NaN coordinates. :func:spatial_graph_algorithms.metrics.evaluate automatically excludes those rows from CPD/KNN computations.

Parameters:

Name Type Description Default
name str

One of :func:list_experimental_datasets.

 required
data_root str or Path

Directory containing experimental dataset subdirectories. Default is "data".

 'data'

Returns:

Type Description
SpatialGraph

Loaded graph with partial positions attached and optional node_metadata for positioned nodes.

Raises:

Type Description
ValueError

If name is unknown.
FileNotFoundError

If required dataset files are missing.

Examples:

>>> load_experimental_dataset("unknown")
Traceback (most recent call last):
ValueError: ...
Source code in src/spatial_graph_algorithms/compare/datasets.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def load_experimental_dataset(
    name: str,
    *,
    data_root: str | Path = "data",
) -> SpatialGraph:
    """Load an experimental graph and attach available ground-truth positions.

    Nodes absent from ``original_positions.csv`` receive NaN coordinates.
    :func:`spatial_graph_algorithms.metrics.evaluate` automatically excludes
    those rows from CPD/KNN computations.

    Parameters
    ----------
    name : str
        One of :func:`list_experimental_datasets`.
    data_root : str or Path
        Directory containing experimental dataset subdirectories. Default is
        ``"data"``.

    Returns
    -------
    SpatialGraph
        Loaded graph with partial ``positions`` attached and optional
        ``node_metadata`` for positioned nodes.

    Raises
    ------
    ValueError
        If *name* is unknown.
    FileNotFoundError
        If required dataset files are missing.

    Examples
    --------
    >>> load_experimental_dataset("unknown")  # doctest: +IGNORE_EXCEPTION_DETAIL
    Traceback (most recent call last):
    ValueError: ...
    """
    if name not in EXPERIMENTAL_DATASETS:
        available = ", ".join(list_experimental_datasets())
        raise ValueError(f"Unknown experimental dataset {name!r}. Available: {available}.")

    root = _resolve_dataset_root(name, data_root=Path(data_root))
    edge_path = root / "edge_list.csv.gz"
    if not edge_path.exists():
        edge_path = root / "edge_list.csv"
    pos_path = root / "original_positions.csv"
    if not edge_path.exists():
        raise FileNotFoundError(f"Experimental edge list not found: {edge_path}")
    if not pos_path.exists():
        raise FileNotFoundError(f"Experimental positions file not found: {pos_path}")

    sg = load_edge_list(str(edge_path))
    pos_df = pd.read_csv(pos_path)
    node_id_col = _position_node_id_column(pos_df, path=pos_path)

    positions = np.full((sg.n_nodes, 2), np.nan, dtype=float)
    node_metadata = pd.DataFrame(index=range(sg.n_nodes))
    for _, row in pos_df.iterrows():
        node_id = row[node_id_col]
        if node_id not in sg.node_id_map:
            continue
        idx = int(sg.node_id_map[node_id])
        positions[idx, 0] = float(row["x"])
        positions[idx, 1] = float(row["y"])
        if "label" in pos_df.columns:
            node_metadata.loc[idx, "label"] = row.get("label")

    node_metadata["has_position"] = ~np.isnan(positions).any(axis=1)
    return sg.replace(positions=positions, node_metadata=node_metadata.reset_index(drop=True))

spatial_graph_algorithms.compare.report.status_counts(result)

Count result rows by status.

Parameters:

Name Type Description Default
result ComparisonResult

Result table to summarise.

 required

Returns:

Type Description
DataFrame

Columns status and n_rows sorted by count descending.

Examples:

>>> import pandas as pd
>>> from spatial_graph_algorithms.compare import ComparisonResult
>>> status_counts(ComparisonResult(pd.DataFrame({"status": ["ok", "ok"]})))["n_rows"].iloc[0]
2
Source code in src/spatial_graph_algorithms/compare/report.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def status_counts(result: ComparisonResult) -> pd.DataFrame:
    """Count result rows by status.

    Parameters
    ----------
    result : ComparisonResult
        Result table to summarise.

    Returns
    -------
    pandas.DataFrame
        Columns ``status`` and ``n_rows`` sorted by count descending.

    Examples
    --------
    >>> import pandas as pd
    >>> from spatial_graph_algorithms.compare import ComparisonResult
    >>> status_counts(ComparisonResult(pd.DataFrame({"status": ["ok", "ok"]})))["n_rows"].iloc[0]
    2
    """
    if "status" not in result.df.columns:
        return pd.DataFrame({"status": [], "n_rows": []})
    return (
        result.df["status"]
        .value_counts(dropna=False)
        .rename_axis("status")
        .reset_index(name="n_rows")
    )

spatial_graph_algorithms.compare.report.failed_rows(result)

Return non-successful rows with identifying columns and errors.

Parameters:

Name Type Description Default
result ComparisonResult

Result table to inspect.

 required

Returns:

Type Description
DataFrame

Failed rows restricted to stable identification and error columns.

Examples:

>>> import pandas as pd
>>> from spatial_graph_algorithms.compare import ComparisonResult
>>> failed_rows(ComparisonResult(pd.DataFrame({"status": ["ok"]}))).empty
True
Source code in src/spatial_graph_algorithms/compare/report.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def failed_rows(result: ComparisonResult) -> pd.DataFrame:
    """Return non-successful rows with identifying columns and errors.

    Parameters
    ----------
    result : ComparisonResult
        Result table to inspect.

    Returns
    -------
    pandas.DataFrame
        Failed rows restricted to stable identification and error columns.

    Examples
    --------
    >>> import pandas as pd
    >>> from spatial_graph_algorithms.compare import ComparisonResult
    >>> failed_rows(ComparisonResult(pd.DataFrame({"status": ["ok"]}))).empty
    True
    """
    if "status" not in result.df.columns:
        return pd.DataFrame()
    failed = result.df[result.df["status"] != "ok"].copy()
    cols = [
        "suite",
        "difficulty",
        "graph_id",
        "graph_label",
        "seed",
        "denoise_label",
        "reconstruction_label",
        "method",
        "denoise_method",
        "status",
        "error",
    ]
    present = [col for col in cols if col in failed.columns]
    return failed[present].reset_index(drop=True)

spatial_graph_algorithms.compare.report.gym_leaderboard(result, *, task, suite=None, primary_metric=None, group_by=None)

Aggregate and rank benchmark gym results.

Parameters:

Name Type Description Default
result ComparisonResult

Raw benchmark result.

 required
task ('denoise', 'reconstruction', 'pipeline')

Task schema to use for method columns and default metrics.

 "denoise"
suite BenchmarkSuite

Suite manifest used to read task-level primary and secondary metrics.

 None
primary_metric str

Metric used for ranking. Overrides the suite manifest and defaults.

 None
group_by list of str

Extra grouping columns. Defaults to ["difficulty"] when present, producing one headline row per method. Pass explicit graph-spec columns for detailed breakdowns.

 None

Returns:

Type Description
DataFrame

Ranked aggregate table with mean/std metric columns and row counts.

Raises:

Type Description
ValueError

If task is unknown or the primary metric is unavailable.

Examples:

>>> import pandas as pd
>>> from spatial_graph_algorithms.compare import ComparisonResult
>>> df = pd.DataFrame({"status": ["ok"], "method": ["mds"], "cpd": [0.8]})
>>> gym_leaderboard(ComparisonResult(df), task="reconstruction")["rank"].iloc[0]
1
Source code in src/spatial_graph_algorithms/compare/report.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def gym_leaderboard(
    result: ComparisonResult,
    *,
    task: GymTask,
    suite: BenchmarkSuite | None = None,
    primary_metric: str | None = None,
    group_by: list[str] | None = None,
) -> pd.DataFrame:
    """Aggregate and rank benchmark gym results.

    Parameters
    ----------
    result : ComparisonResult
        Raw benchmark result.
    task : {"denoise", "reconstruction", "pipeline"}
        Task schema to use for method columns and default metrics.
    suite : BenchmarkSuite, optional
        Suite manifest used to read task-level primary and secondary metrics.
    primary_metric : str, optional
        Metric used for ranking. Overrides the suite manifest and defaults.
    group_by : list of str, optional
        Extra grouping columns. Defaults to ``["difficulty"]`` when present,
        producing one headline row per method. Pass explicit graph-spec columns
        for detailed breakdowns.

    Returns
    -------
    pandas.DataFrame
        Ranked aggregate table with mean/std metric columns and row counts.

    Raises
    ------
    ValueError
        If *task* is unknown or the primary metric is unavailable.

    Examples
    --------
    >>> import pandas as pd
    >>> from spatial_graph_algorithms.compare import ComparisonResult
    >>> df = pd.DataFrame({"status": ["ok"], "method": ["mds"], "cpd": [0.8]})
    >>> gym_leaderboard(ComparisonResult(df), task="reconstruction")["rank"].iloc[0]
    1
    """
    if task not in _DEFAULT_PRIMARY:
        raise ValueError(f"Unsupported gym task: {task!r}")

    df = _analysis_frame(result, task=task)
    ok = df[df.get("status") == "ok"].copy() if "status" in df.columns else df.copy()
    if ok.empty:
        return pd.DataFrame()

    metric = primary_metric or _suite_primary_metric(suite, task) or _DEFAULT_PRIMARY[task]
    if metric not in ok.columns:
        raise ValueError(f"Primary metric {metric!r} not found in result columns.")

    method_cols = _method_columns(task, ok)
    group_cols = _group_columns(ok, explicit=group_by)
    metrics = _metric_columns(ok, task=task, suite=suite, primary_metric=metric)

    grouped = ok.groupby(group_cols + method_cols, dropna=False)
    agg = grouped[metrics].agg(["mean", "std"]).reset_index()
    agg.columns = [
        "_".join(str(part) for part in col if part)
        if isinstance(col, tuple)
        else str(col)
        for col in agg.columns
    ]
    counts = _status_count_by_group(df, group_cols + method_cols)
    out = agg.merge(counts, on=group_cols + method_cols, how="left")

    primary_mean = f"{metric}_mean"
    time_col = _time_column(task, out)
    sort_cols = [primary_mean]
    ascending = [False]
    if time_col is not None:
        sort_cols.append(time_col)
        ascending.append(True)
    out = out.sort_values(sort_cols, ascending=ascending).reset_index(drop=True)
    out.insert(0, "rank", np.arange(1, len(out) + 1))
    return out

spatial_graph_algorithms.compare.report.gym_plots(result, *, task, suite=None)

Create a small set of labelled benchmark report plots.

Parameters:

Name Type Description Default
result ComparisonResult

Benchmark result to plot.

 required
task ('denoise', 'reconstruction', 'pipeline')

Task schema.

 "denoise"
suite BenchmarkSuite

Suite manifest used for metric defaults.

 None

Returns:

Type Description
dict[str, Figure]

Mapping from stable plot name to figure.

Raises:

Type Description
ValueError

If task is unknown.

Examples:

>>> import pandas as pd
>>> from spatial_graph_algorithms.compare import ComparisonResult
>>> df = pd.DataFrame({"status": ["ok"], "method": ["mds"], "cpd": [0.8]})
>>> plots = gym_plots(ComparisonResult(df), task="reconstruction")
>>> "quality_bar" in plots
True
Source code in src/spatial_graph_algorithms/compare/report.py 
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
def gym_plots(
    result: ComparisonResult,
    *,
    task: GymTask,
    suite: BenchmarkSuite | None = None,
) -> dict[str, Figure]:
    """Create a small set of labelled benchmark report plots.

    Parameters
    ----------
    result : ComparisonResult
        Benchmark result to plot.
    task : {"denoise", "reconstruction", "pipeline"}
        Task schema.
    suite : BenchmarkSuite, optional
        Suite manifest used for metric defaults.

    Returns
    -------
    dict[str, matplotlib.figure.Figure]
        Mapping from stable plot name to figure.

    Raises
    ------
    ValueError
        If *task* is unknown.

    Examples
    --------
    >>> import pandas as pd
    >>> from spatial_graph_algorithms.compare import ComparisonResult
    >>> df = pd.DataFrame({"status": ["ok"], "method": ["mds"], "cpd": [0.8]})
    >>> plots = gym_plots(ComparisonResult(df), task="reconstruction")
    >>> "quality_bar" in plots
    True
    """
    if task not in _DEFAULT_PRIMARY:
        raise ValueError(f"Unsupported gym task: {task!r}")

    leaderboard = gym_leaderboard(result, task=task, suite=suite)
    plots: dict[str, Figure] = {}
    if leaderboard.empty:
        plots["status_counts"] = plot_status_counts(result)
        return plots

    primary = _suite_primary_metric(suite, task) or _DEFAULT_PRIMARY[task]
    plots["quality_bar"] = plot_metric_bar(
        leaderboard,
        task=task,
        metric=primary,
        title=f"{primary} by method",
    )
    plots["status_counts"] = plot_status_counts(result)
    plots["scaling"] = plot_scaling(result, task=task)
    if task == "denoise":
        if "f1_mean" in leaderboard.columns:
            plots["f1_bar"] = plot_metric_bar(
                leaderboard,
                task=task,
                metric="f1",
                title="Denoising F1 by method",
            )
        plots["precision_recall"] = plot_precision_recall(result)
        plots["timing_bar"] = plot_metric_bar(
            leaderboard,
            task=task,
            metric="score_seconds",
            title="Scoring time by method",
            higher_is_better=False,
            log_y=True,
        )
    elif task == "reconstruction":
        if "knn_mean" in leaderboard.columns:
            plots["knn_bar"] = plot_metric_bar(
                leaderboard,
                task=task,
                metric="knn",
                title="KNN preservation by method",
            )
        plots["quality_vs_time"] = plot_quality_vs_time(
            leaderboard,
            task=task,
            quality_metric=primary,
            time_metric="reconstruction_seconds",
        )
    elif task == "pipeline":
        if "delta_knn_mean" in leaderboard.columns:
            plots["delta_knn_bar"] = plot_metric_bar(
                leaderboard,
                task=task,
                metric="delta_knn",
                title="Change in KNN after denoising",
                zero_line=True,
            )
        plots["delta_bar"] = plot_metric_bar(
            leaderboard,
            task=task,
            metric=primary,
            title=f"{primary} after denoising",
            zero_line=True,
        )
        plots["quality_vs_time"] = plot_quality_vs_time(
            leaderboard,
            task=task,
            quality_metric=primary,
            time_metric="total_seconds",
        )
    return plots

spatial_graph_algorithms.compare.report.gym_report(results, *, outdir, suite=None, title=None)

Write a Markdown benchmark report with CSV tables and PNG plots.

Parameters:

Name Type Description Default
results mapping

Mapping from task name to :class:ComparisonResult.

 required
outdir str or Path

Destination directory. It is created when missing.

 required
suite BenchmarkSuite

Suite metadata used for report description and manifest metrics.

 None
title str

Report title. Defaults to the suite name when available.

 None

Returns:

Type Description
Path

Path to the written report.md.

Raises:

Type Description
ValueError

If results contains an unsupported task name.

Examples:

>>> import pandas as pd
>>> from tempfile import TemporaryDirectory
>>> from spatial_graph_algorithms.compare import ComparisonResult
>>> with TemporaryDirectory() as tmp:
...     result = ComparisonResult(pd.DataFrame({
...         "status": ["ok"], "method": ["mds"], "cpd": [0.8],
...     }))
...     path = gym_report({"reconstruction": result}, outdir=tmp)
>>> path.name
'report.md'
Source code in src/spatial_graph_algorithms/compare/report.py 
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
def gym_report(
    results: Mapping[str, ComparisonResult],
    *,
    outdir: str | Path,
    suite: BenchmarkSuite | None = None,
    title: str | None = None,
) -> Path:
    """Write a Markdown benchmark report with CSV tables and PNG plots.

    Parameters
    ----------
    results : mapping
        Mapping from task name to :class:`ComparisonResult`.
    outdir : str or Path
        Destination directory. It is created when missing.
    suite : BenchmarkSuite, optional
        Suite metadata used for report description and manifest metrics.
    title : str, optional
        Report title. Defaults to the suite name when available.

    Returns
    -------
    pathlib.Path
        Path to the written ``report.md``.

    Raises
    ------
    ValueError
        If *results* contains an unsupported task name.

    Examples
    --------
    >>> import pandas as pd
    >>> from tempfile import TemporaryDirectory
    >>> from spatial_graph_algorithms.compare import ComparisonResult
    >>> with TemporaryDirectory() as tmp:
    ...     result = ComparisonResult(pd.DataFrame({
    ...         "status": ["ok"], "method": ["mds"], "cpd": [0.8],
    ...     }))
    ...     path = gym_report({"reconstruction": result}, outdir=tmp)
    >>> path.name
    'report.md'
    """
    out = Path(outdir)
    plots_dir = out / "plots"
    out.mkdir(parents=True, exist_ok=True)
    plots_dir.mkdir(parents=True, exist_ok=True)

    report_title = title or (f"Gym Report: {suite.name}" if suite is not None else "Gym Report")
    lines = [f"# {report_title}", ""]
    if suite is not None:
        lines.extend(_suite_markdown(suite))

    metadata = {
        "suite": None if suite is None else suite.name,
        "tasks": list(results),
    }
    (out / "metadata.json").write_text(json.dumps(metadata, indent=2, sort_keys=True))

    for task_name, result in results.items():
        task = _validate_task(task_name)
        task_dir = out / task
        task_dir.mkdir(exist_ok=True)
        result.df.to_csv(task_dir / "raw.csv", index=False)
        if task == "pipeline":
            result.delta().to_csv(task_dir / "pipeline_deltas.csv", index=False)

        leaderboard = gym_leaderboard(result, task=task, suite=suite)
        by_graph = gym_leaderboard(
            result,
            task=task,
            suite=suite,
            group_by=_graph_breakdown_columns(_analysis_frame(result, task=task)),
        )
        winners = gym_winners(result, task=task, suite=suite)
        counts = status_counts(result)
        failures = failed_rows(result)
        leaderboard.to_csv(task_dir / "leaderboard.csv", index=False)
        by_graph.to_csv(task_dir / "leaderboard_by_graph.csv", index=False)
        winners.to_csv(task_dir / "winners_by_regime.csv", index=False)
        counts.to_csv(task_dir / "status_counts.csv", index=False)
        failures.to_csv(task_dir / "failed_rows.csv", index=False)

        plots = gym_plots(result, task=task, suite=suite)
        plot_paths = []
        for name, fig in plots.items():
            path = plots_dir / f"{task}_{name}.png"
            fig.savefig(path, dpi=180, bbox_inches="tight")
            plt.close(fig)
            plot_paths.append(path)

        lines.extend(
            _task_markdown(task, leaderboard, winners, by_graph, counts, failures, plot_paths, out)
        )

    report_path = out / "report.md"
    report_path.write_text("\n".join(lines).rstrip() + "\n")
    return report_path