Hugging Face's logo

Spaces:
Tang-Lab
/
benchdash
Running

App Files Community
benchdash / app.py
josephsoo's picture
josephsoo
Link benchmark preprint from dashboard
8f498bb
Raw History Blame Contribute Delete
137 kB
from __future__ import annotations
import re
from pathlib import Path
from typing import Iterable, Sequence
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from dash import Dash, Input, Output, State, dash_table, dcc, html
from flask import abort, send_from_directory
from plotly.colors import sample_colorscale
from plotly.subplots import make_subplots
DATA_DIR = Path(__file__).resolve().parent / "data"
PAPER_MODEL_ORDER = [
"blend",
"blend_ndt",
"cebra",
"dnn",
"dpad",
"gpfa",
"gru",
"langevinflow_ccn",
"ldns",
"lfads_torch",
"lstm",
"marble",
"mint",
"neds",
"neds_pretrained",
"neuro_behavior_conditioning",
"pca",
"rnn",
"smc_rnns",
"svc",
"tndm",
"torchdfine",
"xg",
]
DISPLAY_NAMES = {
"blend": "BLEND-LFADS",
"blend_ndt": "BLEND-NDT",
"cebra": "CEBRA",
"dnn": "DNN",
"dpad": "DPAD",
"gpfa": "GPFA",
"gru": "GRU",
"langevinflow_ccn": "LangevinFlow",
"ldns": "LDNS",
"lfads_torch": "AutoLFADS",
"lstm": "LSTM",
"marble": "MARBLE",
"mint": "MINT",
"neds": "NEDS",
"neds_pretrained": "NEDS-pt",
"neuro_behavior_conditioning": "mVAE",
"pca": "PCA",
"rnn": "RNN",
"smc_rnns": "SMC-RNN",
"svc": "SVM/SVR",
"tndm": "TNDM",
"torchdfine": "DFINE",
"xg": "XGBoost",
}
# Canonical dataset names from Supplementary Table 1. Task names
# belong in descriptions, not in alternate dataset labels.
DATASET_LABELS = {
"monkey": "Macaque center-out reaching",
"allen_neuropixels": "Allen Neuropixels",
"speech": "Attempted speech",
"mc_pacman": "MC PacMan",
"ratinabox": "RatInABox",
}
DATASET_TICK_LABELS = {
"monkey": "Macaque<br>center-out reaching",
"allen_neuropixels": "Allen<br>Neuropixels",
"speech": "Attempted<br>speech",
"mc_pacman": "MC<br>PacMan",
"ratinabox": "RatInABox",
}
DATASET_DESCRIPTIONS = {
"monkey": "Two-dimensional hand-position regression from macaque neural population activity.",
"allen_neuropixels": "Eight-class drifting-grating orientation classification from Allen Neuropixels units.",
"speech": "Eight-class attempted-word classification from threshold-crossing features.",
"mc_pacman": "Continuous force regression from motor-cortical population activity.",
"ratinabox": "Two-dimensional position regression from simulated place, head-direction and speed cells.",
}
DATASETS = list(DATASET_LABELS)
MODELS = PAPER_MODEL_ORDER.copy()
MODEL_INDEX = {model: index for index, model in enumerate(MODELS)}
# Figure 2 color semantics.
PREDICTION_COLOR = "#1565C0"
ROBUSTNESS_COLOR = "#2E7D32"
COMPUTE_COLOR = "#E65100"
# Figure 3 uses a green-blue scale; Figures 4 and 5 use purple scales.
CONSISTENCY_COLOR = "#007C91"
FEATURE_COLOR = "#6A51A3"
TRIAL_COLOR = "#6A1B9A"
NEGATIVE_COLOR = "#B35806"
TEXT_COLOR = "#17202A"
MUTED_COLOR = "#607080"
GRID_COLOR = "#E8EDF1"
PREDICTION_SCALE = [[0.0, "#F3F8FD"], [1.0, PREDICTION_COLOR]]
CONSISTENCY_SCALE = [[0.0, "#F1FAF8"], [1.0, CONSISTENCY_COLOR]]
FEATURE_SCALE = [[0.0, "#F7F2FA"], [1.0, FEATURE_COLOR]]
TRIAL_SCALE = [[0.0, "#F8F2FA"], [1.0, TRIAL_COLOR]]
ATTRIBUTION_BIN_ORDER = ["Top", "Middle", "Bottom"]
ATTRIBUTION_BIN_COLORS = {
"Top": "#238B45",
"Middle": "#74C476",
"Bottom": "#D9F0A3",
}
FEATURE_GROUP_COLORS = {
"monkey": {"Recorded": "#0072B2", "Synthetic control": "#BDBDBD"},
"speech": {"Recorded": "#009E73", "Synthetic control": "#BDBDBD"},
"mc_pacman": {"Recorded": "#D55E00", "Synthetic control": "#BDBDBD"},
"ratinabox": {
"Place": "#CC79A7",
"Head direction": "#56B4E9",
"Speed": "#E69F00",
},
}
MODEL_COLORS = {
model: color
for model, color in zip(
MODELS,
sample_colorscale("Turbo", np.linspace(0.04, 0.96, len(MODELS))),
)
}
CPU_ONLY_MODELS = {"gpfa", "mint", "pca", "svc", "xg"}
CONSISTENCY_ELIGIBLE = {
"blend",
"cebra",
"dpad",
"gpfa",
"ldns",
"lfads_torch",
"marble",
"neuro_behavior_conditioning",
"pca",
"smc_rnns",
"tndm",
"torchdfine",
}
FEATURE_ELIGIBLE = set(MODELS) - {"marble"}
TRIAL_ELIGIBLE = {
"blend",
"blend_ndt",
"cebra",
"dnn",
"gpfa",
"gru",
"langevinflow_ccn",
"ldns",
"lfads_torch",
"lstm",
"marble",
"neds",
"neds_pretrained",
"pca",
"rnn",
"smc_rnns",
"tndm",
}
CONDITION_LABELS = {
"monkey": {
"0": "90°",
"1": "45°",
"2": "0°",
"3": "315°",
"4": "270°",
"5": "225°",
"6": "180°",
"7": "135°",
},
"allen_neuropixels": {
str(index): f"{angle}°"
for index, angle in enumerate([0, 45, 90, 135, 180, 225, 270, 315])
},
"speech": {
"0": "rest",
"1": "ban",
"2": "choice",
"3": "day",
"4": "feel",
"5": "kite",
"6": "though",
"7": "were",
},
}
# Exact reach-direction palette used by Figure 3.
DIRECTION_PALETTE = [
"#B23AEE",
"#3B1C54",
"#2DD4F6",
"#289285",
"#E3D724",
"#00A65A",
"#5B8FF9",
"#F97316",
]
DIRECTION_LEGEND_ORDER = [2, 1, 0, 7, 6, 5, 4, 3]
DIRECTION_LEGEND_LABELS = ["0°", "45°", "90°", "135°", "180°", "225°", "270°", "315°"]
SPEECH_PALETTE = {str(index): color for index, color in enumerate(DIRECTION_PALETTE)}
ALLEN_PALETTE = {str(index): color for index, color in enumerate(DIRECTION_PALETTE)}
MC_PROFILE_LABELS = {
1: "Slow ascending ramp",
2: "Slow descending ramp",
3: "Fast ascending ramp",
4: "Fast descending ramp",
5: "Sine, 0.25 Hz",
6: "Sine, 1 Hz",
7: "Sine, 3 Hz",
8: "Chirp, 0–3 Hz",
}
RATINABOX_SCALE = [
[0.0, "#440154"],
[0.25, "#3B528B"],
[0.5, "#21918C"],
[0.75, "#5EC962"],
[1.0, "#FDE725"],
]
TABLE_LABELS = {
"method": "Method",
"workflow": "Prediction workflow",
"hardware": "Primary hardware",
"task_score": "Held-out task score",
"prediction_percentile": "Within-dataset percentile",
"robustness_auc": "Area under task-score-versus-noise curve",
"training_time_sec": "Training time (s)",
"inference_time_sec": "Inference time (s)",
"peak_ram_gb": "Peak RAM (GB)",
"peak_vram_gb": "Peak GPU memory (GB)",
"unperturbed_score": "Unperturbed score",
"highest_noise_score": "Score at λ = 0.8",
"average_noisy_score": "Mean score across λ",
"latent_consistency_r2": "Latent-consistency R²",
"latent_dim": "Latent dimensions",
"n_recordings": "Recordings",
"n_pairwise": "Directional pairs",
"validation_target": "Validation target",
"validation_metric": "Validation metric",
"validation_score": "Validation score",
"corrupted_trial_auc": "Corrupted-trial ROC-AUC",
"mixed_full": "Mixed trials",
"data_shapley": "After trial-value removal",
"recovery": "Recovery ΔR²",
"target_only": "Current session only",
"all_sessions": "All-session pooling",
"historical_selected": "Trial-value historical selection",
}
NUMERIC_COLUMNS = {
"task_score",
"prediction_percentile",
"robustness_auc",
"training_time_sec",
"inference_time_sec",
"peak_ram_gb",
"peak_vram_gb",
"unperturbed_score",
"highest_noise_score",
"average_noisy_score",
"latent_consistency_r2",
"latent_dim",
"n_recordings",
"n_pairwise",
"validation_score",
"shap_mean_value",
"shap_median_value",
"shap_min_value",
"shap_max_value",
"shap_fraction_positive",
"shap_fraction_negative",
"corrupted_trial_auc",
"iterations",
"final_error",
"perturbation_fraction",
"rotation_angle_deg",
"shapley_mean_value",
"shapley_median_value",
"shapley_min_value",
"shapley_max_value",
"shapley_fraction_positive",
"shapley_fraction_negative",
"mixed_full",
"data_shapley",
"oracle",
"recovery",
"target_only",
"all_sessions",
"historical_selected",
}
DOWNLOADABLE_FILES = {
"dataset_overview.csv",
"dataset_example_neural.csv",
"dataset_example_targets.csv",
"dataset_targets.csv",
"feature_example_raster.csv",
"clean_prediction_summary.csv",
"robustness_summary.csv",
"consistency_summary.csv",
"scalability_summary.csv",
"neuron_shap_summary.csv",
"neuron_attributions.csv",
"trial_shapley_summary.csv",
"trial_shapley_retrain_summary.csv",
"trial_historical_trajectories.csv",
}
TRIAL_HISTORICAL_TRAJECTORY_COLUMNS = [
"model",
"target_session",
"trial_index",
"trial_id",
"direction_index",
"direction_label",
"time_index",
"target_x",
"target_y",
"current_only_x",
"current_only_y",
"historical_selected_x",
"historical_selected_y",
"current_only_r2",
"historical_selected_r2",
]
def load_csv(name: str, **read_csv_kwargs) -> pd.DataFrame:
path = DATA_DIR / name
if not path.exists():
raise FileNotFoundError(f"Missing dashboard data: {path}")
return pd.read_csv(path, **read_csv_kwargs)
def load_historical_trajectories() -> pd.DataFrame:
frame = load_csv("trial_historical_trajectories.csv")
if list(frame.columns) != TRIAL_HISTORICAL_TRAJECTORY_COLUMNS:
raise ValueError(
"trial_historical_trajectories.csv has an invalid schema. "
f"Expected exactly: {TRIAL_HISTORICAL_TRAJECTORY_COLUMNS}"
)
if frame.empty:
raise ValueError("trial_historical_trajectories.csv must contain the Figure 5e RNN example.")
if set(frame["model"].astype(str)) != {"rnn"}:
raise ValueError("trial_historical_trajectories.csv must contain only the Figure 5e RNN example.")
numeric_columns = [
"trial_index",
"direction_index",
"time_index",
"target_x",
"target_y",
"current_only_x",
"current_only_y",
"historical_selected_x",
"historical_selected_y",
"current_only_r2",
"historical_selected_r2",
]
for column in numeric_columns:
values = pd.to_numeric(frame[column], errors="coerce")
if values.isna().any():
raise ValueError(
f"trial_historical_trajectories.csv contains a non-numeric or missing {column} value."
)
if not np.isfinite(values.to_numpy(dtype=float)).all():
raise ValueError(
f"trial_historical_trajectories.csv contains a non-finite {column} value."
)
frame[column] = values
integer_columns = ["trial_index", "direction_index", "time_index"]
for column in integer_columns:
if not np.allclose(frame[column], np.round(frame[column])):
raise ValueError(f"trial_historical_trajectories.csv requires integer {column} values.")
frame[column] = frame[column].astype(int)
if not frame["direction_index"].between(0, 7).all():
raise ValueError("trial_historical_trajectories.csv direction_index values must be in [0, 7].")
if set(frame["direction_index"]) != set(DIRECTION_LEGEND_ORDER):
raise ValueError("trial_historical_trajectories.csv must include all eight reach directions.")
if frame["target_session"].astype(str).nunique() != 1:
raise ValueError("trial_historical_trajectories.csv must contain one target session.")
if frame["current_only_r2"].nunique() != 1 or frame["historical_selected_r2"].nunique() != 1:
raise ValueError("Figure 5e R² values must be constant across trajectory rows.")
trial_metadata = frame.groupby("trial_index").agg(
trial_ids=("trial_id", "nunique"),
directions=("direction_index", "nunique"),
direction_labels=("direction_label", "nunique"),
)
if (trial_metadata != 1).any().any():
raise ValueError("Each Figure 5e trial must have one ID and one reach-direction label.")
if frame.duplicated(["trial_index", "time_index"]).any():
raise ValueError("Figure 5e contains duplicate trial/time rows.")
return frame.sort_values(["direction_index", "trial_index", "time_index"]).reset_index(drop=True)
dataset_overview = load_csv("dataset_overview.csv")
dataset_example_neural = load_csv("dataset_example_neural.csv")
dataset_example_targets = load_csv("dataset_example_targets.csv")
dataset_targets = load_csv(
"dataset_targets.csv",
dtype={"target_label": "string"},
low_memory=False,
)
feature_example_raster = load_csv("feature_example_raster.csv")
prediction = load_csv("clean_prediction_summary.csv")
robustness = load_csv("robustness_summary.csv")
consistency = load_csv("consistency_summary.csv")
scalability = load_csv("scalability_summary.csv")
neuron_shap = load_csv("neuron_shap_summary.csv")
neuron_attributions = load_csv("neuron_attributions.csv")
trial_shapley = load_csv("trial_shapley_summary.csv")
trial_retrain = load_csv("trial_shapley_retrain_summary.csv")
trial_historical_trajectories = load_historical_trajectories()
latent_samples = load_csv("latent_samples.csv")
latent_trajectories = load_csv("latent_trajectories.csv")
def present_rows(df: pd.DataFrame) -> pd.DataFrame:
if df.empty or "status" not in df.columns:
return df.copy()
return df[df["status"].fillna("").astype(str).str.lower() == "present"].copy()
def active_rows(df: pd.DataFrame) -> pd.DataFrame:
if df.empty or "is_active_model" not in df.columns:
return df.copy()
mask = df["is_active_model"].fillna(False).astype(str).str.lower().isin({"true", "1", "yes"})
return df[mask].copy()
def model_label(model: object) -> str:
if model is None or pd.isna(model):
return ""
return DISPLAY_NAMES.get(str(model), str(model))
def dataset_model_label(model: object, dataset: str) -> str:
if str(model) == "svc":
return "SVM" if dataset in {"allen_neuropixels", "speech"} else "SVR"
return model_label(model)
def selected_models(models: Sequence[str] | None) -> list[str]:
if not models:
return MODELS.copy()
chosen = set(models)
return [model for model in MODELS if model in chosen]
def filter_models(df: pd.DataFrame, models: Sequence[str] | None) -> pd.DataFrame:
if df.empty or "model" not in df.columns:
return df.copy()
return df[df["model"].astype(str).isin(selected_models(models))].copy()
def add_method_columns(df: pd.DataFrame) -> pd.DataFrame:
out = df.copy()
if "model" in out.columns:
out["method"] = out["model"].map(model_label)
out["model_order"] = out["model"].map(MODEL_INDEX).fillna(len(MODELS)).astype(int)
return out
def records(df: pd.DataFrame) -> list[dict]:
clean = df.astype(object).where(pd.notna(df), None)
return clean.to_dict("records")
def round_numeric(df: pd.DataFrame, digits: int = 4) -> pd.DataFrame:
out = df.copy()
for column in NUMERIC_COLUMNS.intersection(out.columns):
out[column] = pd.to_numeric(out[column], errors="coerce").round(digits)
return out
def metric_name(metric: object, *, held_out: bool = False) -> str:
text = "" if metric is None or pd.isna(metric) else str(metric).lower()
prefix = "Held-out " if held_out else ""
if text == "r2":
return f"{prefix}R²"
if text == "accuracy":
return f"{prefix}accuracy"
return f"{prefix}{text or 'task score'}"
def value_text(value: object, digits: int = 3) -> str:
if value is None or pd.isna(value):
return "Unavailable"
number = float(value)
if abs(number) >= 1000:
return f"{number:,.0f}"
return f"{number:.{digits}f}"
def parse_float_list(value: object) -> list[float]:
if value is None or pd.isna(value):
return []
values: list[float] = []
for part in str(value).split(";"):
try:
values.append(float(part.strip()))
except ValueError:
raise ValueError(f"Malformed numeric sequence in dashboard data: {value!r}") from None
return values
def aggregate_prediction_order() -> list[str]:
frame = prediction.copy()
frame["score"] = pd.to_numeric(frame["score"], errors="coerce")
grid = pd.MultiIndex.from_product([MODELS, DATASETS], names=["model", "dataset"]).to_frame(index=False)
values = frame[["model", "dataset", "score"]].drop_duplicates(["model", "dataset"])
grid = grid.merge(values, on=["model", "dataset"], how="left")
grid["rank"] = grid.groupby("dataset")["score"].rank(method="average", ascending=False)
grid["rank"] = grid["rank"].fillna(len(MODELS))
mean_rank = grid.groupby("model", as_index=False)["rank"].mean()
mean_rank["model_order"] = mean_rank["model"].map(MODEL_INDEX)
return mean_rank.sort_values(["rank", "model_order"])["model"].tolist()
FIGURE_MODEL_ORDER = aggregate_prediction_order()
def column_defs(columns: Iterable[str]) -> list[dict]:
definitions = []
for column in columns:
item = {"name": TABLE_LABELS.get(column, column), "id": column}
if column in NUMERIC_COLUMNS:
item["type"] = "numeric"
definitions.append(item)
return definitions
def dataframe_table(
table_id: str,
*,
page_size: int = 12,
max_height: str = "620px",
) -> dash_table.DataTable:
return dash_table.DataTable(
id=table_id,
columns=[],
data=[],
page_size=page_size,
sort_action="native",
sort_mode="multi",
filter_action="native",
cell_selectable=False,
style_as_list_view=True,
fixed_rows={"headers": True},
tooltip_delay=250,
tooltip_duration=None,
style_table={"overflowX": "auto", "overflowY": "auto", "maxHeight": max_height},
style_header={
"backgroundColor": "#F3F6F8",
"fontWeight": "700",
"border": "0",
"borderBottom": "1px solid #CFD8DF",
"color": "#26323F",
},
style_cell={
"fontFamily": "Arial, Helvetica, sans-serif",
"fontSize": "13px",
"padding": "9px 11px",
"textAlign": "left",
"minWidth": "92px",
"maxWidth": "300px",
"whiteSpace": "normal",
"height": "auto",
"border": "0",
"borderBottom": "1px solid #EDF1F4",
},
style_cell_conditional=[
{"if": {"column_id": column}, "textAlign": "right"}
for column in NUMERIC_COLUMNS
],
style_data_conditional=[
{"if": {"row_index": "odd"}, "backgroundColor": "#FBFCFD"},
],
)
def panel(
title: str,
*children,
subtitle: str | None = None,
class_name: str = "",
eyebrow: str | None = None,
) -> html.Section:
heading: list = []
if eyebrow:
heading.append(html.Div(eyebrow, className="section-eyebrow"))
heading.append(html.H2(title))
if subtitle:
heading.append(html.P(subtitle, className="panel-subtitle"))
classes = "panel" if not class_name else f"panel {class_name}"
return html.Section([html.Div(heading, className="panel-heading"), *children], className=classes)
def details_table(summary: str, table: dash_table.DataTable) -> html.Details:
return html.Details(
[html.Summary(summary), html.Div(table, className="details-body")],
className="details-table",
)
def metric_card(label: str, value: str, detail: str | None = None, accent: str = "") -> html.Div:
children = [
html.Div(label, className="metric-label"),
html.Div(value, className="metric-value"),
]
if detail:
children.append(html.Div(detail, className="metric-detail"))
classes = "metric-card" if not accent else f"metric-card metric-card-{accent}"
return html.Div(children, className=classes)
def source_link(filename: str, label: str = "Download CSV") -> html.A:
return html.A(
label,
href=f"/download/{filename}",
className="source-link",
target="_blank",
rel="noopener noreferrer",
)
def graph_box(graph_id: str, label: str, *, class_name: str = "") -> html.Div:
classes = "graph-box" if not class_name else f"graph-box {class_name}"
return html.Div(
dcc.Graph(
id=graph_id,
config={
"displaylogo": False,
"responsive": True,
"toImageButtonOptions": {"format": "png", "scale": 2},
},
),
className=classes,
role="region",
**{"aria-label": label},
)
def figure_layout(
fig: go.Figure,
*,
height: int = 430,
legend_below: bool = False,
) -> go.Figure:
legend = dict(
orientation="h",
yanchor="bottom",
y=1.02,
xanchor="left",
x=0,
font=dict(size=11),
)
if legend_below:
legend.update(yanchor="top", y=-0.18)
fig.update_layout(
height=height,
paper_bgcolor="#FFFFFF",
plot_bgcolor="#FFFFFF",
margin=dict(l=54, r=28, t=58, b=58 if not legend_below else 105),
font=dict(family="Arial, Helvetica, sans-serif", size=13, color=TEXT_COLOR),
title=dict(
font=dict(size=16, color=TEXT_COLOR),
x=0.02,
xanchor="left",
pad=dict(l=6, b=8),
),
legend=legend,
hoverlabel=dict(
align="left",
bgcolor="#FFFFFF",
bordercolor="#607080",
font=dict(
family="Arial, Helvetica, sans-serif",
size=12,
color=TEXT_COLOR,
),
),
)
fig.update_xaxes(
showgrid=True,
gridcolor=GRID_COLOR,
zerolinecolor="#CDD6DD",
automargin=True,
)
fig.update_yaxes(
showgrid=True,
gridcolor=GRID_COLOR,
zerolinecolor="#CDD6DD",
automargin=True,
)
return fig
def heatmap_layout(fig: go.Figure, *, height: int) -> go.Figure:
figure_layout(fig, height=height)
fig.update_layout(
margin=dict(l=54, r=28, t=116, b=50),
title=dict(y=0.985, yanchor="top", pad=dict(b=12)),
)
fig.update_xaxes(tickangle=0, tickfont=dict(size=10), automargin=True)
return fig
def empty_figure(message: str, *, height: int = 360) -> go.Figure:
fig = go.Figure()
fig.add_annotation(
text=message,
x=0.5,
y=0.5,
xref="paper",
yref="paper",
showarrow=False,
align="center",
font=dict(size=14, color=MUTED_COLOR),
)
fig.update_xaxes(visible=False)
fig.update_yaxes(visible=False)
return figure_layout(fig, height=height)
def prediction_workflow(model: str, decoder: object, status: object) -> str:
if str(status).lower() != "present":
return "Unavailable"
decoder_name = "" if decoder is None or pd.isna(decoder) else str(decoder)
native_prediction_decoders = {
"native",
"dnn",
"gru",
"lstm",
"mint_pipeline",
"rnn",
"neds_e2e",
"svc",
"svr",
"xgboost_classification",
"xgboost_regression",
}
if decoder_name in native_prediction_decoders:
return "Native prediction output"
# The manuscript distinguishes LDNS task families: continuous
# prediction uses the method recipe's ridge mapping on reconstructed rates
# (alpha = 1e-6), while classification uses the standard logistic readout.
if decoder_name in {"ridge", "logistic", "ldns_rate_sklearn_logistic"}:
return "Shared linear readout"
if decoder_name in {"knn", "ole", "ldns_rate_sklearn_ridge"}:
return "Author-style task readout"
raise ValueError(f"Unrecognized prediction decoder for {model}: {decoder_name!r}")
def prediction_percentiles() -> pd.DataFrame:
frame = present_rows(prediction)[["model", "dataset", "score"]].copy()
frame["score"] = pd.to_numeric(frame["score"], errors="coerce")
frame["prediction_percentile"] = (
frame.groupby("dataset")["score"].rank(method="average", pct=True) * 100.0
)
return frame
def overview_frame(dataset: str, models: Sequence[str] | None) -> pd.DataFrame:
chosen = selected_models(models)
base = add_method_columns(pd.DataFrame({"model": chosen}))
base["method"] = base["model"].map(lambda model: dataset_model_label(model, dataset))
pred = prediction[prediction["dataset"].astype(str) == str(dataset)].copy()
pred["score"] = pd.to_numeric(pred["score"], errors="coerce")
pred = pred.merge(
prediction_percentiles()[["model", "dataset", "prediction_percentile"]],
on=["model", "dataset"],
how="left",
)
pred = pred[
[
"model",
"status",
"metric",
"decoder",
"score",
"prediction_percentile",
"n_train_trials",
"n_test_trials",
"n_neurons",
]
].rename(columns={"score": "task_score", "status": "prediction_status_raw"})
rob = present_rows(robustness)
rob = rob[rob["dataset"].astype(str) == str(dataset)][["model", "raw_auc"]].copy()
rob["raw_auc"] = pd.to_numeric(rob["raw_auc"], errors="coerce")
rob = rob.rename(columns={"raw_auc": "robustness_auc"})
scale = present_rows(scalability)
scale = scale[scale["dataset"].astype(str) == str(dataset)][
["model", "training_time_sec", "inference_time_sec", "peak_ram_gb", "peak_vram_gb"]
].copy()
for column in ["training_time_sec", "inference_time_sec", "peak_ram_gb", "peak_vram_gb"]:
scale[column] = pd.to_numeric(scale[column], errors="coerce")
frame = base.merge(pred, on="model", how="left")
frame = frame.merge(rob, on="model", how="left")
frame = frame.merge(scale, on="model", how="left")
frame["workflow"] = frame.apply(
lambda row: prediction_workflow(row["model"], row.get("decoder"), row.get("prediction_status_raw")),
axis=1,
)
frame.loc[frame["model"].isin(CPU_ONLY_MODELS), "peak_vram_gb"] = np.nan
return round_numeric(frame)
def dataset_cards(selected_dataset: str) -> list[html.Article]:
cards: list[html.Article] = []
ordered = dataset_overview.set_index("dataset").reindex(DATASETS).reset_index()
for row in ordered.itertuples(index=False):
classes = "dataset-card"
if str(row.dataset) == str(selected_dataset):
classes += " dataset-card-selected"
cards.append(
html.Article(
[
html.Div(str(row.dataset_name), className="dataset-card-title"),
html.Div(
f"{row.species} · {row.task}",
className="dataset-card-context",
),
html.Div(
[
html.Div(
[
html.Span(
"Primary array",
className="dataset-fact-label",
title="Trials × time bins × features",
),
html.Strong(str(row.array_shape)),
]
),
html.Div(
[
html.Span("Target", className="dataset-fact-label"),
html.Strong(f"{row.target} · {row.score}"),
]
),
html.Div(
[
html.Span("Sampling", className="dataset-fact-label"),
html.Strong(f"{float(row.bin_ms):g} ms bins"),
]
),
html.Div(
[
html.Span("Consistency cohort", className="dataset-fact-label"),
html.Strong(str(row.recordings)),
]
),
],
className="dataset-facts",
),
html.A(
str(row.source_label),
href=str(row.source_url),
target="_blank",
rel="noopener noreferrer",
className="dataset-source",
),
],
className=classes,
)
)
return cards
def target_space_graph(figure: go.Figure, label: str) -> html.Div:
return html.Div(
dcc.Graph(
id="dataset-target-trajectory",
figure=figure,
config={"displaylogo": False, "responsive": True},
),
className="target-space-graph",
role="img",
**{"aria-label": label},
)
def target_space_legend(items: Sequence[tuple[str, str]]) -> html.Div:
return html.Div(
[
html.Span(
[
html.Span(
className="target-legend-swatch",
style={"backgroundColor": color},
),
label,
],
className="target-legend-item",
)
for label, color in items
],
className="target-space-legend",
)
def target_class_space(dataset: str, frame: pd.DataFrame) -> html.Div:
counts = frame.groupby("condition_id", sort=True).size()
example_class = int(frame.loc[frame["is_example"], "condition_id"].iloc[0])
cards = []
for raw_class, count in counts.items():
class_id = int(raw_class)
color = DIRECTION_PALETTE[class_id % len(DIRECTION_PALETTE)]
label = condition_label(dataset, class_id)
selected = class_id == example_class
if dataset == "allen_neuropixels":
glyph = html.Div(
[html.Span(className="orientation-arrow-head")],
className="orientation-arrow",
style={
"backgroundColor": color,
"color": color,
"transform": f"rotate({-class_id * 45}deg)",
},
)
else:
glyph = html.Div(
label,
className="speech-target-word",
style={"color": color},
)
cards.append(
html.Div(
[
html.Div(glyph, className="target-class-glyph"),
(
html.Div(label, className="target-class-label")
if dataset == "allen_neuropixels"
else None
),
html.Div(f"{int(count)} trials", className="target-class-count"),
html.Span("Example", className="target-example-badge") if selected else None,
],
className=(
"target-class-card target-class-card-selected"
if selected
else "target-class-card"
),
style={"borderTopColor": color},
)
)
context = (
"Eight stimulus orientations"
if dataset == "allen_neuropixels"
else "Seven attempted words and rest"
)
return html.Div(
[
html.Div(
[
html.Strong("Target space"),
html.Span(f"{len(frame):,} trials · {context}"),
],
className="dataset-viz-heading",
),
html.Div(cards, className="target-class-grid"),
],
className="target-space-content",
)
def separated_trajectory_values(
frame: pd.DataFrame,
x_column: str,
y_column: str,
) -> tuple[list[float | None], list[float | None]]:
x_values: list[float | None] = []
y_values: list[float | None] = []
for _, trial in frame.groupby("trial_index", sort=False):
trial = trial.sort_values("time_index")
x_values.extend(trial[x_column].astype(float).tolist())
y_values.extend(trial[y_column].astype(float).tolist())
x_values.append(None)
y_values.append(None)
return x_values, y_values
def add_linked_target_traces(
figure: go.Figure,
x_values: Sequence[float],
y_values: Sequence[float],
customdata: np.ndarray,
hovertemplate: str,
*,
base_width: float,
progress_width: float,
halo_width: float | None = None,
) -> None:
x_values = list(x_values)
y_values = list(y_values)
if halo_width is not None:
figure.add_trace(
go.Scatter(
x=x_values,
y=y_values,
mode="lines",
line=dict(color="#FFFFFF", width=halo_width),
opacity=0.9,
hoverinfo="skip",
showlegend=False,
)
)
figure.add_trace(
go.Scatter(
x=x_values,
y=y_values,
mode="lines",
line=dict(color="#102A3A", width=base_width),
opacity=0.34,
customdata=customdata,
meta={"benchdash_role": "example_trajectory"},
hovertemplate=hovertemplate,
showlegend=False,
)
)
figure.add_trace(
go.Scatter(
x=[x_values[0]],
y=[y_values[0]],
mode="lines",
line=dict(color="#102A3A", width=progress_width),
hoverinfo="skip",
meta={"benchdash_role": "linked_target_progress"},
showlegend=False,
)
)
figure.add_trace(
go.Scatter(
x=[x_values[0]],
y=[y_values[0]],
mode="markers",
marker=dict(
size=14,
color="#D55E00",
line=dict(color="#FFFFFF", width=3),
),
hoverinfo="skip",
meta={"benchdash_role": "linked_target_cursor"},
showlegend=False,
)
)
def target_trajectory_space(
dataset: str,
frame: pd.DataFrame,
example: pd.DataFrame,
) -> html.Div:
for column in ["trial_index", "condition_id", "time_index", "time_ms", "target_0", "target_1"]:
frame[column] = pd.to_numeric(frame[column], errors="coerce")
for column in ["time_index", "time_ms", "target_0", "target_1"]:
example[column] = pd.to_numeric(example[column], errors="coerce")
example = example.sort_values("time_index")
example_customdata = np.column_stack(
[example["time_ms"].to_numpy(), example["time_index"].to_numpy()]
)
figure = go.Figure()
legend_items: list[tuple[str, str]] = []
if dataset == "monkey":
for condition_id, group in frame.groupby("condition_id", sort=True):
condition_id = int(condition_id)
color = DIRECTION_PALETTE[condition_id % len(DIRECTION_PALETTE)]
x_values, y_values = separated_trajectory_values(group, "target_0", "target_1")
figure.add_trace(
go.Scattergl(
x=x_values,
y=y_values,
mode="lines",
line=dict(color=color, width=1),
opacity=0.16,
hoverinfo="skip",
showlegend=False,
)
)
legend_items.append((condition_label(dataset, condition_id), color))
add_linked_target_traces(
figure,
example["target_0"],
example["target_1"],
example_customdata,
(
"Horizontal position=%{x:.3f}<br>Vertical position=%{y:.3f}<br>"
"Paired bin=%{customdata[1]:.0f}<extra>Example</extra>"
),
base_width=3,
progress_width=4,
halo_width=7,
)
figure.update_xaxes(title="Horizontal hand position")
figure.update_yaxes(title="Vertical hand position", scaleanchor="x", scaleratio=1)
caption = f"{frame['trial_index'].nunique():,} hand trajectories · color = reach direction"
aria_label = "All macaque hand-position targets, colored by reach direction."
elif dataset == "mc_pacman":
for condition_id, group in frame.groupby("condition_id", sort=True):
condition_id = int(condition_id)
color = DIRECTION_PALETTE[condition_id % len(DIRECTION_PALETTE)]
x_values, y_values = separated_trajectory_values(group, "time_ms", "target_0")
figure.add_trace(
go.Scattergl(
x=x_values,
y=y_values,
mode="lines",
line=dict(color=color, width=0.8),
opacity=0.075,
hoverinfo="skip",
showlegend=False,
)
)
mean_profile = group.groupby("time_ms", sort=True)["target_0"].mean()
figure.add_trace(
go.Scatter(
x=mean_profile.index,
y=mean_profile.values,
mode="lines",
line=dict(color=color, width=2.2),
hovertemplate=f"{MC_PROFILE_LABELS[condition_id]}<br>Time=%{{x:.0f}} ms<br>Mean force=%{{y:.3f}}<extra></extra>",
showlegend=False,
)
)
legend_items.append((MC_PROFILE_LABELS[condition_id], color))
add_linked_target_traces(
figure,
example["time_ms"],
example["target_0"],
example_customdata,
"Paired bin=%{customdata[1]:.0f}<br>Force=%{y:.3f}<extra>Example</extra>",
base_width=3,
progress_width=4,
)
figure.add_vline(x=0, line_color="#71808D", line_dash="dash")
figure.update_xaxes(title="Time from scoring onset (ms)")
figure.update_yaxes(title="Force")
caption = f"{frame['trial_index'].nunique():,} force trajectories · color = force profile"
aria_label = "All force targets, colored by force profile."
else:
figure.add_trace(
go.Histogram2d(
x=frame["target_0"],
y=frame["target_1"],
nbinsx=44,
nbinsy=44,
colorscale=RATINABOX_SCALE,
showscale=False,
hovertemplate="x=%{x:.3f}<br>y=%{y:.3f}<br>Samples=%{z}<extra></extra>",
)
)
add_linked_target_traces(
figure,
example["target_0"],
example["target_1"],
example_customdata,
(
"x=%{x:.3f}<br>y=%{y:.3f}<br>"
"Paired bin=%{customdata[1]:.0f}<extra>Example</extra>"
),
base_width=2.5,
progress_width=3.5,
halo_width=5,
)
figure.update_xaxes(title="x position", range=[0, 1])
figure.update_yaxes(title="y position", range=[0, 1], scaleanchor="x", scaleratio=1)
caption = f"{len(frame):,} position samples · color = occupancy density"
aria_label = "All simulated position targets shown as spatial occupancy density."
figure_layout(figure, height=410)
figure.update_layout(
margin=dict(l=62, r=24, t=18, b=62),
showlegend=False,
meta={"benchdash_dataset": dataset},
uirevision=f"target-space-{dataset}",
)
return html.Div(
[
html.Div(
[html.Strong("Target space"), html.Span(caption)],
className="dataset-viz-heading",
),
target_space_graph(figure, aria_label),
target_space_legend(legend_items) if legend_items else None,
],
className="target-space-content",
)
def dataset_link_payload(
dataset: str,
example: pd.DataFrame,
target_lag_ms: float,
) -> dict:
if dataset in {"allen_neuropixels", "speech"}:
return {"dataset": dataset, "enabled": False}
for column in ["time_index", "time_ms", "target_0", "target_1"]:
example[column] = pd.to_numeric(example[column], errors="coerce")
example = example.sort_values("time_index")
if dataset == "mc_pacman":
plot_x = example["time_ms"].to_numpy(dtype=float)
plot_y = example["target_0"].to_numpy(dtype=float)
value_kind = "force"
else:
plot_x = example["target_0"].to_numpy(dtype=float)
plot_y = example["target_1"].to_numpy(dtype=float)
value_kind = "position"
times = example["time_ms"].to_numpy(dtype=float)
return {
"dataset": dataset,
"enabled": True,
"time_index": example["time_index"].astype(int).tolist(),
"time_ms": times.tolist(),
"plot_x": plot_x.tolist(),
"plot_y": plot_y.tolist(),
"value_kind": value_kind,
"target_lag_ms": float(target_lag_ms),
"initial_index": int(np.argmin(np.abs(times))),
}
def dataset_example_figures(dataset: str) -> tuple[go.Figure, html.Div, str, dict]:
metadata = dataset_overview[dataset_overview["dataset"].astype(str).eq(dataset)].iloc[0]
neural = dataset_example_neural[
dataset_example_neural["dataset"].astype(str).eq(dataset)
].copy()
targets = dataset_targets[dataset_targets["dataset"].astype(str).eq(dataset)].copy()
targets["is_example"] = targets["is_example"].astype(str).str.lower().eq("true")
example_target = dataset_example_targets[
dataset_example_targets["dataset"].astype(str).eq(dataset)
].copy()
for column in (
"time_index",
"time_ms",
"feature_display_index",
"feature_index",
"neural_value",
):
neural[column] = pd.to_numeric(neural[column], errors="coerce")
values = neural.pivot(
index="feature_display_index", columns="time_index", values="neural_value"
).sort_index()
time_values = (
neural[["time_index", "time_ms"]]
.drop_duplicates()
.sort_values("time_index")["time_ms"]
.to_numpy(dtype=float)
)
feature_ids = (
neural[["feature_display_index", "feature_index"]]
.drop_duplicates()
.sort_values("feature_display_index")["feature_index"]
.to_numpy(dtype=int)
)
feature_customdata = np.repeat(feature_ids[:, None], values.shape[1], axis=1)
time_customdata = np.repeat(
np.arange(values.shape[1], dtype=int)[None, :],
values.shape[0],
axis=0,
)
customdata = np.stack([feature_customdata, time_customdata], axis=-1)
upper = max(int(np.ceil(np.nanmax(values.to_numpy(dtype=float)))), 1)
if upper <= 4:
count_ticks = list(range(upper + 1))
else:
count_ticks = np.unique(
np.rint(np.linspace(0, upper, 4)).astype(int)
).tolist()
neural_figure = go.Figure(
go.Heatmap(
z=values.to_numpy(dtype=float),
x=time_values,
y=np.arange(len(feature_ids)),
customdata=customdata,
colorscale=[[0.0, "#F8FAFB"], [1.0, "#263238"]],
zmin=0,
zmax=upper,
colorbar=dict(
title="Count / bin",
thickness=13,
tickmode="array",
tickvals=count_ticks,
ticktext=[str(value) for value in count_ticks],
),
hovertemplate=(
"Feature=%{customdata[0]}<br>Time=%{x:.0f} ms<br>"
"Count=%{z:.0f}<extra></extra>"
),
)
)
neural_figure.add_vline(x=0, line_color="#71808D", line_dash="dash")
neural_figure.update_xaxes(title="Time from scoring onset (ms)")
neural_figure.update_yaxes(title="Neural features", showticklabels=False)
figure_layout(neural_figure, height=470)
neural_figure.update_layout(
margin=dict(l=58, r=62, t=18, b=62),
meta={"benchdash_dataset": dataset},
uirevision=f"dataset-neural-{dataset}",
)
if dataset in {"allen_neuropixels", "speech"}:
target_component = target_class_space(dataset, targets)
else:
target_component = target_trajectory_space(dataset, targets, example_target.copy())
shown = int(metadata.example_features_shown)
total = int(metadata.array_shape.split("×")[-1].strip())
feature_text = (
f"all {total} neural features are shown"
if shown == total
else f"{shown} of {total} neural features are shown for legibility"
)
if dataset in {"allen_neuropixels", "speech"}:
description = (
f"The highlighted target corresponds to the neural activity shown at left; "
f"{feature_text}."
)
else:
description = f"{feature_text.capitalize()}."
return (
neural_figure,
target_component,
description,
dataset_link_payload(dataset, example_target, float(metadata.target_lag_ms)),
)
def overview_cards(dataset: str, models: Sequence[str] | None) -> list[html.Div]:
frame = overview_frame(dataset, models)
available = frame.dropna(subset=["task_score"]).sort_values(
["task_score", "model_order"], ascending=[False, True]
)
metric = "task score"
if not available.empty and available["metric"].notna().any():
metric = metric_name(available["metric"].dropna().iloc[0], held_out=True)
cards = [
metric_card(
"Dataset",
DATASET_LABELS.get(dataset, dataset),
DATASET_DESCRIPTIONS.get(dataset),
"prediction",
),
metric_card(
"Primary outcome",
metric,
"Higher is better; raw values are shown within each task.",
"prediction",
),
]
if available.empty:
cards.append(metric_card("Leading selected method", "Unavailable", accent="prediction"))
else:
top = available.iloc[0]
cards.append(
metric_card(
"Leading selected method",
str(top["method"]),
f"{metric}: {value_text(top['task_score'])}",
"prediction",
)
)
return cards
def prediction_ranking_figure(dataset: str, models: Sequence[str] | None) -> go.Figure:
frame = overview_frame(dataset, models).dropna(subset=["task_score"])
if frame.empty:
return empty_figure("No held-out prediction results are available for this selection.")
frame = frame.sort_values(["task_score", "model_order"], ascending=[True, False])
metric = metric_name(frame["metric"].dropna().iloc[0], held_out=True)
fig = go.Figure(
go.Bar(
x=frame["task_score"],
y=frame["method"],
orientation="h",
marker=dict(color=PREDICTION_COLOR),
customdata=np.stack([frame["prediction_percentile"], frame["workflow"]], axis=-1),
hovertemplate=(
"Method=%{y}<br>Raw score=%{x:.4f}<br>"
"Within-dataset percentile=%{customdata[0]:.1f}<br>"
"Prediction workflow=%{customdata[1]}<extra></extra>"
),
)
)
fig.update_layout(title="Held-out prediction")
fig.update_xaxes(title=metric)
fig.update_yaxes(title="", showgrid=False)
return figure_layout(fig, height=max(440, 25 * len(frame) + 145))
def mean_rank_order(
values: pd.DataFrame,
value_column: str,
eligible_models: set[str],
) -> list[str]:
frame = values[values["model"].isin(eligible_models)].copy()
frame[value_column] = pd.to_numeric(frame[value_column], errors="coerce")
pivot = frame.pivot_table(
index="model", columns="dataset", values=value_column, aggfunc="first"
).reindex(columns=DATASETS)
n_models = len(pivot)
ranks = pd.concat(
[
pivot[dataset].rank(ascending=False, method="average").fillna(n_models)
for dataset in DATASETS
],
axis=1,
)
pivot["mean_rank"] = ranks.mean(axis=1)
return pivot.sort_values("mean_rank", ascending=True).index.astype(str).tolist()
def percentile_heatmap(
values: pd.DataFrame,
models: Sequence[str] | None,
*,
title: str,
colorscale: list,
raw_column: str,
metric_column: str,
empty_message: str,
eligible_models: set[str] | None = None,
row_order: Sequence[str] | None = None,
) -> go.Figure:
chosen = selected_models(models)
if eligible_models is not None:
chosen = [model for model in chosen if model in eligible_models]
if not chosen:
return empty_figure("No results for this selection.")
frame = values.copy()
if frame.empty:
return empty_figure(empty_message)
frame[raw_column] = pd.to_numeric(frame[raw_column], errors="coerce")
frame["percentile"] = (
frame.groupby("dataset")[raw_column].rank(method="average", pct=True) * 100.0
)
frame = frame[frame["model"].isin(chosen)].copy()
grid = pd.MultiIndex.from_product([chosen, DATASETS], names=["model", "dataset"]).to_frame(index=False)
grid = grid.merge(
frame[["model", "dataset", raw_column, metric_column, "percentile"]],
on=["model", "dataset"],
how="left",
)
canonical_order = list(row_order) if row_order is not None else FIGURE_MODEL_ORDER
ordered_models = [model for model in canonical_order if model in chosen]
ordered_models += [model for model in chosen if model not in set(ordered_models)]
percentile_matrix = grid.pivot(index="model", columns="dataset", values="percentile").reindex(
index=ordered_models, columns=DATASETS
)
raw_matrix = grid.pivot(index="model", columns="dataset", values=raw_column).reindex(
index=ordered_models, columns=DATASETS
)
metric_matrix = grid.pivot(index="model", columns="dataset", values=metric_column).reindex(
index=ordered_models, columns=DATASETS
)
display_text = np.empty(percentile_matrix.shape, dtype=object)
customdata = np.empty((*percentile_matrix.shape, 2), dtype=object)
for row_index, model in enumerate(percentile_matrix.index):
for column_index, dataset in enumerate(percentile_matrix.columns):
percentile = percentile_matrix.iloc[row_index, column_index]
raw_value = raw_matrix.iloc[row_index, column_index]
metric = metric_matrix.iloc[row_index, column_index]
available = pd.notna(raw_value)
display_text[row_index, column_index] = "" if not available else f"{percentile:.0f}"
customdata[row_index, column_index, 0] = (
"" if not available else f"{float(raw_value):.4f}"
)
customdata[row_index, column_index, 1] = "" if pd.isna(metric) else str(metric)
fig = go.Figure(
go.Heatmap(
z=percentile_matrix.to_numpy(dtype=float),
x=[DATASET_TICK_LABELS[dataset] for dataset in percentile_matrix.columns],
y=[model_label(model) for model in percentile_matrix.index],
text=display_text,
texttemplate="%{text}",
textfont=dict(size=11),
customdata=customdata,
colorscale=colorscale,
zmin=0,
zmax=100,
colorbar=dict(title="Percentile", thickness=13, ticksuffix="th"),
hovertemplate=(
"Method=%{y}<br>Dataset=%{x}<br>"
"Within-dataset percentile=%{z:.1f}<br>"
"Raw value=%{customdata[0]}<br>Metric=%{customdata[1]}<extra></extra>"
),
hoverongaps=False,
)
)
missing_rows, missing_columns = np.where(percentile_matrix.isna().to_numpy())
if len(missing_rows):
fig.add_trace(
go.Scatter(
x=[DATASET_TICK_LABELS[percentile_matrix.columns[index]] for index in missing_columns],
y=[model_label(percentile_matrix.index[index]) for index in missing_rows],
mode="markers",
marker=dict(symbol="x", size=8, color="#8A949C", line=dict(width=1)),
showlegend=False,
hoverinfo="skip",
)
)
fig.update_layout(title=title)
fig.update_xaxes(title="", side="top", showgrid=False)
fig.update_yaxes(title="", showgrid=False)
return heatmap_layout(fig, height=max(500, 25 * len(percentile_matrix) + 185))
def prediction_heatmap(models: Sequence[str] | None) -> go.Figure:
values = present_rows(prediction)[["model", "dataset", "score", "metric"]].copy()
values["metric_label"] = values["metric"].map(lambda value: metric_name(value, held_out=True))
return percentile_heatmap(
values,
models,
title="Prediction across tasks",
colorscale=PREDICTION_SCALE,
raw_column="score",
metric_column="metric_label",
empty_message="No held-out prediction results are available.",
)
def robustness_frame(dataset: str, models: Sequence[str] | None) -> pd.DataFrame:
chosen = selected_models(models)
base = add_method_columns(pd.DataFrame({"model": chosen}))
base["method"] = base["model"].map(lambda model: dataset_model_label(model, dataset))
values = present_rows(robustness)
values = values[values["dataset"].astype(str) == str(dataset)].copy()
values = values.rename(
columns={
"score_at_noise0": "unperturbed_score",
"score_at_max_noise": "highest_noise_score",
"raw_auc": "robustness_auc",
"mean_score": "average_noisy_score",
}
)
columns = [
"model",
"metric",
"noise_levels",
"scores",
"unperturbed_score",
"highest_noise_score",
"robustness_auc",
"average_noisy_score",
]
values = values[[column for column in columns if column in values.columns]]
frame = base.merge(values, on="model", how="left")
return round_numeric(frame)
def robustness_figure(dataset: str, models: Sequence[str] | None) -> go.Figure:
frame = robustness_frame(dataset, models).dropna(subset=["robustness_auc"])
if frame.empty:
return empty_figure("No robustness results are available for this selection.")
frame = frame.sort_values("model_order")
fig = go.Figure()
line_dashes = ["solid", "dash", "dot", "dashdot"]
for index, row in enumerate(frame.itertuples()):
levels = parse_float_list(row.noise_levels)
scores = parse_float_list(row.scores)
if len(levels) != len(scores):
raise ValueError(f"Noise levels and scores differ for {row.model} on {dataset}.")
fig.add_trace(
go.Scatter(
x=levels,
y=scores,
mode="lines+markers",
name=row.method,
line=dict(color=MODEL_COLORS[row.model], width=2.2, dash=line_dashes[index % len(line_dashes)]),
marker=dict(size=6, symbol=index % 8),
customdata=np.repeat(row.robustness_auc, len(levels)),
hovertemplate=(
f"Method={row.method}<br>Input-noise level λ=%{{x:.1f}}<br>"
"Task score=%{y:.4f}<br>Area under task-score-versus-noise curve=%{customdata:.4f}<extra></extra>"
),
)
)
metric = metric_name(frame["metric"].dropna().iloc[0])
fig.update_layout(
title="Robustness",
hovermode="closest",
showlegend=len(frame) <= 12,
)
fig.update_xaxes(title="Input-noise level λ", tickvals=[0, 0.2, 0.4, 0.6, 0.8])
fig.update_yaxes(title=metric)
return figure_layout(fig, height=540, legend_below=len(frame) <= 12)
def compute_figures(
dataset: str, models: Sequence[str] | None
) -> tuple[go.Figure, go.Figure, pd.DataFrame]:
frame = overview_frame(dataset, models).dropna(subset=["training_time_sec"])
if frame.empty:
empty = pd.DataFrame(
columns=[
"method",
"hardware",
"task_score",
"training_time_sec",
"inference_time_sec",
"peak_ram_gb",
"peak_vram_gb",
]
)
return empty_figure("No runtime results are available."), empty_figure("No memory results are available."), empty
frame["hardware"] = np.where(frame["model"].isin(CPU_ONLY_MODELS), "CPU", "GPU")
frame = frame.sort_values(["training_time_sec", "model_order"], ascending=[False, True])
runtime = go.Figure()
runtime.add_trace(
go.Bar(
x=frame["training_time_sec"],
y=frame["method"],
orientation="h",
name="Training",
marker=dict(color=COMPUTE_COLOR),
customdata=frame["task_score"],
hovertemplate="Method=%{y}<br>Training time=%{x:.4g} s<br>Held-out score=%{customdata:.4f}<extra></extra>",
)
)
runtime.add_trace(
go.Bar(
x=frame["inference_time_sec"],
y=frame["method"],
orientation="h",
name="Inference",
marker=dict(color="#F6A15D"),
hovertemplate="Method=%{y}<br>Complete held-out split=%{x:.4g} s<extra></extra>",
)
)
runtime.update_layout(
title="Training and inference time",
barmode="group",
)
runtime.update_xaxes(title="Elapsed time (s, log)", type="log")
runtime.update_yaxes(title="", showgrid=False)
figure_layout(runtime, height=max(470, 27 * len(frame) + 155), legend_below=True)
memory = go.Figure()
memory.add_trace(
go.Bar(
x=frame["peak_ram_gb"],
y=frame["method"],
orientation="h",
name="Peak RAM",
marker=dict(color="#E6842A"),
hovertemplate="Method=%{y}<br>Peak RAM=%{x:.3f} GB<extra></extra>",
)
)
memory.add_trace(
go.Bar(
x=frame["peak_vram_gb"],
y=frame["method"],
orientation="h",
name="Peak GPU memory",
marker=dict(color="#F7C68B"),
hovertemplate="Method=%{y}<br>Peak GPU memory=%{x:.3f} GB<extra></extra>",
)
)
memory.update_layout(
title="Peak memory",
barmode="group",
)
memory.update_xaxes(title="Memory (GB)")
memory.update_yaxes(title="", showgrid=False)
figure_layout(memory, height=max(470, 27 * len(frame) + 155), legend_below=True)
table = frame[
[
"method",
"hardware",
"task_score",
"training_time_sec",
"inference_time_sec",
"peak_ram_gb",
"peak_vram_gb",
]
].copy()
return runtime, memory, round_numeric(table)
def feature_spec(dataset: str) -> tuple[str, str, str, float | None]:
if dataset == "allen_neuropixels":
return (
"spearman_corr",
"Drifting-gratings gOSI",
"Spearman’s r",
0.0,
)
if dataset == "ratinabox":
return (
"auc",
"Place cells vs head-direction and speed cells",
"ROC-AUC",
0.5,
)
return (
"auc",
"Recorded neural features vs appended synthetic controls",
"ROC-AUC",
0.5,
)
def feature_frame(dataset: str, models: Sequence[str] | None) -> pd.DataFrame:
score_column, target, metric, _ = feature_spec(dataset)
frame = filter_models(active_rows(neuron_shap), models)
frame = frame[frame["dataset"].astype(str) == str(dataset)].copy()
if frame.empty:
return frame
frame = add_method_columns(frame)
frame["method"] = frame["model"].map(lambda model: dataset_model_label(model, dataset))
frame["validation_score"] = pd.to_numeric(frame[score_column], errors="coerce")
frame["validation_target"] = target
frame["validation_metric"] = metric
return round_numeric(frame)
def feature_figures(
dataset: str, models: Sequence[str] | None
) -> tuple[go.Figure, pd.DataFrame]:
frame = feature_frame(dataset, models)
if frame.empty:
columns = ["method", "validation_score"]
return empty_figure("No results for this selection."), pd.DataFrame(columns=columns)
score_column, target, metric, reference = feature_spec(dataset)
validation = frame.dropna(subset=["validation_score"]).sort_values(
["validation_score", "model_order"], ascending=[True, False]
)
validation_fig = go.Figure(
go.Bar(
x=validation["validation_score"],
y=validation["method"],
orientation="h",
marker=dict(color=FEATURE_COLOR),
hovertemplate=f"Method=%{{y}}<br>{metric}=%{{x:.4f}}<br>Target={target}<extra></extra>",
)
)
if reference is not None:
validation_fig.add_vline(
x=reference,
line_dash="dash",
line_color="#6F7882",
annotation_text="0" if reference == 0 else "Chance = 0.5",
annotation_position="top",
)
validation_fig.update_layout(
title="Feature-attribution validation"
)
validation_fig.update_xaxes(title=metric)
validation_fig.update_yaxes(title="", showgrid=False)
figure_layout(validation_fig, height=max(430, 25 * len(validation) + 145))
columns = ["method", "validation_score"]
return validation_fig, round_numeric(
frame[columns].sort_values("validation_score", ascending=False)
)
def feature_heatmap(models: Sequence[str] | None) -> go.Figure:
rows = []
for dataset in DATASETS:
score_column, _target, metric, _reference = feature_spec(dataset)
frame = active_rows(neuron_shap)
frame = frame[frame["dataset"].astype(str) == dataset]
for row in frame.itertuples():
rows.append(
{
"model": row.model,
"dataset": dataset,
"validation_score": getattr(row, score_column),
"metric_label": metric,
}
)
values = pd.DataFrame(rows)
return percentile_heatmap(
values,
models,
title="Feature validation across tasks",
colorscale=FEATURE_SCALE,
raw_column="validation_score",
metric_column="metric_label",
empty_message="No feature-attribution validation results are available.",
eligible_models=FEATURE_ELIGIBLE,
row_order=mean_rank_order(values, "validation_score", FEATURE_ELIGIBLE),
)
def feature_attribution_frame(dataset: str, model: str | None) -> pd.DataFrame:
if not model:
return pd.DataFrame(columns=neuron_attributions.columns)
frame = neuron_attributions[
neuron_attributions["dataset"].astype(str).eq(str(dataset))
& neuron_attributions["model"].astype(str).eq(str(model))
].copy()
for column in (
"feature_index",
"signed_attribution",
"attribution_rank",
"validation_value",
):
frame[column] = pd.to_numeric(frame[column], errors="coerce")
return frame.sort_values("attribution_rank", kind="stable")
def feature_story_raster_payload(dataset: str) -> dict:
frame = feature_example_raster[
feature_example_raster["dataset"].astype(str).eq(dataset)
].copy()
if frame.empty:
return {"dataset": dataset, "features": [], "counts": []}
numeric_columns = [
"display_index",
"feature_index",
"validation_value",
"group_order",
"n_time",
"t0_index",
"bin_ms",
]
for column in numeric_columns:
frame[column] = pd.to_numeric(frame[column], errors="coerce")
if dataset == "allen_neuropixels":
frame = frame.sort_values(
["validation_value", "feature_index"], ascending=[False, True]
)
else:
frame = frame.sort_values(["group_order", "display_index"])
n_time = int(frame["n_time"].iloc[0])
value_columns = [f"value_{index:03d}" for index in range(n_time)]
counts = (
frame[value_columns]
.apply(pd.to_numeric, errors="coerce")
.to_numpy(dtype=float)
)
if not np.allclose(counts, np.rint(counts)):
raise ValueError(f"Feature raster contains noninteger counts for {dataset}")
counts = np.rint(counts).astype(int)
time_ms = (
np.arange(n_time) - int(frame["t0_index"].iloc[0])
) * float(frame["bin_ms"].iloc[0])
features = [
{
"feature_index": int(row.feature_index),
"group": str(row.feature_group),
"validation_value": (
None
if pd.isna(row.validation_value)
else float(row.validation_value)
),
}
for row in frame.itertuples(index=False)
]
return {
"dataset": dataset,
"dataset_label": DATASET_LABELS[dataset],
"features": features,
"counts": counts.tolist(),
"time_ms": time_ms.tolist(),
"max_count": int(counts.max(initial=0)),
}
def feature_story_attribution_payload(
dataset: str,
model: str | None,
) -> dict:
frame = feature_attribution_frame(dataset, model)
if frame.empty:
return {
"dataset": dataset,
"model": model,
"method": "",
"features": [],
"tied": False,
}
values = pd.to_numeric(frame["signed_attribution"], errors="coerce")
return {
"dataset": dataset,
"model": model,
"method": dataset_model_label(str(model), dataset),
"tied": values.nunique(dropna=True) <= 1,
"features": [
{
"feature_index": int(row.feature_index),
"rank": int(row.attribution_rank),
"signed_attribution": float(row.signed_attribution),
}
for row in frame.itertuples(index=False)
],
}
def trial_frame(dataset: str, models: Sequence[str] | None) -> pd.DataFrame:
frame = filter_models(active_rows(trial_shapley), models)
frame = frame[frame["dataset"].astype(str) == str(dataset)].copy()
if frame.empty:
return frame
frame = add_method_columns(frame)
frame["method"] = frame["model"].map(lambda model: dataset_model_label(model, dataset))
frame = frame.rename(columns={"perturbation_auc": "corrupted_trial_auc"})
return round_numeric(frame)
def trial_detection_figure(dataset: str, models: Sequence[str] | None) -> go.Figure:
frame = trial_frame(dataset, models)
if frame.empty:
return empty_figure("No corrupted-trial detection result is available for this selection.")
frame = frame.dropna(subset=["corrupted_trial_auc"]).sort_values(
["corrupted_trial_auc", "model_order"], ascending=[True, False]
)
fig = go.Figure(
go.Bar(
x=frame["corrupted_trial_auc"],
y=frame["method"],
orientation="h",
marker=dict(color=TRIAL_COLOR),
hovertemplate="Method=%{y}<br>Corrupted-trial ROC-AUC=%{x:.4f}<extra></extra>",
)
)
fig.add_vline(
x=0.5,
line_dash="dash",
line_color="#6F7882",
annotation_text="Chance = 0.5",
annotation_position="top",
)
fig.update_layout(
title="Corrupted-trial detection"
)
fig.update_xaxes(title="Detection ROC-AUC")
fig.update_yaxes(title="", showgrid=False)
return figure_layout(fig, height=max(430, 25 * len(frame) + 145))
def trial_heatmap(models: Sequence[str] | None) -> go.Figure:
values = active_rows(trial_shapley)[["model", "dataset", "perturbation_auc"]].copy()
values["metric_label"] = "Corrupted-trial ROC-AUC"
return percentile_heatmap(
values,
models,
title="Trial detection across tasks",
colorscale=TRIAL_SCALE,
raw_column="perturbation_auc",
metric_column="metric_label",
empty_message="No corrupted-trial detection results are available.",
eligible_models=TRIAL_ELIGIBLE,
row_order=mean_rank_order(values, "perturbation_auc", TRIAL_ELIGIBLE),
)
def retrain_frames(models: Sequence[str] | None) -> tuple[pd.DataFrame, pd.DataFrame]:
chosen = selected_models(models)
frame = active_rows(trial_retrain)
frame = frame[frame["model"].astype(str).isin(chosen)].copy()
within = frame[frame["analysis"] == "within_session_cleaning"].pivot_table(
index="model", columns="condition", values="score", aggfunc="first"
)
historical = frame[frame["analysis"] == "cross_session_old_trial_selection"].pivot_table(
index="model", columns="condition", values="score", aggfunc="first"
)
within = within.reset_index()
historical = historical.reset_index()
if not within.empty:
within = add_method_columns(within)
within["recovery"] = within.get("data_shapley") - within.get("mixed_full")
if not historical.empty:
historical = add_method_columns(historical)
historical = historical.rename(
columns={"oldonly_dshap_negative_removal": "historical_selected"}
)
return within, historical
def equality_bounds(*series: pd.Series) -> tuple[float, float]:
values = pd.concat([pd.to_numeric(item, errors="coerce") for item in series]).dropna()
if values.empty:
return 0.0, 1.0
span = float(values.max() - values.min())
padding = max(span * 0.08, 0.03)
return float(values.min() - padding), float(values.max() + padding)
def relative_change_in_mean(before: pd.Series, after: pd.Series) -> float:
"""Return the percentage change between the two condition means."""
before_mean = float(before.mean())
if abs(before_mean) <= 1e-12:
raise ValueError("Cannot express the change relative to a zero comparator mean.")
return (float(after.mean()) - before_mean) / before_mean * 100.0
def trial_retrain_figures() -> tuple[go.Figure, go.Figure, go.Figure, pd.DataFrame]:
within, historical = retrain_frames(None)
if within.empty:
removal = empty_figure("No macaque within-session removal summary is available.")
relation = empty_figure("No detection-versus-recovery summary is available.")
else:
lower, upper = equality_bounds(within["mixed_full"], within["data_shapley"])
removal = go.Figure()
removal.add_trace(
go.Scatter(
x=within["mixed_full"],
y=within["data_shapley"],
mode="markers",
text=within["method"],
marker=dict(
color=[MODEL_COLORS[model] for model in within["model"]],
size=10,
line=dict(color="#FFFFFF", width=1),
),
customdata=np.stack([within["method"], within["recovery"]], axis=-1),
hovertemplate=(
"Method=%{customdata[0]}<br>Mixed trials R²=%{x:.4f}<br>"
"After trial-value removal R²=%{y:.4f}<br>"
"Recovery ΔR²=%{customdata[1]:+.4f}<extra></extra>"
),
)
)
removal.add_shape(type="line", x0=lower, x1=upper, y0=lower, y1=upper, line=dict(color="#69737D", dash="dash"))
removal_change = relative_change_in_mean(
within["mixed_full"], within["data_shapley"]
)
removal.add_annotation(
text=f"Mean change = {removal_change:+.0f}%",
x=0.03,
y=0.97,
xref="paper",
yref="paper",
xanchor="left",
yanchor="top",
showarrow=False,
bgcolor="rgba(255,255,255,0.88)",
font=dict(size=12, color="#2E7D32" if removal_change >= 0 else NEGATIVE_COLOR),
)
removal.update_layout(title="Trial-value-guided removal")
removal.update_xaxes(title="Before removal: test R²", range=[lower, upper])
removal.update_yaxes(title="After removal: test R²", range=[lower, upper])
figure_layout(removal, height=480)
detection = active_rows(trial_shapley)
detection = detection[detection["dataset"].astype(str) == "monkey"][["model", "perturbation_auc"]]
relation_frame = within.merge(detection, on="model", how="inner").dropna(
subset=["perturbation_auc", "recovery"]
)
relation = go.Figure()
if relation_frame.empty:
relation = empty_figure("No shared detection and recovery entries are available.")
else:
x = relation_frame["perturbation_auc"].astype(float)
y = relation_frame["recovery"].astype(float)
rho = (
x.rank(method="average").corr(y.rank(method="average"))
if len(relation_frame) >= 2
else np.nan
)
relation.add_trace(
go.Scatter(
x=x,
y=y,
mode="markers",
marker=dict(
color=[MODEL_COLORS[model] for model in relation_frame["model"]],
size=10,
line=dict(color="#FFFFFF", width=1),
),
customdata=relation_frame["method"],
hovertemplate=(
"Method=%{customdata}<br>Detection ROC-AUC=%{x:.4f}<br>"
"Recovery ΔR²=%{y:+.4f}<extra></extra>"
),
)
)
if len(relation_frame) >= 2 and float(x.max() - x.min()) > 0:
coefficients = np.polyfit(x, y, 1)
line_x = np.linspace(float(x.min()), float(x.max()), 100)
relation.add_trace(
go.Scatter(
x=line_x,
y=np.polyval(coefficients, line_x),
mode="lines",
line=dict(color=TRIAL_COLOR, width=2),
name="Linear fit",
hoverinfo="skip",
)
)
relation.add_hline(y=0, line_dash="dash", line_color="#69737D")
relation_text = (
f"Spearman’s r = {rho:.2f}; n = {len(relation_frame)}"
if pd.notna(rho)
else f"n = {len(relation_frame)}; select at least two methods for correlation"
)
if pd.notna(rho):
relation_text += "<br>one-sided permutation P = 0.047"
relation.add_annotation(
text=relation_text,
x=0.02,
y=0.98,
xref="paper",
yref="paper",
xanchor="left",
yanchor="top",
showarrow=False,
bgcolor="rgba(255,255,255,0.85)",
font=dict(size=12),
)
relation.update_layout(title="Detection and recovery", showlegend=False)
relation.update_xaxes(title="Detection ROC-AUC")
relation.update_yaxes(title="Recovery (ΔR²)")
figure_layout(relation, height=480)
if historical.empty:
historical_fig = empty_figure("No same-subject historical-selection summary is available.")
else:
lower, upper = equality_bounds(historical["target_only"], historical["historical_selected"])
historical_fig = go.Figure(
go.Scatter(
x=historical["target_only"],
y=historical["historical_selected"],
mode="markers",
marker=dict(
color=[MODEL_COLORS[model] for model in historical["model"]],
size=10,
line=dict(color="#FFFFFF", width=1),
),
customdata=np.stack([historical["method"], historical["all_sessions"]], axis=-1),
hovertemplate=(
"Method=%{customdata[0]}<br>Current session only R²=%{x:.4f}<br>"
"Trial-value historical selection R²=%{y:.4f}<br>"
"All-session pooling R²=%{customdata[1]:.4f}<extra></extra>"
),
)
)
historical_fig.add_shape(type="line", x0=lower, x1=upper, y0=lower, y1=upper, line=dict(color="#69737D", dash="dash"))
historical_change = relative_change_in_mean(
historical["target_only"], historical["historical_selected"]
)
historical_fig.add_annotation(
text=f"Mean change = {historical_change:+.0f}%",
x=0.03,
y=0.97,
xref="paper",
yref="paper",
xanchor="left",
yanchor="top",
showarrow=False,
bgcolor="rgba(255,255,255,0.88)",
font=dict(size=12, color="#2E7D32" if historical_change >= 0 else NEGATIVE_COLOR),
)
historical_fig.update_layout(title="Historical-trial selection")
historical_fig.update_xaxes(title="Current-session test R²", range=[lower, upper])
historical_fig.update_yaxes(
title="Selected historical trials: test R²",
range=[lower, upper],
)
figure_layout(historical_fig, height=480)
table = within[
[column for column in ["model", "method", "mixed_full", "data_shapley", "recovery"] if column in within]
].copy() if not within.empty else pd.DataFrame(columns=["model", "method", "mixed_full", "data_shapley", "recovery"])
historical_columns = ["model", "target_only", "all_sessions", "historical_selected"]
if not historical.empty:
table = table.merge(historical[historical_columns], on="model", how="outer")
table["method"] = table["method"].fillna(table["model"].map(model_label))
table = table.drop(columns=["model"], errors="ignore")
return removal, relation, historical_fig, round_numeric(table)
def historical_trajectory_figure(*, stacked: bool = False) -> go.Figure:
frame = trial_historical_trajectories.copy()
current_r2 = float(frame["current_only_r2"].iloc[0])
historical_r2 = float(frame["historical_selected_r2"].iloc[0])
panels = [
("target_x", "target_y", "Ground truth", 0.84, 2.2),
(
"current_only_x",
"current_only_y",
f"Current session only<br>R² = {current_r2:.2f}",
0.64,
2.5,
),
(
"historical_selected_x",
"historical_selected_y",
f"Selected historical trials<br>R² = {historical_r2:.2f}",
0.64,
2.5,
),
]
if stacked:
fig = make_subplots(
rows=3,
cols=1,
vertical_spacing=0.075,
subplot_titles=[panel[2] for panel in panels],
)
else:
fig = make_subplots(
rows=1,
cols=3,
horizontal_spacing=0.045,
subplot_titles=[panel[2] for panel in panels],
)
direction_labels = dict(zip(DIRECTION_LEGEND_ORDER, DIRECTION_LEGEND_LABELS))
for direction_rank, direction_index in enumerate(DIRECTION_LEGEND_ORDER):
direction_frame = frame[frame["direction_index"] == direction_index]
direction_label = direction_labels[direction_index]
for panel_index, (x_column, y_column, _title, opacity, width) in enumerate(
panels, start=1
):
subplot_row = panel_index if stacked else 1
subplot_col = 1 if stacked else panel_index
x_values: list[object] = []
y_values: list[object] = []
hover_values: list[list[object]] = []
for trial_index in sorted(direction_frame["trial_index"].unique()):
trial = direction_frame[
direction_frame["trial_index"] == trial_index
].sort_values("time_index")
x_values.extend(trial[x_column].tolist())
y_values.extend(trial[y_column].tolist())
hover_values.extend(
[
[trial_id, direction_label, time_index]
for trial_id, time_index in zip(
trial["trial_id"].astype(str), trial["time_index"]
)
]
)
x_values.append(None)
y_values.append(None)
hover_values.append([None, None, None])
fig.add_trace(
go.Scatter(
x=x_values,
y=y_values,
mode="lines",
name=direction_label,
legendgroup=f"direction-{direction_index}",
legendrank=direction_rank,
showlegend=False,
opacity=opacity,
line=dict(color=DIRECTION_PALETTE[direction_index], width=width),
customdata=hover_values,
connectgaps=False,
hovertemplate=(
"Trial=%{customdata[0]}<br>Reach direction=%{customdata[1]}<br>"
"Time bin=%{customdata[2]}<br>x=%{x:.3f}<br>y=%{y:.3f}<extra></extra>"
),
),
row=subplot_row,
col=subplot_col,
)
first_points = (
frame.sort_values("time_index").groupby("trial_index", as_index=False).first()
)
for panel_index, (x_column, y_column, _title, _opacity, _width) in enumerate(
panels, start=1
):
subplot_row = panel_index if stacked else 1
subplot_col = 1 if stacked else panel_index
fig.add_trace(
go.Scatter(
x=[float(first_points[x_column].mean())],
y=[float(first_points[y_column].mean())],
mode="markers",
marker=dict(size=8, color="#222222"),
showlegend=False,
hovertemplate="Mean trajectory origin<extra></extra>",
),
row=subplot_row,
col=subplot_col,
)
x_values = pd.concat(
[frame["target_x"], frame["current_only_x"], frame["historical_selected_x"]],
ignore_index=True,
)
y_values = pd.concat(
[frame["target_y"], frame["current_only_y"], frame["historical_selected_y"]],
ignore_index=True,
)
x_span = max(float(x_values.max() - x_values.min()), 1.0)
y_span = max(float(y_values.max() - y_values.min()), 1.0)
x_range = [float(x_values.min() - 0.06 * x_span), float(x_values.max() + 0.06 * x_span)]
y_range = [float(y_values.min() - 0.06 * y_span), float(y_values.max() + 0.06 * y_span)]
for panel_index in range(1, 4):
subplot_row = panel_index if stacked else 1
subplot_col = 1 if stacked else panel_index
x_axis_id = "x" if panel_index == 1 else f"x{panel_index}"
fig.update_xaxes(
range=x_range,
showgrid=False,
zeroline=False,
showticklabels=False,
ticks="",
row=subplot_row,
col=subplot_col,
)
fig.update_yaxes(
range=y_range,
showgrid=False,
zeroline=False,
showticklabels=False,
ticks="",
scaleanchor=x_axis_id,
scaleratio=1,
row=subplot_row,
col=subplot_col,
)
figure_layout(fig, height=930 if stacked else 480)
fig.update_layout(
title="Held-out trajectories · RNN",
margin=dict(l=24, r=24, t=76 if not stacked else 68, b=24),
showlegend=False,
)
fig.for_each_annotation(
lambda annotation: annotation.update(font=dict(size=12, color=TEXT_COLOR))
)
return fig
def historical_direction_legend() -> html.Div:
labels = dict(zip(DIRECTION_LEGEND_ORDER, DIRECTION_LEGEND_LABELS))
return html.Div(
[
html.Span("Reach direction", className="trajectory-legend-title"),
*[
html.Span(
[
html.Span(
className="trajectory-legend-swatch",
style={"backgroundColor": DIRECTION_PALETTE[index]},
),
labels[index],
],
className="trajectory-legend-item",
)
for index in DIRECTION_LEGEND_ORDER
],
],
className="trajectory-legend",
)
def condition_sort_key(value: object) -> tuple[int, float | str]:
try:
return (0, float(value))
except (TypeError, ValueError):
return (1, str(value))
def condition_label(dataset: str, value: object, color_mode: str = "condition") -> str:
if value is None or pd.isna(value):
return "Unknown"
text = str(int(float(value))) if re.fullmatch(r"-?\d+(\.0+)?", str(value)) else str(value)
if dataset == "ratinabox":
index = int(float(text))
if color_mode == "x":
return f"x bin {index}"
if color_mode == "y":
return f"y bin {index}"
return f"x{index % 10}, y{index // 10}"
return CONDITION_LABELS.get(dataset, {}).get(text, text)
def condition_axis_label(dataset: str, color_mode: str = "condition") -> str:
if dataset == "ratinabox":
if color_mode == "x":
return "X-position bin"
if color_mode == "y":
return "Y-position bin"
return "Spatial bin"
return {
"monkey": "Reach direction",
"allen_neuropixels": "Stimulus orientation",
"speech": "Attempted word",
}.get(dataset, "Task condition")
def session_display_label(dataset: str, session: object) -> str:
text = "" if session is None or pd.isna(session) else str(session)
if dataset == "monkey":
match = re.search(r"sub-([A-Za-z])_ses-CO-", text)
return f"Monkey {match.group(1)}" if match else text
if dataset == "ratinabox":
sessions = ["ratinabox_nav", "ratinabox_nav_s123", "ratinabox_nav_s456", "ratinabox_nav_s789"]
return f"Simulation {sessions.index(text) + 1}" if text in sessions else text
if dataset == "speech":
return f"Participant {text.upper()}"
if dataset == "allen_neuropixels":
return f"Recording {text}"
return text or "Recording"
def add_latent_color_columns(df: pd.DataFrame, dataset: str, color_mode: str) -> pd.DataFrame:
out = df.copy()
values = pd.to_numeric(out["condition"], errors="coerce")
if values.isna().any() or (values < 0).any():
raise ValueError("Latent samples contain invalid task-condition labels.")
if dataset == "ratinabox" and color_mode == "x":
color_values = values.astype(int) % 10
elif dataset == "ratinabox" and color_mode == "y":
color_values = values.astype(int) // 10
else:
color_values = values.astype(int)
out["color_value"] = color_values.astype(str)
out["color_num"] = color_values
out["color_label"] = [condition_label(dataset, value, color_mode) for value in color_values]
return out
def latent_space_figure(
dataset: str,
model: str | None,
color_mode: str,
*,
stacked: bool = False,
) -> go.Figure:
if dataset == "mc_pacman":
return empty_figure(
"Cross-recording consistency is not available for this dataset.",
height=500,
)
if not model:
return empty_figure("Select an available method to view aligned coordinates.", height=500)
samples = latent_samples[
(latent_samples["dataset"].astype(str) == str(dataset))
& (latent_samples["model"].astype(str) == str(model))
].copy()
if samples.empty:
return empty_figure("No results for this selection.", height=500)
for column in ["x", "y", "z"]:
samples[column] = pd.to_numeric(samples[column], errors="coerce")
samples = add_latent_color_columns(samples, dataset, color_mode).dropna(subset=["x", "y", "z"])
trajectories = latent_trajectories[
(latent_trajectories["dataset"].astype(str) == str(dataset))
& (latent_trajectories["model"].astype(str) == str(model))
].copy()
if not trajectories.empty:
for column in ["x", "y", "z", "time_index"]:
trajectories[column] = pd.to_numeric(trajectories[column], errors="coerce")
trajectories = add_latent_color_columns(trajectories, dataset, color_mode).dropna(
subset=["x", "y", "z"]
)
else:
trajectories["color_value"] = pd.Series(dtype=str)
sessions = list(dict.fromkeys(samples["session_label"].astype(str)))
columns = 1 if stacked else (2 if len(sessions) > 1 else 1)
rows = int(np.ceil(len(sessions) / columns))
fig = make_subplots(
rows=rows,
cols=columns,
specs=[[{"type": "scene"} for _ in range(columns)] for _ in range(rows)],
subplot_titles=[session_display_label(dataset, session) for session in sessions],
horizontal_spacing=0.04,
vertical_spacing=0.1,
)
condition_values = sorted(samples["color_value"].unique(), key=condition_sort_key)
categorical = dataset != "ratinabox"
if dataset == "monkey":
colors = {value: DIRECTION_PALETTE[int(value) % len(DIRECTION_PALETTE)] for value in condition_values}
elif dataset == "speech":
colors = {value: SPEECH_PALETTE.get(value, "#777777") for value in condition_values}
else:
colors = {value: ALLEN_PALETTE.get(value, "#777777") for value in condition_values}
condition_name = condition_axis_label(dataset, color_mode)
for session_index, session in enumerate(sessions):
row_index = session_index // columns + 1
column_index = session_index % columns + 1
session_samples = samples[samples["session_label"].astype(str) == session]
display_session = session_display_label(dataset, session)
if categorical:
for condition in condition_values:
points = session_samples[session_samples["color_value"] == condition]
if points.empty:
continue
means = trajectories[
(trajectories["session_label"].astype(str) == session)
& (trajectories["color_value"] == condition)
].sort_values("time_index")
label = condition_label(dataset, condition, color_mode)
fig.add_trace(
go.Scatter3d(
x=points["x"],
y=points["y"],
z=points["z"],
mode="markers",
name=label,
legendgroup=condition,
showlegend=session_index == 0,
marker=dict(size=2.6, opacity=0.42 if not means.empty else 0.76, color=colors[condition]),
customdata=np.stack(
[
np.repeat(display_session, len(points)),
points["color_label"],
points["trial_index"],
points["time_index"],
np.repeat(condition_name, len(points)),
],
axis=-1,
),
hoverinfo="none",
),
row=row_index,
col=column_index,
)
if not means.empty:
fig.add_trace(
go.Scatter3d(
x=means["x"],
y=means["y"],
z=means["z"],
mode="lines",
name=label,
legendgroup=condition,
showlegend=False,
line=dict(color=colors[condition], width=5),
hoverinfo="skip",
customdata=means["time_index"],
),
row=row_index,
col=column_index,
)
else:
fig.add_trace(
go.Scatter3d(
x=session_samples["x"],
y=session_samples["y"],
z=session_samples["z"],
mode="markers",
name=display_session,
showlegend=False,
marker=dict(
size=2.8,
opacity=0.74,
color=session_samples["color_num"],
colorscale=RATINABOX_SCALE,
cmin=0,
cmax=9 if color_mode in {"x", "y"} else 99,
showscale=session_index == 0,
colorbar=dict(title=condition_name, thickness=12),
),
customdata=np.stack(
[
np.repeat(display_session, len(session_samples)),
session_samples["color_label"],
session_samples["trial_index"],
session_samples["time_index"],
np.repeat(condition_name, len(session_samples)),
],
axis=-1,
),
hoverinfo="none",
),
row=row_index,
col=column_index,
)
extent = max(
float(np.nanpercentile(np.abs(samples[["x", "y", "z"]].to_numpy()), 99)),
1.0,
) * 1.08
for scene_index in range(len(sessions)):
scene_id = "scene" if scene_index == 0 else f"scene{scene_index + 1}"
fig.update_layout(
**{
scene_id: dict(
xaxis=dict(range=[-extent, extent], visible=False),
yaxis=dict(range=[-extent, extent], visible=False),
zaxis=dict(range=[-extent, extent], visible=False),
aspectmode="cube",
bgcolor="#FFFFFF",
camera=dict(eye=dict(x=1.5, y=1.4, z=1.0)),
)
}
)
score_rows = active_rows(consistency)
score_rows = score_rows[
(score_rows["dataset"].astype(str) == str(dataset))
& (score_rows["model"].astype(str) == str(model))
]
score = pd.to_numeric(score_rows.get("mean_r2"), errors="coerce").dropna()
title = model_label(model)
if not score.empty:
title += f" · R² = {float(score.iloc[0]):.3f}"
fig.update_layout(
title=title,
height=(max(520, 300 * rows + 110) if stacked else (700 if rows > 1 else 500)),
paper_bgcolor="#FFFFFF",
plot_bgcolor="#FFFFFF",
margin=dict(l=8, r=8, t=72, b=88),
font=dict(family="Arial, Helvetica, sans-serif", size=12, color=TEXT_COLOR),
legend=dict(
orientation="h",
yanchor="top",
y=-0.04,
xanchor="center",
x=0.5,
entrywidth=46,
entrywidthmode="pixels",
),
showlegend=not stacked,
)
fig.for_each_annotation(lambda annotation: annotation.update(font=dict(size=12, color="#526171")))
return fig
def latent_mobile_legend(dataset: str, color_mode: str) -> html.Div:
frame = latent_samples[latent_samples["dataset"].astype(str).eq(dataset)].copy()
frame = add_latent_color_columns(frame, dataset, color_mode)
label = condition_axis_label(dataset, color_mode)
if dataset == "ratinabox":
colors = [RATINABOX_SCALE[0][1], RATINABOX_SCALE[len(RATINABOX_SCALE) // 2][1], RATINABOX_SCALE[-1][1]]
return html.Div(
[
html.Span(label, className="latent-mobile-legend-title"),
*[
html.Span(
className="latent-mobile-legend-dot",
style={"backgroundColor": color},
)
for color in colors
],
],
className="latent-mobile-legend",
)
values = sorted(frame["color_value"].dropna().unique(), key=condition_sort_key)
if dataset == "monkey":
colors = {
value: DIRECTION_PALETTE[int(value) % len(DIRECTION_PALETTE)]
for value in values
}
elif dataset == "speech":
colors = {value: SPEECH_PALETTE.get(value, "#777777") for value in values}
else:
colors = {value: ALLEN_PALETTE.get(value, "#777777") for value in values}
return html.Div(
[
html.Span(label, className="latent-mobile-legend-title"),
*[
html.Span(
[
html.Span(
className="latent-mobile-legend-dot",
style={"backgroundColor": colors[value]},
),
condition_label(dataset, value, color_mode),
],
className="latent-mobile-legend-item",
)
for value in values
],
],
className="latent-mobile-legend",
)
def consistency_frame(dataset: str, models: Sequence[str] | None) -> pd.DataFrame:
frame = filter_models(active_rows(consistency), models)
frame = frame[frame["dataset"].astype(str) == str(dataset)].copy()
if frame.empty:
return frame
frame = add_method_columns(frame)
frame = frame.rename(
columns={
"mean_r2": "latent_consistency_r2",
"n_sessions": "n_recordings",
}
)
return round_numeric(frame)
def consistency_figures(
dataset: str,
models: Sequence[str] | None,
) -> tuple[go.Figure, go.Figure, pd.DataFrame]:
frame = consistency_frame(dataset, models)
if frame.empty:
message = (
"Cross-recording consistency is not available for this dataset."
if dataset == "mc_pacman"
else "No latent-consistency result is available for this selection."
)
columns = ["method", "latent_consistency_r2"]
return empty_figure(message), consistency_heatmap(models), pd.DataFrame(columns=columns)
bar = frame.sort_values(["latent_consistency_r2", "model_order"], ascending=[True, False])
bar_fig = go.Figure(
go.Bar(
x=bar["latent_consistency_r2"],
y=bar["method"],
orientation="h",
marker=dict(color=CONSISTENCY_COLOR),
hovertemplate="Method=%{y}<br>Latent-consistency R²=%{x:.4f}<extra></extra>",
)
)
bar_fig.update_layout(title="Latent consistency")
bar_fig.update_xaxes(title="Consistency R²", range=[0, 1.02])
bar_fig.update_yaxes(title="", showgrid=False)
figure_layout(bar_fig, height=max(400, 27 * len(bar) + 145))
columns = ["method", "latent_consistency_r2"]
return bar_fig, consistency_heatmap(models), frame[columns].sort_values("latent_consistency_r2", ascending=False)
def consistency_heatmap(models: Sequence[str] | None) -> go.Figure:
chosen = selected_models(models)
frame = filter_models(active_rows(consistency), chosen)
if frame.empty:
return empty_figure("No cross-recording latent-consistency results are available.")
frame["mean_r2"] = pd.to_numeric(frame["mean_r2"], errors="coerce")
row_order = [model for model in FIGURE_MODEL_ORDER if model in CONSISTENCY_ELIGIBLE and model in chosen]
datasets = [dataset for dataset in DATASETS if dataset != "mc_pacman"]
pivot = frame.pivot_table(index="model", columns="dataset", values="mean_r2", aggfunc="first").reindex(
index=row_order, columns=datasets
)
text = np.empty(pivot.shape, dtype=object)
for row in range(pivot.shape[0]):
for column in range(pivot.shape[1]):
value = pivot.iloc[row, column]
text[row, column] = "" if pd.isna(value) else f"{value:.2f}"
fig = go.Figure(
go.Heatmap(
z=pivot.to_numpy(dtype=float),
x=[DATASET_TICK_LABELS[dataset] for dataset in pivot.columns],
y=[model_label(model) for model in pivot.index],
text=text,
texttemplate="%{text}",
colorscale=CONSISTENCY_SCALE,
zmin=0,
zmax=1,
colorbar=dict(title="R²", thickness=13),
hovertemplate="Method=%{y}<br>Dataset=%{x}<br>Latent-consistency R²=%{z:.4f}<extra></extra>",
hoverongaps=False,
)
)
missing_rows, missing_columns = np.where(pivot.isna().to_numpy())
if len(missing_rows):
fig.add_trace(
go.Scatter(
x=[DATASET_TICK_LABELS[pivot.columns[index]] for index in missing_columns],
y=[model_label(pivot.index[index]) for index in missing_rows],
mode="markers",
marker=dict(symbol="x", size=8, color="#8A949C", line=dict(width=1)),
showlegend=False,
hoverinfo="skip",
)
)
fig.update_layout(title="Latent consistency across tasks")
fig.update_xaxes(title="", side="top", showgrid=False)
fig.update_yaxes(title="", showgrid=False)
return heatmap_layout(fig, height=max(470, 27 * len(pivot) + 180))
app = Dash(__name__, title="BEND-BCI Interactive Benchmark")
server = app.server
@server.route("/download/<path:filename>")
def download_data(filename: str):
if filename not in DOWNLOADABLE_FILES:
abort(404)
return send_from_directory(DATA_DIR, filename, as_attachment=True)
app.layout = html.Div(
[
html.Header(
[
html.Div(
[
html.A("BEND-BCI", href="/", className="site-brand"),
html.Nav(
[
html.Span(
["Code & data", html.Small("coming soon")],
className="nav-placeholder",
title="Code and data will be released with the benchmark.",
),
html.A(
"Paper",
href="https://www.biorxiv.org/content/10.64898/2026.07.21.739953v1",
target="_blank",
rel="noopener noreferrer",
),
html.Span(
["Submit a model", html.Small("coming soon")],
className="nav-placeholder",
title="A model-submission workflow is planned.",
),
],
className="hero-links",
**{"aria-label": "Resources"},
),
],
className="site-nav",
),
html.Div(
[
html.H1(
"Benchmarking Neural Decoders for Brain-Computer Interfaces "
"and Neural Population Analysis"
),
],
className="hero-copy",
),
],
className="hero",
),
html.Div(
[
html.Div(
[
html.Label("Dataset", htmlFor="dataset-filter"),
dcc.Dropdown(
id="dataset-filter",
options=[{"label": DATASET_LABELS[dataset], "value": dataset} for dataset in DATASETS],
value=DATASETS[0],
clearable=False,
searchable=False,
),
],
className="control",
),
html.Div(
[
html.Label("Methods", htmlFor="method-filter"),
dcc.Dropdown(
id="method-filter",
options=[{"label": model_label(model), "value": model} for model in FIGURE_MODEL_ORDER],
value=[],
multi=True,
placeholder="All methods",
),
],
className="control method-control",
),
],
className="toolbar",
),
html.Main(
dcc.Tabs(
id="tabs",
value="overview",
className="tabs",
children=[
dcc.Tab(
label="Datasets",
value="datasets",
className="tab",
selected_className="tab tab-selected",
children=[
panel(
"Benchmark datasets",
html.Div(id="dataset-cards", className="dataset-card-grid"),
subtitle="Motor, visual, speech and spatial decoding across recordings, participants and simulations. Array dimensions are trials × time bins × features.",
class_name="axis-datasets",
),
panel(
"Neural activity and task targets",
dcc.Store(id="dataset-link-data"),
html.Span(
id="dataset-link-render-token",
className="feature-story-render-token",
),
html.Div(
[
html.Div(
[
html.Div(
[
html.Strong("Example neural activity"),
html.Span("One benchmark trial"),
],
className="dataset-viz-heading",
),
graph_box(
"dataset-neural-example",
"Example trial neural activity for the selected dataset.",
),
],
className="dataset-example-card dataset-neural-card",
),
html.Div(
id="dataset-link-controls",
className="dataset-link-controls",
),
html.Div(
id="dataset-target-space",
className="dataset-example-card dataset-target-card",
),
],
className="dataset-example-grid",
),
html.Div(id="dataset-example-description", className="dataset-example-note"),
html.Div(
[
source_link("dataset_overview.csv", "Dataset manifest"),
source_link("dataset_example_neural.csv", "Example neural data"),
source_link("dataset_targets.csv", "All targets"),
],
className="download-grid panel-downloads",
),
class_name="axis-datasets",
),
],
),
dcc.Tab(
label="Overview",
value="overview",
className="tab",
selected_className="tab tab-selected",
children=[
panel(
"Predictive performance",
html.Div(id="overview-cards", className="metric-strip"),
html.Div(
[
graph_box("prediction-ranking", "Raw held-out prediction scores for the selected dataset."),
graph_box(
"prediction-heatmap",
"Within-dataset prediction percentiles across five tasks.",
class_name="heatmap-graph",
),
],
className="chart-grid two",
),
details_table("View data", dataframe_table("overview-table", page_size=23)),
source_link("clean_prediction_summary.csv"),
subtitle="Raw task scores are shown by dataset; the cross-dataset view uses within-dataset percentiles.",
class_name="axis-prediction",
),
panel(
"Robustness to noisy inputs",
graph_box("robustness-curve", "Task score as input-noise level increases."),
details_table("View data", dataframe_table("robustness-table", page_size=23)),
source_link("robustness_summary.csv"),
subtitle="Performance as controlled noise is added to held-out neural inputs.",
class_name="axis-robustness",
),
panel(
"Computational cost",
html.Div(
[
graph_box("runtime-bars", "Training and complete-held-out-split inference times."),
graph_box("memory-bars", "Peak RAM and GPU memory."),
],
className="chart-grid two",
),
details_table("View data", dataframe_table("compute-table", page_size=23)),
source_link("scalability_summary.csv"),
subtitle="Training time, inference time, RAM and GPU memory under the benchmark protocol.",
class_name="axis-compute",
),
],
),
dcc.Tab(
label="Latent consistency",
value="consistency",
className="tab",
selected_className="tab tab-selected",
children=[
panel(
"Representation consistency across recordings",
html.Div(
[
html.Div(
[
html.Label("Method", htmlFor="consistency-method"),
dcc.Dropdown(id="consistency-method", clearable=False),
],
className="control",
),
html.Div(
[
html.Label("Color by", htmlFor="latent-color-mode"),
dcc.Dropdown(id="latent-color-mode", clearable=False),
],
id="latent-color-control",
className="control",
),
],
className="inline-controls",
),
html.Div(
"Hover a point to inspect its recording, condition, trial and time bin.",
id="latent-hover-detail",
className="latent-hover-detail",
**{"aria-live": "polite"},
),
graph_box(
"latent-space",
"Aligned latent representations for each recording.",
class_name="latent-graph latent-space-desktop",
),
graph_box(
"latent-space-mobile",
"Aligned latent representations stacked for narrow screens.",
class_name="latent-graph latent-space-mobile",
),
html.Div(id="latent-mobile-legend"),
html.Div(
[
graph_box("consistency-bars", "Latent-consistency R-squared for the selected dataset."),
graph_box(
"consistency-heatmap",
"Latent-consistency R-squared across four tasks.",
class_name="heatmap-graph",
),
],
className="chart-grid two",
),
details_table("View data", dataframe_table("consistency-table", page_size=12)),
source_link("consistency_summary.csv"),
subtitle="Plots show whitened latent spaces aligned to a common display frame. Consistency measures linear alignment of matched task landmarks across recordings, participants or simulations.",
class_name="axis-consistency",
)
],
),
dcc.Tab(
label="Feature attribution",
value="feature",
className="tab",
selected_className="tab tab-selected",
children=[
panel(
"Feature-level attribution",
html.Div(
[
html.Label("Method", htmlFor="feature-method"),
dcc.Dropdown(id="feature-method", clearable=False),
],
className="control feature-method-control",
),
dcc.Store(id="feature-story-raster-data"),
dcc.Store(id="feature-story-attribution-data"),
html.Div(
id="feature-story-viz",
className="feature-story-viz",
role="group",
tabIndex=0,
**{
"aria-label": (
"Linked example neural activity and signed "
"Kernel SHAP feature ranking."
)
},
),
html.Span(
id="feature-story-render-token",
className="feature-story-render-token",
),
html.Div(
"Hover or tap a raster row or attribution dot to trace the same feature.",
id="feature-selection-detail",
className="feature-selection-detail",
**{"aria-live": "polite"},
),
html.Div(
[
source_link("feature_example_raster.csv", "Example raster CSV"),
source_link("neuron_attributions.csv", "Feature-level CSV"),
],
className="download-grid panel-downloads",
),
subtitle="Follow each input feature from an example trial to its signed Kernel SHAP rank.",
class_name="axis-feature",
),
panel(
"Feature-attribution validation",
html.Div(
[
graph_box("feature-validation-bars", "Feature-attribution validation metric for the selected dataset."),
graph_box(
"feature-heatmap",
"Within-dataset feature-attribution validation percentiles across tasks.",
class_name="heatmap-graph",
),
],
className="chart-grid two",
),
html.Div(
html.Span(id="feature-definition"),
className="method-note",
),
details_table("View data", dataframe_table("feature-table", page_size=23)),
source_link("neuron_shap_summary.csv"),
subtitle="Agreement with predefined, dataset-specific validation targets.",
class_name="axis-feature",
)
],
),
dcc.Tab(
label="Trial valuation",
value="trial",
className="tab",
selected_className="tab tab-selected",
children=[
panel(
"Corrupted-trial detection",
html.Div(
[
graph_box("trial-detection-bars", "Corrupted-trial detection ROC-AUC for the selected dataset."),
graph_box(
"trial-heatmap",
"Within-dataset corrupted-trial detection percentiles across tasks.",
class_name="heatmap-graph",
),
],
className="chart-grid two",
),
html.Div(
[
"One third of training trials were rotated 75° in population-activity space while targets were unchanged. ROC-AUC uses negative trial value as the detection score.",
],
className="method-note",
),
details_table("View data", dataframe_table("trial-table", page_size=23)),
source_link("trial_shapley_summary.csv"),
subtitle="ROC-AUC measures whether lower trial values identify training trials with rotated neural activity.",
class_name="axis-trial",
),
panel(
"Macaque center-out reaching training-data interventions",
html.Div(
[
graph_box("trial-removal", "Held-out R-squared before and after trial-value-guided corrupted-trial removal."),
graph_box("trial-recovery", "Relationship between corrupted-trial detection and recovery after removal."),
],
className="chart-grid two",
),
graph_box("trial-historical", "Same-subject historical-trial selection compared with current-session training."),
graph_box(
"trial-historical-trajectories",
"Held-out RNN target-session trajectories for ground truth, current-session training, and nonnegative-valued historical-trial selection.",
class_name="historical-trajectory-graph historical-trajectory-desktop",
),
graph_box(
"trial-historical-trajectories-mobile",
"Held-out RNN target-session trajectories, stacked for narrow screens.",
class_name="historical-trajectory-graph historical-trajectory-mobile",
),
historical_direction_legend(),
details_table("View data", dataframe_table("trial-retrain-table", page_size=17)),
html.Div(
[
source_link("trial_shapley_retrain_summary.csv", "Summary CSV"),
source_link("trial_historical_trajectories.csv", "RNN trajectory CSV"),
],
className="download-grid panel-downloads",
),
subtitle="Removing negative-valued trials improved 13 of 17 methods (mean R² 0.775→0.798). Selecting nonnegative-valued historical trials raised mean held-out R² to 0.617, versus 0.532 for current-only training and 0.546 for all-session pooling.",
class_name="axis-trial",
),
],
),
],
),
className="main-content",
),
html.Footer(
[
html.Span("BEND-BCI · Tang Lab"),
html.Span(
["Code & data", html.Small("coming soon")],
className="nav-placeholder",
title="Code and data will be released with the benchmark.",
),
],
className="provenance-footer",
),
],
className="app-shell",
)
app.clientside_callback(
"""
function(desktopHover, desktopClick, mobileHover, mobileClick) {
const fallback = "Hover a point to inspect its recording, condition, trial and time bin.";
const triggered = dash_clientside.callback_context.triggered[0];
if (!triggered) {
return fallback;
}
const prop = triggered.prop_id || "";
let eventData = null;
if (prop.startsWith("latent-space-mobile.clickData")) {
eventData = mobileClick;
} else if (prop.startsWith("latent-space-mobile.hoverData")) {
eventData = mobileHover;
} else if (prop.startsWith("latent-space.clickData")) {
eventData = desktopClick;
} else {
eventData = desktopHover;
}
const point = eventData && eventData.points && eventData.points[0];
const values = point && point.customdata;
if (!Array.isArray(values) || values.length < 5) {
return fallback;
}
return `${values[0]} · ${values[4]}: ${values[1]} · Trial ${values[2]} · Time bin ${values[3]}`;
}
""",
Output("latent-hover-detail", "children"),
Input("latent-space", "hoverData"),
Input("latent-space", "clickData"),
Input("latent-space-mobile", "hoverData"),
Input("latent-space-mobile", "clickData"),
)
@app.callback(
Output("dataset-cards", "children"),
Output("dataset-neural-example", "figure"),
Output("dataset-target-space", "children"),
Output("dataset-example-description", "children"),
Output("dataset-link-data", "data"),
Input("dataset-filter", "value"),
)
def update_dataset_examples(dataset: str):
dataset = dataset or DATASETS[0]
neural, target, description, link_data = dataset_example_figures(dataset)
return dataset_cards(dataset), neural, target, description, link_data
app.clientside_callback(
"""
function(linkData, activeTab) {
if (!window.benchdashDatasetLink) {
return window.dash_clientside.no_update;
}
return window.benchdashDatasetLink.schedule(linkData, activeTab);
}
""",
Output("dataset-link-render-token", "children"),
Input("dataset-link-data", "data"),
Input("tabs", "value"),
)
@app.callback(
Output("overview-cards", "children"),
Output("overview-table", "columns"),
Output("overview-table", "data"),
Output("prediction-ranking", "figure"),
Output("prediction-heatmap", "figure"),
Output("robustness-curve", "figure"),
Output("robustness-table", "columns"),
Output("robustness-table", "data"),
Output("runtime-bars", "figure"),
Output("memory-bars", "figure"),
Output("compute-table", "columns"),
Output("compute-table", "data"),
Input("dataset-filter", "value"),
Input("method-filter", "value"),
)
def update_overview(dataset: str, models: list[str] | None):
dataset = dataset or DATASETS[0]
frame = overview_frame(dataset, models)
overview_columns = [
"method",
"workflow",
"task_score",
"prediction_percentile",
"robustness_auc",
"training_time_sec",
"inference_time_sec",
"peak_ram_gb",
"peak_vram_gb",
]
overview_table = frame.dropna(subset=["task_score"]).sort_values(
["task_score", "model_order"],
ascending=[False, True],
na_position="last",
)[overview_columns]
robustness_table = robustness_frame(dataset, models)
robustness_columns = [
"method",
"unperturbed_score",
"highest_noise_score",
"robustness_auc",
"average_noisy_score",
]
robustness_table = robustness_table[robustness_columns].dropna(subset=["robustness_auc"]).sort_values(
"robustness_auc", ascending=False
)
runtime, memory, compute_table = compute_figures(dataset, models)
return (
overview_cards(dataset, models),
column_defs(overview_columns),
records(round_numeric(overview_table)),
prediction_ranking_figure(dataset, models),
prediction_heatmap(models),
robustness_figure(dataset, models),
column_defs(robustness_columns),
records(round_numeric(robustness_table)),
runtime,
memory,
column_defs(compute_table.columns),
records(compute_table),
)
@app.callback(
Output("latent-color-mode", "options"),
Output("latent-color-mode", "value"),
Output("latent-color-control", "style"),
Input("dataset-filter", "value"),
)
def update_latent_color_control(dataset: str):
if dataset == "ratinabox":
return (
[
{"label": "Spatial bin", "value": "condition"},
{"label": "X-position bin", "value": "x"},
{"label": "Y-position bin", "value": "y"},
],
"x",
{},
)
return ([{"label": condition_axis_label(dataset), "value": "condition"}], "condition", {"display": "none"})
@app.callback(
Output("consistency-method", "options"),
Output("consistency-method", "value"),
Output("consistency-method", "disabled"),
Input("dataset-filter", "value"),
Input("method-filter", "value"),
State("consistency-method", "value"),
)
def update_consistency_selector(dataset: str, models: list[str] | None, current: str | None):
dataset = dataset or DATASETS[0]
if dataset == "mc_pacman":
return [], None, True
frame = consistency_frame(dataset, models)
latent_pairs = set(zip(latent_samples["model"].astype(str), latent_samples["dataset"].astype(str)))
available = []
if not frame.empty:
available = [
model
for model in frame.sort_values("latent_consistency_r2", ascending=False)["model"].astype(str)
if (model, dataset) in latent_pairs
]
options = [{"label": model_label(model), "value": model} for model in available]
value = current if current in available else (available[0] if available else None)
return options, value, not bool(options)
@app.callback(
Output("latent-space", "figure"),
Output("latent-space-mobile", "figure"),
Output("latent-mobile-legend", "children"),
Output("consistency-bars", "figure"),
Output("consistency-heatmap", "figure"),
Output("consistency-table", "columns"),
Output("consistency-table", "data"),
Input("dataset-filter", "value"),
Input("method-filter", "value"),
Input("consistency-method", "value"),
Input("latent-color-mode", "value"),
)
def update_consistency(
dataset: str,
models: list[str] | None,
method: str | None,
color_mode: str | None,
):
dataset = dataset or DATASETS[0]
bars, heatmap, table = consistency_figures(dataset, models)
resolved_color_mode = color_mode or "condition"
return (
latent_space_figure(dataset, method, resolved_color_mode),
latent_space_figure(dataset, method, resolved_color_mode, stacked=True),
latent_mobile_legend(dataset, resolved_color_mode),
bars,
heatmap,
column_defs(table.columns),
records(round_numeric(table)),
)
@app.callback(
Output("feature-method", "options"),
Output("feature-method", "value"),
Output("feature-method", "disabled"),
Input("dataset-filter", "value"),
Input("method-filter", "value"),
State("feature-method", "value"),
)
def update_feature_selector(
dataset: str,
models: list[str] | None,
current: str | None,
):
dataset = dataset or DATASETS[0]
frame = feature_frame(dataset, models).dropna(subset=["validation_score"])
available_pairs = set(
zip(
neuron_attributions["model"].astype(str),
neuron_attributions["dataset"].astype(str),
)
)
available = [
model
for model in frame.sort_values(
["validation_score", "model_order"], ascending=[False, True]
)["model"].astype(str)
if (model, dataset) in available_pairs
]
options = [
{"label": dataset_model_label(model, dataset), "value": model}
for model in available
]
value = current if current in available else (available[0] if available else None)
return options, value, not bool(options)
@app.callback(
Output("feature-story-raster-data", "data"),
Input("dataset-filter", "value"),
)
def update_feature_story_raster(dataset: str):
return feature_story_raster_payload(dataset or DATASETS[0])
@app.callback(
Output("feature-story-attribution-data", "data"),
Input("dataset-filter", "value"),
Input("feature-method", "value"),
)
def update_feature_story_attribution(
dataset: str,
method: str | None,
):
return feature_story_attribution_payload(dataset or DATASETS[0], method)
app.clientside_callback(
"""
function(rasterData, attributionData, activeTab) {
if (!rasterData || !attributionData ||
activeTab !== "feature" ||
rasterData.dataset !== attributionData.dataset ||
!window.benchdashFeatureStory) {
return window.dash_clientside.no_update;
}
window.benchdashFeatureStory.schedule(
"feature-story-viz",
"feature-selection-detail",
rasterData,
attributionData
);
return `${rasterData.dataset}:${attributionData.model}:feature`;
}
""",
Output("feature-story-render-token", "children"),
Input("feature-story-raster-data", "data"),
Input("feature-story-attribution-data", "data"),
Input("tabs", "value"),
)
@app.callback(
Output("feature-definition", "children"),
Output("feature-validation-bars", "figure"),
Output("feature-heatmap", "figure"),
Output("feature-table", "columns"),
Output("feature-table", "data"),
Input("dataset-filter", "value"),
Input("method-filter", "value"),
)
def update_feature(dataset: str, models: list[str] | None):
dataset = dataset or DATASETS[0]
validation_fig, table = feature_figures(dataset, models)
_column, _target, metric, _reference = feature_spec(dataset)
if dataset == "allen_neuropixels":
definition = (
"Spearman’s r measures association between feature-attribution values "
"and each unit’s gOSI measured from drifting gratings."
)
elif dataset == "ratinabox":
definition = (
"ROC-AUC measures whether place cells rank above head-direction and speed "
"cells. Chance ROC-AUC is 0.5."
)
else:
definition = (
"ROC-AUC measures whether recorded neural features rank above appended "
"synthetic controls. Chance ROC-AUC is 0.5."
)
table_columns = column_defs(table.columns)
for column in table_columns:
if column["id"] == "validation_score":
column["name"] = metric
return (
definition,
validation_fig,
feature_heatmap(models),
table_columns,
records(table),
)
@app.callback(
Output("trial-detection-bars", "figure"),
Output("trial-heatmap", "figure"),
Output("trial-table", "columns"),
Output("trial-table", "data"),
Output("trial-removal", "figure"),
Output("trial-recovery", "figure"),
Output("trial-historical", "figure"),
Output("trial-historical-trajectories", "figure"),
Output("trial-historical-trajectories-mobile", "figure"),
Output("trial-retrain-table", "columns"),
Output("trial-retrain-table", "data"),
Input("dataset-filter", "value"),
Input("method-filter", "value"),
)
def update_trial(dataset: str, models: list[str] | None):
dataset = dataset or DATASETS[0]
frame = trial_frame(dataset, models)
table_columns = ["method", "corrupted_trial_auc"]
table = frame[[column for column in table_columns if column in frame.columns]].sort_values(
"corrupted_trial_auc", ascending=False
) if not frame.empty else pd.DataFrame(columns=table_columns)
removal, relation, historical, retrain_table = trial_retrain_figures()
return (
trial_detection_figure(dataset, models),
trial_heatmap(models),
column_defs(table.columns),
records(round_numeric(table)),
removal,
relation,
historical,
historical_trajectory_figure(),
historical_trajectory_figure(stacked=True),
column_defs(retrain_table.columns),
records(retrain_table),
)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=7860, debug=False)