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", } DATASET_LABELS = { "monkey": "Macaque center-out reaching", "allen_neuropixels": "Allen visual coding", "speech": "Attempted speech", "mc_pacman": "MC PacMan force decoding", "ratinabox": "RatInABox navigation", } DATASET_TICK_LABELS = { "monkey": "Macaque
center-out reaching", "allen_neuropixels": "Allen
visual coding", "speech": "Attempted
speech", "mc_pacman": "MC PacMan
force decoding", "ratinabox": "RatInABox
navigation", } 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]] 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": "Do nothing", "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 = { "3": "#E69F00", "2": "#56B4E9", "4": "#009E73", "7": "#999933", "6": "#0072B2", "1": "#D55E00", "5": "#CC79A7", "0": "#000000", } ALLEN_PALETTE = { str(index): color for index, color in enumerate( ["#E69F00", "#56B4E9", "#009E73", "#F0E442", "#0072B2", "#D55E00", "#CC79A7", "#000000"] ) } 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 = { "clean_prediction_summary.csv", "robustness_summary.csv", "consistency_summary.csv", "scalability_summary.csv", "neuron_shap_summary.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) -> pd.DataFrame: path = DATA_DIR / name if not path.exists(): raise FileNotFoundError(f"Missing dashboard data: {path}") return pd.read_csv(path) 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) 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") 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.01, xanchor="left"), legend=legend, hoverlabel=dict(font=dict(family="Arial, Helvetica, sans-serif", size=12)), ) 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" # Manuscript v7 deliberately 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 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 (seconds, log scale)", 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 orientation selectivity", "Spearman’s ρ", 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_target", "validation_metric", "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_target", "validation_metric", "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 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="ROC-AUC from negative trial value") 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 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 = ( float(within["recovery"].mean()) / float(within["mixed_full"].abs().mean()) * 100.0 ) 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 ρ = {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.035" 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 = ( float((historical["historical_selected"] - historical["target_only"]).mean()) / float(historical["target_only"].abs().mean()) * 100.0 ) 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() -> 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, ), ] 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 ): 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=panel_index == 1, 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=1, col=panel_index, ) 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 ): 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=1, col=panel_index, ) 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): 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=1, col=panel_index, ) fig.update_yaxes( range=y_range, showgrid=False, zeroline=False, showticklabels=False, ticks="", scaleanchor=x_axis_id, scaleratio=1, row=1, col=panel_index, ) figure_layout(fig, height=510, legend_below=True) fig.update_layout( title="Held-out trajectories · RNN", margin=dict(l=28, r=28, t=76, b=118), legend=dict( orientation="h", yanchor="top", y=-0.10, xanchor="center", x=0.5, title="Reach direction", traceorder="normal", font=dict(size=11), ), ) fig.for_each_annotation( lambda annotation: annotation.update(font=dict(size=12, color=TEXT_COLOR)) ) return fig 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) -> 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 = 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"], ], axis=-1, ), hovertemplate=( "Recording=%{customdata[0]}
" f"{condition_name}=%{{customdata[1]}}
Trial=%{{customdata[2]}}; time bin=%{{customdata[3]}}" "" ), ), 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), hovertemplate=f"{condition_name}={label}
Time bin=%{{customdata}}", 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"], ], axis=-1, ), hovertemplate=( "Recording=%{customdata[0]}
" f"{condition_name}=%{{customdata[1]}}
Trial=%{{customdata[2]}}; time bin=%{{customdata[3]}}" "" ), ), 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=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", ), ) fig.for_each_annotation(lambda annotation: annotation.update(font=dict(size=12, color="#526171"))) return fig 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", "n_recordings", "latent_dim", "n_pairwise"] 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), customdata=np.stack([bar["n_recordings"], bar["latent_dim"], bar["n_pairwise"]], axis=-1), hovertemplate=( "Method=%{y}
Latent-consistency R²=%{x:.4f}
" "Recordings=%{customdata[0]:.0f}
Latent dimensions=%{customdata[1]:.0f}
" "Directional pairs=%{customdata[2]:.0f}" ), ) ) bar_fig.update_layout(title="Latent consistency") bar_fig.update_xaxes(title="Latent-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", "n_recordings", "latent_dim", "n_pairwise"] 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.A( "Code & data", href="https://github.com/TangLab-UBC/behavior_benchmarking", target="_blank", rel="noopener noreferrer", ), html.Span( ["Paper", html.Small("coming soon")], className="nav-placeholder", title="Manuscript link will be added on release.", ), html.Span( ["Submit a model", html.Small("planned")], className="nav-placeholder", title="A model-submission workflow is planned.", ), ], className="hero-links", **{"aria-label": "Resources"}, ), ], className="site-nav", ), html.Div( [ html.H1("Neural decoder selection beyond held-out performance"), html.P( "Interactive results for 23 methods across motor, visual, speech and spatial decoding tasks.", className="lede", ), ], 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="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", ), graph_box("latent-space", "Aligned latent representations for each recording.", class_name="latent-graph"), 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-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", ), 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.797). 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.A( "Code & data", href="https://github.com/TangLab-UBC/behavior_benchmarking", target="_blank", rel="noopener noreferrer", ), ], className="provenance-footer", ), ], className="app-shell", ) @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("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) return ( latent_space_figure(dataset, method, color_mode or "condition"), bars, heatmap, column_defs(table.columns), records(round_numeric(table)), ) @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 ρ measures association with drifting-gratings orientation " "selectivity, a biological proxy." ) 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." ) return ( definition, validation_fig, feature_heatmap(models), column_defs(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-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(), column_defs(retrain_table.columns), records(retrain_table), ) if __name__ == "__main__": app.run(host="0.0.0.0", port=7860, debug=False)