| 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 Neuropixels", |
| "speech": "Attempted speech", |
| "mc_pacman": "MC PacMan", |
| "ratinabox": "RatInABox", |
| } |
|
|
| DATASET_TICK_LABELS = { |
| "monkey": "Macaque<br>center-out reaching", |
| "allen_neuropixels": "Allen<br>Neuropixels", |
| "speech": "Attempted<br>speech", |
| "mc_pacman": "MC<br>PacMan", |
| "ratinabox": "RatInABox", |
| } |
|
|
| DATASET_DESCRIPTIONS = { |
| "monkey": "Two-dimensional hand-position regression from macaque neural population activity.", |
| "allen_neuropixels": "Eight-class drifting-grating orientation classification from Allen Neuropixels units.", |
| "speech": "Eight-class attempted-word classification from threshold-crossing features.", |
| "mc_pacman": "Continuous force regression from motor-cortical population activity.", |
| "ratinabox": "Two-dimensional position regression from simulated place, head-direction and speed cells.", |
| } |
|
|
| DATASETS = list(DATASET_LABELS) |
| MODELS = PAPER_MODEL_ORDER.copy() |
| MODEL_INDEX = {model: index for index, model in enumerate(MODELS)} |
|
|
| |
| PREDICTION_COLOR = "#1565C0" |
| ROBUSTNESS_COLOR = "#2E7D32" |
| COMPUTE_COLOR = "#E65100" |
| |
| 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", |
| }, |
| } |
|
|
| |
| 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" |
| |
| |
| |
| if decoder_name in {"ridge", "logistic", "ldns_rate_sklearn_logistic"}: |
| return "Shared linear readout" |
| if decoder_name in {"knn", "ole", "ldns_rate_sklearn_ridge"}: |
| return "Author-style task readout" |
| raise ValueError(f"Unrecognized prediction decoder for {model}: {decoder_name!r}") |
|
|
|
|
| def prediction_percentiles() -> pd.DataFrame: |
| frame = present_rows(prediction)[["model", "dataset", "score"]].copy() |
| frame["score"] = pd.to_numeric(frame["score"], errors="coerce") |
| frame["prediction_percentile"] = ( |
| frame.groupby("dataset")["score"].rank(method="average", pct=True) * 100.0 |
| ) |
| return frame |
|
|
|
|
| def overview_frame(dataset: str, models: Sequence[str] | None) -> pd.DataFrame: |
| chosen = selected_models(models) |
| base = add_method_columns(pd.DataFrame({"model": chosen})) |
| base["method"] = base["model"].map(lambda model: dataset_model_label(model, dataset)) |
| pred = prediction[prediction["dataset"].astype(str) == str(dataset)].copy() |
| pred["score"] = pd.to_numeric(pred["score"], errors="coerce") |
| pred = pred.merge( |
| prediction_percentiles()[["model", "dataset", "prediction_percentile"]], |
| on=["model", "dataset"], |
| how="left", |
| ) |
| pred = pred[ |
| [ |
| "model", |
| "status", |
| "metric", |
| "decoder", |
| "score", |
| "prediction_percentile", |
| "n_train_trials", |
| "n_test_trials", |
| "n_neurons", |
| ] |
| ].rename(columns={"score": "task_score", "status": "prediction_status_raw"}) |
|
|
| rob = present_rows(robustness) |
| rob = rob[rob["dataset"].astype(str) == str(dataset)][["model", "raw_auc"]].copy() |
| rob["raw_auc"] = pd.to_numeric(rob["raw_auc"], errors="coerce") |
| rob = rob.rename(columns={"raw_auc": "robustness_auc"}) |
|
|
| scale = present_rows(scalability) |
| scale = scale[scale["dataset"].astype(str) == str(dataset)][ |
| ["model", "training_time_sec", "inference_time_sec", "peak_ram_gb", "peak_vram_gb"] |
| ].copy() |
| for column in ["training_time_sec", "inference_time_sec", "peak_ram_gb", "peak_vram_gb"]: |
| scale[column] = pd.to_numeric(scale[column], errors="coerce") |
|
|
| frame = base.merge(pred, on="model", how="left") |
| frame = frame.merge(rob, on="model", how="left") |
| frame = frame.merge(scale, on="model", how="left") |
| frame["workflow"] = frame.apply( |
| lambda row: prediction_workflow(row["model"], row.get("decoder"), row.get("prediction_status_raw")), |
| axis=1, |
| ) |
| frame.loc[frame["model"].isin(CPU_ONLY_MODELS), "peak_vram_gb"] = np.nan |
| return round_numeric(frame) |
|
|
|
|
| def dataset_cards(selected_dataset: str) -> list[html.Article]: |
| cards: list[html.Article] = [] |
| ordered = dataset_overview.set_index("dataset").reindex(DATASETS).reset_index() |
| for row in ordered.itertuples(index=False): |
| classes = "dataset-card" |
| if str(row.dataset) == str(selected_dataset): |
| classes += " dataset-card-selected" |
| cards.append( |
| html.Article( |
| [ |
| html.Div(str(row.dataset_name), className="dataset-card-title"), |
| html.Div( |
| f"{row.species} · {row.task}", |
| className="dataset-card-context", |
| ), |
| html.Div( |
| [ |
| html.Div( |
| [ |
| html.Span( |
| "Primary array", |
| className="dataset-fact-label", |
| title="Trials × time bins × features", |
| ), |
| html.Strong(str(row.array_shape)), |
| ] |
| ), |
| html.Div( |
| [ |
| html.Span("Target", className="dataset-fact-label"), |
| html.Strong(f"{row.target} · {row.score}"), |
| ] |
| ), |
| html.Div( |
| [ |
| html.Span("Sampling", className="dataset-fact-label"), |
| html.Strong(f"{float(row.bin_ms):g} ms bins"), |
| ] |
| ), |
| html.Div( |
| [ |
| html.Span("Consistency cohort", className="dataset-fact-label"), |
| html.Strong(str(row.recordings)), |
| ] |
| ), |
| ], |
| className="dataset-facts", |
| ), |
| html.A( |
| str(row.source_label), |
| href=str(row.source_url), |
| target="_blank", |
| rel="noopener noreferrer", |
| className="dataset-source", |
| ), |
| ], |
| className=classes, |
| ) |
| ) |
| return cards |
|
|
|
|
| def target_space_graph(figure: go.Figure, label: str) -> html.Div: |
| return html.Div( |
| dcc.Graph( |
| id="dataset-target-trajectory", |
| figure=figure, |
| config={"displaylogo": False, "responsive": True}, |
| ), |
| className="target-space-graph", |
| role="img", |
| **{"aria-label": label}, |
| ) |
|
|
|
|
| def target_space_legend(items: Sequence[tuple[str, str]]) -> html.Div: |
| return html.Div( |
| [ |
| html.Span( |
| [ |
| html.Span( |
| className="target-legend-swatch", |
| style={"backgroundColor": color}, |
| ), |
| label, |
| ], |
| className="target-legend-item", |
| ) |
| for label, color in items |
| ], |
| className="target-space-legend", |
| ) |
|
|
|
|
| def target_class_space(dataset: str, frame: pd.DataFrame) -> html.Div: |
| counts = frame.groupby("condition_id", sort=True).size() |
| example_class = int(frame.loc[frame["is_example"], "condition_id"].iloc[0]) |
| cards = [] |
| for raw_class, count in counts.items(): |
| class_id = int(raw_class) |
| color = DIRECTION_PALETTE[class_id % len(DIRECTION_PALETTE)] |
| label = condition_label(dataset, class_id) |
| selected = class_id == example_class |
| if dataset == "allen_neuropixels": |
| glyph = html.Div( |
| [html.Span(className="orientation-arrow-head")], |
| className="orientation-arrow", |
| style={ |
| "backgroundColor": color, |
| "color": color, |
| "transform": f"rotate({-class_id * 45}deg)", |
| }, |
| ) |
| else: |
| glyph = html.Div( |
| label, |
| className="speech-target-word", |
| style={"color": color}, |
| ) |
| cards.append( |
| html.Div( |
| [ |
| html.Div(glyph, className="target-class-glyph"), |
| ( |
| html.Div(label, className="target-class-label") |
| if dataset == "allen_neuropixels" |
| else None |
| ), |
| html.Div(f"{int(count)} trials", className="target-class-count"), |
| html.Span("Example", className="target-example-badge") if selected else None, |
| ], |
| className=( |
| "target-class-card target-class-card-selected" |
| if selected |
| else "target-class-card" |
| ), |
| style={"borderTopColor": color}, |
| ) |
| ) |
| context = ( |
| "Eight stimulus orientations" |
| if dataset == "allen_neuropixels" |
| else "Seven attempted words and rest" |
| ) |
| return html.Div( |
| [ |
| html.Div( |
| [ |
| html.Strong("Target space"), |
| html.Span(f"{len(frame):,} trials · {context}"), |
| ], |
| className="dataset-viz-heading", |
| ), |
| html.Div(cards, className="target-class-grid"), |
| ], |
| className="target-space-content", |
| ) |
|
|
|
|
| def separated_trajectory_values( |
| frame: pd.DataFrame, |
| x_column: str, |
| y_column: str, |
| ) -> tuple[list[float | None], list[float | None]]: |
| x_values: list[float | None] = [] |
| y_values: list[float | None] = [] |
| for _, trial in frame.groupby("trial_index", sort=False): |
| trial = trial.sort_values("time_index") |
| x_values.extend(trial[x_column].astype(float).tolist()) |
| y_values.extend(trial[y_column].astype(float).tolist()) |
| x_values.append(None) |
| y_values.append(None) |
| return x_values, y_values |
|
|
|
|
| def add_linked_target_traces( |
| figure: go.Figure, |
| x_values: Sequence[float], |
| y_values: Sequence[float], |
| customdata: np.ndarray, |
| hovertemplate: str, |
| *, |
| base_width: float, |
| progress_width: float, |
| halo_width: float | None = None, |
| ) -> None: |
| x_values = list(x_values) |
| y_values = list(y_values) |
| if halo_width is not None: |
| figure.add_trace( |
| go.Scatter( |
| x=x_values, |
| y=y_values, |
| mode="lines", |
| line=dict(color="#FFFFFF", width=halo_width), |
| opacity=0.9, |
| hoverinfo="skip", |
| showlegend=False, |
| ) |
| ) |
| figure.add_trace( |
| go.Scatter( |
| x=x_values, |
| y=y_values, |
| mode="lines", |
| line=dict(color="#102A3A", width=base_width), |
| opacity=0.34, |
| customdata=customdata, |
| meta={"benchdash_role": "example_trajectory"}, |
| hovertemplate=hovertemplate, |
| showlegend=False, |
| ) |
| ) |
| figure.add_trace( |
| go.Scatter( |
| x=[x_values[0]], |
| y=[y_values[0]], |
| mode="lines", |
| line=dict(color="#102A3A", width=progress_width), |
| hoverinfo="skip", |
| meta={"benchdash_role": "linked_target_progress"}, |
| showlegend=False, |
| ) |
| ) |
| figure.add_trace( |
| go.Scatter( |
| x=[x_values[0]], |
| y=[y_values[0]], |
| mode="markers", |
| marker=dict( |
| size=14, |
| color="#D55E00", |
| line=dict(color="#FFFFFF", width=3), |
| ), |
| hoverinfo="skip", |
| meta={"benchdash_role": "linked_target_cursor"}, |
| showlegend=False, |
| ) |
| ) |
|
|
|
|
| def target_trajectory_space( |
| dataset: str, |
| frame: pd.DataFrame, |
| example: pd.DataFrame, |
| ) -> html.Div: |
| for column in ["trial_index", "condition_id", "time_index", "time_ms", "target_0", "target_1"]: |
| frame[column] = pd.to_numeric(frame[column], errors="coerce") |
| for column in ["time_index", "time_ms", "target_0", "target_1"]: |
| example[column] = pd.to_numeric(example[column], errors="coerce") |
| example = example.sort_values("time_index") |
| example_customdata = np.column_stack( |
| [example["time_ms"].to_numpy(), example["time_index"].to_numpy()] |
| ) |
| figure = go.Figure() |
| legend_items: list[tuple[str, str]] = [] |
|
|
| if dataset == "monkey": |
| for condition_id, group in frame.groupby("condition_id", sort=True): |
| condition_id = int(condition_id) |
| color = DIRECTION_PALETTE[condition_id % len(DIRECTION_PALETTE)] |
| x_values, y_values = separated_trajectory_values(group, "target_0", "target_1") |
| figure.add_trace( |
| go.Scattergl( |
| x=x_values, |
| y=y_values, |
| mode="lines", |
| line=dict(color=color, width=1), |
| opacity=0.16, |
| hoverinfo="skip", |
| showlegend=False, |
| ) |
| ) |
| legend_items.append((condition_label(dataset, condition_id), color)) |
| add_linked_target_traces( |
| figure, |
| example["target_0"], |
| example["target_1"], |
| example_customdata, |
| ( |
| "Horizontal position=%{x:.3f}<br>Vertical position=%{y:.3f}<br>" |
| "Paired bin=%{customdata[1]:.0f}<extra>Example</extra>" |
| ), |
| base_width=3, |
| progress_width=4, |
| halo_width=7, |
| ) |
| figure.update_xaxes(title="Horizontal hand position") |
| figure.update_yaxes(title="Vertical hand position", scaleanchor="x", scaleratio=1) |
| caption = f"{frame['trial_index'].nunique():,} hand trajectories · color = reach direction" |
| aria_label = "All macaque hand-position targets, colored by reach direction." |
| elif dataset == "mc_pacman": |
| for condition_id, group in frame.groupby("condition_id", sort=True): |
| condition_id = int(condition_id) |
| color = DIRECTION_PALETTE[condition_id % len(DIRECTION_PALETTE)] |
| x_values, y_values = separated_trajectory_values(group, "time_ms", "target_0") |
| figure.add_trace( |
| go.Scattergl( |
| x=x_values, |
| y=y_values, |
| mode="lines", |
| line=dict(color=color, width=0.8), |
| opacity=0.075, |
| hoverinfo="skip", |
| showlegend=False, |
| ) |
| ) |
| mean_profile = group.groupby("time_ms", sort=True)["target_0"].mean() |
| figure.add_trace( |
| go.Scatter( |
| x=mean_profile.index, |
| y=mean_profile.values, |
| mode="lines", |
| line=dict(color=color, width=2.2), |
| hovertemplate=f"{MC_PROFILE_LABELS[condition_id]}<br>Time=%{{x:.0f}} ms<br>Mean force=%{{y:.3f}}<extra></extra>", |
| showlegend=False, |
| ) |
| ) |
| legend_items.append((MC_PROFILE_LABELS[condition_id], color)) |
| add_linked_target_traces( |
| figure, |
| example["time_ms"], |
| example["target_0"], |
| example_customdata, |
| "Paired bin=%{customdata[1]:.0f}<br>Force=%{y:.3f}<extra>Example</extra>", |
| base_width=3, |
| progress_width=4, |
| ) |
| figure.add_vline(x=0, line_color="#71808D", line_dash="dash") |
| figure.update_xaxes(title="Time from scoring onset (ms)") |
| figure.update_yaxes(title="Force") |
| caption = f"{frame['trial_index'].nunique():,} force trajectories · color = force profile" |
| aria_label = "All force targets, colored by force profile." |
| else: |
| figure.add_trace( |
| go.Histogram2d( |
| x=frame["target_0"], |
| y=frame["target_1"], |
| nbinsx=44, |
| nbinsy=44, |
| colorscale=RATINABOX_SCALE, |
| showscale=False, |
| hovertemplate="x=%{x:.3f}<br>y=%{y:.3f}<br>Samples=%{z}<extra></extra>", |
| ) |
| ) |
| add_linked_target_traces( |
| figure, |
| example["target_0"], |
| example["target_1"], |
| example_customdata, |
| ( |
| "x=%{x:.3f}<br>y=%{y:.3f}<br>" |
| "Paired bin=%{customdata[1]:.0f}<extra>Example</extra>" |
| ), |
| base_width=2.5, |
| progress_width=3.5, |
| halo_width=5, |
| ) |
| figure.update_xaxes(title="x position", range=[0, 1]) |
| figure.update_yaxes(title="y position", range=[0, 1], scaleanchor="x", scaleratio=1) |
| caption = f"{len(frame):,} position samples · color = occupancy density" |
| aria_label = "All simulated position targets shown as spatial occupancy density." |
|
|
| figure_layout(figure, height=410) |
| figure.update_layout( |
| margin=dict(l=62, r=24, t=18, b=62), |
| showlegend=False, |
| meta={"benchdash_dataset": dataset}, |
| uirevision=f"target-space-{dataset}", |
| ) |
| return html.Div( |
| [ |
| html.Div( |
| [html.Strong("Target space"), html.Span(caption)], |
| className="dataset-viz-heading", |
| ), |
| target_space_graph(figure, aria_label), |
| target_space_legend(legend_items) if legend_items else None, |
| ], |
| className="target-space-content", |
| ) |
|
|
|
|
| def dataset_link_payload( |
| dataset: str, |
| example: pd.DataFrame, |
| target_lag_ms: float, |
| ) -> dict: |
| if dataset in {"allen_neuropixels", "speech"}: |
| return {"dataset": dataset, "enabled": False} |
| for column in ["time_index", "time_ms", "target_0", "target_1"]: |
| example[column] = pd.to_numeric(example[column], errors="coerce") |
| example = example.sort_values("time_index") |
| if dataset == "mc_pacman": |
| plot_x = example["time_ms"].to_numpy(dtype=float) |
| plot_y = example["target_0"].to_numpy(dtype=float) |
| value_kind = "force" |
| else: |
| plot_x = example["target_0"].to_numpy(dtype=float) |
| plot_y = example["target_1"].to_numpy(dtype=float) |
| value_kind = "position" |
| times = example["time_ms"].to_numpy(dtype=float) |
| return { |
| "dataset": dataset, |
| "enabled": True, |
| "time_index": example["time_index"].astype(int).tolist(), |
| "time_ms": times.tolist(), |
| "plot_x": plot_x.tolist(), |
| "plot_y": plot_y.tolist(), |
| "value_kind": value_kind, |
| "target_lag_ms": float(target_lag_ms), |
| "initial_index": int(np.argmin(np.abs(times))), |
| } |
|
|
|
|
| def dataset_example_figures(dataset: str) -> tuple[go.Figure, html.Div, str, dict]: |
| metadata = dataset_overview[dataset_overview["dataset"].astype(str).eq(dataset)].iloc[0] |
| neural = dataset_example_neural[ |
| dataset_example_neural["dataset"].astype(str).eq(dataset) |
| ].copy() |
| targets = dataset_targets[dataset_targets["dataset"].astype(str).eq(dataset)].copy() |
| targets["is_example"] = targets["is_example"].astype(str).str.lower().eq("true") |
| example_target = dataset_example_targets[ |
| dataset_example_targets["dataset"].astype(str).eq(dataset) |
| ].copy() |
| for column in ( |
| "time_index", |
| "time_ms", |
| "feature_display_index", |
| "feature_index", |
| "neural_value", |
| ): |
| neural[column] = pd.to_numeric(neural[column], errors="coerce") |
|
|
| values = neural.pivot( |
| index="feature_display_index", columns="time_index", values="neural_value" |
| ).sort_index() |
| time_values = ( |
| neural[["time_index", "time_ms"]] |
| .drop_duplicates() |
| .sort_values("time_index")["time_ms"] |
| .to_numpy(dtype=float) |
| ) |
| feature_ids = ( |
| neural[["feature_display_index", "feature_index"]] |
| .drop_duplicates() |
| .sort_values("feature_display_index")["feature_index"] |
| .to_numpy(dtype=int) |
| ) |
| feature_customdata = np.repeat(feature_ids[:, None], values.shape[1], axis=1) |
| time_customdata = np.repeat( |
| np.arange(values.shape[1], dtype=int)[None, :], |
| values.shape[0], |
| axis=0, |
| ) |
| customdata = np.stack([feature_customdata, time_customdata], axis=-1) |
| upper = max(int(np.ceil(np.nanmax(values.to_numpy(dtype=float)))), 1) |
| if upper <= 4: |
| count_ticks = list(range(upper + 1)) |
| else: |
| count_ticks = np.unique( |
| np.rint(np.linspace(0, upper, 4)).astype(int) |
| ).tolist() |
| neural_figure = go.Figure( |
| go.Heatmap( |
| z=values.to_numpy(dtype=float), |
| x=time_values, |
| y=np.arange(len(feature_ids)), |
| customdata=customdata, |
| colorscale=[[0.0, "#F8FAFB"], [1.0, "#263238"]], |
| zmin=0, |
| zmax=upper, |
| colorbar=dict( |
| title="Count / bin", |
| thickness=13, |
| tickmode="array", |
| tickvals=count_ticks, |
| ticktext=[str(value) for value in count_ticks], |
| ), |
| hovertemplate=( |
| "Feature=%{customdata[0]}<br>Time=%{x:.0f} ms<br>" |
| "Count=%{z:.0f}<extra></extra>" |
| ), |
| ) |
| ) |
| neural_figure.add_vline(x=0, line_color="#71808D", line_dash="dash") |
| neural_figure.update_xaxes(title="Time from scoring onset (ms)") |
| neural_figure.update_yaxes(title="Neural features", showticklabels=False) |
| figure_layout(neural_figure, height=470) |
| neural_figure.update_layout( |
| margin=dict(l=58, r=62, t=18, b=62), |
| meta={"benchdash_dataset": dataset}, |
| uirevision=f"dataset-neural-{dataset}", |
| ) |
|
|
| if dataset in {"allen_neuropixels", "speech"}: |
| target_component = target_class_space(dataset, targets) |
| else: |
| target_component = target_trajectory_space(dataset, targets, example_target.copy()) |
|
|
| shown = int(metadata.example_features_shown) |
| total = int(metadata.array_shape.split("×")[-1].strip()) |
| feature_text = ( |
| f"all {total} neural features are shown" |
| if shown == total |
| else f"{shown} of {total} neural features are shown for legibility" |
| ) |
| if dataset in {"allen_neuropixels", "speech"}: |
| description = ( |
| f"The highlighted target corresponds to the neural activity shown at left; " |
| f"{feature_text}." |
| ) |
| else: |
| description = f"{feature_text.capitalize()}." |
| return ( |
| neural_figure, |
| target_component, |
| description, |
| dataset_link_payload(dataset, example_target, float(metadata.target_lag_ms)), |
| ) |
|
|
|
|
| def overview_cards(dataset: str, models: Sequence[str] | None) -> list[html.Div]: |
| frame = overview_frame(dataset, models) |
| available = frame.dropna(subset=["task_score"]).sort_values( |
| ["task_score", "model_order"], ascending=[False, True] |
| ) |
| metric = "task score" |
| if not available.empty and available["metric"].notna().any(): |
| metric = metric_name(available["metric"].dropna().iloc[0], held_out=True) |
| cards = [ |
| metric_card( |
| "Dataset", |
| DATASET_LABELS.get(dataset, dataset), |
| DATASET_DESCRIPTIONS.get(dataset), |
| "prediction", |
| ), |
| metric_card( |
| "Primary outcome", |
| metric, |
| "Higher is better; raw values are shown within each task.", |
| "prediction", |
| ), |
| ] |
| if available.empty: |
| cards.append(metric_card("Leading selected method", "Unavailable", accent="prediction")) |
| else: |
| top = available.iloc[0] |
| cards.append( |
| metric_card( |
| "Leading selected method", |
| str(top["method"]), |
| f"{metric}: {value_text(top['task_score'])}", |
| "prediction", |
| ) |
| ) |
| return cards |
|
|
|
|
| def prediction_ranking_figure(dataset: str, models: Sequence[str] | None) -> go.Figure: |
| frame = overview_frame(dataset, models).dropna(subset=["task_score"]) |
| if frame.empty: |
| return empty_figure("No held-out prediction results are available for this selection.") |
| frame = frame.sort_values(["task_score", "model_order"], ascending=[True, False]) |
| metric = metric_name(frame["metric"].dropna().iloc[0], held_out=True) |
| fig = go.Figure( |
| go.Bar( |
| x=frame["task_score"], |
| y=frame["method"], |
| orientation="h", |
| marker=dict(color=PREDICTION_COLOR), |
| customdata=np.stack([frame["prediction_percentile"], frame["workflow"]], axis=-1), |
| hovertemplate=( |
| "Method=%{y}<br>Raw score=%{x:.4f}<br>" |
| "Within-dataset percentile=%{customdata[0]:.1f}<br>" |
| "Prediction workflow=%{customdata[1]}<extra></extra>" |
| ), |
| ) |
| ) |
| fig.update_layout(title="Held-out prediction") |
| fig.update_xaxes(title=metric) |
| fig.update_yaxes(title="", showgrid=False) |
| return figure_layout(fig, height=max(440, 25 * len(frame) + 145)) |
|
|
|
|
| def mean_rank_order( |
| values: pd.DataFrame, |
| value_column: str, |
| eligible_models: set[str], |
| ) -> list[str]: |
| frame = values[values["model"].isin(eligible_models)].copy() |
| frame[value_column] = pd.to_numeric(frame[value_column], errors="coerce") |
| pivot = frame.pivot_table( |
| index="model", columns="dataset", values=value_column, aggfunc="first" |
| ).reindex(columns=DATASETS) |
| n_models = len(pivot) |
| ranks = pd.concat( |
| [ |
| pivot[dataset].rank(ascending=False, method="average").fillna(n_models) |
| for dataset in DATASETS |
| ], |
| axis=1, |
| ) |
| pivot["mean_rank"] = ranks.mean(axis=1) |
| return pivot.sort_values("mean_rank", ascending=True).index.astype(str).tolist() |
|
|
|
|
| def percentile_heatmap( |
| values: pd.DataFrame, |
| models: Sequence[str] | None, |
| *, |
| title: str, |
| colorscale: list, |
| raw_column: str, |
| metric_column: str, |
| empty_message: str, |
| eligible_models: set[str] | None = None, |
| row_order: Sequence[str] | None = None, |
| ) -> go.Figure: |
| chosen = selected_models(models) |
| if eligible_models is not None: |
| chosen = [model for model in chosen if model in eligible_models] |
| if not chosen: |
| return empty_figure("No results for this selection.") |
| frame = values.copy() |
| if frame.empty: |
| return empty_figure(empty_message) |
| frame[raw_column] = pd.to_numeric(frame[raw_column], errors="coerce") |
| frame["percentile"] = ( |
| frame.groupby("dataset")[raw_column].rank(method="average", pct=True) * 100.0 |
| ) |
| frame = frame[frame["model"].isin(chosen)].copy() |
| grid = pd.MultiIndex.from_product([chosen, DATASETS], names=["model", "dataset"]).to_frame(index=False) |
| grid = grid.merge( |
| frame[["model", "dataset", raw_column, metric_column, "percentile"]], |
| on=["model", "dataset"], |
| how="left", |
| ) |
| canonical_order = list(row_order) if row_order is not None else FIGURE_MODEL_ORDER |
| ordered_models = [model for model in canonical_order if model in chosen] |
| ordered_models += [model for model in chosen if model not in set(ordered_models)] |
| percentile_matrix = grid.pivot(index="model", columns="dataset", values="percentile").reindex( |
| index=ordered_models, columns=DATASETS |
| ) |
| raw_matrix = grid.pivot(index="model", columns="dataset", values=raw_column).reindex( |
| index=ordered_models, columns=DATASETS |
| ) |
| metric_matrix = grid.pivot(index="model", columns="dataset", values=metric_column).reindex( |
| index=ordered_models, columns=DATASETS |
| ) |
| display_text = np.empty(percentile_matrix.shape, dtype=object) |
| customdata = np.empty((*percentile_matrix.shape, 2), dtype=object) |
| for row_index, model in enumerate(percentile_matrix.index): |
| for column_index, dataset in enumerate(percentile_matrix.columns): |
| percentile = percentile_matrix.iloc[row_index, column_index] |
| raw_value = raw_matrix.iloc[row_index, column_index] |
| metric = metric_matrix.iloc[row_index, column_index] |
| available = pd.notna(raw_value) |
| display_text[row_index, column_index] = "" if not available else f"{percentile:.0f}" |
| customdata[row_index, column_index, 0] = ( |
| "" if not available else f"{float(raw_value):.4f}" |
| ) |
| customdata[row_index, column_index, 1] = "" if pd.isna(metric) else str(metric) |
| fig = go.Figure( |
| go.Heatmap( |
| z=percentile_matrix.to_numpy(dtype=float), |
| x=[DATASET_TICK_LABELS[dataset] for dataset in percentile_matrix.columns], |
| y=[model_label(model) for model in percentile_matrix.index], |
| text=display_text, |
| texttemplate="%{text}", |
| textfont=dict(size=11), |
| customdata=customdata, |
| colorscale=colorscale, |
| zmin=0, |
| zmax=100, |
| colorbar=dict(title="Percentile", thickness=13, ticksuffix="th"), |
| hovertemplate=( |
| "Method=%{y}<br>Dataset=%{x}<br>" |
| "Within-dataset percentile=%{z:.1f}<br>" |
| "Raw value=%{customdata[0]}<br>Metric=%{customdata[1]}<extra></extra>" |
| ), |
| hoverongaps=False, |
| ) |
| ) |
| missing_rows, missing_columns = np.where(percentile_matrix.isna().to_numpy()) |
| if len(missing_rows): |
| fig.add_trace( |
| go.Scatter( |
| x=[DATASET_TICK_LABELS[percentile_matrix.columns[index]] for index in missing_columns], |
| y=[model_label(percentile_matrix.index[index]) for index in missing_rows], |
| mode="markers", |
| marker=dict(symbol="x", size=8, color="#8A949C", line=dict(width=1)), |
| showlegend=False, |
| hoverinfo="skip", |
| ) |
| ) |
| fig.update_layout(title=title) |
| fig.update_xaxes(title="", side="top", showgrid=False) |
| fig.update_yaxes(title="", showgrid=False) |
| return heatmap_layout(fig, height=max(500, 25 * len(percentile_matrix) + 185)) |
|
|
|
|
| def prediction_heatmap(models: Sequence[str] | None) -> go.Figure: |
| values = present_rows(prediction)[["model", "dataset", "score", "metric"]].copy() |
| values["metric_label"] = values["metric"].map(lambda value: metric_name(value, held_out=True)) |
| return percentile_heatmap( |
| values, |
| models, |
| title="Prediction across tasks", |
| colorscale=PREDICTION_SCALE, |
| raw_column="score", |
| metric_column="metric_label", |
| empty_message="No held-out prediction results are available.", |
| ) |
|
|
|
|
| def robustness_frame(dataset: str, models: Sequence[str] | None) -> pd.DataFrame: |
| chosen = selected_models(models) |
| base = add_method_columns(pd.DataFrame({"model": chosen})) |
| base["method"] = base["model"].map(lambda model: dataset_model_label(model, dataset)) |
| values = present_rows(robustness) |
| values = values[values["dataset"].astype(str) == str(dataset)].copy() |
| values = values.rename( |
| columns={ |
| "score_at_noise0": "unperturbed_score", |
| "score_at_max_noise": "highest_noise_score", |
| "raw_auc": "robustness_auc", |
| "mean_score": "average_noisy_score", |
| } |
| ) |
| columns = [ |
| "model", |
| "metric", |
| "noise_levels", |
| "scores", |
| "unperturbed_score", |
| "highest_noise_score", |
| "robustness_auc", |
| "average_noisy_score", |
| ] |
| values = values[[column for column in columns if column in values.columns]] |
| frame = base.merge(values, on="model", how="left") |
| return round_numeric(frame) |
|
|
|
|
| def robustness_figure(dataset: str, models: Sequence[str] | None) -> go.Figure: |
| frame = robustness_frame(dataset, models).dropna(subset=["robustness_auc"]) |
| if frame.empty: |
| return empty_figure("No robustness results are available for this selection.") |
| frame = frame.sort_values("model_order") |
| fig = go.Figure() |
| line_dashes = ["solid", "dash", "dot", "dashdot"] |
| for index, row in enumerate(frame.itertuples()): |
| levels = parse_float_list(row.noise_levels) |
| scores = parse_float_list(row.scores) |
| if len(levels) != len(scores): |
| raise ValueError(f"Noise levels and scores differ for {row.model} on {dataset}.") |
| fig.add_trace( |
| go.Scatter( |
| x=levels, |
| y=scores, |
| mode="lines+markers", |
| name=row.method, |
| line=dict(color=MODEL_COLORS[row.model], width=2.2, dash=line_dashes[index % len(line_dashes)]), |
| marker=dict(size=6, symbol=index % 8), |
| customdata=np.repeat(row.robustness_auc, len(levels)), |
| hovertemplate=( |
| f"Method={row.method}<br>Input-noise level λ=%{{x:.1f}}<br>" |
| "Task score=%{y:.4f}<br>Area under task-score-versus-noise curve=%{customdata:.4f}<extra></extra>" |
| ), |
| ) |
| ) |
| metric = metric_name(frame["metric"].dropna().iloc[0]) |
| fig.update_layout( |
| title="Robustness", |
| hovermode="closest", |
| showlegend=len(frame) <= 12, |
| ) |
| fig.update_xaxes(title="Input-noise level λ", tickvals=[0, 0.2, 0.4, 0.6, 0.8]) |
| fig.update_yaxes(title=metric) |
| return figure_layout(fig, height=540, legend_below=len(frame) <= 12) |
|
|
|
|
| def compute_figures( |
| dataset: str, models: Sequence[str] | None |
| ) -> tuple[go.Figure, go.Figure, pd.DataFrame]: |
| frame = overview_frame(dataset, models).dropna(subset=["training_time_sec"]) |
| if frame.empty: |
| empty = pd.DataFrame( |
| columns=[ |
| "method", |
| "hardware", |
| "task_score", |
| "training_time_sec", |
| "inference_time_sec", |
| "peak_ram_gb", |
| "peak_vram_gb", |
| ] |
| ) |
| return empty_figure("No runtime results are available."), empty_figure("No memory results are available."), empty |
| frame["hardware"] = np.where(frame["model"].isin(CPU_ONLY_MODELS), "CPU", "GPU") |
| frame = frame.sort_values(["training_time_sec", "model_order"], ascending=[False, True]) |
|
|
| runtime = go.Figure() |
| runtime.add_trace( |
| go.Bar( |
| x=frame["training_time_sec"], |
| y=frame["method"], |
| orientation="h", |
| name="Training", |
| marker=dict(color=COMPUTE_COLOR), |
| customdata=frame["task_score"], |
| hovertemplate="Method=%{y}<br>Training time=%{x:.4g} s<br>Held-out score=%{customdata:.4f}<extra></extra>", |
| ) |
| ) |
| runtime.add_trace( |
| go.Bar( |
| x=frame["inference_time_sec"], |
| y=frame["method"], |
| orientation="h", |
| name="Inference", |
| marker=dict(color="#F6A15D"), |
| hovertemplate="Method=%{y}<br>Complete held-out split=%{x:.4g} s<extra></extra>", |
| ) |
| ) |
| runtime.update_layout( |
| title="Training and inference time", |
| barmode="group", |
| ) |
| runtime.update_xaxes(title="Elapsed time (s, log)", type="log") |
| runtime.update_yaxes(title="", showgrid=False) |
| figure_layout(runtime, height=max(470, 27 * len(frame) + 155), legend_below=True) |
|
|
| memory = go.Figure() |
| memory.add_trace( |
| go.Bar( |
| x=frame["peak_ram_gb"], |
| y=frame["method"], |
| orientation="h", |
| name="Peak RAM", |
| marker=dict(color="#E6842A"), |
| hovertemplate="Method=%{y}<br>Peak RAM=%{x:.3f} GB<extra></extra>", |
| ) |
| ) |
| memory.add_trace( |
| go.Bar( |
| x=frame["peak_vram_gb"], |
| y=frame["method"], |
| orientation="h", |
| name="Peak GPU memory", |
| marker=dict(color="#F7C68B"), |
| hovertemplate="Method=%{y}<br>Peak GPU memory=%{x:.3f} GB<extra></extra>", |
| ) |
| ) |
| memory.update_layout( |
| title="Peak memory", |
| barmode="group", |
| ) |
| memory.update_xaxes(title="Memory (GB)") |
| memory.update_yaxes(title="", showgrid=False) |
| figure_layout(memory, height=max(470, 27 * len(frame) + 155), legend_below=True) |
|
|
| table = frame[ |
| [ |
| "method", |
| "hardware", |
| "task_score", |
| "training_time_sec", |
| "inference_time_sec", |
| "peak_ram_gb", |
| "peak_vram_gb", |
| ] |
| ].copy() |
| return runtime, memory, round_numeric(table) |
|
|
|
|
| def feature_spec(dataset: str) -> tuple[str, str, str, float | None]: |
| if dataset == "allen_neuropixels": |
| return ( |
| "spearman_corr", |
| "Drifting-gratings gOSI", |
| "Spearman’s r", |
| 0.0, |
| ) |
| if dataset == "ratinabox": |
| return ( |
| "auc", |
| "Place cells vs head-direction and speed cells", |
| "ROC-AUC", |
| 0.5, |
| ) |
| return ( |
| "auc", |
| "Recorded neural features vs appended synthetic controls", |
| "ROC-AUC", |
| 0.5, |
| ) |
|
|
|
|
| def feature_frame(dataset: str, models: Sequence[str] | None) -> pd.DataFrame: |
| score_column, target, metric, _ = feature_spec(dataset) |
| frame = filter_models(active_rows(neuron_shap), models) |
| frame = frame[frame["dataset"].astype(str) == str(dataset)].copy() |
| if frame.empty: |
| return frame |
| frame = add_method_columns(frame) |
| frame["method"] = frame["model"].map(lambda model: dataset_model_label(model, dataset)) |
| frame["validation_score"] = pd.to_numeric(frame[score_column], errors="coerce") |
| frame["validation_target"] = target |
| frame["validation_metric"] = metric |
| return round_numeric(frame) |
|
|
|
|
| def feature_figures( |
| dataset: str, models: Sequence[str] | None |
| ) -> tuple[go.Figure, pd.DataFrame]: |
| frame = feature_frame(dataset, models) |
| if frame.empty: |
| columns = ["method", "validation_score"] |
| return empty_figure("No results for this selection."), pd.DataFrame(columns=columns) |
| score_column, target, metric, reference = feature_spec(dataset) |
| validation = frame.dropna(subset=["validation_score"]).sort_values( |
| ["validation_score", "model_order"], ascending=[True, False] |
| ) |
| validation_fig = go.Figure( |
| go.Bar( |
| x=validation["validation_score"], |
| y=validation["method"], |
| orientation="h", |
| marker=dict(color=FEATURE_COLOR), |
| hovertemplate=f"Method=%{{y}}<br>{metric}=%{{x:.4f}}<br>Target={target}<extra></extra>", |
| ) |
| ) |
| if reference is not None: |
| validation_fig.add_vline( |
| x=reference, |
| line_dash="dash", |
| line_color="#6F7882", |
| annotation_text="0" if reference == 0 else "Chance = 0.5", |
| annotation_position="top", |
| ) |
| validation_fig.update_layout( |
| title="Feature-attribution validation" |
| ) |
| validation_fig.update_xaxes(title=metric) |
| validation_fig.update_yaxes(title="", showgrid=False) |
| figure_layout(validation_fig, height=max(430, 25 * len(validation) + 145)) |
|
|
| columns = ["method", "validation_score"] |
| return validation_fig, round_numeric( |
| frame[columns].sort_values("validation_score", ascending=False) |
| ) |
|
|
|
|
| def feature_heatmap(models: Sequence[str] | None) -> go.Figure: |
| rows = [] |
| for dataset in DATASETS: |
| score_column, _target, metric, _reference = feature_spec(dataset) |
| frame = active_rows(neuron_shap) |
| frame = frame[frame["dataset"].astype(str) == dataset] |
| for row in frame.itertuples(): |
| rows.append( |
| { |
| "model": row.model, |
| "dataset": dataset, |
| "validation_score": getattr(row, score_column), |
| "metric_label": metric, |
| } |
| ) |
| values = pd.DataFrame(rows) |
| return percentile_heatmap( |
| values, |
| models, |
| title="Feature validation across tasks", |
| colorscale=FEATURE_SCALE, |
| raw_column="validation_score", |
| metric_column="metric_label", |
| empty_message="No feature-attribution validation results are available.", |
| eligible_models=FEATURE_ELIGIBLE, |
| row_order=mean_rank_order(values, "validation_score", FEATURE_ELIGIBLE), |
| ) |
|
|
|
|
| def feature_attribution_frame(dataset: str, model: str | None) -> pd.DataFrame: |
| if not model: |
| return pd.DataFrame(columns=neuron_attributions.columns) |
| frame = neuron_attributions[ |
| neuron_attributions["dataset"].astype(str).eq(str(dataset)) |
| & neuron_attributions["model"].astype(str).eq(str(model)) |
| ].copy() |
| for column in ( |
| "feature_index", |
| "signed_attribution", |
| "attribution_rank", |
| "validation_value", |
| ): |
| frame[column] = pd.to_numeric(frame[column], errors="coerce") |
| return frame.sort_values("attribution_rank", kind="stable") |
|
|
|
|
| def feature_story_raster_payload(dataset: str) -> dict: |
| frame = feature_example_raster[ |
| feature_example_raster["dataset"].astype(str).eq(dataset) |
| ].copy() |
| if frame.empty: |
| return {"dataset": dataset, "features": [], "counts": []} |
|
|
| numeric_columns = [ |
| "display_index", |
| "feature_index", |
| "validation_value", |
| "group_order", |
| "n_time", |
| "t0_index", |
| "bin_ms", |
| ] |
| for column in numeric_columns: |
| frame[column] = pd.to_numeric(frame[column], errors="coerce") |
| if dataset == "allen_neuropixels": |
| frame = frame.sort_values( |
| ["validation_value", "feature_index"], ascending=[False, True] |
| ) |
| else: |
| frame = frame.sort_values(["group_order", "display_index"]) |
|
|
| n_time = int(frame["n_time"].iloc[0]) |
| value_columns = [f"value_{index:03d}" for index in range(n_time)] |
| counts = ( |
| frame[value_columns] |
| .apply(pd.to_numeric, errors="coerce") |
| .to_numpy(dtype=float) |
| ) |
| if not np.allclose(counts, np.rint(counts)): |
| raise ValueError(f"Feature raster contains noninteger counts for {dataset}") |
| counts = np.rint(counts).astype(int) |
| time_ms = ( |
| np.arange(n_time) - int(frame["t0_index"].iloc[0]) |
| ) * float(frame["bin_ms"].iloc[0]) |
| features = [ |
| { |
| "feature_index": int(row.feature_index), |
| "group": str(row.feature_group), |
| "validation_value": ( |
| None |
| if pd.isna(row.validation_value) |
| else float(row.validation_value) |
| ), |
| } |
| for row in frame.itertuples(index=False) |
| ] |
| return { |
| "dataset": dataset, |
| "dataset_label": DATASET_LABELS[dataset], |
| "features": features, |
| "counts": counts.tolist(), |
| "time_ms": time_ms.tolist(), |
| "max_count": int(counts.max(initial=0)), |
| } |
|
|
|
|
| def feature_story_attribution_payload( |
| dataset: str, |
| model: str | None, |
| ) -> dict: |
| frame = feature_attribution_frame(dataset, model) |
| if frame.empty: |
| return { |
| "dataset": dataset, |
| "model": model, |
| "method": "", |
| "features": [], |
| "tied": False, |
| } |
| values = pd.to_numeric(frame["signed_attribution"], errors="coerce") |
| return { |
| "dataset": dataset, |
| "model": model, |
| "method": dataset_model_label(str(model), dataset), |
| "tied": values.nunique(dropna=True) <= 1, |
| "features": [ |
| { |
| "feature_index": int(row.feature_index), |
| "rank": int(row.attribution_rank), |
| "signed_attribution": float(row.signed_attribution), |
| } |
| for row in frame.itertuples(index=False) |
| ], |
| } |
|
|
|
|
| def trial_frame(dataset: str, models: Sequence[str] | None) -> pd.DataFrame: |
| frame = filter_models(active_rows(trial_shapley), models) |
| frame = frame[frame["dataset"].astype(str) == str(dataset)].copy() |
| if frame.empty: |
| return frame |
| frame = add_method_columns(frame) |
| frame["method"] = frame["model"].map(lambda model: dataset_model_label(model, dataset)) |
| frame = frame.rename(columns={"perturbation_auc": "corrupted_trial_auc"}) |
| return round_numeric(frame) |
|
|
|
|
| def trial_detection_figure(dataset: str, models: Sequence[str] | None) -> go.Figure: |
| frame = trial_frame(dataset, models) |
| if frame.empty: |
| return empty_figure("No corrupted-trial detection result is available for this selection.") |
| frame = frame.dropna(subset=["corrupted_trial_auc"]).sort_values( |
| ["corrupted_trial_auc", "model_order"], ascending=[True, False] |
| ) |
| fig = go.Figure( |
| go.Bar( |
| x=frame["corrupted_trial_auc"], |
| y=frame["method"], |
| orientation="h", |
| marker=dict(color=TRIAL_COLOR), |
| hovertemplate="Method=%{y}<br>Corrupted-trial ROC-AUC=%{x:.4f}<extra></extra>", |
| ) |
| ) |
| fig.add_vline( |
| x=0.5, |
| line_dash="dash", |
| line_color="#6F7882", |
| annotation_text="Chance = 0.5", |
| annotation_position="top", |
| ) |
| fig.update_layout( |
| title="Corrupted-trial detection" |
| ) |
| fig.update_xaxes(title="Detection ROC-AUC") |
| fig.update_yaxes(title="", showgrid=False) |
| return figure_layout(fig, height=max(430, 25 * len(frame) + 145)) |
|
|
|
|
| def trial_heatmap(models: Sequence[str] | None) -> go.Figure: |
| values = active_rows(trial_shapley)[["model", "dataset", "perturbation_auc"]].copy() |
| values["metric_label"] = "Corrupted-trial ROC-AUC" |
| return percentile_heatmap( |
| values, |
| models, |
| title="Trial detection across tasks", |
| colorscale=TRIAL_SCALE, |
| raw_column="perturbation_auc", |
| metric_column="metric_label", |
| empty_message="No corrupted-trial detection results are available.", |
| eligible_models=TRIAL_ELIGIBLE, |
| row_order=mean_rank_order(values, "perturbation_auc", TRIAL_ELIGIBLE), |
| ) |
|
|
|
|
| def retrain_frames(models: Sequence[str] | None) -> tuple[pd.DataFrame, pd.DataFrame]: |
| chosen = selected_models(models) |
| frame = active_rows(trial_retrain) |
| frame = frame[frame["model"].astype(str).isin(chosen)].copy() |
| within = frame[frame["analysis"] == "within_session_cleaning"].pivot_table( |
| index="model", columns="condition", values="score", aggfunc="first" |
| ) |
| historical = frame[frame["analysis"] == "cross_session_old_trial_selection"].pivot_table( |
| index="model", columns="condition", values="score", aggfunc="first" |
| ) |
| within = within.reset_index() |
| historical = historical.reset_index() |
| if not within.empty: |
| within = add_method_columns(within) |
| within["recovery"] = within.get("data_shapley") - within.get("mixed_full") |
| if not historical.empty: |
| historical = add_method_columns(historical) |
| historical = historical.rename( |
| columns={"oldonly_dshap_negative_removal": "historical_selected"} |
| ) |
| return within, historical |
|
|
|
|
| def equality_bounds(*series: pd.Series) -> tuple[float, float]: |
| values = pd.concat([pd.to_numeric(item, errors="coerce") for item in series]).dropna() |
| if values.empty: |
| return 0.0, 1.0 |
| span = float(values.max() - values.min()) |
| padding = max(span * 0.08, 0.03) |
| return float(values.min() - padding), float(values.max() + padding) |
|
|
|
|
| def relative_change_in_mean(before: pd.Series, after: pd.Series) -> float: |
| """Return the percentage change between the two condition means.""" |
| before_mean = float(before.mean()) |
| if abs(before_mean) <= 1e-12: |
| raise ValueError("Cannot express the change relative to a zero comparator mean.") |
| return (float(after.mean()) - before_mean) / before_mean * 100.0 |
|
|
|
|
| def trial_retrain_figures() -> tuple[go.Figure, go.Figure, go.Figure, pd.DataFrame]: |
| within, historical = retrain_frames(None) |
| if within.empty: |
| removal = empty_figure("No macaque within-session removal summary is available.") |
| relation = empty_figure("No detection-versus-recovery summary is available.") |
| else: |
| lower, upper = equality_bounds(within["mixed_full"], within["data_shapley"]) |
| removal = go.Figure() |
| removal.add_trace( |
| go.Scatter( |
| x=within["mixed_full"], |
| y=within["data_shapley"], |
| mode="markers", |
| text=within["method"], |
| marker=dict( |
| color=[MODEL_COLORS[model] for model in within["model"]], |
| size=10, |
| line=dict(color="#FFFFFF", width=1), |
| ), |
| customdata=np.stack([within["method"], within["recovery"]], axis=-1), |
| hovertemplate=( |
| "Method=%{customdata[0]}<br>Mixed trials R²=%{x:.4f}<br>" |
| "After trial-value removal R²=%{y:.4f}<br>" |
| "Recovery ΔR²=%{customdata[1]:+.4f}<extra></extra>" |
| ), |
| ) |
| ) |
| removal.add_shape(type="line", x0=lower, x1=upper, y0=lower, y1=upper, line=dict(color="#69737D", dash="dash")) |
| removal_change = relative_change_in_mean( |
| within["mixed_full"], within["data_shapley"] |
| ) |
| removal.add_annotation( |
| text=f"Mean change = {removal_change:+.0f}%", |
| x=0.03, |
| y=0.97, |
| xref="paper", |
| yref="paper", |
| xanchor="left", |
| yanchor="top", |
| showarrow=False, |
| bgcolor="rgba(255,255,255,0.88)", |
| font=dict(size=12, color="#2E7D32" if removal_change >= 0 else NEGATIVE_COLOR), |
| ) |
| removal.update_layout(title="Trial-value-guided removal") |
| removal.update_xaxes(title="Before removal: test R²", range=[lower, upper]) |
| removal.update_yaxes(title="After removal: test R²", range=[lower, upper]) |
| figure_layout(removal, height=480) |
|
|
| detection = active_rows(trial_shapley) |
| detection = detection[detection["dataset"].astype(str) == "monkey"][["model", "perturbation_auc"]] |
| relation_frame = within.merge(detection, on="model", how="inner").dropna( |
| subset=["perturbation_auc", "recovery"] |
| ) |
| relation = go.Figure() |
| if relation_frame.empty: |
| relation = empty_figure("No shared detection and recovery entries are available.") |
| else: |
| x = relation_frame["perturbation_auc"].astype(float) |
| y = relation_frame["recovery"].astype(float) |
| rho = ( |
| x.rank(method="average").corr(y.rank(method="average")) |
| if len(relation_frame) >= 2 |
| else np.nan |
| ) |
| relation.add_trace( |
| go.Scatter( |
| x=x, |
| y=y, |
| mode="markers", |
| marker=dict( |
| color=[MODEL_COLORS[model] for model in relation_frame["model"]], |
| size=10, |
| line=dict(color="#FFFFFF", width=1), |
| ), |
| customdata=relation_frame["method"], |
| hovertemplate=( |
| "Method=%{customdata}<br>Detection ROC-AUC=%{x:.4f}<br>" |
| "Recovery ΔR²=%{y:+.4f}<extra></extra>" |
| ), |
| ) |
| ) |
| if len(relation_frame) >= 2 and float(x.max() - x.min()) > 0: |
| coefficients = np.polyfit(x, y, 1) |
| line_x = np.linspace(float(x.min()), float(x.max()), 100) |
| relation.add_trace( |
| go.Scatter( |
| x=line_x, |
| y=np.polyval(coefficients, line_x), |
| mode="lines", |
| line=dict(color=TRIAL_COLOR, width=2), |
| name="Linear fit", |
| hoverinfo="skip", |
| ) |
| ) |
| relation.add_hline(y=0, line_dash="dash", line_color="#69737D") |
| relation_text = ( |
| f"Spearman’s r = {rho:.2f}; n = {len(relation_frame)}" |
| if pd.notna(rho) |
| else f"n = {len(relation_frame)}; select at least two methods for correlation" |
| ) |
| if pd.notna(rho): |
| relation_text += "<br>one-sided permutation P = 0.047" |
| relation.add_annotation( |
| text=relation_text, |
| x=0.02, |
| y=0.98, |
| xref="paper", |
| yref="paper", |
| xanchor="left", |
| yanchor="top", |
| showarrow=False, |
| bgcolor="rgba(255,255,255,0.85)", |
| font=dict(size=12), |
| ) |
| relation.update_layout(title="Detection and recovery", showlegend=False) |
| relation.update_xaxes(title="Detection ROC-AUC") |
| relation.update_yaxes(title="Recovery (ΔR²)") |
| figure_layout(relation, height=480) |
|
|
| if historical.empty: |
| historical_fig = empty_figure("No same-subject historical-selection summary is available.") |
| else: |
| lower, upper = equality_bounds(historical["target_only"], historical["historical_selected"]) |
| historical_fig = go.Figure( |
| go.Scatter( |
| x=historical["target_only"], |
| y=historical["historical_selected"], |
| mode="markers", |
| marker=dict( |
| color=[MODEL_COLORS[model] for model in historical["model"]], |
| size=10, |
| line=dict(color="#FFFFFF", width=1), |
| ), |
| customdata=np.stack([historical["method"], historical["all_sessions"]], axis=-1), |
| hovertemplate=( |
| "Method=%{customdata[0]}<br>Current session only R²=%{x:.4f}<br>" |
| "Trial-value historical selection R²=%{y:.4f}<br>" |
| "All-session pooling R²=%{customdata[1]:.4f}<extra></extra>" |
| ), |
| ) |
| ) |
| historical_fig.add_shape(type="line", x0=lower, x1=upper, y0=lower, y1=upper, line=dict(color="#69737D", dash="dash")) |
| historical_change = relative_change_in_mean( |
| historical["target_only"], historical["historical_selected"] |
| ) |
| historical_fig.add_annotation( |
| text=f"Mean change = {historical_change:+.0f}%", |
| x=0.03, |
| y=0.97, |
| xref="paper", |
| yref="paper", |
| xanchor="left", |
| yanchor="top", |
| showarrow=False, |
| bgcolor="rgba(255,255,255,0.88)", |
| font=dict(size=12, color="#2E7D32" if historical_change >= 0 else NEGATIVE_COLOR), |
| ) |
| historical_fig.update_layout(title="Historical-trial selection") |
| historical_fig.update_xaxes(title="Current-session test R²", range=[lower, upper]) |
| historical_fig.update_yaxes( |
| title="Selected historical trials: test R²", |
| range=[lower, upper], |
| ) |
| figure_layout(historical_fig, height=480) |
|
|
| table = within[ |
| [column for column in ["model", "method", "mixed_full", "data_shapley", "recovery"] if column in within] |
| ].copy() if not within.empty else pd.DataFrame(columns=["model", "method", "mixed_full", "data_shapley", "recovery"]) |
| historical_columns = ["model", "target_only", "all_sessions", "historical_selected"] |
| if not historical.empty: |
| table = table.merge(historical[historical_columns], on="model", how="outer") |
| table["method"] = table["method"].fillna(table["model"].map(model_label)) |
| table = table.drop(columns=["model"], errors="ignore") |
| return removal, relation, historical_fig, round_numeric(table) |
|
|
|
|
| def historical_trajectory_figure(*, stacked: bool = False) -> go.Figure: |
| frame = trial_historical_trajectories.copy() |
| current_r2 = float(frame["current_only_r2"].iloc[0]) |
| historical_r2 = float(frame["historical_selected_r2"].iloc[0]) |
| panels = [ |
| ("target_x", "target_y", "Ground truth", 0.84, 2.2), |
| ( |
| "current_only_x", |
| "current_only_y", |
| f"Current session only<br>R² = {current_r2:.2f}", |
| 0.64, |
| 2.5, |
| ), |
| ( |
| "historical_selected_x", |
| "historical_selected_y", |
| f"Selected historical trials<br>R² = {historical_r2:.2f}", |
| 0.64, |
| 2.5, |
| ), |
| ] |
| if stacked: |
| fig = make_subplots( |
| rows=3, |
| cols=1, |
| vertical_spacing=0.075, |
| subplot_titles=[panel[2] for panel in panels], |
| ) |
| else: |
| fig = make_subplots( |
| rows=1, |
| cols=3, |
| horizontal_spacing=0.045, |
| subplot_titles=[panel[2] for panel in panels], |
| ) |
|
|
| direction_labels = dict(zip(DIRECTION_LEGEND_ORDER, DIRECTION_LEGEND_LABELS)) |
| for direction_rank, direction_index in enumerate(DIRECTION_LEGEND_ORDER): |
| direction_frame = frame[frame["direction_index"] == direction_index] |
| direction_label = direction_labels[direction_index] |
| for panel_index, (x_column, y_column, _title, opacity, width) in enumerate( |
| panels, start=1 |
| ): |
| subplot_row = panel_index if stacked else 1 |
| subplot_col = 1 if stacked else panel_index |
| x_values: list[object] = [] |
| y_values: list[object] = [] |
| hover_values: list[list[object]] = [] |
| for trial_index in sorted(direction_frame["trial_index"].unique()): |
| trial = direction_frame[ |
| direction_frame["trial_index"] == trial_index |
| ].sort_values("time_index") |
| x_values.extend(trial[x_column].tolist()) |
| y_values.extend(trial[y_column].tolist()) |
| hover_values.extend( |
| [ |
| [trial_id, direction_label, time_index] |
| for trial_id, time_index in zip( |
| trial["trial_id"].astype(str), trial["time_index"] |
| ) |
| ] |
| ) |
| x_values.append(None) |
| y_values.append(None) |
| hover_values.append([None, None, None]) |
| fig.add_trace( |
| go.Scatter( |
| x=x_values, |
| y=y_values, |
| mode="lines", |
| name=direction_label, |
| legendgroup=f"direction-{direction_index}", |
| legendrank=direction_rank, |
| showlegend=False, |
| opacity=opacity, |
| line=dict(color=DIRECTION_PALETTE[direction_index], width=width), |
| customdata=hover_values, |
| connectgaps=False, |
| hovertemplate=( |
| "Trial=%{customdata[0]}<br>Reach direction=%{customdata[1]}<br>" |
| "Time bin=%{customdata[2]}<br>x=%{x:.3f}<br>y=%{y:.3f}<extra></extra>" |
| ), |
| ), |
| row=subplot_row, |
| col=subplot_col, |
| ) |
|
|
| first_points = ( |
| frame.sort_values("time_index").groupby("trial_index", as_index=False).first() |
| ) |
| for panel_index, (x_column, y_column, _title, _opacity, _width) in enumerate( |
| panels, start=1 |
| ): |
| subplot_row = panel_index if stacked else 1 |
| subplot_col = 1 if stacked else panel_index |
| fig.add_trace( |
| go.Scatter( |
| x=[float(first_points[x_column].mean())], |
| y=[float(first_points[y_column].mean())], |
| mode="markers", |
| marker=dict(size=8, color="#222222"), |
| showlegend=False, |
| hovertemplate="Mean trajectory origin<extra></extra>", |
| ), |
| row=subplot_row, |
| col=subplot_col, |
| ) |
|
|
| x_values = pd.concat( |
| [frame["target_x"], frame["current_only_x"], frame["historical_selected_x"]], |
| ignore_index=True, |
| ) |
| y_values = pd.concat( |
| [frame["target_y"], frame["current_only_y"], frame["historical_selected_y"]], |
| ignore_index=True, |
| ) |
| x_span = max(float(x_values.max() - x_values.min()), 1.0) |
| y_span = max(float(y_values.max() - y_values.min()), 1.0) |
| x_range = [float(x_values.min() - 0.06 * x_span), float(x_values.max() + 0.06 * x_span)] |
| y_range = [float(y_values.min() - 0.06 * y_span), float(y_values.max() + 0.06 * y_span)] |
| for panel_index in range(1, 4): |
| subplot_row = panel_index if stacked else 1 |
| subplot_col = 1 if stacked else panel_index |
| x_axis_id = "x" if panel_index == 1 else f"x{panel_index}" |
| fig.update_xaxes( |
| range=x_range, |
| showgrid=False, |
| zeroline=False, |
| showticklabels=False, |
| ticks="", |
| row=subplot_row, |
| col=subplot_col, |
| ) |
| fig.update_yaxes( |
| range=y_range, |
| showgrid=False, |
| zeroline=False, |
| showticklabels=False, |
| ticks="", |
| scaleanchor=x_axis_id, |
| scaleratio=1, |
| row=subplot_row, |
| col=subplot_col, |
| ) |
|
|
| figure_layout(fig, height=930 if stacked else 480) |
| fig.update_layout( |
| title="Held-out trajectories · RNN", |
| margin=dict(l=24, r=24, t=76 if not stacked else 68, b=24), |
| showlegend=False, |
| ) |
| fig.for_each_annotation( |
| lambda annotation: annotation.update(font=dict(size=12, color=TEXT_COLOR)) |
| ) |
| return fig |
|
|
|
|
| def historical_direction_legend() -> html.Div: |
| labels = dict(zip(DIRECTION_LEGEND_ORDER, DIRECTION_LEGEND_LABELS)) |
| return html.Div( |
| [ |
| html.Span("Reach direction", className="trajectory-legend-title"), |
| *[ |
| html.Span( |
| [ |
| html.Span( |
| className="trajectory-legend-swatch", |
| style={"backgroundColor": DIRECTION_PALETTE[index]}, |
| ), |
| labels[index], |
| ], |
| className="trajectory-legend-item", |
| ) |
| for index in DIRECTION_LEGEND_ORDER |
| ], |
| ], |
| className="trajectory-legend", |
| ) |
|
|
|
|
| def condition_sort_key(value: object) -> tuple[int, float | str]: |
| try: |
| return (0, float(value)) |
| except (TypeError, ValueError): |
| return (1, str(value)) |
|
|
|
|
| def condition_label(dataset: str, value: object, color_mode: str = "condition") -> str: |
| if value is None or pd.isna(value): |
| return "Unknown" |
| text = str(int(float(value))) if re.fullmatch(r"-?\d+(\.0+)?", str(value)) else str(value) |
| if dataset == "ratinabox": |
| index = int(float(text)) |
| if color_mode == "x": |
| return f"x bin {index}" |
| if color_mode == "y": |
| return f"y bin {index}" |
| return f"x{index % 10}, y{index // 10}" |
| return CONDITION_LABELS.get(dataset, {}).get(text, text) |
|
|
|
|
| def condition_axis_label(dataset: str, color_mode: str = "condition") -> str: |
| if dataset == "ratinabox": |
| if color_mode == "x": |
| return "X-position bin" |
| if color_mode == "y": |
| return "Y-position bin" |
| return "Spatial bin" |
| return { |
| "monkey": "Reach direction", |
| "allen_neuropixels": "Stimulus orientation", |
| "speech": "Attempted word", |
| }.get(dataset, "Task condition") |
|
|
|
|
| def session_display_label(dataset: str, session: object) -> str: |
| text = "" if session is None or pd.isna(session) else str(session) |
| if dataset == "monkey": |
| match = re.search(r"sub-([A-Za-z])_ses-CO-", text) |
| return f"Monkey {match.group(1)}" if match else text |
| if dataset == "ratinabox": |
| sessions = ["ratinabox_nav", "ratinabox_nav_s123", "ratinabox_nav_s456", "ratinabox_nav_s789"] |
| return f"Simulation {sessions.index(text) + 1}" if text in sessions else text |
| if dataset == "speech": |
| return f"Participant {text.upper()}" |
| if dataset == "allen_neuropixels": |
| return f"Recording {text}" |
| return text or "Recording" |
|
|
|
|
| def add_latent_color_columns(df: pd.DataFrame, dataset: str, color_mode: str) -> pd.DataFrame: |
| out = df.copy() |
| values = pd.to_numeric(out["condition"], errors="coerce") |
| if values.isna().any() or (values < 0).any(): |
| raise ValueError("Latent samples contain invalid task-condition labels.") |
| if dataset == "ratinabox" and color_mode == "x": |
| color_values = values.astype(int) % 10 |
| elif dataset == "ratinabox" and color_mode == "y": |
| color_values = values.astype(int) // 10 |
| else: |
| color_values = values.astype(int) |
| out["color_value"] = color_values.astype(str) |
| out["color_num"] = color_values |
| out["color_label"] = [condition_label(dataset, value, color_mode) for value in color_values] |
| return out |
|
|
|
|
| def latent_space_figure( |
| dataset: str, |
| model: str | None, |
| color_mode: str, |
| *, |
| stacked: bool = False, |
| ) -> go.Figure: |
| if dataset == "mc_pacman": |
| return empty_figure( |
| "Cross-recording consistency is not available for this dataset.", |
| height=500, |
| ) |
| if not model: |
| return empty_figure("Select an available method to view aligned coordinates.", height=500) |
| samples = latent_samples[ |
| (latent_samples["dataset"].astype(str) == str(dataset)) |
| & (latent_samples["model"].astype(str) == str(model)) |
| ].copy() |
| if samples.empty: |
| return empty_figure("No results for this selection.", height=500) |
| for column in ["x", "y", "z"]: |
| samples[column] = pd.to_numeric(samples[column], errors="coerce") |
| samples = add_latent_color_columns(samples, dataset, color_mode).dropna(subset=["x", "y", "z"]) |
| trajectories = latent_trajectories[ |
| (latent_trajectories["dataset"].astype(str) == str(dataset)) |
| & (latent_trajectories["model"].astype(str) == str(model)) |
| ].copy() |
| if not trajectories.empty: |
| for column in ["x", "y", "z", "time_index"]: |
| trajectories[column] = pd.to_numeric(trajectories[column], errors="coerce") |
| trajectories = add_latent_color_columns(trajectories, dataset, color_mode).dropna( |
| subset=["x", "y", "z"] |
| ) |
| else: |
| trajectories["color_value"] = pd.Series(dtype=str) |
|
|
| sessions = list(dict.fromkeys(samples["session_label"].astype(str))) |
| columns = 1 if stacked else (2 if len(sessions) > 1 else 1) |
| rows = int(np.ceil(len(sessions) / columns)) |
| fig = make_subplots( |
| rows=rows, |
| cols=columns, |
| specs=[[{"type": "scene"} for _ in range(columns)] for _ in range(rows)], |
| subplot_titles=[session_display_label(dataset, session) for session in sessions], |
| horizontal_spacing=0.04, |
| vertical_spacing=0.1, |
| ) |
| condition_values = sorted(samples["color_value"].unique(), key=condition_sort_key) |
| categorical = dataset != "ratinabox" |
| if dataset == "monkey": |
| colors = {value: DIRECTION_PALETTE[int(value) % len(DIRECTION_PALETTE)] for value in condition_values} |
| elif dataset == "speech": |
| colors = {value: SPEECH_PALETTE.get(value, "#777777") for value in condition_values} |
| else: |
| colors = {value: ALLEN_PALETTE.get(value, "#777777") for value in condition_values} |
| condition_name = condition_axis_label(dataset, color_mode) |
|
|
| for session_index, session in enumerate(sessions): |
| row_index = session_index // columns + 1 |
| column_index = session_index % columns + 1 |
| session_samples = samples[samples["session_label"].astype(str) == session] |
| display_session = session_display_label(dataset, session) |
| if categorical: |
| for condition in condition_values: |
| points = session_samples[session_samples["color_value"] == condition] |
| if points.empty: |
| continue |
| means = trajectories[ |
| (trajectories["session_label"].astype(str) == session) |
| & (trajectories["color_value"] == condition) |
| ].sort_values("time_index") |
| label = condition_label(dataset, condition, color_mode) |
| fig.add_trace( |
| go.Scatter3d( |
| x=points["x"], |
| y=points["y"], |
| z=points["z"], |
| mode="markers", |
| name=label, |
| legendgroup=condition, |
| showlegend=session_index == 0, |
| marker=dict(size=2.6, opacity=0.42 if not means.empty else 0.76, color=colors[condition]), |
| customdata=np.stack( |
| [ |
| np.repeat(display_session, len(points)), |
| points["color_label"], |
| points["trial_index"], |
| points["time_index"], |
| np.repeat(condition_name, len(points)), |
| ], |
| axis=-1, |
| ), |
| hoverinfo="none", |
| ), |
| row=row_index, |
| col=column_index, |
| ) |
| if not means.empty: |
| fig.add_trace( |
| go.Scatter3d( |
| x=means["x"], |
| y=means["y"], |
| z=means["z"], |
| mode="lines", |
| name=label, |
| legendgroup=condition, |
| showlegend=False, |
| line=dict(color=colors[condition], width=5), |
| hoverinfo="skip", |
| customdata=means["time_index"], |
| ), |
| row=row_index, |
| col=column_index, |
| ) |
| else: |
| fig.add_trace( |
| go.Scatter3d( |
| x=session_samples["x"], |
| y=session_samples["y"], |
| z=session_samples["z"], |
| mode="markers", |
| name=display_session, |
| showlegend=False, |
| marker=dict( |
| size=2.8, |
| opacity=0.74, |
| color=session_samples["color_num"], |
| colorscale=RATINABOX_SCALE, |
| cmin=0, |
| cmax=9 if color_mode in {"x", "y"} else 99, |
| showscale=session_index == 0, |
| colorbar=dict(title=condition_name, thickness=12), |
| ), |
| customdata=np.stack( |
| [ |
| np.repeat(display_session, len(session_samples)), |
| session_samples["color_label"], |
| session_samples["trial_index"], |
| session_samples["time_index"], |
| np.repeat(condition_name, len(session_samples)), |
| ], |
| axis=-1, |
| ), |
| hoverinfo="none", |
| ), |
| row=row_index, |
| col=column_index, |
| ) |
|
|
| extent = max( |
| float(np.nanpercentile(np.abs(samples[["x", "y", "z"]].to_numpy()), 99)), |
| 1.0, |
| ) * 1.08 |
| for scene_index in range(len(sessions)): |
| scene_id = "scene" if scene_index == 0 else f"scene{scene_index + 1}" |
| fig.update_layout( |
| **{ |
| scene_id: dict( |
| xaxis=dict(range=[-extent, extent], visible=False), |
| yaxis=dict(range=[-extent, extent], visible=False), |
| zaxis=dict(range=[-extent, extent], visible=False), |
| aspectmode="cube", |
| bgcolor="#FFFFFF", |
| camera=dict(eye=dict(x=1.5, y=1.4, z=1.0)), |
| ) |
| } |
| ) |
| score_rows = active_rows(consistency) |
| score_rows = score_rows[ |
| (score_rows["dataset"].astype(str) == str(dataset)) |
| & (score_rows["model"].astype(str) == str(model)) |
| ] |
| score = pd.to_numeric(score_rows.get("mean_r2"), errors="coerce").dropna() |
| title = model_label(model) |
| if not score.empty: |
| title += f" · R² = {float(score.iloc[0]):.3f}" |
| fig.update_layout( |
| title=title, |
| height=(max(520, 300 * rows + 110) if stacked else (700 if rows > 1 else 500)), |
| paper_bgcolor="#FFFFFF", |
| plot_bgcolor="#FFFFFF", |
| margin=dict(l=8, r=8, t=72, b=88), |
| font=dict(family="Arial, Helvetica, sans-serif", size=12, color=TEXT_COLOR), |
| legend=dict( |
| orientation="h", |
| yanchor="top", |
| y=-0.04, |
| xanchor="center", |
| x=0.5, |
| entrywidth=46, |
| entrywidthmode="pixels", |
| ), |
| showlegend=not stacked, |
| ) |
| fig.for_each_annotation(lambda annotation: annotation.update(font=dict(size=12, color="#526171"))) |
| return fig |
|
|
|
|
| def latent_mobile_legend(dataset: str, color_mode: str) -> html.Div: |
| frame = latent_samples[latent_samples["dataset"].astype(str).eq(dataset)].copy() |
| frame = add_latent_color_columns(frame, dataset, color_mode) |
| label = condition_axis_label(dataset, color_mode) |
| if dataset == "ratinabox": |
| colors = [RATINABOX_SCALE[0][1], RATINABOX_SCALE[len(RATINABOX_SCALE) // 2][1], RATINABOX_SCALE[-1][1]] |
| return html.Div( |
| [ |
| html.Span(label, className="latent-mobile-legend-title"), |
| *[ |
| html.Span( |
| className="latent-mobile-legend-dot", |
| style={"backgroundColor": color}, |
| ) |
| for color in colors |
| ], |
| ], |
| className="latent-mobile-legend", |
| ) |
|
|
| values = sorted(frame["color_value"].dropna().unique(), key=condition_sort_key) |
| if dataset == "monkey": |
| colors = { |
| value: DIRECTION_PALETTE[int(value) % len(DIRECTION_PALETTE)] |
| for value in values |
| } |
| elif dataset == "speech": |
| colors = {value: SPEECH_PALETTE.get(value, "#777777") for value in values} |
| else: |
| colors = {value: ALLEN_PALETTE.get(value, "#777777") for value in values} |
| return html.Div( |
| [ |
| html.Span(label, className="latent-mobile-legend-title"), |
| *[ |
| html.Span( |
| [ |
| html.Span( |
| className="latent-mobile-legend-dot", |
| style={"backgroundColor": colors[value]}, |
| ), |
| condition_label(dataset, value, color_mode), |
| ], |
| className="latent-mobile-legend-item", |
| ) |
| for value in values |
| ], |
| ], |
| className="latent-mobile-legend", |
| ) |
|
|
|
|
| def consistency_frame(dataset: str, models: Sequence[str] | None) -> pd.DataFrame: |
| frame = filter_models(active_rows(consistency), models) |
| frame = frame[frame["dataset"].astype(str) == str(dataset)].copy() |
| if frame.empty: |
| return frame |
| frame = add_method_columns(frame) |
| frame = frame.rename( |
| columns={ |
| "mean_r2": "latent_consistency_r2", |
| "n_sessions": "n_recordings", |
| } |
| ) |
| return round_numeric(frame) |
|
|
|
|
| def consistency_figures( |
| dataset: str, |
| models: Sequence[str] | None, |
| ) -> tuple[go.Figure, go.Figure, pd.DataFrame]: |
| frame = consistency_frame(dataset, models) |
| if frame.empty: |
| message = ( |
| "Cross-recording consistency is not available for this dataset." |
| if dataset == "mc_pacman" |
| else "No latent-consistency result is available for this selection." |
| ) |
| columns = ["method", "latent_consistency_r2"] |
| return empty_figure(message), consistency_heatmap(models), pd.DataFrame(columns=columns) |
| bar = frame.sort_values(["latent_consistency_r2", "model_order"], ascending=[True, False]) |
| bar_fig = go.Figure( |
| go.Bar( |
| x=bar["latent_consistency_r2"], |
| y=bar["method"], |
| orientation="h", |
| marker=dict(color=CONSISTENCY_COLOR), |
| hovertemplate="Method=%{y}<br>Latent-consistency R²=%{x:.4f}<extra></extra>", |
| ) |
| ) |
| bar_fig.update_layout(title="Latent consistency") |
| bar_fig.update_xaxes(title="Consistency R²", range=[0, 1.02]) |
| bar_fig.update_yaxes(title="", showgrid=False) |
| figure_layout(bar_fig, height=max(400, 27 * len(bar) + 145)) |
| columns = ["method", "latent_consistency_r2"] |
| return bar_fig, consistency_heatmap(models), frame[columns].sort_values("latent_consistency_r2", ascending=False) |
|
|
|
|
| def consistency_heatmap(models: Sequence[str] | None) -> go.Figure: |
| chosen = selected_models(models) |
| frame = filter_models(active_rows(consistency), chosen) |
| if frame.empty: |
| return empty_figure("No cross-recording latent-consistency results are available.") |
| frame["mean_r2"] = pd.to_numeric(frame["mean_r2"], errors="coerce") |
| row_order = [model for model in FIGURE_MODEL_ORDER if model in CONSISTENCY_ELIGIBLE and model in chosen] |
| datasets = [dataset for dataset in DATASETS if dataset != "mc_pacman"] |
| pivot = frame.pivot_table(index="model", columns="dataset", values="mean_r2", aggfunc="first").reindex( |
| index=row_order, columns=datasets |
| ) |
| text = np.empty(pivot.shape, dtype=object) |
| for row in range(pivot.shape[0]): |
| for column in range(pivot.shape[1]): |
| value = pivot.iloc[row, column] |
| text[row, column] = "" if pd.isna(value) else f"{value:.2f}" |
| fig = go.Figure( |
| go.Heatmap( |
| z=pivot.to_numpy(dtype=float), |
| x=[DATASET_TICK_LABELS[dataset] for dataset in pivot.columns], |
| y=[model_label(model) for model in pivot.index], |
| text=text, |
| texttemplate="%{text}", |
| colorscale=CONSISTENCY_SCALE, |
| zmin=0, |
| zmax=1, |
| colorbar=dict(title="R²", thickness=13), |
| hovertemplate="Method=%{y}<br>Dataset=%{x}<br>Latent-consistency R²=%{z:.4f}<extra></extra>", |
| hoverongaps=False, |
| ) |
| ) |
| missing_rows, missing_columns = np.where(pivot.isna().to_numpy()) |
| if len(missing_rows): |
| fig.add_trace( |
| go.Scatter( |
| x=[DATASET_TICK_LABELS[pivot.columns[index]] for index in missing_columns], |
| y=[model_label(pivot.index[index]) for index in missing_rows], |
| mode="markers", |
| marker=dict(symbol="x", size=8, color="#8A949C", line=dict(width=1)), |
| showlegend=False, |
| hoverinfo="skip", |
| ) |
| ) |
| fig.update_layout(title="Latent consistency across tasks") |
| fig.update_xaxes(title="", side="top", showgrid=False) |
| fig.update_yaxes(title="", showgrid=False) |
| return heatmap_layout(fig, height=max(470, 27 * len(pivot) + 180)) |
|
|
|
|
| app = Dash(__name__, title="BEND-BCI Interactive Benchmark") |
| server = app.server |
|
|
|
|
| @server.route("/download/<path:filename>") |
| def download_data(filename: str): |
| if filename not in DOWNLOADABLE_FILES: |
| abort(404) |
| return send_from_directory(DATA_DIR, filename, as_attachment=True) |
|
|
|
|
| app.layout = html.Div( |
| [ |
| html.Header( |
| [ |
| html.Div( |
| [ |
| html.A("BEND-BCI", href="/", className="site-brand"), |
| html.Nav( |
| [ |
| html.Span( |
| ["Code & data", html.Small("coming soon")], |
| className="nav-placeholder", |
| title="Code and data will be released with the benchmark.", |
| ), |
| html.A( |
| "Paper", |
| href="https://www.biorxiv.org/content/10.64898/2026.07.21.739953v1", |
| target="_blank", |
| rel="noopener noreferrer", |
| ), |
| html.Span( |
| ["Submit a model", html.Small("coming soon")], |
| className="nav-placeholder", |
| title="A model-submission workflow is planned.", |
| ), |
| ], |
| className="hero-links", |
| **{"aria-label": "Resources"}, |
| ), |
| ], |
| className="site-nav", |
| ), |
| html.Div( |
| [ |
| html.H1( |
| "Benchmarking Neural Decoders for Brain-Computer Interfaces " |
| "and Neural Population Analysis" |
| ), |
| ], |
| className="hero-copy", |
| ), |
| ], |
| className="hero", |
| ), |
| html.Div( |
| [ |
| html.Div( |
| [ |
| html.Label("Dataset", htmlFor="dataset-filter"), |
| dcc.Dropdown( |
| id="dataset-filter", |
| options=[{"label": DATASET_LABELS[dataset], "value": dataset} for dataset in DATASETS], |
| value=DATASETS[0], |
| clearable=False, |
| searchable=False, |
| ), |
| ], |
| className="control", |
| ), |
| html.Div( |
| [ |
| html.Label("Methods", htmlFor="method-filter"), |
| dcc.Dropdown( |
| id="method-filter", |
| options=[{"label": model_label(model), "value": model} for model in FIGURE_MODEL_ORDER], |
| value=[], |
| multi=True, |
| placeholder="All methods", |
| ), |
| ], |
| className="control method-control", |
| ), |
| ], |
| className="toolbar", |
| ), |
| html.Main( |
| dcc.Tabs( |
| id="tabs", |
| value="overview", |
| className="tabs", |
| children=[ |
| dcc.Tab( |
| label="Datasets", |
| value="datasets", |
| className="tab", |
| selected_className="tab tab-selected", |
| children=[ |
| panel( |
| "Benchmark datasets", |
| html.Div(id="dataset-cards", className="dataset-card-grid"), |
| subtitle="Motor, visual, speech and spatial decoding across recordings, participants and simulations. Array dimensions are trials × time bins × features.", |
| class_name="axis-datasets", |
| ), |
| panel( |
| "Neural activity and task targets", |
| dcc.Store(id="dataset-link-data"), |
| html.Span( |
| id="dataset-link-render-token", |
| className="feature-story-render-token", |
| ), |
| html.Div( |
| [ |
| html.Div( |
| [ |
| html.Div( |
| [ |
| html.Strong("Example neural activity"), |
| html.Span("One benchmark trial"), |
| ], |
| className="dataset-viz-heading", |
| ), |
| graph_box( |
| "dataset-neural-example", |
| "Example trial neural activity for the selected dataset.", |
| ), |
| ], |
| className="dataset-example-card dataset-neural-card", |
| ), |
| html.Div( |
| id="dataset-link-controls", |
| className="dataset-link-controls", |
| ), |
| html.Div( |
| id="dataset-target-space", |
| className="dataset-example-card dataset-target-card", |
| ), |
| ], |
| className="dataset-example-grid", |
| ), |
| html.Div(id="dataset-example-description", className="dataset-example-note"), |
| html.Div( |
| [ |
| source_link("dataset_overview.csv", "Dataset manifest"), |
| source_link("dataset_example_neural.csv", "Example neural data"), |
| source_link("dataset_targets.csv", "All targets"), |
| ], |
| className="download-grid panel-downloads", |
| ), |
| class_name="axis-datasets", |
| ), |
| ], |
| ), |
| dcc.Tab( |
| label="Overview", |
| value="overview", |
| className="tab", |
| selected_className="tab tab-selected", |
| children=[ |
| panel( |
| "Predictive performance", |
| html.Div(id="overview-cards", className="metric-strip"), |
| html.Div( |
| [ |
| graph_box("prediction-ranking", "Raw held-out prediction scores for the selected dataset."), |
| graph_box( |
| "prediction-heatmap", |
| "Within-dataset prediction percentiles across five tasks.", |
| class_name="heatmap-graph", |
| ), |
| ], |
| className="chart-grid two", |
| ), |
| details_table("View data", dataframe_table("overview-table", page_size=23)), |
| source_link("clean_prediction_summary.csv"), |
| subtitle="Raw task scores are shown by dataset; the cross-dataset view uses within-dataset percentiles.", |
| class_name="axis-prediction", |
| ), |
| panel( |
| "Robustness to noisy inputs", |
| graph_box("robustness-curve", "Task score as input-noise level increases."), |
| details_table("View data", dataframe_table("robustness-table", page_size=23)), |
| source_link("robustness_summary.csv"), |
| subtitle="Performance as controlled noise is added to held-out neural inputs.", |
| class_name="axis-robustness", |
| ), |
| panel( |
| "Computational cost", |
| html.Div( |
| [ |
| graph_box("runtime-bars", "Training and complete-held-out-split inference times."), |
| graph_box("memory-bars", "Peak RAM and GPU memory."), |
| ], |
| className="chart-grid two", |
| ), |
| details_table("View data", dataframe_table("compute-table", page_size=23)), |
| source_link("scalability_summary.csv"), |
| subtitle="Training time, inference time, RAM and GPU memory under the benchmark protocol.", |
| class_name="axis-compute", |
| ), |
| ], |
| ), |
| dcc.Tab( |
| label="Latent consistency", |
| value="consistency", |
| className="tab", |
| selected_className="tab tab-selected", |
| children=[ |
| panel( |
| "Representation consistency across recordings", |
| html.Div( |
| [ |
| html.Div( |
| [ |
| html.Label("Method", htmlFor="consistency-method"), |
| dcc.Dropdown(id="consistency-method", clearable=False), |
| ], |
| className="control", |
| ), |
| html.Div( |
| [ |
| html.Label("Color by", htmlFor="latent-color-mode"), |
| dcc.Dropdown(id="latent-color-mode", clearable=False), |
| ], |
| id="latent-color-control", |
| className="control", |
| ), |
| ], |
| className="inline-controls", |
| ), |
| html.Div( |
| "Hover a point to inspect its recording, condition, trial and time bin.", |
| id="latent-hover-detail", |
| className="latent-hover-detail", |
| **{"aria-live": "polite"}, |
| ), |
| graph_box( |
| "latent-space", |
| "Aligned latent representations for each recording.", |
| class_name="latent-graph latent-space-desktop", |
| ), |
| graph_box( |
| "latent-space-mobile", |
| "Aligned latent representations stacked for narrow screens.", |
| class_name="latent-graph latent-space-mobile", |
| ), |
| html.Div(id="latent-mobile-legend"), |
| html.Div( |
| [ |
| graph_box("consistency-bars", "Latent-consistency R-squared for the selected dataset."), |
| graph_box( |
| "consistency-heatmap", |
| "Latent-consistency R-squared across four tasks.", |
| class_name="heatmap-graph", |
| ), |
| ], |
| className="chart-grid two", |
| ), |
| details_table("View data", dataframe_table("consistency-table", page_size=12)), |
| source_link("consistency_summary.csv"), |
| subtitle="Plots show whitened latent spaces aligned to a common display frame. Consistency measures linear alignment of matched task landmarks across recordings, participants or simulations.", |
| class_name="axis-consistency", |
| ) |
| ], |
| ), |
| dcc.Tab( |
| label="Feature attribution", |
| value="feature", |
| className="tab", |
| selected_className="tab tab-selected", |
| children=[ |
| panel( |
| "Feature-level attribution", |
| html.Div( |
| [ |
| html.Label("Method", htmlFor="feature-method"), |
| dcc.Dropdown(id="feature-method", clearable=False), |
| ], |
| className="control feature-method-control", |
| ), |
| dcc.Store(id="feature-story-raster-data"), |
| dcc.Store(id="feature-story-attribution-data"), |
| html.Div( |
| id="feature-story-viz", |
| className="feature-story-viz", |
| role="group", |
| tabIndex=0, |
| **{ |
| "aria-label": ( |
| "Linked example neural activity and signed " |
| "Kernel SHAP feature ranking." |
| ) |
| }, |
| ), |
| html.Span( |
| id="feature-story-render-token", |
| className="feature-story-render-token", |
| ), |
| html.Div( |
| "Hover or tap a raster row or attribution dot to trace the same feature.", |
| id="feature-selection-detail", |
| className="feature-selection-detail", |
| **{"aria-live": "polite"}, |
| ), |
| html.Div( |
| [ |
| source_link("feature_example_raster.csv", "Example raster CSV"), |
| source_link("neuron_attributions.csv", "Feature-level CSV"), |
| ], |
| className="download-grid panel-downloads", |
| ), |
| subtitle="Follow each input feature from an example trial to its signed Kernel SHAP rank.", |
| class_name="axis-feature", |
| ), |
| panel( |
| "Feature-attribution validation", |
| html.Div( |
| [ |
| graph_box("feature-validation-bars", "Feature-attribution validation metric for the selected dataset."), |
| graph_box( |
| "feature-heatmap", |
| "Within-dataset feature-attribution validation percentiles across tasks.", |
| class_name="heatmap-graph", |
| ), |
| ], |
| className="chart-grid two", |
| ), |
| html.Div( |
| html.Span(id="feature-definition"), |
| className="method-note", |
| ), |
| details_table("View data", dataframe_table("feature-table", page_size=23)), |
| source_link("neuron_shap_summary.csv"), |
| subtitle="Agreement with predefined, dataset-specific validation targets.", |
| class_name="axis-feature", |
| ) |
| ], |
| ), |
| dcc.Tab( |
| label="Trial valuation", |
| value="trial", |
| className="tab", |
| selected_className="tab tab-selected", |
| children=[ |
| panel( |
| "Corrupted-trial detection", |
| html.Div( |
| [ |
| graph_box("trial-detection-bars", "Corrupted-trial detection ROC-AUC for the selected dataset."), |
| graph_box( |
| "trial-heatmap", |
| "Within-dataset corrupted-trial detection percentiles across tasks.", |
| class_name="heatmap-graph", |
| ), |
| ], |
| className="chart-grid two", |
| ), |
| html.Div( |
| [ |
| "One third of training trials were rotated 75° in population-activity space while targets were unchanged. ROC-AUC uses negative trial value as the detection score.", |
| ], |
| className="method-note", |
| ), |
| details_table("View data", dataframe_table("trial-table", page_size=23)), |
| source_link("trial_shapley_summary.csv"), |
| subtitle="ROC-AUC measures whether lower trial values identify training trials with rotated neural activity.", |
| class_name="axis-trial", |
| ), |
| panel( |
| "Macaque center-out reaching training-data interventions", |
| html.Div( |
| [ |
| graph_box("trial-removal", "Held-out R-squared before and after trial-value-guided corrupted-trial removal."), |
| graph_box("trial-recovery", "Relationship between corrupted-trial detection and recovery after removal."), |
| ], |
| className="chart-grid two", |
| ), |
| graph_box("trial-historical", "Same-subject historical-trial selection compared with current-session training."), |
| graph_box( |
| "trial-historical-trajectories", |
| "Held-out RNN target-session trajectories for ground truth, current-session training, and nonnegative-valued historical-trial selection.", |
| class_name="historical-trajectory-graph historical-trajectory-desktop", |
| ), |
| graph_box( |
| "trial-historical-trajectories-mobile", |
| "Held-out RNN target-session trajectories, stacked for narrow screens.", |
| class_name="historical-trajectory-graph historical-trajectory-mobile", |
| ), |
| historical_direction_legend(), |
| details_table("View data", dataframe_table("trial-retrain-table", page_size=17)), |
| html.Div( |
| [ |
| source_link("trial_shapley_retrain_summary.csv", "Summary CSV"), |
| source_link("trial_historical_trajectories.csv", "RNN trajectory CSV"), |
| ], |
| className="download-grid panel-downloads", |
| ), |
| subtitle="Removing negative-valued trials improved 13 of 17 methods (mean R² 0.775→0.798). Selecting nonnegative-valued historical trials raised mean held-out R² to 0.617, versus 0.532 for current-only training and 0.546 for all-session pooling.", |
| class_name="axis-trial", |
| ), |
| ], |
| ), |
| ], |
| ), |
| className="main-content", |
| ), |
| html.Footer( |
| [ |
| html.Span("BEND-BCI · Tang Lab"), |
| html.Span( |
| ["Code & data", html.Small("coming soon")], |
| className="nav-placeholder", |
| title="Code and data will be released with the benchmark.", |
| ), |
| ], |
| className="provenance-footer", |
| ), |
| ], |
| className="app-shell", |
| ) |
|
|
|
|
| app.clientside_callback( |
| """ |
| function(desktopHover, desktopClick, mobileHover, mobileClick) { |
| const fallback = "Hover a point to inspect its recording, condition, trial and time bin."; |
| const triggered = dash_clientside.callback_context.triggered[0]; |
| if (!triggered) { |
| return fallback; |
| } |
| const prop = triggered.prop_id || ""; |
| let eventData = null; |
| if (prop.startsWith("latent-space-mobile.clickData")) { |
| eventData = mobileClick; |
| } else if (prop.startsWith("latent-space-mobile.hoverData")) { |
| eventData = mobileHover; |
| } else if (prop.startsWith("latent-space.clickData")) { |
| eventData = desktopClick; |
| } else { |
| eventData = desktopHover; |
| } |
| const point = eventData && eventData.points && eventData.points[0]; |
| const values = point && point.customdata; |
| if (!Array.isArray(values) || values.length < 5) { |
| return fallback; |
| } |
| return `${values[0]} · ${values[4]}: ${values[1]} · Trial ${values[2]} · Time bin ${values[3]}`; |
| } |
| """, |
| Output("latent-hover-detail", "children"), |
| Input("latent-space", "hoverData"), |
| Input("latent-space", "clickData"), |
| Input("latent-space-mobile", "hoverData"), |
| Input("latent-space-mobile", "clickData"), |
| ) |
|
|
|
|
| @app.callback( |
| Output("dataset-cards", "children"), |
| Output("dataset-neural-example", "figure"), |
| Output("dataset-target-space", "children"), |
| Output("dataset-example-description", "children"), |
| Output("dataset-link-data", "data"), |
| Input("dataset-filter", "value"), |
| ) |
| def update_dataset_examples(dataset: str): |
| dataset = dataset or DATASETS[0] |
| neural, target, description, link_data = dataset_example_figures(dataset) |
| return dataset_cards(dataset), neural, target, description, link_data |
|
|
|
|
| app.clientside_callback( |
| """ |
| function(linkData, activeTab) { |
| if (!window.benchdashDatasetLink) { |
| return window.dash_clientside.no_update; |
| } |
| return window.benchdashDatasetLink.schedule(linkData, activeTab); |
| } |
| """, |
| Output("dataset-link-render-token", "children"), |
| Input("dataset-link-data", "data"), |
| Input("tabs", "value"), |
| ) |
|
|
|
|
| @app.callback( |
| Output("overview-cards", "children"), |
| Output("overview-table", "columns"), |
| Output("overview-table", "data"), |
| Output("prediction-ranking", "figure"), |
| Output("prediction-heatmap", "figure"), |
| Output("robustness-curve", "figure"), |
| Output("robustness-table", "columns"), |
| Output("robustness-table", "data"), |
| Output("runtime-bars", "figure"), |
| Output("memory-bars", "figure"), |
| Output("compute-table", "columns"), |
| Output("compute-table", "data"), |
| Input("dataset-filter", "value"), |
| Input("method-filter", "value"), |
| ) |
| def update_overview(dataset: str, models: list[str] | None): |
| dataset = dataset or DATASETS[0] |
| frame = overview_frame(dataset, models) |
| overview_columns = [ |
| "method", |
| "workflow", |
| "task_score", |
| "prediction_percentile", |
| "robustness_auc", |
| "training_time_sec", |
| "inference_time_sec", |
| "peak_ram_gb", |
| "peak_vram_gb", |
| ] |
| overview_table = frame.dropna(subset=["task_score"]).sort_values( |
| ["task_score", "model_order"], |
| ascending=[False, True], |
| na_position="last", |
| )[overview_columns] |
| robustness_table = robustness_frame(dataset, models) |
| robustness_columns = [ |
| "method", |
| "unperturbed_score", |
| "highest_noise_score", |
| "robustness_auc", |
| "average_noisy_score", |
| ] |
| robustness_table = robustness_table[robustness_columns].dropna(subset=["robustness_auc"]).sort_values( |
| "robustness_auc", ascending=False |
| ) |
| runtime, memory, compute_table = compute_figures(dataset, models) |
| return ( |
| overview_cards(dataset, models), |
| column_defs(overview_columns), |
| records(round_numeric(overview_table)), |
| prediction_ranking_figure(dataset, models), |
| prediction_heatmap(models), |
| robustness_figure(dataset, models), |
| column_defs(robustness_columns), |
| records(round_numeric(robustness_table)), |
| runtime, |
| memory, |
| column_defs(compute_table.columns), |
| records(compute_table), |
| ) |
|
|
|
|
| @app.callback( |
| Output("latent-color-mode", "options"), |
| Output("latent-color-mode", "value"), |
| Output("latent-color-control", "style"), |
| Input("dataset-filter", "value"), |
| ) |
| def update_latent_color_control(dataset: str): |
| if dataset == "ratinabox": |
| return ( |
| [ |
| {"label": "Spatial bin", "value": "condition"}, |
| {"label": "X-position bin", "value": "x"}, |
| {"label": "Y-position bin", "value": "y"}, |
| ], |
| "x", |
| {}, |
| ) |
| return ([{"label": condition_axis_label(dataset), "value": "condition"}], "condition", {"display": "none"}) |
|
|
|
|
| @app.callback( |
| Output("consistency-method", "options"), |
| Output("consistency-method", "value"), |
| Output("consistency-method", "disabled"), |
| Input("dataset-filter", "value"), |
| Input("method-filter", "value"), |
| State("consistency-method", "value"), |
| ) |
| def update_consistency_selector(dataset: str, models: list[str] | None, current: str | None): |
| dataset = dataset or DATASETS[0] |
| if dataset == "mc_pacman": |
| return [], None, True |
| frame = consistency_frame(dataset, models) |
| latent_pairs = set(zip(latent_samples["model"].astype(str), latent_samples["dataset"].astype(str))) |
| available = [] |
| if not frame.empty: |
| available = [ |
| model |
| for model in frame.sort_values("latent_consistency_r2", ascending=False)["model"].astype(str) |
| if (model, dataset) in latent_pairs |
| ] |
| options = [{"label": model_label(model), "value": model} for model in available] |
| value = current if current in available else (available[0] if available else None) |
| return options, value, not bool(options) |
|
|
|
|
| @app.callback( |
| Output("latent-space", "figure"), |
| Output("latent-space-mobile", "figure"), |
| Output("latent-mobile-legend", "children"), |
| Output("consistency-bars", "figure"), |
| Output("consistency-heatmap", "figure"), |
| Output("consistency-table", "columns"), |
| Output("consistency-table", "data"), |
| Input("dataset-filter", "value"), |
| Input("method-filter", "value"), |
| Input("consistency-method", "value"), |
| Input("latent-color-mode", "value"), |
| ) |
| def update_consistency( |
| dataset: str, |
| models: list[str] | None, |
| method: str | None, |
| color_mode: str | None, |
| ): |
| dataset = dataset or DATASETS[0] |
| bars, heatmap, table = consistency_figures(dataset, models) |
| resolved_color_mode = color_mode or "condition" |
| return ( |
| latent_space_figure(dataset, method, resolved_color_mode), |
| latent_space_figure(dataset, method, resolved_color_mode, stacked=True), |
| latent_mobile_legend(dataset, resolved_color_mode), |
| bars, |
| heatmap, |
| column_defs(table.columns), |
| records(round_numeric(table)), |
| ) |
|
|
|
|
| @app.callback( |
| Output("feature-method", "options"), |
| Output("feature-method", "value"), |
| Output("feature-method", "disabled"), |
| Input("dataset-filter", "value"), |
| Input("method-filter", "value"), |
| State("feature-method", "value"), |
| ) |
| def update_feature_selector( |
| dataset: str, |
| models: list[str] | None, |
| current: str | None, |
| ): |
| dataset = dataset or DATASETS[0] |
| frame = feature_frame(dataset, models).dropna(subset=["validation_score"]) |
| available_pairs = set( |
| zip( |
| neuron_attributions["model"].astype(str), |
| neuron_attributions["dataset"].astype(str), |
| ) |
| ) |
| available = [ |
| model |
| for model in frame.sort_values( |
| ["validation_score", "model_order"], ascending=[False, True] |
| )["model"].astype(str) |
| if (model, dataset) in available_pairs |
| ] |
| options = [ |
| {"label": dataset_model_label(model, dataset), "value": model} |
| for model in available |
| ] |
| value = current if current in available else (available[0] if available else None) |
| return options, value, not bool(options) |
|
|
|
|
| @app.callback( |
| Output("feature-story-raster-data", "data"), |
| Input("dataset-filter", "value"), |
| ) |
| def update_feature_story_raster(dataset: str): |
| return feature_story_raster_payload(dataset or DATASETS[0]) |
|
|
|
|
| @app.callback( |
| Output("feature-story-attribution-data", "data"), |
| Input("dataset-filter", "value"), |
| Input("feature-method", "value"), |
| ) |
| def update_feature_story_attribution( |
| dataset: str, |
| method: str | None, |
| ): |
| return feature_story_attribution_payload(dataset or DATASETS[0], method) |
|
|
|
|
| app.clientside_callback( |
| """ |
| function(rasterData, attributionData, activeTab) { |
| if (!rasterData || !attributionData || |
| activeTab !== "feature" || |
| rasterData.dataset !== attributionData.dataset || |
| !window.benchdashFeatureStory) { |
| return window.dash_clientside.no_update; |
| } |
| window.benchdashFeatureStory.schedule( |
| "feature-story-viz", |
| "feature-selection-detail", |
| rasterData, |
| attributionData |
| ); |
| return `${rasterData.dataset}:${attributionData.model}:feature`; |
| } |
| """, |
| Output("feature-story-render-token", "children"), |
| Input("feature-story-raster-data", "data"), |
| Input("feature-story-attribution-data", "data"), |
| Input("tabs", "value"), |
| ) |
|
|
|
|
| @app.callback( |
| Output("feature-definition", "children"), |
| Output("feature-validation-bars", "figure"), |
| Output("feature-heatmap", "figure"), |
| Output("feature-table", "columns"), |
| Output("feature-table", "data"), |
| Input("dataset-filter", "value"), |
| Input("method-filter", "value"), |
| ) |
| def update_feature(dataset: str, models: list[str] | None): |
| dataset = dataset or DATASETS[0] |
| validation_fig, table = feature_figures(dataset, models) |
| _column, _target, metric, _reference = feature_spec(dataset) |
| if dataset == "allen_neuropixels": |
| definition = ( |
| "Spearman’s r measures association between feature-attribution values " |
| "and each unit’s gOSI measured from drifting gratings." |
| ) |
| elif dataset == "ratinabox": |
| definition = ( |
| "ROC-AUC measures whether place cells rank above head-direction and speed " |
| "cells. Chance ROC-AUC is 0.5." |
| ) |
| else: |
| definition = ( |
| "ROC-AUC measures whether recorded neural features rank above appended " |
| "synthetic controls. Chance ROC-AUC is 0.5." |
| ) |
| table_columns = column_defs(table.columns) |
| for column in table_columns: |
| if column["id"] == "validation_score": |
| column["name"] = metric |
| return ( |
| definition, |
| validation_fig, |
| feature_heatmap(models), |
| table_columns, |
| records(table), |
| ) |
|
|
|
|
| @app.callback( |
| Output("trial-detection-bars", "figure"), |
| Output("trial-heatmap", "figure"), |
| Output("trial-table", "columns"), |
| Output("trial-table", "data"), |
| Output("trial-removal", "figure"), |
| Output("trial-recovery", "figure"), |
| Output("trial-historical", "figure"), |
| Output("trial-historical-trajectories", "figure"), |
| Output("trial-historical-trajectories-mobile", "figure"), |
| Output("trial-retrain-table", "columns"), |
| Output("trial-retrain-table", "data"), |
| Input("dataset-filter", "value"), |
| Input("method-filter", "value"), |
| ) |
| def update_trial(dataset: str, models: list[str] | None): |
| dataset = dataset or DATASETS[0] |
| frame = trial_frame(dataset, models) |
| table_columns = ["method", "corrupted_trial_auc"] |
| table = frame[[column for column in table_columns if column in frame.columns]].sort_values( |
| "corrupted_trial_auc", ascending=False |
| ) if not frame.empty else pd.DataFrame(columns=table_columns) |
| removal, relation, historical, retrain_table = trial_retrain_figures() |
| return ( |
| trial_detection_figure(dataset, models), |
| trial_heatmap(models), |
| column_defs(table.columns), |
| records(round_numeric(table)), |
| removal, |
| relation, |
| historical, |
| historical_trajectory_figure(), |
| historical_trajectory_figure(stacked=True), |
| column_defs(retrain_table.columns), |
| records(retrain_table), |
| ) |
|
|
| if __name__ == "__main__": |
| app.run(host="0.0.0.0", port=7860, debug=False) |
|
|