"""Staged evaluation pipeline with progress reporting. The analysis steps, their order and their seeds are those of ``ppi_benchmark_controls.run_framework`` (v1.0.0), so results are identical on the same inputs. What changes is the packaging. Each stage reports progress through a callback, results are returned as plain data (tables + chart-ready series) instead of matplotlib figures, a species stage and an interpretation stage are added, and all files land in one output directory that the web app zips for download. """ from __future__ import annotations import json import os import time from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Callable import numpy as np import pandas as pd from sklearn.metrics import precision_recall_curve, roc_auc_score, roc_curve from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler from threadpoolctl import threadpool_limits from . import models as _models from . import species as _species from .audit import audit_and_canonicalize, connected_components from .constants import ENGINE_VERSION, FRAMEWORK_VERSION, SCHEMA_VERSION from .features import aac_ctd_features, pair_matrix, scalar_pair_matrix from .interpret import interpret from .io import (InputError, InputIssue, cross_reference_issues, read_clusters, read_embeddings, read_pairs_with_issues, read_proteins_with_issues, read_species_map) from .models import build_model, classification_metrics, fit_predict from .negatives import (NEGATIVE_BUILDERS, balance_row, compute_protein_stats, endpoint_values, known_pair_set, sample_random_negatives) from .schema import to_jsonable from .splits import _proteins_of, leakage_checks, make_splits, partition_protein_pool from .uncertainty import benjamini_hochberg, component_bootstrap_ci, component_groups, component_jackknife_ci PRESETS = { "quick": {"models": ["lr", "mlp"], "n_boot": 200, "extra_negatives": False}, "standard": {"models": ["lr", "rf", "mlp"], "n_boot": 1000, "extra_negatives": True}, "full": {"models": ["lr", "rf", "xgb", "mlp"], "n_boot": 2000, "extra_negatives": True}, } STAGES = [("audit", "Auditing and canonicalizing", 4), ("negatives", "Constructing putative negatives", 8), ("splits", "Building protein-novelty splits", 4), ("features", "Computing protein representations", 6), ("novelty", "Training along the novelty curve", 32), ("controls", "Shortcut controls and revalidation", 22), ("uncertainty", "Component-aware uncertainty", 10), ("species", "Species diagnostics", 12), ("interpret", "Interpreting and packaging", 2)] _TOTAL = sum(w for *_, w in STAGES) @dataclass class RunConfig: preset: str = "quick" seed: int = 42 models: list[str] | None = None n_boot: int | None = None extra_negatives: bool | None = None test_fraction: float = 0.2 species_mode: str = "auto" # auto | off species_min_positive: int = 50 n_jobs: int = 4 embeddings_name: str | None = None embeddings_dataset: str = "mean" def resolved(self) -> dict: base = dict(PRESETS.get(self.preset, PRESETS["quick"])) for k in ("models", "n_boot", "extra_negatives"): if getattr(self, k) is not None: base[k] = getattr(self, k) return {**asdict(self), **base} @dataclass class Inputs: proteins: str | Path pairs: str | Path species: str | Path | None = None clusters: str | Path | None = None embeddings: str | Path | None = None mappings: dict = field(default_factory=dict) # {"proteins": {...}, "pairs": {...}, "species": {...}} labels: dict = field(default_factory=dict) # display names of the uploaded files class Progress: def __init__(self, callback: Callable | None): self.callback = callback self.done = 0.0 self.t0 = time.time() def stage(self, key: str, fraction: float = 0.0, message: str | None = None): offset = 0 for k, label, w in STAGES: if k == key: pct = (offset + w * min(max(fraction, 0), 1)) / _TOTAL if self.callback: self.callback({"stage": key, "stage_label": label, "stage_fraction": fraction, "progress": round(pct, 4), "message": message or label, "elapsed": round(time.time() - self.t0, 1)}) return offset += w def sub(self, key: str): return lambda fraction, message=None: self.stage(key, fraction, message) # --------------------------------------------------------------------------- # loading and validation # --------------------------------------------------------------------------- def load_inputs(inputs: Inputs): """Parse all input files. Returns (proteins, pairs, species_map, issues); raises InputError if anything blocks the run.""" issues: list[InputIssue] = [] try: proteins, iss = read_proteins_with_issues(inputs.proteins, inputs.mappings.get("proteins")) issues += iss pairs, iss = read_pairs_with_issues(inputs.pairs, inputs.mappings.get("pairs")) issues += iss except InputError as e: raise InputError(issues + e.issues) from None if not any(i.severity == "error" for i in issues): issues += cross_reference_issues(proteins, pairs) # line numbers were only needed for error messages; pandas deep-copies attrs into # every derived frame/row, which makes row-wise samplers dramatically slower pairs.attrs.clear() species_map = None if inputs.species: species_map = read_species_map(inputs.species, inputs.mappings.get("species")) elif "species" in proteins: species_map = dict(zip(proteins["protein_id"], proteins["species"])) if species_map is not None: species_map = {k: v for k, v in species_map.items() if v and v.lower() not in {"na", "nan", "none", "unknown", "unresolved", "unassigned", "?"}} covered = proteins["protein_id"].isin(set(species_map)).mean() n_species = len(set(species_map.values())) issues.append(InputIssue("species_coverage", "info" if covered > 0.95 else "warning", f"Species assigned for {covered:.1%} of proteins ({n_species} species)", "species")) if n_species < 2: issues.append(InputIssue("single_species", "info", "Only one species found: species diagnostics will be skipped.", "species")) if any(i.severity == "error" for i in issues): raise InputError(issues) return proteins[["protein_id", "sequence"]], pairs, species_map, issues def validate(inputs: Inputs) -> dict: """Audit only (no model training). Never raises for input problems; they are returned.""" try: proteins, pairs, species_map, issues = load_inputs(inputs) except InputError as e: return to_jsonable({"ok": False, "issues": [i.to_dict() for i in e.issues], "audit": None}) _, canon, audit = audit_and_canonicalize(proteins, pairs) issues = issues + _audit_issues(audit) pos = canon[canon["label"] == 1] if len(pos) < 20: issues.append(InputIssue("too_few_positives", "error", f"Only {len(pos)} positive pairs after cleaning; at least 20 are required.", "pairs")) sp_summary = None if species_map: sp = _species.species_series(proteins, species_map) sp_summary = _species.composition(proteins, pos, canon[canon["label"] == 0], sp) ok = not any(i.severity == "error" for i in issues) return to_jsonable({"ok": ok, "issues": [i.to_dict() for i in issues], "audit": audit, "species": sp_summary}) def _audit_issues(audit: dict) -> list[InputIssue]: iss, out = audit["issues"], [] if iss["duplicate_pairs_removed"]: out.append(InputIssue("duplicate_pairs", "warning", f"{iss['duplicate_pairs_removed']} duplicate pairs removed " f"({iss['reversed_duplicate_rows']} were reversed duplicates such as B–A after A–B)", "pairs")) if iss["self_pairs_removed"]: out.append(InputIssue("self_pairs", "warning", f"{iss['self_pairs_removed']} self-pairs (A–A) removed", "pairs")) if iss["conflicting_label_pairs_removed"]: out.append(InputIssue("conflicting_labels", "warning", f"{iss['conflicting_label_pairs_removed']} pairs had both label 0 and 1 and were removed", "pairs")) c = audit["canonical"] if c["identical_sequence_groups_size_gt1"]: out.append(InputIssue("identical_sequences", "info", f"{c['proteins_in_identical_sequence_groups']} proteins share an identical sequence " f"({c['identical_sequence_groups_size_gt1']} groups); each group is kept on one side of every split", "proteins")) return out # --------------------------------------------------------------------------- # chart-data helpers # --------------------------------------------------------------------------- def _curve(y, p, max_points: int = 120) -> dict: fpr, tpr, _ = roc_curve(y, p) prec, rec, _ = precision_recall_curve(y, p) def thin(a, b): idx = np.unique(np.linspace(0, len(a) - 1, min(len(a), max_points)).astype(int)) return [[round(float(a[i]), 4), round(float(b[i]), 4)] for i in idx] return {"roc": thin(fpr, tpr), "pr": thin(rec, prec)} def _score_hist(y, p, bins: int = 25) -> dict: edges = np.linspace(0, 1, bins + 1) return {"edges": edges.round(3).tolist(), "positive": np.histogram(p[y == 1], edges)[0].tolist(), "negative": np.histogram(p[y == 0], edges)[0].tolist()} def _int_distribution(values: np.ndarray, cap: int) -> list[int]: v = np.clip(values.astype(int), 0, cap) return np.bincount(v, minlength=cap + 1).tolist() def _network_sample(pos: pd.DataFrame, stats_tbl: pd.DataFrame, species_map, max_nodes: int = 450, seed: int = 42) -> dict: """A connected, readable slice of the positive graph for visualisation: BFS from the highest-degree protein of the largest components until ``max_nodes`` nodes.""" if pos.empty: return {"nodes": [], "edges": []} adj: dict[str, list[str]] = {} for a, b in zip(pos["protein_a"], pos["protein_b"]): adj.setdefault(a, []).append(b) adj.setdefault(b, []).append(a) comps = connected_components(pos) chosen: list[str] = [] seen: set = set() rng = np.random.default_rng(seed) for comp in comps: if len(chosen) >= max_nodes: break start = max(comp, key=lambda n: (len(adj[n]), n)) queue = [start] seen.add(start) while queue and len(chosen) < max_nodes: node = queue.pop(0) chosen.append(node) nb = [x for x in adj[node] if x not in seen] rng.shuffle(nb) for x in nb: seen.add(x) queue.append(x) keep = set(chosen) edges = [[a, b] for a, b in zip(pos["protein_a"], pos["protein_b"]) if a in keep and b in keep] comp_of = {n: i for i, c in enumerate(comps) for n in c} nodes = [{"id": n, "degree": int(stats_tbl.at[n, "positive_degree"]), "component": comp_of[n], "species": (species_map or {}).get(n)} for n in chosen] return {"nodes": nodes, "edges": edges, "total_nodes": len(adj), "total_edges": int(len(pos))} # --------------------------------------------------------------------------- # the run # --------------------------------------------------------------------------- def run_pipeline(inputs: Inputs, config: RunConfig | None = None, out_dir: str | Path | None = None, progress: Callable | None = None) -> dict: config = config or RunConfig() cfg = config.resolved() _models.N_JOBS = cfg["n_jobs"] with threadpool_limits(limits=max(1, int(cfg["n_jobs"]))): return _run(inputs, cfg, Path(out_dir) if out_dir else None, Progress(progress)) def _run(inputs: Inputs, cfg: dict, out: Path | None, prog: Progress) -> dict: t0 = time.time() seed, test_fraction, n_boot = cfg["seed"], cfg["test_fraction"], cfg["n_boot"] if out: for sub in ("tables", "splits", "negatives", "predictions"): (out / sub).mkdir(parents=True, exist_ok=True) def save(df: pd.DataFrame, rel: str, **kw): if out is not None: path = out / rel path.parent.mkdir(parents=True, exist_ok=True) df.to_csv(path, sep="\t" if rel.endswith(".tsv") else ",", index=kw.pop("index", False), **kw) # ---- 1. audit ------------------------------------------------------------------ prog.stage("audit", 0.0) proteins_raw, pairs_raw, species_map, input_issues = load_inputs(inputs) proteins, pairs, audit = audit_and_canonicalize(proteins_raw, pairs_raw) input_issues += _audit_issues(audit) save(pairs, "canonical_pairs.tsv") pos = pairs[pairs["label"] == 1].reset_index(drop=True) neg0 = pairs[pairs["label"] == 0].reset_index(drop=True) if len(pos) < 20: raise InputError([InputIssue("too_few_positives", "error", f"Only {len(pos)} positive pairs after cleaning; at least 20 are required.", "pairs")]) stats_tbl = compute_protein_stats(proteins, pos) save(stats_tbl, "tables/protein_statistics.csv", index=True) deg, length = stats_tbl["positive_degree"], stats_tbl["length"] deg_cap = int(max(5, np.percentile(deg, 99))) len_edges = np.unique(np.round(np.geomspace(max(10, length.min()), max(length.max(), 11), 40))) dataset_charts = { "degree_hist": _int_distribution(deg.to_numpy(), deg_cap), "degree_cap": deg_cap, "length_hist": {"edges": len_edges.tolist(), "counts": np.histogram(length, len_edges)[0].tolist()}, "network": _network_sample(pos, stats_tbl, species_map, seed=seed), } prog.stage("audit", 1.0, f"{audit['canonical']['n_pairs']:,} canonical pairs") known = known_pair_set(pos) supplied_negatives = len(neg0) > 0 sp = _species.species_series(proteins, species_map) if species_map else None use_species = bool(sp is not None and cfg["species_mode"] != "off" and (sp != _species.UNASSIGNED).any() and sp[sp != _species.UNASSIGNED].nunique() >= 2) # ---- 2. negative sets --------------------------------------------------------- prog.stage("negatives", 0.0) neg_sets = {"N1": sample_random_negatives(list(proteins["protein_id"]), known, len(pos), seed)} if not supplied_negatives: neg0 = neg_sets["N1"].copy() pairs = pd.concat([pos, neg0], ignore_index=True) neg_sets = {"N0": neg0, **neg_sets} names = ["N2", "N3", "N5"] + (["N2-exact", "N2-nearest"] if cfg["extra_negatives"] else []) for i, name in enumerate(names): prog.stage("negatives", (i + 1) / (len(names) + 2), f"Building {name}") neg_sets[name] = NEGATIVE_BUILDERS[name](pos, stats_tbl, known, seed) if use_species: neg_sets["N2-species"] = _species.species_matched_negatives(pos, stats_tbl, known, seed, sp) for name, df in neg_sets.items(): save(df, f"negatives/{name}.tsv") pos_deg = endpoint_values(pos, deg) balance = [] for name, df in neg_sets.items(): row = {"negative_set": name, "n_negatives": len(df), "positives_without_negative": df.attrs.get("n_skipped", 0)} row.update(balance_row(pos_deg, endpoint_values(df, deg))) row["length_smd"] = balance_row(endpoint_values(pos, length), endpoint_values(df, length))["smd"] balance.append(row) balance = pd.DataFrame(balance) save(balance, "tables/negative_set_balance.csv") degree_dists = {"positives": _int_distribution(pos_deg, deg_cap), **{k: _int_distribution(endpoint_values(v, deg), deg_cap) for k, v in neg_sets.items()}} # ---- 3. splits ------------------------------------------------------------------ prog.stage("splits", 0.0) clusters = read_clusters(inputs.clusters) if inputs.clusters else None splits = make_splits(pairs, proteins, seed, test_fraction, 0.2, clusters) id_to_group = dict(zip(proteins["protein_id"], proteins["sequence_group_id"])) split_rows = [] for name, (tr, te) in splits.items(): checks = leakage_checks(name, tr, te, id_to_group, clusters) save(tr, f"splits/{name}/train.tsv") save(te, f"splits/{name}/test.tsv") split_rows.append({"regime": name, "train_pairs": len(tr), "test_pairs": len(te), "train_positive": int(tr.label.sum()), "test_positive": int(te.label.sum()), "train_proteins": len(_proteins_of(tr)), "test_proteins": len(_proteins_of(te)), "test_components": len(connected_components(te)), **checks}) split_df = pd.DataFrame(split_rows) save(split_df, "tables/split_summary.csv") # ---- 4. representations ---------------------------------------------------------- prog.stage("features", 0.0, "AAC+CTD descriptors") reps = {"AAC+CTD": aac_ctd_features(proteins)} if inputs.embeddings: emb = read_embeddings(inputs.embeddings, cfg["embeddings_dataset"]) missing = set(proteins["protein_id"]) - set(emb["protein_id"]) if missing: raise InputError([InputIssue("embedding_missing_proteins", "error", f"Embedding file lacks {len(missing)} proteins", "embeddings", examples=sorted(missing)[:10])]) reps[cfg["embeddings_name"] or "Uploaded embedding"] = emb available = [] for m in cfg["models"]: if m == "xgb": try: import xgboost # noqa: F401 except ImportError: continue available.append(m) prog.stage("features", 1.0) all_preds, curves = [], {} # ---- 5. protein-novelty curve --------------------------------------------------- nov_rows = [] total = max(1, len(reps) * len(splits) * len(available)) step = 0 for rep_name, feats in reps.items(): for regime, (tr, te) in splits.items(): if te.empty or te.label.nunique() < 2: continue x_tr, y_tr = pair_matrix(tr, feats) x_te, y_te = pair_matrix(te, feats) for model in available: prog.stage("novelty", step / total, f"{rep_name} · {regime} · {model.upper()}") step += 1 p = fit_predict(model, x_tr, y_tr, x_te, seed) nov_rows.append({"representation": rep_name, "regime": regime, "model": model, "train_pairs": len(tr), "test_pairs": len(te), **classification_metrics(y_te, p)}) curves[f"novelty|{rep_name}|{regime}|{model}"] = _curve(y_te, p) all_preds.append(te[["protein_a", "protein_b", "label"]].assign( experiment="novelty_curve", representation=rep_name, regime=regime, negative_set="N0", model=model, probability=p)) novelty = pd.DataFrame(nov_rows) save(novelty, "tables/novelty_curve_metrics.csv") # ---- 6. topology/length-only controls ------------------------------------------ prog.stage("controls", 0.0, "Degree-only and length-only controls") pos_tr, pos_te = train_test_split(pos, test_size=test_fraction, random_state=seed) tr_sets, te_sets = {}, {} for name, df in neg_sets.items(): if len(df) < 5: continue n_tr, n_te = train_test_split(df, test_size=test_fraction, random_state=seed) tr_sets[name] = pd.concat([pos_tr, n_tr], ignore_index=True) te_sets[name] = pd.concat([pos_te, n_te], ignore_index=True) topo_rows = [] for feature, scalar in (("degree", deg), ("length", length)): for model in ("lr", "rf"): x_tr, y_tr = scalar_pair_matrix(tr_sets["N0"], scalar) scaler = StandardScaler().fit(x_tr) clf = build_model(model, seed, control=True).fit(scaler.transform(x_tr), y_tr) for name, te in te_sets.items(): x_te, y_te = scalar_pair_matrix(te, scalar) p = clf.predict_proba(scaler.transform(x_te))[:, 1] topo_rows.append({"feature": feature, "model": model, "trained_on": "N0", "evaluated_on": name, **classification_metrics(y_te, p)}) if model == "lr": curves[f"topology|{feature}|{name}"] = _curve(y_te, p) topology = pd.DataFrame(topo_rows) save(topology, "tables/topology_controls.csv") # ---- consistent-negative revalidation --------------------------------------- train_pool, test_pool = partition_protein_pool(proteins, test_fraction, seed) r3_tr, r3_te = splits["R3"] reval_names = ["N0", "N2", "N5"] + (["N2-species"] if use_species else []) reval_sets = {} for name in reval_names: if name == "N0": r0 = (tr_sets["N0"], te_sets["N0"]) r3 = (r3_tr, r3_te) else: r0 = (tr_sets[name], te_sets[name]) tr_pos, te_pos = r3_tr[r3_tr.label == 1], r3_te[r3_te.label == 1] if name == "N2-species": ntr = _species.species_matched_negatives(tr_pos.reset_index(drop=True), stats_tbl, known, seed, sp, train_pool) nte = _species.species_matched_negatives(te_pos.reset_index(drop=True), stats_tbl, known, seed, sp, test_pool) else: ntr = NEGATIVE_BUILDERS[name](tr_pos.reset_index(drop=True), stats_tbl, known, seed, train_pool) nte = NEGATIVE_BUILDERS[name](te_pos.reset_index(drop=True), stats_tbl, known, seed, test_pool) r3 = (pd.concat([tr_pos, ntr], ignore_index=True), pd.concat([te_pos, nte], ignore_index=True)) if not _proteins_of(r3[0]).isdisjoint(_proteins_of(r3[1])): raise RuntimeError(f"R3/{name}: protein disjointness violated after negative substitution") reval_sets[("R0", name)] = r0 reval_sets[("R3", name)] = r3 reval_rows, score_hists = [], {} total = max(1, len(reps) * len(reval_sets)) step = 0 for rep_name, feats in reps.items(): for (regime, name), (tr, te) in reval_sets.items(): prog.stage("controls", 0.2 + 0.8 * step / total, f"Consistent negatives · {rep_name} · {regime}/{name}") step += 1 if te.label.nunique() < 2 or tr.label.nunique() < 2: continue x_tr, y_tr = pair_matrix(tr, feats) x_te, y_te = pair_matrix(te, feats) p = fit_predict("mlp", x_tr, y_tr, x_te, seed) reval_rows.append({"representation": rep_name, "regime": regime, "negative_set": name, "model": "mlp", "train_pairs": len(tr), "test_pairs": len(te), **classification_metrics(y_te, p)}) curves[f"revalidation|{rep_name}|{regime}|{name}"] = _curve(y_te, p) score_hists[f"{rep_name}|{regime}|{name}"] = _score_hist(y_te, p) all_preds.append(te[["protein_a", "protein_b", "label"]].assign( experiment="consistent_negative", representation=rep_name, regime=regime, negative_set=name, model="mlp", probability=p)) revalidation = pd.DataFrame(reval_rows) save(revalidation, "tables/consistent_negative_revalidation.csv") # ---- 7. graph-aware uncertainty ------------------------------------------------ prog.stage("uncertainty", 0.0, f"Component bootstrap (n={n_boot}) and jackknife") preds = pd.concat(all_preds, ignore_index=True) unc_rows, comp_sizes = [], {} sel = preds[(preds.experiment == "consistent_negative") & (preds.regime == "R3")] groups_iter = list(sel.groupby(["representation", "negative_set"], sort=False)) for i, ((rep_name, name), g) in enumerate(groups_iter): prog.stage("uncertainty", i / max(len(groups_iter), 1), f"Uncertainty · {rep_name} · R3/{name}") g = g.reset_index(drop=True) y, p = g["label"].to_numpy(), g["probability"].to_numpy() groups = component_groups(g) row = {"regime": "R3", "negative_set": name, "representation": rep_name, "model": "mlp", "n_test": len(g), "n_components": len(groups), "largest_component_pairs": max(len(x) for x in groups), "roc_auc": roc_auc_score(y, p)} row.update(component_bootstrap_ci(y, p, groups, n_boot, seed)) row.update(component_jackknife_ci(y, p, groups)) unc_rows.append(row) comp_sizes[f"{rep_name}|{name}"] = sorted((len(x) for x in groups), reverse=True) uncertainty = pd.DataFrame(unc_rows) if len(uncertainty): uncertainty["boot_p_bh"] = benjamini_hochberg(uncertainty["boot_p"].tolist()) uncertainty["jack_p_bh"] = benjamini_hochberg(uncertainty["jack_p"].tolist()) uncertainty["above_chance_both_methods"] = ((uncertainty.roc_auc > 0.5) & (uncertainty.boot_p_bh < 0.05) & (uncertainty.jack_p_bh < 0.05)) save(uncertainty, "tables/component_aware_uncertainty.csv") if out is not None: preds.to_csv(out / "predictions" / "sample_level_predictions.csv.gz", index=False, compression="gzip") # ---- 8. species -------------------------------------------------------------------- species_res = None if use_species: prog.stage("species", 0.0, "Species composition and cross-species negatives") comp = _species.composition(proteins, pos, neg_sets["N0"], sp) cross = [_species.cross_species_row("positives", pos, sp)] + \ [_species.cross_species_row(k, v, sp) for k, v in neg_sets.items()] shortcut = _species.same_species_shortcut(te_sets, sp) within = _species.within_species(proteins, pairs, sp, reps, seed, test_fraction, cfg["species_min_positive"], ("lr", "mlp"), prog.sub("species")) save(pd.DataFrame(comp), "tables/species_composition.csv") save(pd.DataFrame(cross), "tables/species_cross_fraction.csv") save(pd.DataFrame(shortcut), "tables/species_same_species_shortcut.csv") save(within["metrics"], "tables/species_within_metrics.csv") save(within["balance"], "tables/species_within_balance.csv") species_res = {"composition": comp, "cross_species": cross, "same_species_shortcut": shortcut, "within_metrics": within["metrics"], "within_balance": within["balance"], "skipped": within["skipped"]} else: prog.stage("species", 1.0, "No species information: skipped") # ---- 9. interpretation + results document --------------------------------------- prog.stage("interpret", 0.0) results = { "schema_version": SCHEMA_VERSION, "engine_version": ENGINE_VERSION, "framework_version": FRAMEWORK_VERSION, "generated_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "runtime_seconds": round(time.time() - t0, 1), "config": {k: v for k, v in cfg.items() if k != "n_jobs"}, "inputs": {"files": inputs.labels, "supplied_negatives": supplied_negatives, "species": use_species, "clusters": bool(clusters), "representations": list(reps), "models": available}, "input_issues": [i.to_dict() for i in input_issues], "audit": audit, "dataset_charts": dataset_charts, "balance": balance, "degree_distributions": degree_dists, "splits": split_df, "novelty": novelty, "topology": topology, "revalidation": revalidation, "uncertainty": uncertainty, "component_sizes": comp_sizes, "curves": curves, "score_histograms": score_hists, "species": species_res, } results = to_jsonable(results) results.update(interpret(results)) if out is not None: (out / "results.json").write_text(json.dumps(results, separators=(",", ":"))) from .report import write_report_html write_report_html(results, out / "report.html") prog.stage("interpret", 1.0, "Done") return results def default_threads() -> int: return max(1, min(4, (os.cpu_count() or 2) // 2))