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
center-out reaching", "allen_neuropixels": "Allen
Neuropixels", "speech": "Attempted
speech", "mc_pacman": "MC
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}
Vertical position=%{y:.3f}
" "Paired bin=%{customdata[1]:.0f}Example" ), 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]}
Time=%{{x:.0f}} ms
Mean force=%{{y:.3f}}", 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}
Force=%{y:.3f}Example", 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}
y=%{y:.3f}
Samples=%{z}", ) ) add_linked_target_traces( figure, example["target_0"], example["target_1"], example_customdata, ( "x=%{x:.3f}
y=%{y:.3f}
" "Paired bin=%{customdata[1]:.0f}Example" ), 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]}
Time=%{x:.0f} ms
" "Count=%{z:.0f}" ), ) ) 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}
Raw score=%{x:.4f}
" "Within-dataset percentile=%{customdata[0]:.1f}
" "Prediction workflow=%{customdata[1]}" ), ) ) 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}
Dataset=%{x}
" "Within-dataset percentile=%{z:.1f}
" "Raw value=%{customdata[0]}
Metric=%{customdata[1]}" ), 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}
Input-noise level λ=%{{x:.1f}}
" "Task score=%{y:.4f}
Area under task-score-versus-noise curve=%{customdata:.4f}" ), ) ) 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}
Training time=%{x:.4g} s
Held-out score=%{customdata:.4f}", ) ) runtime.add_trace( go.Bar( x=frame["inference_time_sec"], y=frame["method"], orientation="h", name="Inference", marker=dict(color="#F6A15D"), hovertemplate="Method=%{y}
Complete held-out split=%{x:.4g} s", ) ) 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}
Peak RAM=%{x:.3f} GB", ) ) 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}
Peak GPU memory=%{x:.3f} GB", ) ) 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}}
{metric}=%{{x:.4f}}
Target={target}", ) ) 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}
Corrupted-trial ROC-AUC=%{x:.4f}", ) ) 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]}
Mixed trials R²=%{x:.4f}
" "After trial-value removal R²=%{y:.4f}
" "Recovery ΔR²=%{customdata[1]:+.4f}" ), ) ) 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}
Detection ROC-AUC=%{x:.4f}
" "Recovery ΔR²=%{y:+.4f}" ), ) ) 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 += "
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]}
Current session only R²=%{x:.4f}
" "Trial-value historical selection R²=%{y:.4f}
" "All-session pooling R²=%{customdata[1]:.4f}" ), ) ) 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
R² = {current_r2:.2f}", 0.64, 2.5, ), ( "historical_selected_x", "historical_selected_y", f"Selected historical trials
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]}
Reach direction=%{customdata[1]}
" "Time bin=%{customdata[2]}
x=%{x:.3f}
y=%{y:.3f}" ), ), 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", ), 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}
Latent-consistency R²=%{x:.4f}", ) ) 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}
Dataset=%{x}
Latent-consistency R²=%{z:.4f}", 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/") 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)