3918 lines
176 KiB
Python
3918 lines
176 KiB
Python
"""ML Channel Builder plugin: N-vs-T diversity/density grid experiment (fNIRS, LDA).
|
||
|
||
Pick a condition, label participants as Class A / Class B, and sweep every combination of
|
||
(training participants N) x (trials per participant T) x (seed), ranking feature-metric
|
||
subsets by held-out participant accuracy. All grid settings are exposed in the GUI. The best
|
||
metric set can then be exported as a joblib model bundle. A second tool (ModelInferenceGUI)
|
||
loads that bundle and classifies newly collected data in the main window; the same logic is
|
||
available in code via `load_model_bundle`, `classify_dataframe` and `classify_participant`.
|
||
"""
|
||
|
||
# ==========================================================================
|
||
# Imports
|
||
# ==========================================================================
|
||
import html
|
||
import json
|
||
import math
|
||
import multiprocessing
|
||
import os
|
||
import platform
|
||
import re
|
||
import subprocess
|
||
import threading
|
||
import time
|
||
import traceback
|
||
from collections import Counter
|
||
from dataclasses import dataclass
|
||
from itertools import combinations
|
||
from pathlib import Path
|
||
|
||
import joblib
|
||
import numpy as np
|
||
import pandas as pd
|
||
from openpyxl import load_workbook
|
||
from openpyxl.formatting.rule import ColorScaleRule
|
||
from openpyxl.styles import PatternFill
|
||
from openpyxl.utils import get_column_letter
|
||
from openpyxl.worksheet.table import Table, TableStyleInfo
|
||
from PySide6.QtCore import Qt, QThread, Signal
|
||
from PySide6.QtGui import QAction, QColor, QTextCursor
|
||
from PySide6.QtWidgets import (
|
||
QApplication, QCheckBox, QComboBox, QDoubleSpinBox, QFileDialog, QFrame, QGridLayout, QGroupBox,
|
||
QHBoxLayout, QHeaderView, QLabel, QLineEdit, QMenu, QMessageBox, QPushButton,
|
||
QScrollArea, QSpinBox, QSplitter, QTableWidget, QTableWidgetItem, QTabWidget, QTextEdit, QVBoxLayout,
|
||
QWidget,
|
||
)
|
||
from scipy import stats
|
||
from sklearn.discriminant_analysis import LinearDiscriminantAnalysis
|
||
from sklearn.impute import SimpleImputer
|
||
from sklearn.metrics import balanced_accuracy_score
|
||
from sklearn.pipeline import Pipeline
|
||
from sklearn.preprocessing import StandardScaler
|
||
|
||
try: # Windows-only; absent on macOS/Linux
|
||
import winreg
|
||
except ImportError:
|
||
winreg = None
|
||
|
||
try: # embedded plots for the Model Explanation tab (the tab falls back to tables without it)
|
||
from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg, NavigationToolbar2QT
|
||
from matplotlib.figure import Figure
|
||
from matplotlib.patches import Patch, Rectangle
|
||
MPL_OK = True
|
||
except Exception:
|
||
MPL_OK = False
|
||
|
||
|
||
# ==========================================================================
|
||
# Constants
|
||
# ==========================================================================
|
||
# Maps short metric keys to feature-name suffixes
|
||
METRIC_REGISTRY = {
|
||
"Peak_Amp": "Peak_Amp",
|
||
"TTP": "Time_to_Peak",
|
||
"AUC": "AUC",
|
||
"Rising_Slope": "Rising_Slope",
|
||
"Recovery_Slope": "Recovery_Slope",
|
||
"FWHM": "FWHM",
|
||
"Onset_Latency": "Onset_Latency",
|
||
"P2P_Amp": "Peak_to_Peak_Amp",
|
||
"Signal_Std": "Signal_Std",
|
||
"RMS": "RMS",
|
||
}
|
||
|
||
# Chromophore tags recognised in feature names (matched as whole "_"-separated tokens)
|
||
KNOWN_CHROMOPHORES = ("hbo", "hbr", "hbt", "o2hb", "hhb")
|
||
|
||
META_COLS = {"Participant", "Trial", "Condition", "File", "Kind", "TrialNum"}
|
||
|
||
CLASS_NEG, CLASS_POS = "Class A", "Class B" # LDA decision score > 0 <=> CLASS_POS
|
||
UNASSIGNED = "Unassigned"
|
||
LIST_SEP = " | " # separator for participant lists stored in one results cell (paths may contain commas)
|
||
KIND_TRIAL, KIND_PARTICIPANT = "trial", "participant"
|
||
|
||
MIN_PER_CLASS = 6 # required labelled participants per class
|
||
MAX_WORKERS = 30 # upper cap for the worker-process spinbox default
|
||
DEFAULT_WORKERS = min(os.cpu_count() or 4, MAX_WORKERS)
|
||
DEFAULT_BUDGET_TOLERANCE = 0.05 # +/- fraction around each target total-trial budget
|
||
|
||
BUNDLE_VERSION = 1 # version of the exported joblib bundle layout
|
||
|
||
|
||
# ==========================================================================
|
||
# System info
|
||
# ==========================================================================
|
||
def get_cpu_model_name():
|
||
"""Detect detailed processor info, collapsing Windows registry whitespace padding."""
|
||
try:
|
||
system = platform.system()
|
||
|
||
if system == "Windows" and winreg is not None:
|
||
with winreg.OpenKey(
|
||
winreg.HKEY_LOCAL_MACHINE,
|
||
r"HARDWARE\DESCRIPTION\System\CentralProcessor\0",
|
||
) as key:
|
||
raw_cpu_name, _ = winreg.QueryValueEx(key, "ProcessorNameString")
|
||
cpu_name = " ".join(raw_cpu_name.split())
|
||
|
||
command = (
|
||
"Get-CimInstance Win32_Processor | "
|
||
"Select-Object NumberOfCores, NumberOfLogicalProcessors, MaxClockSpeed | "
|
||
"ConvertTo-Json -Compress"
|
||
)
|
||
cpu = json.loads(subprocess.check_output(
|
||
["powershell", "-NoProfile", "-Command", command],
|
||
text=True,
|
||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
||
))
|
||
base_ghz = cpu["MaxClockSpeed"] / 1000
|
||
return (f"{cpu_name} @ {base_ghz:.1f}GHz "
|
||
f"{cpu['NumberOfCores']}c/{cpu['NumberOfLogicalProcessors']}t")
|
||
|
||
if system == "Darwin":
|
||
return subprocess.check_output(
|
||
["sysctl", "-n", "machdep.cpu.brand_string"], text=True
|
||
).strip()
|
||
|
||
if system == "Linux":
|
||
with open("/proc/cpuinfo") as f:
|
||
for line in f:
|
||
if "model name" in line:
|
||
return line.split(":")[1].strip()
|
||
except Exception:
|
||
pass
|
||
|
||
return platform.processor() or "Multi-Core CPU"
|
||
|
||
|
||
# ==========================================================================
|
||
# Data parsing (no Qt)
|
||
# ==========================================================================
|
||
def trial_number(trial_id):
|
||
try:
|
||
return int(str(trial_id).split("_trial_")[1].split("_")[0])
|
||
except Exception:
|
||
return 0
|
||
|
||
|
||
def _strip_fir(name):
|
||
return name[4:] if name.startswith("FIR_") else name
|
||
|
||
|
||
def split_feature_name(name):
|
||
"""Return (condition, trial_id or None, feature_key).
|
||
|
||
'FIR_Reach_trial_3_ROI_X_hbo_Peak_Amp' -> ('Reach', 'Reach_trial_3', 'ROI_X_hbo_Peak_Amp')
|
||
'FIR_Reach_ROI_X_hbo_Peak_Amp' -> ('Reach', None, 'ROI_X_hbo_Peak_Amp')
|
||
"""
|
||
if "_trial_" in name:
|
||
cond_part, rest = name.split("_trial_", 1)
|
||
trial_num, _, key = rest.partition("_")
|
||
cond = _strip_fir(cond_part)
|
||
return cond, f"{cond}_trial_{trial_num}", key
|
||
|
||
parts = name.split("_")
|
||
if len(parts) >= 3 and parts[0] == "FIR":
|
||
return parts[1], None, "_".join(parts[2:])
|
||
return "Overall", None, name
|
||
|
||
|
||
def parse_fir_dict(fir_dict):
|
||
"""Turn main_window.fir_feature_dict into a long table.
|
||
|
||
One row per (participant, trial) -- or per (participant, condition) when no
|
||
trial-specific features exist (Step 21.5 not run).
|
||
|
||
Returns (df, kind, info). kind is KIND_TRIAL if ANY file has trial features,
|
||
else KIND_PARTICIPANT, else None. Files lacking trial features are dropped from a
|
||
trial-kind dataset and listed in info['dropped'].
|
||
"""
|
||
rows = {}
|
||
for file_path, packet in (fir_dict or {}).items():
|
||
if not isinstance(packet, dict):
|
||
continue
|
||
feats = packet.get("features", np.array([]))
|
||
names = packet.get("feature_names", [])
|
||
if len(feats) == 0 or len(feats) != len(names):
|
||
continue
|
||
|
||
abs_path = str(Path(file_path).resolve())
|
||
for val, name in zip(feats, names):
|
||
cond, trial_id, key = split_feature_name(str(name))
|
||
tid = trial_id or f"{cond}_participant"
|
||
rec = rows.setdefault((abs_path, tid), {
|
||
"Participant": abs_path,
|
||
"Trial": tid,
|
||
"Condition": cond,
|
||
"File": file_path,
|
||
"Kind": KIND_TRIAL if trial_id else KIND_PARTICIPANT,
|
||
"TrialNum": trial_number(tid),
|
||
})
|
||
rec[key] = float(val)
|
||
|
||
if not rows:
|
||
return pd.DataFrame(), None, {"dropped": []}
|
||
|
||
df = pd.DataFrame(list(rows.values()))
|
||
feat_cols = [c for c in df.columns if c not in META_COLS]
|
||
df[feat_cols] = df[feat_cols].replace([np.inf, -np.inf], np.nan)
|
||
|
||
info = {"dropped": []}
|
||
if (df["Kind"] == KIND_TRIAL).any():
|
||
kind = KIND_TRIAL
|
||
bad = df[df["Kind"] == KIND_PARTICIPANT]["Participant"].unique().tolist()
|
||
if bad:
|
||
info["dropped"] = bad
|
||
df = df[df["Kind"] == KIND_TRIAL].copy()
|
||
else:
|
||
kind = KIND_PARTICIPANT
|
||
|
||
return df.reset_index(drop=True), kind, info
|
||
|
||
|
||
# ==========================================================================
|
||
# Feature-column selection (no Qt)
|
||
# ==========================================================================
|
||
def name_tokens(name):
|
||
"""Lower-cased alphanumeric tokens of a feature name ('ROI_X_hbo_Peak_Amp' -> {roi, x, hbo, ...})."""
|
||
return set(re.split(r"[^a-z0-9]+", name.lower())) - {""}
|
||
|
||
|
||
def detect_chromophores_in(columns):
|
||
"""Chromophore tags (upper-case) that appear as a whole token in any column name."""
|
||
found = set()
|
||
for col in columns:
|
||
found |= name_tokens(col) & set(KNOWN_CHROMOPHORES)
|
||
return sorted(t.upper() for t in found)
|
||
|
||
|
||
def select_columns(all_cols, metrics, chromos):
|
||
"""Feature columns ending in a chosen metric suffix and tagged with a chosen chromophore.
|
||
|
||
A chromophore matches only as a whole token, so 'hbo' selects 'ROI_X_hbo_Peak_Amp'
|
||
but not a column that merely contains the letters inside another word.
|
||
"""
|
||
wanted = {c.lower() for c in chromos}
|
||
cols = []
|
||
for m in metrics:
|
||
suffix = f"_{METRIC_REGISTRY[m]}"
|
||
for c in all_cols:
|
||
if not c.endswith(suffix):
|
||
continue
|
||
if m == "Peak_Amp" and c.endswith("_to_Peak_Amp"):
|
||
continue
|
||
if name_tokens(c) & wanted:
|
||
cols.append(c)
|
||
return sorted(set(cols))
|
||
|
||
|
||
def describe_feature_columns(all_cols, chromos):
|
||
"""Diagnostics for the log: columns per chromophore, total, and metrics with no columns.
|
||
|
||
Returns (per_chromophore_counts, total_columns, metrics_without_columns).
|
||
Counts cover all metrics, so they show exactly what each chromophore contributes.
|
||
"""
|
||
all_metrics = list(METRIC_REGISTRY)
|
||
per_chromo = {ch: len(select_columns(all_cols, all_metrics, [ch])) for ch in chromos}
|
||
total = len(select_columns(all_cols, all_metrics, chromos))
|
||
empty = [m for m in all_metrics if not select_columns(all_cols, [m], chromos)]
|
||
return per_chromo, total, empty
|
||
|
||
|
||
# ==========================================================================
|
||
# Evaluation core (no Qt)
|
||
# ==========================================================================
|
||
def make_pipeline():
|
||
return Pipeline([
|
||
("imputer", SimpleImputer(strategy="median")), # fitted on training data only
|
||
("scaler", StandardScaler()),
|
||
("lda", LinearDiscriminantAnalysis(solver="lsqr", shrinkage="auto")),
|
||
])
|
||
|
||
|
||
def _fit(train_df, cols, classes):
|
||
X = train_df[cols].to_numpy(float)
|
||
y = train_df["Participant"].map(classes).to_numpy()
|
||
if len(set(y)) < 2:
|
||
raise ValueError("Training data contains only one class.")
|
||
pipe = make_pipeline()
|
||
pipe.fit(X, y)
|
||
if list(pipe.classes_) != [CLASS_NEG, CLASS_POS]:
|
||
raise ValueError(f"Unexpected class order {list(pipe.classes_)}")
|
||
return pipe
|
||
|
||
|
||
def evaluate_feature_set(train_df, test_df, cols, classes):
|
||
"""Fit on train_df, score test_df. Returns participant- and trial-level metrics.
|
||
|
||
Participant prediction = sign of the mean trial decision score (> 0 -> Class B).
|
||
"""
|
||
pipe = _fit(train_df, cols, classes)
|
||
scores = pipe.decision_function(test_df[cols].to_numpy(float))
|
||
|
||
trial_true = test_df["Participant"].map(classes).to_numpy()
|
||
trial_pred = np.where(scores > 0, CLASS_POS, CLASS_NEG)
|
||
|
||
mean_scores = (
|
||
pd.DataFrame({"Participant": test_df["Participant"].to_numpy(), "Score": scores})
|
||
.groupby("Participant")["Score"].mean()
|
||
)
|
||
part_true = mean_scores.index.map(classes).to_numpy()
|
||
part_pred = np.where(mean_scores.to_numpy() > 0, CLASS_POS, CLASS_NEG)
|
||
|
||
part_acc = float(np.mean(part_true == part_pred))
|
||
try:
|
||
part_bal = float(balanced_accuracy_score(part_true, part_pred))
|
||
except Exception:
|
||
part_bal = part_acc
|
||
wrong = [str(p) for p in mean_scores.index[part_true != part_pred]]
|
||
|
||
return {
|
||
"Wrong_Participants": wrong,
|
||
"Part_Correct": int(np.sum(part_true == part_pred)),
|
||
"Part_Total": int(len(part_true)),
|
||
"Part_Acc": part_acc,
|
||
"Part_Bal_Acc": part_bal,
|
||
"Trial_Acc": float(np.mean(trial_true == trial_pred)),
|
||
"N_Features": len(cols),
|
||
}
|
||
|
||
|
||
def metric_sets_per_task(top_k, n_metrics=len(METRIC_REGISTRY)):
|
||
"""How many metric sets one grid configuration evaluates (used for the GUI estimate).
|
||
|
||
singles + all pairs + every subset of size >= 3 drawn from the top_k single metrics
|
||
(subsets of size 1 and 2 of the top_k were already covered, so they are not repeated).
|
||
"""
|
||
k = min(top_k, n_metrics)
|
||
expanded = sum(math.comb(k, s) for s in range(3, k + 1))
|
||
return n_metrics + math.comb(n_metrics, 2) + expanded
|
||
|
||
|
||
def parse_metric_list(text):
|
||
"""'rms, ttp' -> (('RMS', 'TTP'), []). Returns (valid metric keys in order, unknown tokens)."""
|
||
keys = {k.lower(): k for k in METRIC_REGISTRY}
|
||
good, bad = [], []
|
||
for tok in re.split(r"[,;\s]+", str(text).strip()):
|
||
if not tok:
|
||
continue
|
||
key = keys.get(tok.lower())
|
||
if key:
|
||
good.append(key)
|
||
else:
|
||
bad.append(tok)
|
||
return tuple(dict.fromkeys(good)), bad
|
||
|
||
|
||
def run_metric_sweep(train_df, test_df, classes, all_feature_cols, chromos, top_k=5,
|
||
fixed_metrics=()):
|
||
"""Metric sweep for one train/test split.
|
||
|
||
With fixed_metrics (stage 2) NO sweep happens: exactly that one metric set is evaluated,
|
||
so nothing about the held-out participants influences which features are used.
|
||
|
||
Phase 1: every single metric (10).
|
||
Phase 2: every pair (45).
|
||
Phase 3: take the top_k single metrics from phase 1 and evaluate every subset of them
|
||
of size 3..top_k (for top_k = 5: 10 + 5 + 1 = 16 new sets). Subsets of size 1-2
|
||
were already done in phases 1-2. Order within a set is irrelevant to the model,
|
||
so these are combinations, not permutations.
|
||
|
||
Returns a DataFrame ranked by (Part_Bal_Acc, Trial_Acc), empty if nothing evaluated.
|
||
"""
|
||
metric_cols = {m: select_columns(all_feature_cols, [m], chromos) for m in METRIC_REGISTRY}
|
||
usable = [m for m, cols in metric_cols.items() if cols]
|
||
results, seen = [], set()
|
||
|
||
def run_combo(metrics):
|
||
key = ", ".join(sorted(metrics))
|
||
if key in seen:
|
||
return
|
||
cols = sorted(set().union(*(metric_cols[m] for m in metrics)))
|
||
try:
|
||
rec = evaluate_feature_set(train_df, test_df, cols, classes)
|
||
except Exception:
|
||
return
|
||
seen.add(key)
|
||
results.append({"Metrics": key, **rec})
|
||
|
||
def ranked():
|
||
return (pd.DataFrame(results)
|
||
.sort_values(["Part_Bal_Acc", "Trial_Acc"], ascending=False, kind="mergesort")
|
||
.reset_index(drop=True))
|
||
|
||
if fixed_metrics: # Stage 2: one pre-specified set
|
||
if all(m in usable for m in fixed_metrics):
|
||
run_combo(list(fixed_metrics))
|
||
return ranked() if results else pd.DataFrame()
|
||
|
||
for m in usable: # Phase 1: singles
|
||
run_combo([m])
|
||
if not results:
|
||
return pd.DataFrame()
|
||
|
||
top_singles = ranked().head(top_k)["Metrics"].tolist() # best top_k single metrics
|
||
|
||
for combo in combinations(usable, 2): # Phase 2: all pairs
|
||
run_combo(list(combo))
|
||
|
||
for size in range(3, len(top_singles) + 1): # Phase 3: subsets of the top_k
|
||
for combo in combinations(top_singles, size):
|
||
run_combo(list(combo))
|
||
|
||
return ranked()
|
||
|
||
|
||
def _participant_rows(sub_df, t_max=None):
|
||
"""A participant's rows for one condition; with t_max, only the first t_max trials.
|
||
|
||
"First" means lowest TrialNum, i.e. true chronological order (trial 2 before trial 10).
|
||
Whole-participant data (no trial rows) is always returned unchanged.
|
||
"""
|
||
if t_max is None or (sub_df["Kind"] == KIND_PARTICIPANT).all():
|
||
return sub_df
|
||
ordered = sub_df.sort_values("TrialNum", kind="stable")
|
||
return ordered.iloc[:t_max]
|
||
|
||
|
||
def _run_grid_task(task):
|
||
"""Worker-process entry point: run one configuration, return its top-3 metric sets."""
|
||
cond_df = task["cond_df"]
|
||
classes = task["class_assign"]
|
||
train_parts = task["train_parts"]
|
||
test_parts = task["test_participants"]
|
||
|
||
def rows_for(sub, t_max=None):
|
||
return _participant_rows(cond_df[cond_df["Participant"] == sub], t_max)
|
||
|
||
train_df = pd.concat([rows_for(s, task["t_max"]) for s in train_parts], ignore_index=True)
|
||
test_df = pd.concat([rows_for(s) for s in test_parts], ignore_index=True)
|
||
|
||
ranked = run_metric_sweep(
|
||
train_df, test_df, classes, task["all_feature_cols"], task["selected_chromos"],
|
||
top_k=task["top_k_expand"], fixed_metrics=task["fixed_metrics"],
|
||
)
|
||
if ranked.empty:
|
||
return []
|
||
|
||
test_str = LIST_SEP.join(sorted(str(p) for p in test_parts))
|
||
return [
|
||
{
|
||
"Run_ID": int(task["run_id"]),
|
||
"Seed": int(task["seed"]),
|
||
"Rank": rank,
|
||
"Test_K_Per_Class": int(task["test_k"]),
|
||
"Metrics": str(row["Metrics"]),
|
||
"N_Features": int(row["N_Features"]),
|
||
"Balanced": bool(task["balanced"]),
|
||
"N_Train_Subjects": len(train_parts),
|
||
"T_Trials_Per_Subject": int(task["t_max"]),
|
||
"Train_Trials_Actual": int(len(train_df)),
|
||
"Part_Correct": int(row["Part_Correct"]),
|
||
"Part_Total": int(row["Part_Total"]),
|
||
"Part_Accuracy": float(row["Part_Acc"]),
|
||
"Part_Bal_Accuracy": float(row["Part_Bal_Acc"]),
|
||
"Trial_Accuracy": float(row["Trial_Acc"]),
|
||
"Test_Participants": test_str,
|
||
"Misclassified": LIST_SEP.join(sorted(row["Wrong_Participants"])),
|
||
}
|
||
for rank, (_, row) in enumerate(ranked.head(3).iterrows(), start=1)
|
||
]
|
||
|
||
|
||
# ==========================================================================
|
||
# Best-model selection, export and reuse (no Qt)
|
||
# ==========================================================================
|
||
def rank_metric_sets(summary_df):
|
||
"""Leaderboard of metric sets across every grid configuration.
|
||
|
||
Top1_Count : configurations in which the set ranked first
|
||
Top3_Count : configurations in which it made the top 3
|
||
Mean_Part_Bal_Acc : mean held-out participant balanced accuracy over those appearances
|
||
|
||
Sorted by Top1_Count, then Top3_Count, then mean accuracy. Frequency comes first because
|
||
a mean over only the appearances is biased upward (a set is only logged when it did well).
|
||
"""
|
||
board = (
|
||
summary_df.groupby("Metrics")
|
||
.agg(Top1_Count=("Rank", lambda r: int((r == 1).sum())),
|
||
Top3_Count=("Rank", "size"),
|
||
Mean_Part_Bal_Acc=("Part_Bal_Accuracy", "mean"))
|
||
.reset_index()
|
||
.sort_values(["Top1_Count", "Top3_Count", "Mean_Part_Bal_Acc"], ascending=False)
|
||
.reset_index(drop=True)
|
||
)
|
||
return board
|
||
|
||
|
||
def leaderboard_lines(board, n=5):
|
||
"""Log lines for the top n metric sets, padded so every column lines up (monospace log)."""
|
||
top = board.head(n)
|
||
name_w = max(len(str(m)) for m in top["Metrics"])
|
||
c1_w = max(len(str(c)) for c in top["Top1_Count"])
|
||
c3_w = max(len(str(c)) for c in top["Top3_Count"])
|
||
lines = ["\n[Best metric sets across all configurations]:"]
|
||
for i, row in enumerate(top.itertuples(index=False), start=1):
|
||
lines.append(
|
||
f" {i:>{len(str(len(top)))}}. {row.Metrics:<{name_w}} | "
|
||
f"1st in {row.Top1_Count:>{c1_w}} / top-3 in {row.Top3_Count:>{c3_w}} configs | "
|
||
f"mean bal. acc {row.Mean_Part_Bal_Acc * 100:6.2f}%")
|
||
return lines
|
||
|
||
|
||
def problem_participants(summary_df, classes, n=5):
|
||
"""Held-out participants most often misclassified.
|
||
|
||
Uses the rank-1 (best) metric set of every configuration. Times_Tested counts the
|
||
configurations in which the participant was held out; Times_Wrong counts those where the
|
||
best set got them wrong. Returns the n worst by error rate (empty DataFrame if no data).
|
||
"""
|
||
top = summary_df[summary_df["Rank"] == 1]
|
||
if top.empty or "Misclassified" not in top.columns:
|
||
return pd.DataFrame()
|
||
|
||
def split(cell):
|
||
return [p for p in cell.split(LIST_SEP) if p] if isinstance(cell, str) else []
|
||
|
||
tested, wrong = Counter(), Counter()
|
||
for t_cell, w_cell in zip(top["Test_Participants"], top["Misclassified"]):
|
||
tested.update(split(t_cell))
|
||
wrong.update(split(w_cell))
|
||
|
||
board = pd.DataFrame([
|
||
{"Participant": p, "True_Class": classes.get(p, "?"), "Times_Tested": cnt,
|
||
"Times_Wrong": wrong.get(p, 0), "Error_Rate": wrong.get(p, 0) / cnt}
|
||
for p, cnt in tested.items()
|
||
])
|
||
return (board.sort_values(["Error_Rate", "Times_Wrong"], ascending=False, kind="mergesort")
|
||
.head(n).reset_index(drop=True))
|
||
|
||
|
||
def problem_participant_lines(board):
|
||
"""Log lines for problem_participants(); paths are shortened to parent/file."""
|
||
def short(p):
|
||
parts = Path(p).parts
|
||
return "/".join(parts[-2:]) if len(parts) > 1 else p
|
||
|
||
if board.empty:
|
||
return []
|
||
name_w = max(len(short(p)) for p in board["Participant"])
|
||
lines = ["\n[Most frequently misclassified held-out participants "
|
||
"(by each configuration's best metric set)]:"]
|
||
for i, r in enumerate(board.itertuples(index=False), start=1):
|
||
lines.append(
|
||
f" {i}. {short(r.Participant):<{name_w}} | {r.True_Class} | "
|
||
f"wrong in {r.Times_Wrong}/{r.Times_Tested} tests ({r.Error_Rate * 100:.1f}%)")
|
||
return lines
|
||
|
||
|
||
def train_final_model(df, condition, classes, metrics_str, all_feature_cols, chromos,
|
||
class_hints=None, selection=None):
|
||
"""Fit the chosen metric set on ALL labelled participants (all trials) and bundle it.
|
||
|
||
Returns a dict ready for joblib.dump. Reload it with load_model_bundle and apply it
|
||
with classify_participant.
|
||
"""
|
||
metrics = [m.strip() for m in metrics_str.split(",")]
|
||
cols = select_columns(all_feature_cols, metrics, chromos)
|
||
if not cols:
|
||
raise ValueError(f"No feature columns match metrics [{metrics_str}] and chromophores {list(chromos)}.")
|
||
|
||
cond_df = df[df["Condition"] == condition]
|
||
frames = [_participant_rows(cond_df[cond_df["Participant"] == s]) for s in classes]
|
||
train_df = pd.concat([f for f in frames if len(f)], ignore_index=True)
|
||
pipe = _fit(train_df, cols, classes)
|
||
|
||
per_class = train_df.drop_duplicates("Participant")["Participant"].map(classes).value_counts()
|
||
return {
|
||
"bundle_version": BUNDLE_VERSION,
|
||
"created": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"pipeline": pipe,
|
||
"feature_columns": cols, # order matters: the pipeline expects exactly this order
|
||
"metrics": metrics,
|
||
"chromophores": list(chromos),
|
||
"condition": condition,
|
||
"class_neg": CLASS_NEG,
|
||
"class_pos": CLASS_POS,
|
||
"class_hints": class_hints or {},
|
||
"n_train_participants": {CLASS_NEG: int(per_class.get(CLASS_NEG, 0)),
|
||
CLASS_POS: int(per_class.get(CLASS_POS, 0))},
|
||
"n_train_rows": int(len(train_df)),
|
||
"selection": selection or {},
|
||
"decision_rule": ("Participant score = mean LDA decision score over their trials; "
|
||
"> 0 -> class_pos, else class_neg."),
|
||
}
|
||
|
||
|
||
def load_model_bundle(path):
|
||
"""Load a bundle written by the plugin. Only load files you trust (joblib uses pickle)."""
|
||
bundle = joblib.load(path)
|
||
if not isinstance(bundle, dict) or "pipeline" not in bundle or "feature_columns" not in bundle:
|
||
raise ValueError("Not a valid ML Channel Builder model bundle.")
|
||
if bundle.get("bundle_version") != BUNDLE_VERSION:
|
||
raise ValueError(f"Unsupported bundle version {bundle.get('bundle_version')} "
|
||
f"(expected {BUNDLE_VERSION}).")
|
||
return bundle
|
||
|
||
|
||
def missing_feature_columns(bundle, df):
|
||
"""Model feature columns that are absent from df (the data cannot be scored without them)."""
|
||
return [c for c in bundle["feature_columns"] if c not in df.columns]
|
||
|
||
|
||
def trial_scores(bundle, rows_df):
|
||
"""LDA decision score for every row (> 0 -> class_pos). Missing values are imputed by the
|
||
pipeline using the training medians; entirely missing columns are an error."""
|
||
missing = missing_feature_columns(bundle, rows_df)
|
||
if missing:
|
||
raise ValueError(f"{len(missing)} required feature column(s) missing, e.g. {missing[:3]}")
|
||
if rows_df.empty:
|
||
raise ValueError("No trial rows supplied.")
|
||
X = rows_df[bundle["feature_columns"]].replace([np.inf, -np.inf], np.nan).to_numpy(float)
|
||
return bundle["pipeline"].decision_function(X)
|
||
|
||
|
||
def classify_dataframe(bundle, df):
|
||
"""Score every row of df (already filtered to bundle['condition']) and summarise per participant.
|
||
|
||
Returns (trials_df, participants_df). A participant's prediction is the sign of the mean
|
||
of their trial scores, exactly as in the grid evaluation. Missing_Frac is the share of the
|
||
model's feature values that were missing (and therefore imputed) for that participant.
|
||
"""
|
||
cols = bundle["feature_columns"]
|
||
neg, pos = bundle["class_neg"], bundle["class_pos"]
|
||
if "TrialNum" in df.columns:
|
||
df = df.sort_values(["Participant", "TrialNum"], kind="stable")
|
||
scores = trial_scores(bundle, df)
|
||
|
||
trials = df[["Participant", "Trial", "Condition"]].copy()
|
||
trials["Score"] = scores
|
||
trials["Predicted_Class"] = np.where(scores > 0, pos, neg)
|
||
trials["Missing_Frac"] = df[cols].replace([np.inf, -np.inf], np.nan).isna().mean(axis=1).to_numpy()
|
||
|
||
parts = (
|
||
trials.groupby("Participant", sort=True)
|
||
.agg(N_Trials=("Score", "size"),
|
||
Mean_Score=("Score", "mean"),
|
||
Frac_Trials_Class_B=("Score", lambda s: float((s > 0).mean())),
|
||
Missing_Frac=("Missing_Frac", "mean"))
|
||
.reset_index()
|
||
)
|
||
parts["Predicted_Class"] = np.where(parts["Mean_Score"] > 0, pos, neg)
|
||
return trials.drop(columns="Missing_Frac").reset_index(drop=True), parts
|
||
|
||
|
||
def classify_participant(bundle, trial_df):
|
||
"""Classify one participant from their trial rows.
|
||
|
||
trial_df: that participant's rows for bundle['condition'], in the format produced by
|
||
parse_fir_dict (one row per trial, feature columns named as in training). Example:
|
||
|
||
df, _, _ = parse_fir_dict(main_window.fir_feature_dict)
|
||
rows = df[(df["File"] == my_file) & (df["Condition"] == bundle["condition"])]
|
||
result = classify_participant(load_model_bundle("best_model.joblib"), rows)
|
||
"""
|
||
scores = trial_scores(bundle, trial_df)
|
||
mean_score = float(np.mean(scores))
|
||
return {
|
||
"Predicted_Class": bundle["class_pos"] if mean_score > 0 else bundle["class_neg"],
|
||
"Mean_Score": mean_score,
|
||
"Frac_Trials_Class_B": float(np.mean(scores > 0)),
|
||
"N_Trials": int(len(scores)),
|
||
}
|
||
|
||
|
||
# ==========================================================================
|
||
# Iso-budget analysis and Excel export (no Qt)
|
||
# ==========================================================================
|
||
@dataclass(frozen=True)
|
||
class BudgetTier:
|
||
"""A matched total-trial budget (N x T) compared across few-participants/many-trials
|
||
(low N) versus many-participants/few-trials (high N).
|
||
|
||
Only (N, T) cells whose total lands within target +/- tolerance count, so both arms are
|
||
compared at (nearly) the same number of training trials. N is the TOTAL number of
|
||
training participants (both classes).
|
||
"""
|
||
name: str
|
||
target: int
|
||
tolerance: float # fraction, e.g. 0.05 = +/-5 %
|
||
low_max_n: int
|
||
low_min_t: int
|
||
high_min_n: int
|
||
high_max_t: int
|
||
fill: str # Excel highlight colour
|
||
|
||
@property
|
||
def min_total(self):
|
||
return math.ceil(self.target * (1 - self.tolerance) - 1e-9)
|
||
|
||
@property
|
||
def max_total(self):
|
||
return math.floor(self.target * (1 + self.tolerance) + 1e-9)
|
||
|
||
def masks(self, n, t):
|
||
"""(low_mask, high_mask) for scalars or pandas Series of N and T."""
|
||
total = n * t
|
||
in_budget = (total >= self.min_total) & (total <= self.max_total)
|
||
low = in_budget & (n <= self.low_max_n) & (t >= self.low_min_t)
|
||
high = in_budget & (n >= self.high_min_n) & (t <= self.high_max_t)
|
||
return low, high
|
||
|
||
@property
|
||
def criteria(self):
|
||
return (f"Low-Density: N ≤ {self.low_max_n}, T ≥ {self.low_min_t} | "
|
||
f"High-Density: N ≥ {self.high_min_n}, T ≤ {self.high_max_t} "
|
||
f"(Matched Training Volume: {self.min_total}-{self.max_total} Total Trials)")
|
||
|
||
|
||
# (target, low_max_n, low_min_t, high_min_n, high_max_t, fill)
|
||
_TIER_SPECS = (
|
||
(72, 8, 8, 16, 5, "FCE4D6"), # e.g. 6x12, 8x9 vs 18x4, 24x3 (soft orange)
|
||
(96, 8, 12, 16, 6, "D9E1F2"), # e.g. 6x16, 8x12 vs 16x6, 24x4 (soft blue)
|
||
(144, 10, 14, 20, 8, "E2EFDA"), # e.g. 8x18, 10x14 vs 20x7, 24x6 (soft green)
|
||
)
|
||
|
||
|
||
def make_budget_tiers(tolerance=DEFAULT_BUDGET_TOLERANCE):
|
||
return tuple(
|
||
BudgetTier(
|
||
name=f"Permutation {i}: Target Budget ~{target} Trials (±{tolerance * 100:g}%)",
|
||
target=target, tolerance=tolerance, low_max_n=low_n, low_min_t=low_t,
|
||
high_min_n=high_n, high_max_t=high_t, fill=fill)
|
||
for i, (target, low_n, low_t, high_n, high_t, fill) in enumerate(_TIER_SPECS, start=1))
|
||
|
||
|
||
DEFAULT_TIERS = make_budget_tiers()
|
||
|
||
|
||
def cell_in_budget(tiers, n, t):
|
||
"""True if the (total N, T) cell belongs to the low or high arm of any budget tier."""
|
||
return any(any(tier.masks(n, t)) for tier in tiers)
|
||
|
||
|
||
SHEET_GRID = "Grid Results"
|
||
SHEET_ACC = "Strategy Pivot (Balanced)"
|
||
SHEET_TRIALS = "Total Training Trials"
|
||
|
||
|
||
def _significance_label(p):
|
||
if p < 0.001:
|
||
return f"Statistically Significant (p = {p:.2e} < 0.001) ***"
|
||
if p < 0.01:
|
||
return f"Statistically Significant (p = {p:.4f} < 0.01) **"
|
||
if p < 0.05:
|
||
return f"Statistically Significant (p = {p:.4f} < 0.05) *"
|
||
return f"Not Statistically Significant (p = {p:.4f} >= 0.05)"
|
||
|
||
|
||
def build_pivots(summary_df):
|
||
"""Balanced-runs pivots (N x T): mean participant balanced accuracy and total trials."""
|
||
balanced = summary_df[summary_df["Balanced"]]
|
||
kwargs = dict(index="N_Train_Subjects", columns="T_Trials_Per_Subject", aggfunc="mean")
|
||
return (
|
||
balanced.pivot_table(values="Part_Bal_Accuracy", **kwargs),
|
||
balanced.pivot_table(values="Total_Trials", **kwargs),
|
||
)
|
||
|
||
|
||
def iso_budget_report(summary_df, tiers=DEFAULT_TIERS, fixed=False):
|
||
"""Log lines comparing high-N/low-T against low-N/high-T at matched trial budgets.
|
||
|
||
fixed=True labels the numbers as coming from a pre-specified metric set (stage 2).
|
||
|
||
Paired by seed: every seed uses ONE held-out set for all (N, T) cells, so for each seed
|
||
the rank-1 balanced accuracy is averaged within each arm and the two arms are compared on
|
||
that seed. The test is then a paired test over seeds (n = number of seeds), which removes
|
||
held-out-set difficulty and does not count the top-3 rows of a configuration as
|
||
independent samples.
|
||
"""
|
||
best = summary_df[summary_df["Balanced"] & (summary_df["Rank"] == 1)]
|
||
n, t = best["N_Train_Subjects"], best["T_Trials_Per_Subject"]
|
||
lines = []
|
||
|
||
for tier in tiers:
|
||
low_mask, high_mask = tier.masks(n, t)
|
||
low_by_seed = best.loc[low_mask].groupby("Seed")["Part_Bal_Accuracy"].mean().rename("low")
|
||
high_by_seed = best.loc[high_mask].groupby("Seed")["Part_Bal_Accuracy"].mean().rename("high")
|
||
paired = pd.concat([low_by_seed, high_by_seed], axis=1).dropna()
|
||
|
||
lines += [f"\n-> {tier.name}", f" Criteria: {tier.criteria}"]
|
||
if len(paired) < 2:
|
||
lines.append(" -> Insufficient paired seeds for statistical testing in this budget tier.")
|
||
continue
|
||
|
||
diff = paired["high"] - paired["low"]
|
||
k = len(paired)
|
||
mean_d, sd_d = float(diff.mean()), float(diff.std(ddof=1))
|
||
lines += [
|
||
f" • Low-Diversity / High-Density: Mean Acc = {paired['low'].mean() * 100:.2f}% "
|
||
f"({'pre-specified metric set' if fixed else 'best set per seed'}, {k} paired seeds)",
|
||
f" • High-Diversity / Low-Density: Mean Acc = {paired['high'].mean() * 100:.2f}%",
|
||
]
|
||
|
||
if sd_d > 0:
|
||
t_stat, p_val = stats.ttest_1samp(diff, 0.0)
|
||
half = float(stats.t.ppf(0.975, k - 1)) * sd_d / math.sqrt(k)
|
||
lines.append(
|
||
f" -> Paired difference (high - low): {mean_d * 100:+.2f}% "
|
||
f"[95% CI {(mean_d - half) * 100:+.2f}, {(mean_d + half) * 100:+.2f}] | "
|
||
f"t({k - 1}) = {t_stat:.2f}, Cohen's dz = {mean_d / sd_d:.2f}")
|
||
try:
|
||
w_p = stats.wilcoxon(diff).pvalue
|
||
lines.append(f" -> Wilcoxon signed-rank p = {w_p:.4g} "
|
||
f"(seeds: high better {int((diff > 0).sum())}, "
|
||
f"low better {int((diff < 0).sum())}, tied {int((diff == 0).sum())})")
|
||
except ValueError:
|
||
pass
|
||
lines.append(f" -> Verdict: {_significance_label(p_val)}")
|
||
else:
|
||
lines.append(f" -> Paired difference (high - low): {mean_d * 100:+.2f}% "
|
||
f"(identical in every seed; no variance to test)")
|
||
|
||
return lines
|
||
|
||
|
||
def _format_grid_sheet(ws):
|
||
narrow = {"Run_ID", "Seed", "Rank", "Test_K_Per_Class", "N_Features", "Balanced"}
|
||
percent_like = {"Part_Accuracy", "Part_Bal_Accuracy", "Trial_Accuracy"}
|
||
count_like = {"Total_Trials", "Train_Trials_Actual"}
|
||
|
||
for col_idx in range(1, ws.max_column + 1):
|
||
header = ws.cell(row=1, column=col_idx).value
|
||
if header in narrow:
|
||
width = 9.25
|
||
elif header == "Metrics":
|
||
width = 18.5 * 1.75
|
||
else:
|
||
width = 18.5
|
||
ws.column_dimensions[get_column_letter(col_idx)].width = width
|
||
|
||
if ws.max_row <= 1:
|
||
return
|
||
|
||
table = Table(displayName="DensityGridResults",
|
||
ref=f"A1:{get_column_letter(ws.max_column)}{ws.max_row}")
|
||
table.tableStyleInfo = TableStyleInfo(
|
||
name="TableStyleMedium9", showFirstColumn=False, showLastColumn=False,
|
||
showRowStripes=True, showColumnStripes=True)
|
||
ws.add_table(table)
|
||
|
||
for col_idx in range(1, ws.max_column + 1):
|
||
header = ws.cell(row=1, column=col_idx).value
|
||
if header in percent_like or header in count_like:
|
||
fmt = "#,##0" if header in count_like else "0.0000"
|
||
for row_idx in range(2, ws.max_row + 1):
|
||
cell = ws.cell(row=row_idx, column=col_idx)
|
||
if isinstance(cell.value, (int, float)):
|
||
cell.number_format = fmt
|
||
|
||
|
||
def _format_pivot_sheet(ws, number_format, heatmap=False):
|
||
for col_idx in range(1, ws.max_column + 1):
|
||
ws.column_dimensions[get_column_letter(col_idx)].width = 14.0
|
||
|
||
if ws.max_row <= 1 or ws.max_column <= 1:
|
||
return
|
||
|
||
if heatmap:
|
||
ws.conditional_formatting.add(
|
||
f"B2:{get_column_letter(ws.max_column)}{ws.max_row}",
|
||
ColorScaleRule(
|
||
start_type="num", start_value=0.50, start_color="F8696B",
|
||
mid_type="num", mid_value=0.75, mid_color="FFEB84",
|
||
end_type="num", end_value=1.00, end_color="63BE7B"),
|
||
)
|
||
|
||
for r in range(2, ws.max_row + 1):
|
||
for c in range(2, ws.max_column + 1):
|
||
cell = ws.cell(row=r, column=c)
|
||
if isinstance(cell.value, (int, float)):
|
||
cell.number_format = number_format
|
||
|
||
|
||
def _highlight_budget_tiers(ws, tiers):
|
||
"""Colour each (N, T) cell by the first budget tier it belongs to."""
|
||
fills = [PatternFill(start_color=t.fill, end_color=t.fill, fill_type="solid")
|
||
for t in tiers]
|
||
for r in range(2, ws.max_row + 1):
|
||
n = ws.cell(row=r, column=1).value
|
||
if not isinstance(n, (int, float)):
|
||
continue
|
||
for c in range(2, ws.max_column + 1):
|
||
t = ws.cell(row=1, column=c).value
|
||
if not isinstance(t, (int, float)):
|
||
continue
|
||
for tier, fill in zip(tiers, fills):
|
||
if any(tier.masks(n, t)):
|
||
ws.cell(row=r, column=c).fill = fill
|
||
break
|
||
|
||
|
||
def export_results_xlsx(path, summary_df, pivot_acc=None, pivot_trials=None, tiers=DEFAULT_TIERS):
|
||
"""Write grid results plus the two pivot tabs, formatted and tier-highlighted."""
|
||
def usable(p):
|
||
return p is not None and not p.empty
|
||
|
||
with pd.ExcelWriter(path, engine="openpyxl") as writer:
|
||
summary_df.to_excel(writer, index=False, sheet_name=SHEET_GRID)
|
||
if usable(pivot_acc):
|
||
pivot_acc.to_excel(writer, sheet_name=SHEET_ACC)
|
||
if usable(pivot_trials):
|
||
pivot_trials.to_excel(writer, sheet_name=SHEET_TRIALS)
|
||
|
||
wb = load_workbook(path)
|
||
_format_grid_sheet(wb[SHEET_GRID])
|
||
if usable(pivot_acc):
|
||
_format_pivot_sheet(wb[SHEET_ACC], "0.0000", heatmap=True)
|
||
_highlight_budget_tiers(wb[SHEET_ACC], tiers)
|
||
if usable(pivot_trials):
|
||
_format_pivot_sheet(wb[SHEET_TRIALS], "#,##0")
|
||
_highlight_budget_tiers(wb[SHEET_TRIALS], tiers)
|
||
wb.save(path)
|
||
|
||
|
||
# ==========================================================================
|
||
# Grid settings and planning (no Qt)
|
||
# ==========================================================================
|
||
# --------------------------------------------------------------------------
|
||
# Null-run and learning-curve reporting (no Qt)
|
||
# --------------------------------------------------------------------------
|
||
def per_seed_accuracy(summary_df):
|
||
"""Mean rank-1 balanced participant accuracy of each seed (over its balanced cells)."""
|
||
best = summary_df[summary_df["Balanced"] & (summary_df["Rank"] == 1)]
|
||
return best.groupby("Seed")["Part_Bal_Accuracy"].mean()
|
||
|
||
|
||
def settings_differences(a, b):
|
||
"""Names of the settings that differ between two runs (the ones that matter for comparing)."""
|
||
keys = ("n_min", "n_max", "t_min", "t_max", "test_k", "budget_only", "budget_tolerance",
|
||
"fixed_metrics", "top_k_expand", "include_unbalanced")
|
||
return [k for k in keys if getattr(a, k) != getattr(b, k)]
|
||
|
||
|
||
def null_report_lines(null_df, null_settings, real_df=None, real_settings=None):
|
||
"""Summarise a label-shuffle run and compare it with the latest real-label run."""
|
||
ns = per_seed_accuracy(null_df)
|
||
if ns.empty:
|
||
return ["The null run produced no balanced results."]
|
||
q95, q99 = ns.quantile([0.95, 0.99])
|
||
excess = float(ns.mean()) - 0.5
|
||
lines = [
|
||
"Class labels were shuffled across participants (same class sizes each seed).",
|
||
"With no real signal every accuracy should sit at 50%; anything above that is bias "
|
||
"built into the pipeline.",
|
||
f" Null accuracy per seed (mean over cells): mean {ns.mean() * 100:.1f}% | "
|
||
f"SD {ns.std(ddof=1) * 100:.1f} | 95th pct {q95 * 100:.1f}% | 99th pct {q99 * 100:.1f}% | "
|
||
f"max {ns.max() * 100:.1f}% ({len(ns)} permutations)",
|
||
f" Excess over chance: {excess * 100:+.1f} points"
|
||
+ (" (selecting among metric sets inflates even random labels)"
|
||
if not null_settings.fixed_metrics and excess > 0.02 else ""),
|
||
]
|
||
if real_df is None or real_df.empty or real_settings is None:
|
||
lines.append(" No real-label run in this session to compare with; run the real labels with "
|
||
"the same settings.")
|
||
return lines
|
||
real_mean = float(per_seed_accuracy(real_df).mean())
|
||
k = int((ns >= real_mean).sum())
|
||
lines += [
|
||
f" Real labels (last real run): mean {real_mean * 100:.1f}%",
|
||
f" Null permutations at or above the real mean: {k} of {len(ns)} -> permutation "
|
||
f"p <= {(1 + k) / (1 + len(ns)):.4f}",
|
||
" (Conservative: each null value is a single split, while the real mean averages many.)",
|
||
]
|
||
diff = settings_differences(null_settings, real_settings)
|
||
if diff:
|
||
lines.append(f" Warning: these settings differ from the real run ({', '.join(diff)}), so the "
|
||
f"comparison is only rough.")
|
||
return lines
|
||
|
||
|
||
def learning_curve_table(summary_df):
|
||
"""(table, T) of accuracy vs training people, or None unless the run used a single T."""
|
||
best = summary_df[summary_df["Balanced"] & (summary_df["Rank"] == 1)]
|
||
if best.empty or best["T_Trials_Per_Subject"].nunique() != 1 or best["N_Train_Subjects"].nunique() < 4:
|
||
return None
|
||
g = best.groupby("N_Train_Subjects")["Part_Bal_Accuracy"]
|
||
tbl = pd.DataFrame({"N_Total": g.mean().index.to_numpy(), "Mean": g.mean().to_numpy(),
|
||
"SE": (g.std(ddof=1) / np.sqrt(g.size())).to_numpy(),
|
||
"Seeds": g.size().to_numpy()})
|
||
tbl["N_Per_Class"] = tbl["N_Total"] / 2.0
|
||
return tbl.sort_values("N_Total").reset_index(drop=True), int(best["T_Trials_Per_Subject"].iloc[0])
|
||
|
||
|
||
def learning_curve_fit(tbl):
|
||
"""Least-squares fit acc = a - b / N_per_class (descriptive; 1/N is the shape the variance
|
||
model predicts for the error of the class means). Returns (a, b)."""
|
||
slope, intercept = np.polyfit(1.0 / tbl["N_Per_Class"].to_numpy(), tbl["Mean"].to_numpy(), 1)
|
||
return float(intercept), float(-slope)
|
||
|
||
|
||
def learning_curve_lines(summary_df):
|
||
"""Log lines for a single-T run: the curve, its fitted plateau, and whether it is still rising."""
|
||
res = learning_curve_table(summary_df)
|
||
if res is None:
|
||
return []
|
||
tbl, t_val = res
|
||
lines = [f"\n[Learning curve at T = {t_val} trials per person: accuracy vs training people per class]"]
|
||
for r in tbl.itertuples(index=False):
|
||
lines.append(f" {r.N_Per_Class:>5g} per class (N = {int(r.N_Total):>2d}): {r.Mean * 100:5.1f}% "
|
||
f"± {r.SE * 100:.1f} (SE over {int(r.Seeds)} seeds)")
|
||
a, b = learning_curve_fit(tbl)
|
||
if b > 0:
|
||
lines.append(f" Fit (descriptive): accuracy ≈ {a * 100:.1f}% − {b * 100:.1f}% / N_per_class. "
|
||
f"Asymptote ≈ {a * 100:.1f}%; within 1 point of it from about {b / 0.01:.0f} per class, "
|
||
f"within 0.5 point from about {b / 0.005:.0f}.")
|
||
else:
|
||
lines.append(" The fit shows no rise with the number of training people.")
|
||
best = summary_df[summary_df["Balanced"] & (summary_df["Rank"] == 1)]
|
||
piv = best.pivot_table(index="Seed", columns="N_Train_Subjects", values="Part_Bal_Accuracy")
|
||
levels = sorted(piv.columns)
|
||
if len(levels) >= 4:
|
||
hi, lo = levels[-1], levels[-4]
|
||
d = (piv[hi] - piv[lo]).dropna()
|
||
if len(d) > 2 and d.std(ddof=1) > 0:
|
||
half = float(stats.t.ppf(0.975, len(d) - 1)) * float(d.std(ddof=1)) / math.sqrt(len(d))
|
||
lines.append(
|
||
f" Still rising? {lo / 2:g} -> {hi / 2:g} per class (paired by seed): "
|
||
f"{d.mean() * 100:+.1f} points [95% CI {(d.mean() - half) * 100:+.1f}, "
|
||
f"{(d.mean() + half) * 100:+.1f}]. Seeds resample the same people, so this interval "
|
||
f"reflects split noise, not uncertainty about new people.")
|
||
return lines
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class GridSettings:
|
||
"""Every tunable of the N-vs-T grid experiment (all exposed in the GUI)."""
|
||
n_min: int = 3 # training participants PER CLASS, lower bound
|
||
n_max: int = 14 # ... upper bound
|
||
t_min: int = 2 # trials per training participant, lower bound
|
||
t_max: int = 18 # ... upper bound
|
||
test_k: int = 3 # held-out participants per class
|
||
n_seeds: int = 300 # Monte Carlo seeds
|
||
seed_start: int = 0
|
||
include_unbalanced: bool = False # also run "all remaining Class B" training sets
|
||
budget_only: bool = True # only (N, T) cells used by the budget tiers
|
||
budget_tolerance: float = DEFAULT_BUDGET_TOLERANCE
|
||
top_k_expand: int = 5 # expand this many best single metrics into all 3+ subsets (0/1/2 = off)
|
||
fixed_metrics: tuple = () # stage 2: evaluate ONLY this metric set (no sweep, no selection)
|
||
shuffle_labels: bool = False # null run: give each seed a random relabelling of the people
|
||
workers: int = DEFAULT_WORKERS
|
||
|
||
@property
|
||
def tiers(self):
|
||
return make_budget_tiers(self.budget_tolerance)
|
||
|
||
@property
|
||
def balance_modes(self):
|
||
return (True, False) if self.include_unbalanced else (True,)
|
||
|
||
|
||
def _is_pruned(n_sub, t_max):
|
||
"""Skip extreme corners (ultra-low N with ultra-low T, ultra-high both)."""
|
||
return (n_sub <= 3 and t_max <= 3) or (n_sub >= 12 and t_max >= 18)
|
||
|
||
|
||
def plan_cells(settings, pool_a, pool_b):
|
||
"""The (N, T) cells a run will cover, given the per-class pool sizes left after hold-out.
|
||
|
||
Returns a list of (balanced, n_sub, n_b, [t values]). N in the budget check is the total
|
||
number of training participants (class A + class B), like N_Train_Subjects in the results.
|
||
"""
|
||
tiers = settings.tiers
|
||
cells = []
|
||
for balanced in settings.balance_modes:
|
||
for n_sub in range(settings.n_min, settings.n_max + 1):
|
||
if n_sub > pool_a or (balanced and n_sub > pool_b):
|
||
continue
|
||
n_b = n_sub if balanced else pool_b
|
||
n_total = n_sub + n_b
|
||
ts = [t for t in range(settings.t_min, settings.t_max + 1)
|
||
if not _is_pruned(n_sub, t)
|
||
and (not settings.budget_only or cell_in_budget(tiers, n_total, t))]
|
||
if ts:
|
||
cells.append((balanced, n_sub, n_b, ts))
|
||
return cells
|
||
|
||
|
||
def estimate_workload(settings, n_class_a, n_class_b):
|
||
"""(configurations, metric sets per configuration) for the current settings and labels."""
|
||
if min(n_class_a, n_class_b) <= settings.test_k:
|
||
return 0, 0
|
||
cells = plan_cells(settings, n_class_a - settings.test_k, n_class_b - settings.test_k)
|
||
configs = settings.n_seeds * sum(len(ts) for *_, ts in cells)
|
||
sets_each = 1 if settings.fixed_metrics else metric_sets_per_task(settings.top_k_expand)
|
||
return configs, sets_each
|
||
|
||
|
||
# ==========================================================================
|
||
# Background worker: N vs T grid
|
||
# ==========================================================================
|
||
def _format_seconds(sec):
|
||
"""'45s', '7m 16s' or '2h 05m 09s' - larger units only when needed."""
|
||
sec = int(sec)
|
||
hours, rem = divmod(sec, 3600)
|
||
mins, secs = divmod(rem, 60)
|
||
if hours:
|
||
return f"{hours}h {mins:02d}m {secs:02d}s"
|
||
if mins:
|
||
return f"{mins}m {secs:02d}s"
|
||
return f"{secs}s"
|
||
|
||
|
||
class DiversityDensityGridWorker(QThread):
|
||
log_signal = Signal(str)
|
||
progress_log_signal = Signal(str)
|
||
finished_signal = Signal(object) # summary DataFrame (partial if cancelled)
|
||
|
||
def __init__(self, df, class_assign, all_feature_cols, selected_chromos, condition, settings):
|
||
super().__init__()
|
||
self.df = df
|
||
self.class_assign = class_assign # participant -> CLASS_NEG / CLASS_POS (assigned only)
|
||
self.all_feature_cols = all_feature_cols
|
||
self.selected_chromos = selected_chromos
|
||
self.condition = condition
|
||
self.settings = settings
|
||
self._cancel = threading.Event()
|
||
self.was_cancelled = False
|
||
self.first_error = None # traceback of the first failed configuration, for the log
|
||
|
||
def cancel(self):
|
||
"""Ask the worker to stop: all worker processes are killed, results so far are kept."""
|
||
self._cancel.set()
|
||
|
||
def _build_tasks(self):
|
||
s = self.settings
|
||
class_a = [p for p, c in self.class_assign.items() if c == CLASS_NEG]
|
||
class_b = [p for p, c in self.class_assign.items() if c == CLASS_POS]
|
||
if min(len(class_a), len(class_b)) <= s.test_k:
|
||
return []
|
||
|
||
cond_df = self.df[self.df["Condition"] == self.condition]
|
||
cells = plan_cells(s, len(class_a) - s.test_k, len(class_b) - s.test_k)
|
||
tasks = []
|
||
|
||
people = list(self.class_assign)
|
||
real_labels = [self.class_assign[p] for p in people]
|
||
|
||
for seed in range(s.seed_start, s.seed_start + s.n_seeds):
|
||
rng = np.random.default_rng(seed)
|
||
assign, seed_a, seed_b = self.class_assign, class_a, class_b
|
||
if s.shuffle_labels:
|
||
# Null run: the same class sizes, but the labels land on random people
|
||
# (own RNG stream, so the split draws below are unaffected).
|
||
perm = np.random.default_rng(seed + 1_000_003).permutation(len(people))
|
||
assign = {p: real_labels[i] for p, i in zip(people, perm)}
|
||
seed_a = [p for p in people if assign[p] == CLASS_NEG]
|
||
seed_b = [p for p in people if assign[p] == CLASS_POS]
|
||
test_a = rng.choice(seed_a, s.test_k, replace=False).tolist()
|
||
test_b = rng.choice(seed_b, s.test_k, replace=False).tolist()
|
||
pool_a = [p for p in seed_a if p not in test_a]
|
||
pool_b = [p for p in seed_b if p not in test_b]
|
||
|
||
for balanced, n_sub, n_b, t_values in cells:
|
||
train_parts = (rng.choice(pool_a, n_sub, replace=False).tolist()
|
||
+ rng.choice(pool_b, n_b, replace=False).tolist())
|
||
for t_max in t_values:
|
||
tasks.append({
|
||
"run_id": len(tasks) + 1,
|
||
"seed": seed,
|
||
"test_k": s.test_k,
|
||
"balanced": balanced,
|
||
"train_parts": train_parts,
|
||
"test_participants": test_a + test_b,
|
||
"t_max": t_max,
|
||
"cond_df": cond_df,
|
||
"class_assign": assign,
|
||
"all_feature_cols": self.all_feature_cols,
|
||
"selected_chromos": self.selected_chromos,
|
||
"top_k_expand": s.top_k_expand,
|
||
"fixed_metrics": s.fixed_metrics,
|
||
})
|
||
return tasks
|
||
|
||
def _run_pool(self, tasks):
|
||
"""Run tasks in a process pool. Returns (results, completed_count, failures).
|
||
|
||
Polls with a short timeout so a cancel request is noticed quickly; on cancel (or
|
||
completion) the pool is terminated, which kills every worker process immediately.
|
||
"""
|
||
total = len(tasks)
|
||
max_workers = max(1, self.settings.workers)
|
||
self.log_signal.emit(f"Launching process pool using {max_workers} parallel workers...")
|
||
|
||
results, done, failures = [], 0, 0
|
||
start = time.time()
|
||
bar_len = 25
|
||
|
||
pool = multiprocessing.Pool(processes=max_workers)
|
||
try:
|
||
iterator = pool.imap_unordered(_run_grid_task, tasks)
|
||
while done < total:
|
||
if self._cancel.is_set():
|
||
break
|
||
try:
|
||
res = iterator.next(timeout=0.25)
|
||
except multiprocessing.TimeoutError:
|
||
continue
|
||
except StopIteration:
|
||
break
|
||
except Exception:
|
||
failures += 1
|
||
if self.first_error is None:
|
||
self.first_error = traceback.format_exc()
|
||
else:
|
||
results.extend(res or [])
|
||
done += 1
|
||
|
||
elapsed = time.time() - start
|
||
eta = elapsed / done * (total - done)
|
||
filled = bar_len * done // total
|
||
bar = "█" * filled + "-" * (bar_len - filled)
|
||
self.progress_log_signal.emit(
|
||
f"Progress: |{bar}| {done}/{total} ({done / total * 100:.1f}%) | "
|
||
f"Elapsed: {_format_seconds(elapsed)} | ETA: ~{_format_seconds(eta)}")
|
||
finally:
|
||
pool.terminate()
|
||
pool.join()
|
||
|
||
return results, done, failures
|
||
|
||
def run(self):
|
||
try:
|
||
self.log_signal.emit(
|
||
f"=== Starting Multi-Process Targeted Grid Sweep on {get_cpu_model_name()} ===")
|
||
|
||
tasks = self._build_tasks()
|
||
self.log_signal.emit(f"Generated {len(tasks)} configuration tasks.")
|
||
if not tasks:
|
||
self.finished_signal.emit(pd.DataFrame())
|
||
return
|
||
|
||
results, done, failures = self._run_pool(tasks)
|
||
if failures:
|
||
self.log_signal.emit(
|
||
f"\nWarning: {failures} of {done} finished configurations failed and were skipped.")
|
||
if self.first_error:
|
||
self.log_signal.emit(f"First error:\n{self.first_error}")
|
||
|
||
self.was_cancelled = self._cancel.is_set() and done < len(tasks)
|
||
if self.was_cancelled:
|
||
self.log_signal.emit(
|
||
f"\n=== Cancelled: all workers stopped after {done} of {len(tasks)} configurations. "
|
||
f"Reporting partial results. ===")
|
||
else:
|
||
self.log_signal.emit(
|
||
f"\n=== Grid Experiment Complete! Logged top 3 metric sets across "
|
||
f"{len(tasks)} configurations. ===")
|
||
self.finished_signal.emit(pd.DataFrame(results))
|
||
except Exception as e:
|
||
self.log_signal.emit(f"Error during grid execution: {e}\n{traceback.format_exc()}")
|
||
self.finished_signal.emit(pd.DataFrame())
|
||
|
||
|
||
# ==========================================================================
|
||
# Model explanation (no Qt)
|
||
# ==========================================================================
|
||
METRIC_ORDER = {m: i for i, m in enumerate(METRIC_REGISTRY)}
|
||
CHROMO_COLOURS = {"HBO": "#c0392b", "HBR": "#2e86c1", "HBT": "#7d3c98"}
|
||
CLASS_COLOURS = {CLASS_NEG: "#3a6ea5", CLASS_POS: "#d98c1f"}
|
||
|
||
|
||
def parse_feature_key(col):
|
||
"""Split a feature column like 'ROI_X_hbo_Peak_Amp' into (ROI, CHROMOPHORE, metric key)."""
|
||
for metric, suffix in sorted(METRIC_REGISTRY.items(), key=lambda kv: -len(kv[1])):
|
||
if not col.endswith(f"_{suffix}"):
|
||
continue
|
||
if metric == "Peak_Amp" and col.endswith("_to_Peak_Amp"):
|
||
continue
|
||
stem = col[: -(len(suffix) + 1)]
|
||
tokens = stem.split("_")
|
||
chromo = next((t.upper() for t in tokens if t.lower() in KNOWN_CHROMOPHORES), "?")
|
||
roi = "_".join(t for t in tokens if t.lower() not in KNOWN_CHROMOPHORES) or stem
|
||
return roi, chromo, metric
|
||
return col, "?", "?"
|
||
|
||
|
||
def short_id(path):
|
||
"""Compact participant label for plots ('sub-15' when the file name has one)."""
|
||
name = Path(path).name
|
||
m = re.search(r"sub-[A-Za-z0-9]+", name)
|
||
return m.group(0) if m else name[:14]
|
||
|
||
|
||
def _bh_qvalues(p):
|
||
"""Benjamini-Hochberg adjusted p-values (NaNs ignored)."""
|
||
p = np.asarray(p, float)
|
||
q = np.full(p.shape, np.nan)
|
||
ok = np.isfinite(p)
|
||
n = int(ok.sum())
|
||
if n == 0:
|
||
return q
|
||
pv = p[ok]
|
||
order = np.argsort(pv)
|
||
scaled = pv[order] * n / np.arange(1, n + 1)
|
||
scaled = np.minimum.accumulate(scaled[::-1])[::-1]
|
||
out = np.empty(n)
|
||
out[order] = np.minimum(scaled, 1.0)
|
||
q[ok] = out
|
||
return q
|
||
|
||
|
||
def _labelled_rows(df, condition, classes):
|
||
"""All trial rows of every labelled participant for one condition."""
|
||
cond_df = df[df["Condition"] == condition]
|
||
frames = [_participant_rows(cond_df[cond_df["Participant"] == s]) for s in classes]
|
||
return pd.concat([f for f in frames if len(f)], ignore_index=True)
|
||
|
||
|
||
def explain_model(bundle, df, classes):
|
||
"""Why does this model separate Class A from Class B?
|
||
|
||
Per feature (participant-level, so every participant counts once):
|
||
Mean_A / Mean_B / Delta / Pct_Diff raw group means (B - A)
|
||
Cohen_d, AUC (B > A), P_Value, q_BH effect size, rank-based separability, Mann-Whitney
|
||
p and its Benjamini-Hochberg adjustment
|
||
LDA_Weight weight on the standardised feature
|
||
Contribution weight x (mean standardised B - mean standardised A);
|
||
the contributions SUM to the gap between the two
|
||
groups' mean decision scores (exact, the model is linear)
|
||
Share contribution as % of that gap
|
||
Contributions of correlated features can offset each other, so shares can be negative or
|
||
exceed 100%; the ablation in the GUI is the check on what really matters.
|
||
"""
|
||
cols = bundle["feature_columns"]
|
||
rows = _labelled_rows(df, bundle["condition"], classes)
|
||
X = rows[cols].replace([np.inf, -np.inf], np.nan)
|
||
part_key = rows["Participant"].to_numpy()
|
||
|
||
part_means = X.groupby(part_key).mean()
|
||
labels = np.array([classes[p] for p in part_means.index])
|
||
is_a = labels == CLASS_NEG
|
||
|
||
pipe = bundle["pipeline"]
|
||
Z = pipe[:-1].transform(X.to_numpy(float)) # imputed + standardised trial rows
|
||
z_means = pd.DataFrame(Z).groupby(part_key).mean().reindex(part_means.index).to_numpy()
|
||
delta_z = z_means[~is_a].mean(axis=0) - z_means[is_a].mean(axis=0)
|
||
weights = pipe.named_steps["lda"].coef_.ravel() # > 0 pushes towards Class B
|
||
contrib = weights * delta_z
|
||
gap = float(contrib.sum())
|
||
|
||
recs = []
|
||
for j, col in enumerate(cols):
|
||
a = part_means.loc[is_a, col].dropna().to_numpy()
|
||
b = part_means.loc[~is_a, col].dropna().to_numpy()
|
||
ma = float(a.mean()) if len(a) else np.nan
|
||
mb = float(b.mean()) if len(b) else np.nan
|
||
d = auc = p = np.nan
|
||
if len(a) > 1 and len(b) > 1:
|
||
pooled = np.sqrt(((len(a) - 1) * a.var(ddof=1) + (len(b) - 1) * b.var(ddof=1))
|
||
/ (len(a) + len(b) - 2))
|
||
d = (mb - ma) / pooled if pooled > 0 else np.nan
|
||
try:
|
||
res = stats.mannwhitneyu(b, a, alternative="two-sided")
|
||
auc = float(res.statistic) / (len(a) * len(b))
|
||
p = float(res.pvalue)
|
||
except ValueError:
|
||
pass
|
||
roi, chromo, metric = parse_feature_key(col)
|
||
recs.append({
|
||
"Feature": col, "ROI": roi, "Chromophore": chromo, "Metric": metric,
|
||
"Mean_A": ma, "Mean_B": mb, "Delta": mb - ma,
|
||
"Pct_Diff": (mb - ma) / abs(ma) * 100 if ma not in (0, np.nan) and np.isfinite(ma) else np.nan,
|
||
"Cohen_d": d, "AUC": auc, "P_Value": p,
|
||
"LDA_Weight": float(weights[j]), "Contribution": float(contrib[j]),
|
||
"Share": float(contrib[j]) / gap * 100 if gap else np.nan,
|
||
})
|
||
st = pd.DataFrame(recs)
|
||
st["q_BH"] = _bh_qvalues(st["P_Value"])
|
||
|
||
roi_tbl = (
|
||
st.groupby("ROI")
|
||
.agg(Contribution=("Contribution", "sum"),
|
||
Mean_Abs_d=("Cohen_d", lambda s: float(np.nanmean(np.abs(s))) if s.notna().any() else np.nan),
|
||
N_Features=("Feature", "size"))
|
||
.sort_values("Contribution", ascending=False)
|
||
)
|
||
roi_tbl["Share"] = roi_tbl["Contribution"] / gap * 100 if gap else np.nan
|
||
|
||
return {"bundle": bundle, "stats": st, "roi_table": roi_tbl, "part_means": part_means,
|
||
"labels": labels, "gap": gap, "n_a": int(is_a.sum()), "n_b": int((~is_a).sum())}
|
||
|
||
|
||
def lopo_participant_scores(X, y, groups, col_idx):
|
||
"""Leave-one-participant-out: each participant's mean decision score from a model trained
|
||
on all the other participants (all their trials), using only the columns in col_idx."""
|
||
Xs = X[:, col_idx]
|
||
scores = {}
|
||
for p in pd.unique(groups):
|
||
test = groups == p
|
||
pipe = make_pipeline()
|
||
pipe.fit(Xs[~test], y[~test])
|
||
scores[p] = float(np.mean(pipe.decision_function(Xs[test])))
|
||
return pd.Series(scores)
|
||
|
||
|
||
def lopo_balanced_accuracy(scores, classes):
|
||
true = np.array([classes[p] for p in scores.index])
|
||
pred = np.where(scores.to_numpy() > 0, CLASS_POS, CLASS_NEG)
|
||
return float(balanced_accuracy_score(true, pred))
|
||
|
||
|
||
def _fmt_p(v):
|
||
if not np.isfinite(v):
|
||
return "—"
|
||
return f"{v:.2e}" if v < 1e-3 else f"{v:.4f}"
|
||
|
||
|
||
def explanation_html(expl, ablation=None):
|
||
"""Plain-language summary of the model for the explanation tab."""
|
||
esc = html.escape
|
||
b, st, roi_tbl, gap = expl["bundle"], expl["stats"], expl["roi_table"], expl["gap"]
|
||
metrics = ", ".join(b["metrics"])
|
||
lines = [
|
||
f"<b>Model:</b> {esc(metrics)} | {len(st)} features = {st['ROI'].nunique()} ROI(s) × "
|
||
f"{st['Chromophore'].nunique()} chromophore(s) × {len(b['metrics'])} metric(s) | trained on "
|
||
f"{expl['n_a']} {CLASS_NEG} + {expl['n_b']} {CLASS_POS} participants.",
|
||
f"<b>How it decides:</b> every trial gets an LDA score; a participant is {CLASS_POS} when "
|
||
f"their mean trial score is > 0. The two groups' mean scores differ by <b>{gap:.2f}</b> "
|
||
f"units, and that gap equals the sum of the per-feature contributions (weight × group "
|
||
f"difference, standardised) listed in the table.",
|
||
]
|
||
|
||
if len(roi_tbl):
|
||
top = [f"<b>{esc(str(r))}</b> ({v:.0f}%)" for r, v in roi_tbl["Share"].head(3).items()
|
||
if np.isfinite(v)]
|
||
if top:
|
||
lines.append("<b>ROIs carrying the separation (share of the gap):</b> " + ", ".join(top) + ".")
|
||
|
||
items = []
|
||
for rec in st.sort_values("Contribution", ascending=False).head(3).itertuples():
|
||
pct = f" ({rec.Pct_Diff:+.0f}%)" if np.isfinite(rec.Pct_Diff) else ""
|
||
items.append(
|
||
f"<li><b>{esc(rec.ROI)} · {rec.Chromophore} · {rec.Metric}</b>: {CLASS_NEG} {rec.Mean_A:.4g} vs "
|
||
f"{CLASS_POS} {rec.Mean_B:.4g}{pct}; d = {rec.Cohen_d:.2f}, AUC = {rec.AUC:.2f}, "
|
||
f"q = {rec.q_BH:.3g}; carries {rec.Share:.0f}% of the gap.</li>")
|
||
lines.append("<b>Strongest features:</b><ul style='margin:0'>" + "".join(items) + "</ul>")
|
||
|
||
higher = int((st["Delta"] > 0).sum())
|
||
lines.append(f"{CLASS_POS} has the higher {esc(metrics)} in {higher} of {len(st)} features "
|
||
f"(participant-level means).")
|
||
lines.append("<i>Features are correlated, so shares are a guide (they can be negative or exceed "
|
||
"100%). Run the ablation to see what the model really depends on.</i>")
|
||
|
||
if ablation is not None:
|
||
rows, base = ablation["rows"], ablation["baseline"]
|
||
lines.append(f"<b>Leave-one-participant-out balanced accuracy, all features:</b> {base * 100:.1f}%")
|
||
for kind in ("ROI", "Chromophore"):
|
||
sub = rows[rows["Kind"] == kind]
|
||
if sub.empty:
|
||
continue
|
||
worst = sub.loc[sub["Without"].idxmin()]
|
||
best_only = sub.loc[sub["Only"].idxmax()]
|
||
lines.append(
|
||
f"{kind}: removing <b>{esc(str(worst['Group']))}</b> hurts most "
|
||
f"({worst['Without'] * 100:.1f}%, {(worst['Without'] - base) * 100:+.1f} pts); "
|
||
f"<b>{esc(str(best_only['Group']))}</b> alone reaches {best_only['Only'] * 100:.1f}%.")
|
||
return "<br>".join(lines)
|
||
|
||
|
||
# ==========================================================================
|
||
# Model explanation widget (tab 3)
|
||
# ==========================================================================
|
||
class _PopWindow(QWidget):
|
||
"""Floating top-level window that hosts a plot panel; reports when the user closes it."""
|
||
|
||
def __init__(self, title, on_close):
|
||
super().__init__(None)
|
||
self.setWindowTitle(title)
|
||
self._on_close = on_close
|
||
lay = QVBoxLayout(self)
|
||
lay.setContentsMargins(4, 4, 4, 4)
|
||
|
||
def closeEvent(self, event):
|
||
cb, self._on_close = self._on_close, None
|
||
if cb:
|
||
cb()
|
||
event.accept()
|
||
|
||
|
||
class PopOutController:
|
||
"""Moves a plot panel between its docked spot and its own window (e.g. on a second monitor).
|
||
|
||
The view selector stays in the main window, so the plot can be changed while it is
|
||
floating, and the docked space it leaves behind is given back to the table.
|
||
"""
|
||
|
||
def __init__(self, panel, host_layout, button, title, stretch=1, on_change=None, placeholder_stretch=None):
|
||
self.panel, self.host, self.button, self.title = panel, host_layout, button, title
|
||
self.stretch = stretch
|
||
self.placeholder_stretch = stretch if placeholder_stretch is None else placeholder_stretch
|
||
self.on_change = on_change
|
||
self.window = None
|
||
self._placeholder = None
|
||
self._index = 0
|
||
button.setText("Pop out plot")
|
||
button.setToolTip("Move the plot into its own window (e.g. onto a second monitor).\n"
|
||
"Close that window, or press this button again, to dock it back.")
|
||
button.clicked.connect(self.toggle)
|
||
|
||
def toggle(self):
|
||
if self.window is None:
|
||
self.pop_out()
|
||
else:
|
||
self.dock()
|
||
|
||
def pop_out(self):
|
||
self._index = max(0, self.host.indexOf(self.panel))
|
||
self.host.removeWidget(self.panel)
|
||
self._placeholder = QLabel("The plot is in its own window.\nClose that window or press "
|
||
"'Dock plot' to bring it back.")
|
||
self._placeholder.setAlignment(Qt.AlignCenter)
|
||
self._placeholder.setStyleSheet("color: #888888;")
|
||
self.host.insertWidget(self._index, self._placeholder, self.placeholder_stretch)
|
||
|
||
win = _PopWindow(self.title, lambda: self.dock(close_window=False))
|
||
win.layout().addWidget(self.panel)
|
||
self.panel.show()
|
||
win.resize(1100, 800)
|
||
win.show()
|
||
self.window = win
|
||
self.button.setText("Dock plot")
|
||
if self.on_change:
|
||
self.on_change(True)
|
||
|
||
def dock(self, close_window=True):
|
||
win, self.window = self.window, None
|
||
if win is None:
|
||
return
|
||
win._on_close = None
|
||
win.layout().removeWidget(self.panel)
|
||
if self._placeholder is not None:
|
||
self.host.removeWidget(self._placeholder)
|
||
self._placeholder.deleteLater()
|
||
self._placeholder = None
|
||
self.host.insertWidget(self._index, self.panel, self.stretch)
|
||
self.panel.show()
|
||
self.button.setText("Pop out plot")
|
||
if close_window:
|
||
win.close()
|
||
win.deleteLater()
|
||
if self.on_change:
|
||
self.on_change(False)
|
||
|
||
def close(self):
|
||
"""Dock the plot back and close its window (used when the main window closes)."""
|
||
self.dock()
|
||
|
||
|
||
class NumItem(QTableWidgetItem):
|
||
"""Table cell that shows formatted text but sorts by its numeric value."""
|
||
|
||
def __init__(self, value, text):
|
||
super().__init__(text)
|
||
v = float(value)
|
||
self._v = v if np.isfinite(v) else float("-inf")
|
||
self.setFlags(self.flags() & ~Qt.ItemIsEditable)
|
||
self.setTextAlignment(Qt.AlignRight | Qt.AlignVCenter)
|
||
|
||
def __lt__(self, other):
|
||
o = getattr(other, "_v", None)
|
||
return self._v < o if o is not None else super().__lt__(other)
|
||
|
||
|
||
class AblationWorker(QThread):
|
||
"""Leave-one-participant-out accuracy of the full model, and with each ROI / chromophore
|
||
kept alone or removed. Runs off the GUI thread."""
|
||
progress = Signal(str)
|
||
done = Signal(object) # result dict, or None if stopped
|
||
|
||
def __init__(self, rows, cols, classes):
|
||
super().__init__()
|
||
self.cols = list(cols)
|
||
self.classes = classes
|
||
self.X = rows[self.cols].replace([np.inf, -np.inf], np.nan).to_numpy(float)
|
||
self.groups = rows["Participant"].to_numpy()
|
||
self.y = np.array([classes[p] for p in self.groups])
|
||
self._halt = threading.Event()
|
||
|
||
def stop(self):
|
||
self._halt.set()
|
||
|
||
def run(self):
|
||
try:
|
||
parsed = [parse_feature_key(c) for c in self.cols]
|
||
all_idx = np.arange(len(self.cols))
|
||
variants = []
|
||
for kind, pos in (("ROI", 0), ("Chromophore", 1)):
|
||
names = sorted({p[pos] for p in parsed})
|
||
if len(names) < 2:
|
||
continue # nothing to remove
|
||
for name in names:
|
||
idx = np.array([i for i, p in enumerate(parsed) if p[pos] == name])
|
||
variants.append((kind, name, idx, np.setdiff1d(all_idx, idx)))
|
||
|
||
total = 1 + 2 * len(variants)
|
||
step = 0
|
||
|
||
def score(idx):
|
||
nonlocal step
|
||
step += 1
|
||
self.progress.emit(f"Ablation: fit set {step}/{total} ...")
|
||
return lopo_participant_scores(self.X, self.y, self.groups, idx)
|
||
|
||
base_scores = score(all_idx)
|
||
base_acc = lopo_balanced_accuracy(base_scores, self.classes)
|
||
out = []
|
||
for kind, name, idx, rest in variants:
|
||
if self._halt.is_set():
|
||
self.done.emit(None)
|
||
return
|
||
only = lopo_balanced_accuracy(score(idx), self.classes)
|
||
without = lopo_balanced_accuracy(score(rest), self.classes)
|
||
out.append({"Kind": kind, "Group": name, "Only": only, "Without": without,
|
||
"N_Features": int(len(idx))})
|
||
self.done.emit({"baseline": base_acc, "scores": base_scores, "rows": pd.DataFrame(out)})
|
||
except Exception as e:
|
||
self.progress.emit(f"Ablation failed: {e}")
|
||
self.done.emit(None)
|
||
|
||
|
||
class ModelExplainWidget(QWidget):
|
||
"""Tab 3: explains the model that 'Export Best Model' would write (or any top-10 metric set)."""
|
||
|
||
COLS = [("ROI", "ROI", "text"), ("Chromophore", "Chromophore", "text"), ("Metric", "Metric", "text"),
|
||
("Mean Class A", "Mean_A", "num"), ("Mean Class B", "Mean_B", "num"),
|
||
("Δ (B−A)", "Delta", "num"), ("% diff", "Pct_Diff", "pct"),
|
||
("Cohen's d", "Cohen_d", "num"), ("AUC (B>A)", "AUC", "num"),
|
||
("p", "P_Value", "p"), ("q (BH)", "q_BH", "p"),
|
||
("LDA weight", "LDA_Weight", "num"), ("Contribution", "Contribution", "num"),
|
||
("Share %", "Share", "pct")]
|
||
VIEWS = ["Selected feature: Class A vs Class B",
|
||
"Effect-size heatmap: ROI × chromophore/metric (click a cell)",
|
||
"LDA contribution per feature (click a bar)",
|
||
"ROI importance (summed contribution)",
|
||
"Ablation: accuracy with only / without each ROI & chromophore",
|
||
"Participant scores (leave-one-participant-out)"]
|
||
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
self.run = None
|
||
self.board = None
|
||
self.expl = None
|
||
self.ablation = None
|
||
self.selected = None
|
||
self.worker = None
|
||
self._click = None
|
||
self._main_ax = None
|
||
self._hover = None # hover tooltip state for the current plot (see _register_hover)
|
||
self._build_ui()
|
||
|
||
# ------------------------------------------------------------------ UI
|
||
def _build_ui(self):
|
||
root = QVBoxLayout(self)
|
||
|
||
top = QHBoxLayout()
|
||
top.addWidget(QLabel("Explain metric set:"))
|
||
self.metric_combo = QComboBox()
|
||
self.metric_combo.currentIndexChanged.connect(self.on_metric_changed)
|
||
top.addWidget(self.metric_combo, 1)
|
||
self.ablation_btn = QPushButton("Run ablation (leave-one-participant-out)")
|
||
self.ablation_btn.setToolTip(
|
||
"Refits the model many times: with every ROI / chromophore removed, and with each kept alone,\n"
|
||
"scoring each participant with a model that never saw them.")
|
||
self.ablation_btn.setEnabled(False)
|
||
self.ablation_btn.clicked.connect(self.run_ablation)
|
||
top.addWidget(self.ablation_btn)
|
||
root.addLayout(top)
|
||
|
||
self.status_label = QLabel("Run the grid experiment to load the best model here.")
|
||
root.addWidget(self.status_label)
|
||
|
||
self.info = QTextEdit()
|
||
self.info.setReadOnly(True)
|
||
self.info.setMaximumHeight(230)
|
||
root.addWidget(self.info)
|
||
|
||
splitter = QSplitter(Qt.Horizontal)
|
||
self.table = QTableWidget()
|
||
self.table.setColumnCount(len(self.COLS))
|
||
self.table.setHorizontalHeaderLabels([c[0] for c in self.COLS])
|
||
self.table.setSelectionBehavior(QTableWidget.SelectRows)
|
||
self.table.setSelectionMode(QTableWidget.SingleSelection)
|
||
self.table.setEditTriggers(QTableWidget.NoEditTriggers)
|
||
self.table.currentCellChanged.connect(self.on_table_cell_changed)
|
||
splitter.addWidget(self.table)
|
||
|
||
right = QWidget()
|
||
right_l = QVBoxLayout(right)
|
||
right_l.setContentsMargins(0, 0, 0, 0)
|
||
view_row = QHBoxLayout()
|
||
view_row.addWidget(QLabel("View:"))
|
||
self.view_combo = QComboBox()
|
||
self.view_combo.addItems(self.VIEWS)
|
||
self.view_combo.currentIndexChanged.connect(lambda _i: self.draw())
|
||
view_row.addWidget(self.view_combo, 1)
|
||
self.popout_btn = QPushButton()
|
||
view_row.addWidget(self.popout_btn)
|
||
right_l.addLayout(view_row)
|
||
|
||
self.plot_panel = QWidget()
|
||
panel_l = QVBoxLayout(self.plot_panel)
|
||
panel_l.setContentsMargins(0, 0, 0, 0)
|
||
if MPL_OK:
|
||
self.fig = Figure(figsize=(6, 5), dpi=100)
|
||
self.canvas = FigureCanvasQTAgg(self.fig)
|
||
self.toolbar = NavigationToolbar2QT(self.canvas, self)
|
||
panel_l.addWidget(self.toolbar)
|
||
panel_l.addWidget(self.canvas, 1)
|
||
self.canvas.mpl_connect("button_press_event", self.on_canvas_click)
|
||
self.canvas.mpl_connect("motion_notify_event", self.on_canvas_move)
|
||
else:
|
||
self.fig = self.canvas = self.toolbar = None
|
||
panel_l.addWidget(QLabel("matplotlib is not installed, so plots are unavailable "
|
||
"(the table and summary still work)."), 1)
|
||
right_l.addWidget(self.plot_panel, 1)
|
||
splitter.addWidget(right)
|
||
self.splitter = splitter
|
||
splitter.setSizes([700, 650])
|
||
root.addWidget(splitter, 1)
|
||
self.popout = PopOutController(
|
||
self.plot_panel, right_l, self.popout_btn, "Model Explanation - plot", stretch=1,
|
||
on_change=lambda popped: self.splitter.setSizes([1050, 220] if popped else [700, 650]))
|
||
|
||
def close_popout(self):
|
||
self.popout.close()
|
||
|
||
# ------------------------------------------------------------- lifecycle
|
||
def shutdown(self):
|
||
if self.worker is not None and self.worker.isRunning():
|
||
self.worker.stop()
|
||
self.worker.wait(20000)
|
||
|
||
def clear(self):
|
||
self.shutdown()
|
||
self.run = self.board = self.expl = self.ablation = self.selected = None
|
||
self.metric_combo.blockSignals(True)
|
||
self.metric_combo.clear()
|
||
self.metric_combo.blockSignals(False)
|
||
self.table.setRowCount(0)
|
||
self.info.clear()
|
||
self.ablation_btn.setEnabled(False)
|
||
self.metric_combo.setEnabled(True)
|
||
self.status_label.setText("Run the grid experiment to load the best model here.")
|
||
self.draw()
|
||
|
||
def set_context(self, run, board):
|
||
"""Called when a grid run finishes: load the model that would be exported."""
|
||
self.shutdown()
|
||
self.run = run
|
||
self.board = board.head(10).reset_index(drop=True)
|
||
self.metric_combo.blockSignals(True)
|
||
self.metric_combo.clear()
|
||
for i, row in enumerate(self.board.itertuples(index=False)):
|
||
tag = " ★ best (this is what 'Export Best Model' saves)" if i == 0 else ""
|
||
self.metric_combo.addItem(
|
||
f"{row.Metrics} | 1st in {row.Top1_Count}, top-3 in {row.Top3_Count}{tag}", row.Metrics)
|
||
self.metric_combo.blockSignals(False)
|
||
self.build_explanation()
|
||
|
||
def on_metric_changed(self, _index):
|
||
if self.run is not None:
|
||
self.build_explanation()
|
||
|
||
def build_explanation(self):
|
||
run = self.run
|
||
metrics_str = self.metric_combo.currentData()
|
||
if run is None or not metrics_str:
|
||
return
|
||
bundle = train_final_model(run["df"], run["condition"], run["classes"], metrics_str,
|
||
run["feature_cols"], run["chromos"], class_hints=run["class_hints"])
|
||
self.expl = explain_model(bundle, run["df"], run["classes"])
|
||
self.ablation = None
|
||
self.selected = None
|
||
self.populate_table()
|
||
self.info.setHtml(explanation_html(self.expl))
|
||
self.ablation_btn.setEnabled(True)
|
||
self.status_label.setText(
|
||
f"Explaining [{metrics_str}] | condition {run['condition']} | "
|
||
f"chromophores {', '.join(run['chromos'])}")
|
||
self.draw()
|
||
|
||
# ----------------------------------------------------------------- table
|
||
@staticmethod
|
||
def _fmt(v, kind):
|
||
if not np.isfinite(v):
|
||
return "—"
|
||
if kind == "p":
|
||
return _fmt_p(v)
|
||
if kind == "pct":
|
||
return f"{v:.1f}"
|
||
return f"{v:.4g}"
|
||
|
||
def populate_table(self):
|
||
st = self.expl["stats"]
|
||
t = self.table
|
||
t.setSortingEnabled(False)
|
||
t.setRowCount(len(st))
|
||
for r, rec in enumerate(st.to_dict("records")):
|
||
for c, (_header, key, kind) in enumerate(self.COLS):
|
||
v = rec[key]
|
||
if kind == "text":
|
||
item = QTableWidgetItem(str(v))
|
||
item.setFlags(item.flags() & ~Qt.ItemIsEditable)
|
||
if c == 0:
|
||
item.setData(Qt.UserRole, rec["Feature"])
|
||
else:
|
||
item = NumItem(v, self._fmt(v, kind))
|
||
t.setItem(r, c, item)
|
||
t.setSortingEnabled(True)
|
||
contrib_col = [c[1] for c in self.COLS].index("Contribution")
|
||
t.sortItems(contrib_col, Qt.DescendingOrder)
|
||
t.resizeColumnsToContents()
|
||
if t.rowCount():
|
||
t.setCurrentCell(0, 0)
|
||
|
||
def on_table_cell_changed(self, row, _col, _prow, _pcol):
|
||
item = self.table.item(row, 0) if row >= 0 else None
|
||
self.selected = item.data(Qt.UserRole) if item is not None else None
|
||
if self.view_combo.currentIndex() in (0, 1, 2):
|
||
self.draw()
|
||
|
||
def select_feature(self, name):
|
||
for r in range(self.table.rowCount()):
|
||
item = self.table.item(r, 0)
|
||
if item is not None and item.data(Qt.UserRole) == name:
|
||
self.table.setCurrentCell(r, 0)
|
||
self.table.scrollToItem(item)
|
||
return
|
||
|
||
# -------------------------------------------------------------- ablation
|
||
def run_ablation(self):
|
||
if self.expl is None or (self.worker is not None and self.worker.isRunning()):
|
||
return
|
||
run = self.run
|
||
rows = _labelled_rows(run["df"], run["condition"], run["classes"])
|
||
cols = self.expl["bundle"]["feature_columns"]
|
||
self.worker = AblationWorker(rows, cols, run["classes"])
|
||
self.worker.progress.connect(self.status_label.setText)
|
||
self.worker.done.connect(self.on_ablation_done)
|
||
self.ablation_btn.setEnabled(False)
|
||
self.metric_combo.setEnabled(False)
|
||
self.status_label.setText("Ablation: starting ...")
|
||
self.worker.start()
|
||
|
||
def on_ablation_done(self, result):
|
||
self.metric_combo.setEnabled(True)
|
||
if self.expl is None:
|
||
return
|
||
self.ablation_btn.setEnabled(True)
|
||
if result is None:
|
||
self.status_label.setText("Ablation did not complete.")
|
||
return
|
||
self.ablation = result
|
||
self.status_label.setText(
|
||
f"Ablation done | leave-one-participant-out balanced accuracy "
|
||
f"{result['baseline'] * 100:.1f}%")
|
||
self.info.setHtml(explanation_html(self.expl, result))
|
||
self.view_combo.setCurrentIndex(4) # triggers a redraw
|
||
|
||
# ------------------------------------------------------------------ plots
|
||
def on_canvas_click(self, event):
|
||
if (self._click is None or event.inaxes is None or event.inaxes is not self._main_ax
|
||
or event.xdata is None or event.ydata is None):
|
||
return
|
||
if self.toolbar is not None and str(self.toolbar.mode): # zoom / pan active
|
||
return
|
||
mode, payload = self._click
|
||
name = None
|
||
if mode == "grid":
|
||
r, c = int(round(event.ydata)), int(round(event.xdata))
|
||
if 0 <= r < len(payload) and 0 <= c < len(payload[0]):
|
||
name = payload[r][c]
|
||
elif mode == "bars":
|
||
i = int(round(event.ydata))
|
||
if 0 <= i < len(payload):
|
||
name = payload[i]
|
||
if name:
|
||
self.select_feature(name)
|
||
|
||
def _register_hover(self, ax, xy, texts, max_items=8):
|
||
"""Make the points at data coordinates xy show their texts when the mouse is near.
|
||
|
||
Points that overlap are all listed (nearest first), so bunched participants can be told apart.
|
||
"""
|
||
ann = ax.annotate("", xy=(0, 0), xytext=(14, 14), textcoords="offset points", fontsize=8,
|
||
bbox=dict(boxstyle="round", fc="white", ec="#555555", alpha=0.95),
|
||
arrowprops=dict(arrowstyle="-", color="#555555"), zorder=20,
|
||
annotation_clip=False)
|
||
ann.set_visible(False)
|
||
self._hover = {"ax": ax, "xy": np.asarray(xy, float), "text": list(texts), "ann": ann,
|
||
"max": max_items}
|
||
|
||
def on_canvas_move(self, event):
|
||
h = self._hover
|
||
if h is None:
|
||
return
|
||
ann = h["ann"]
|
||
if event.inaxes is not h["ax"] or event.x is None:
|
||
if ann.get_visible():
|
||
ann.set_visible(False)
|
||
self.canvas.draw_idle()
|
||
return
|
||
disp = h["ax"].transData.transform(h["xy"])
|
||
dist = np.hypot(disp[:, 0] - event.x, disp[:, 1] - event.y)
|
||
near = np.where(dist <= 12)[0]
|
||
if near.size == 0:
|
||
if ann.get_visible():
|
||
ann.set_visible(False)
|
||
self.canvas.draw_idle()
|
||
return
|
||
near = near[np.argsort(dist[near])][: h["max"]]
|
||
ann.xy = tuple(h["xy"][near[0]])
|
||
ann.set_text("\n".join(h["text"][i] for i in near))
|
||
right_half = event.x > h["ax"].bbox.x0 + h["ax"].bbox.width / 2
|
||
ann.xyann = (-14, 14) if right_half else (14, 14) # keep the label inside the plot
|
||
ann.set_ha("right" if right_half else "left")
|
||
ann.set_visible(True)
|
||
self.canvas.draw_idle()
|
||
|
||
def draw(self):
|
||
if not MPL_OK or self.canvas is None:
|
||
return
|
||
fig = self.fig
|
||
fig.clear()
|
||
self._click = None
|
||
self._main_ax = None
|
||
self._hover = None
|
||
if self.expl is None:
|
||
self._placeholder(fig, "Run the grid experiment to load a model here.")
|
||
else:
|
||
draw_fns = [self._draw_feature, self._draw_heatmap, self._draw_contrib,
|
||
self._draw_roi, self._draw_ablation, self._draw_scores]
|
||
try:
|
||
draw_fns[self.view_combo.currentIndex()](fig)
|
||
fig.tight_layout()
|
||
except Exception as e:
|
||
fig.clear()
|
||
self._placeholder(fig, f"Could not draw this view:\n{e}")
|
||
self.canvas.draw_idle()
|
||
|
||
@staticmethod
|
||
def _placeholder(fig, message):
|
||
ax = fig.add_subplot(111)
|
||
ax.axis("off")
|
||
ax.text(0.5, 0.5, message, ha="center", va="center", fontsize=11, wrap=True)
|
||
|
||
def _draw_feature(self, fig):
|
||
name = self.selected
|
||
if not name:
|
||
return self._placeholder(fig, "Select a feature in the table\n(or click a heatmap cell / bar).")
|
||
rec = self.expl["stats"].set_index("Feature").loc[name]
|
||
pm, labels = self.expl["part_means"], self.expl["labels"]
|
||
groups = [(cls, pm.loc[labels == cls, name].dropna()) for cls in (CLASS_NEG, CLASS_POS)]
|
||
|
||
ax = fig.add_subplot(111)
|
||
self._main_ax = ax
|
||
rng = np.random.default_rng(0)
|
||
hover_xy, hover_txt = [], []
|
||
for i, (cls, vals) in enumerate(groups):
|
||
if len(vals) == 0:
|
||
continue
|
||
bp = ax.boxplot(vals.to_numpy(), positions=[i], widths=0.5, showfliers=False,
|
||
patch_artist=True)
|
||
for patch in bp["boxes"]:
|
||
patch.set_facecolor(CLASS_COLOURS[cls])
|
||
patch.set_alpha(0.25)
|
||
for med in bp["medians"]:
|
||
med.set_color("black")
|
||
xs = i + rng.uniform(-0.15, 0.15, len(vals))
|
||
ax.scatter(xs, vals.to_numpy(), color=CLASS_COLOURS[cls], edgecolor="white", s=38, zorder=3)
|
||
ax.hlines(vals.mean(), i - 0.3, i + 0.3, color="black", lw=2.2, zorder=4)
|
||
for x, y, who in zip(xs, vals.to_numpy(), vals.index):
|
||
hover_xy.append((x, y))
|
||
hover_txt.append(f"{short_id(who)} ({cls}) {y:.4g}")
|
||
ax.set_xticks([0, 1])
|
||
ax.set_xticklabels([f"{cls}\n(n={len(v)})" for cls, v in groups])
|
||
ax.set_ylabel(f"{rec['Metric']} · {rec['Chromophore']} (participant mean; bar = group mean)")
|
||
ax.set_title(f"{rec['ROI']} — {rec['Chromophore']} — {rec['Metric']}")
|
||
txt = (f"Δ (B−A) = {rec['Delta']:.4g}\nCohen's d = {rec['Cohen_d']:.2f}\n"
|
||
f"AUC (B>A) = {rec['AUC']:.2f}\np = {_fmt_p(rec['P_Value'])}, "
|
||
f"q(BH) = {_fmt_p(rec['q_BH'])}\ncontributes {rec['Share']:.0f}% of the gap")
|
||
ax.text(0.02, 0.98, txt, transform=ax.transAxes, va="top", fontsize=9,
|
||
bbox=dict(boxstyle="round", fc="white", ec="#999999", alpha=0.9))
|
||
self._register_hover(ax, hover_xy, hover_txt) # hover a dot to see who it is
|
||
|
||
def _draw_heatmap(self, fig):
|
||
st = self.expl["stats"]
|
||
rois = sorted(st["ROI"].unique())
|
||
keys = sorted({(r.Chromophore, r.Metric) for r in st.itertuples(index=False)},
|
||
key=lambda k: (k[0], METRIC_ORDER.get(k[1], 99)))
|
||
r_idx = {r: i for i, r in enumerate(rois)}
|
||
k_idx = {k: j for j, k in enumerate(keys)}
|
||
mat = np.full((len(rois), len(keys)), np.nan)
|
||
names = [[None] * len(keys) for _ in rois]
|
||
for rec in st.itertuples(index=False):
|
||
i, j = r_idx[rec.ROI], k_idx[(rec.Chromophore, rec.Metric)]
|
||
mat[i, j] = rec.Cohen_d
|
||
names[i][j] = rec.Feature
|
||
|
||
finite = np.isfinite(mat)
|
||
vmax = max(0.3, float(np.nanmax(np.abs(mat)))) if finite.any() else 1.0
|
||
ax = fig.add_subplot(111)
|
||
self._main_ax = ax
|
||
im = ax.imshow(np.ma.masked_invalid(mat), cmap="coolwarm", vmin=-vmax, vmax=vmax, aspect="auto")
|
||
ax.set_xticks(range(len(keys)))
|
||
ax.set_xticklabels([f"{c}\n{m}" for c, m in keys], fontsize=8)
|
||
ax.set_yticks(range(len(rois)))
|
||
ax.set_yticklabels(rois, fontsize=8)
|
||
if mat.size <= 120:
|
||
for i in range(mat.shape[0]):
|
||
for j in range(mat.shape[1]):
|
||
if finite[i, j]:
|
||
ax.text(j, i, f"{mat[i, j]:.2f}", ha="center", va="center", fontsize=8)
|
||
for i, row in enumerate(names):
|
||
for j, nm in enumerate(row):
|
||
if nm is not None and nm == self.selected:
|
||
ax.add_patch(Rectangle((j - 0.5, i - 0.5), 1, 1, fill=False, ec="black", lw=2.5))
|
||
fig.colorbar(im, ax=ax, label="Cohen's d (Class B − Class A)", shrink=0.85)
|
||
ax.set_title("Participant-level effect size by ROI — click a cell")
|
||
self._click = ("grid", names)
|
||
|
||
def _draw_contrib(self, fig):
|
||
st = self.expl["stats"].copy()
|
||
st["abs"] = st["Contribution"].abs()
|
||
top = st.sort_values("abs", ascending=False).head(40).sort_values("Contribution")
|
||
multi = st["Metric"].nunique() > 1
|
||
labels = [f"{r.ROI} · {r.Chromophore}" + (f" · {r.Metric}" if multi else "")
|
||
for r in top.itertuples(index=False)]
|
||
y = np.arange(len(top))
|
||
ax = fig.add_subplot(111)
|
||
self._main_ax = ax
|
||
bars = ax.barh(y, top["Contribution"].to_numpy(),
|
||
color=[CHROMO_COLOURS.get(c, "#7f8c8d") for c in top["Chromophore"]])
|
||
for bar, name in zip(bars, top["Feature"]):
|
||
if name == self.selected:
|
||
bar.set_edgecolor("black")
|
||
bar.set_linewidth(2.5)
|
||
ax.set_yticks(y)
|
||
ax.set_yticklabels(labels, fontsize=8)
|
||
ax.axvline(0, color="black", lw=0.8)
|
||
ax.set_xlabel("Contribution to the Class B − Class A score gap (weight × Δz)")
|
||
suffix = " (top 40 by |contribution|)" if len(st) > 40 else ""
|
||
ax.set_title(f"LDA contribution per feature{suffix} — click a bar")
|
||
present = [c for c in CHROMO_COLOURS if c in set(top["Chromophore"])]
|
||
if present:
|
||
ax.legend(handles=[Patch(color=CHROMO_COLOURS[c], label=c) for c in present], fontsize=8)
|
||
self._click = ("bars", top["Feature"].tolist())
|
||
|
||
def _draw_roi(self, fig):
|
||
st, gap = self.expl["stats"], self.expl["gap"]
|
||
piv = st.pivot_table(index="ROI", columns="Chromophore", values="Contribution",
|
||
aggfunc="sum").fillna(0)
|
||
piv["__total"] = piv.sum(axis=1)
|
||
piv = piv.sort_values("__total", ascending=False)
|
||
total = piv.pop("__total")
|
||
x = np.arange(len(piv))
|
||
k = max(piv.shape[1], 1)
|
||
width = 0.8 / k
|
||
ax = fig.add_subplot(111)
|
||
for i, ch in enumerate(piv.columns):
|
||
ax.bar(x + (i - (k - 1) / 2) * width, piv[ch].to_numpy(), width,
|
||
color=CHROMO_COLOURS.get(ch, "#7f8c8d"), label=ch)
|
||
ax.plot(x, total.to_numpy(), "kD", label="ROI total", zorder=5)
|
||
if gap:
|
||
for xi, tv in zip(x, total.to_numpy()):
|
||
ax.annotate(f"{tv / gap * 100:.0f}%", (xi, tv), textcoords="offset points",
|
||
xytext=(0, 7), ha="center", fontsize=8)
|
||
ax.axhline(0, color="black", lw=0.8)
|
||
ax.set_xticks(x)
|
||
ax.set_xticklabels(piv.index, rotation=30, ha="right")
|
||
ax.set_ylabel("Summed LDA contribution")
|
||
ax.set_title("ROI importance (labels = share of the group gap)")
|
||
ax.legend(fontsize=8)
|
||
|
||
def _draw_ablation(self, fig):
|
||
ab = self.ablation
|
||
if ab is None:
|
||
return self._placeholder(
|
||
fig, "Press 'Run ablation (leave-one-participant-out)' above.\n"
|
||
"It refits the model without each ROI / chromophore and with each alone.")
|
||
rows, base = ab["rows"], ab["baseline"]
|
||
kinds = [k for k in ("ROI", "Chromophore") if (rows["Kind"] == k).any()]
|
||
if not kinds:
|
||
return self._placeholder(fig, "Only one ROI and one chromophore: nothing to remove.")
|
||
axes = fig.subplots(len(kinds), 1, squeeze=False)[:, 0]
|
||
for n, (ax, kind) in enumerate(zip(axes, kinds)):
|
||
sub = rows[rows["Kind"] == kind].sort_values("Without")
|
||
y = np.arange(len(sub))
|
||
ax.barh(y + 0.2, sub["Only"], 0.38, color="#27ae60", label="Only this group")
|
||
ax.barh(y - 0.2, sub["Without"], 0.38, color="#c0392b", label="Everything except this group")
|
||
ax.axvline(base, color="black", ls="--", label=f"All features ({base * 100:.1f}%)")
|
||
ax.axvline(0.5, color="#999999", ls=":", label="Chance")
|
||
ax.set_yticks(y)
|
||
ax.set_yticklabels(sub["Group"], fontsize=8)
|
||
ax.set_xlim(0.3, 1.02)
|
||
ax.set_xlabel("Leave-one-participant-out balanced accuracy")
|
||
ax.set_title(f"By {kind}")
|
||
if n == 0:
|
||
ax.legend(fontsize=7, loc="lower right")
|
||
|
||
def _draw_scores(self, fig):
|
||
ab = self.ablation
|
||
if ab is None:
|
||
return self._placeholder(
|
||
fig, "Press 'Run ablation (leave-one-participant-out)' above\n"
|
||
"to compute out-of-sample participant scores.")
|
||
classes = self.run["classes"]
|
||
sc = ab["scores"].sort_values(ascending=False)
|
||
true = [classes[p] for p in sc.index]
|
||
wrong = [(s > 0) != (t == CLASS_POS) for s, t in zip(sc.to_numpy(), true)]
|
||
ax = fig.add_subplot(111)
|
||
x = np.arange(len(sc))
|
||
bars = ax.bar(x, sc.to_numpy(), color=[CLASS_COLOURS[t] for t in true])
|
||
for bar, w in zip(bars, wrong):
|
||
if w:
|
||
bar.set_edgecolor("red")
|
||
bar.set_linewidth(2)
|
||
bar.set_hatch("//")
|
||
self._register_hover(
|
||
ax, list(zip(x, sc.to_numpy())),
|
||
[f"{short_id(p)} ({t}) score {s:+.2f}" + (" MISCLASSIFIED" if w else "")
|
||
for p, t, w, s in zip(sc.index, true, wrong, sc.to_numpy())], max_items=3)
|
||
ax.axhline(0, color="black", lw=0.8)
|
||
ax.set_xticks(x)
|
||
ax.set_xticklabels([short_id(p) for p in sc.index], rotation=90, fontsize=7)
|
||
ax.set_ylabel("Mean LDA score (> 0 → Class B)")
|
||
ax.set_title(f"Out-of-sample participant scores — balanced accuracy {ab['baseline'] * 100:.1f}%")
|
||
ax.legend(handles=[Patch(color=CLASS_COLOURS[CLASS_NEG], label=CLASS_NEG),
|
||
Patch(color=CLASS_COLOURS[CLASS_POS], label=CLASS_POS),
|
||
Patch(fc="white", ec="red", hatch="//", label="Misclassified")], fontsize=8)
|
||
|
||
|
||
# ==========================================================================
|
||
# Variance decomposition and theory (no Qt)
|
||
# ==========================================================================
|
||
ARM_COLOURS = {"low": "#3a6ea5", "high": "#d98c1f", None: "#8c8c8c"}
|
||
|
||
|
||
def lopo_trial_scores(X, y, groups):
|
||
"""Out-of-sample LDA score for EVERY trial (> 0 -> Class B).
|
||
|
||
Each participant's trials are scored by a model trained on all the other participants, so
|
||
no score has seen the person it describes.
|
||
"""
|
||
out = np.full(len(y), np.nan)
|
||
for p in pd.unique(groups):
|
||
test = groups == p
|
||
pipe = make_pipeline()
|
||
pipe.fit(X[~test], y[~test])
|
||
out[test] = pipe.decision_function(X[test])
|
||
return out
|
||
|
||
|
||
def variance_components(x, subj, cls):
|
||
"""Random-effects split of one outcome into class gap, person-to-person and trial-to-trial.
|
||
|
||
x[i, t] = mu_class + u_i + e[i, t] u_i ~ (0, s2b) e ~ (0, s2w)
|
||
|
||
Method-of-moments (one-way ANOVA with class as a fixed factor, unbalanced-safe):
|
||
s2w = within-person mean square, s2b = (between-person MS - s2w) / n0 floored at 0.
|
||
ICC = s2b / (s2b + s2w): the share of within-class variation that is stable per person.
|
||
Returns None when there are too few participants or a class is missing.
|
||
"""
|
||
d = pd.DataFrame({"x": np.asarray(x, float), "s": np.asarray(subj), "c": np.asarray(cls)})
|
||
d = d.dropna(subset=["x"])
|
||
if d.empty:
|
||
return None
|
||
g = d.groupby("s", sort=False)
|
||
n_i, m_i, c_i = g["x"].size(), g["x"].mean(), g["c"].first()
|
||
n_part = len(n_i)
|
||
if CLASS_NEG not in set(c_i) or CLASS_POS not in set(c_i):
|
||
return None
|
||
dfw, dfb = int(n_i.sum() - n_part), n_part - 2
|
||
if dfw <= 0 or dfb <= 0:
|
||
return None
|
||
|
||
msw = float(((d["x"] - d["s"].map(m_i)) ** 2).sum() / dfw)
|
||
class_mean = d.groupby("c")["x"].mean()
|
||
msb = float((n_i * (m_i - c_i.map(class_mean)) ** 2).sum() / dfb)
|
||
num = 0.0
|
||
for c in (CLASS_NEG, CLASS_POS):
|
||
nc = n_i[c_i == c]
|
||
tc = float(nc.sum())
|
||
num += tc - float((nc ** 2).sum()) / tc
|
||
n0 = num / dfb
|
||
s2b = max(0.0, (msb - msw) / n0) if n0 > 0 else 0.0
|
||
s2w = msw
|
||
|
||
mu_a = float(m_i[c_i == CLASS_NEG].mean())
|
||
mu_b = float(m_i[c_i == CLASS_POS].mean())
|
||
delta = mu_b - mu_a
|
||
s2c = delta ** 2 / 4.0
|
||
total = s2c + s2b + s2w
|
||
rho = s2b / (s2b + s2w) if (s2b + s2w) > 0 else float("nan")
|
||
return {
|
||
"n_part": int(n_part), "n_trials": int(n_i.sum()), "mean_trials": float(n_i.mean()),
|
||
"mu_a": mu_a, "mu_b": mu_b, "delta": delta,
|
||
"s2c": s2c, "s2b": s2b, "s2w": s2w, "rho": rho,
|
||
"share_c": s2c / total if total > 0 else float("nan"),
|
||
"share_b": s2b / total if total > 0 else float("nan"),
|
||
"share_w": s2w / total if total > 0 else float("nan"),
|
||
"d_between": abs(delta) / math.sqrt(s2b) if s2b > 0 else float("inf"),
|
||
"t_sat": (1 - rho) / rho if rho and rho > 0 else float("inf"),
|
||
}
|
||
|
||
|
||
def design_effect(rho, t):
|
||
"""Variance inflation from clustering: 1 + (T - 1) * ICC."""
|
||
return 1.0 + (t - 1.0) * rho
|
||
|
||
|
||
def predicted_accuracy(comp, n_per_class, t_train, t_test):
|
||
"""Expected held-out participant accuracy for a one-dimensional score and midpoint rule.
|
||
|
||
A test person's mean score ~ N(mu_class, s2b + s2w / t_test). The threshold is the midpoint
|
||
of the two estimated class means, each estimated from n_per_class people with t_train trials
|
||
(variance (s2b + s2w / t_train) / n_per_class). So
|
||
accuracy = Phi( (delta / 2) / sqrt(v_test + v_train / (2 n)) ).
|
||
It ignores the extra error of estimating many LDA weights, so use it for direction and
|
||
shape, not as an exact accuracy.
|
||
"""
|
||
v_train = comp["s2b"] + comp["s2w"] / max(t_train, 1.0)
|
||
v_test = comp["s2b"] + comp["s2w"] / max(t_test, 1.0)
|
||
sd = math.sqrt(v_test + v_train / (2.0 * max(n_per_class, 1e-9)))
|
||
return float(stats.norm.cdf(abs(comp["delta"]) / 2.0 / sd)) if sd > 0 else float("nan")
|
||
|
||
|
||
def theory_tables(comp, summary_df, tiers):
|
||
"""(cells, tier_table): predicted vs observed accuracy per (N, T) cell and per budget tier.
|
||
|
||
Observed = mean rank-1 balanced participant accuracy of balanced runs in that cell.
|
||
"""
|
||
best = summary_df[summary_df["Balanced"] & (summary_df["Rank"] == 1)]
|
||
if best.empty:
|
||
return pd.DataFrame(), pd.DataFrame()
|
||
obs = (best.groupby(["N_Train_Subjects", "T_Trials_Per_Subject"])["Part_Bal_Accuracy"]
|
||
.mean().reset_index())
|
||
rho, t_test = comp["rho"], comp["mean_trials"]
|
||
rows = []
|
||
for r in obs.itertuples(index=False):
|
||
n_tot, t = int(r.N_Train_Subjects), int(r.T_Trials_Per_Subject)
|
||
n_cls = n_tot / 2.0
|
||
target = arm = None
|
||
for tier in tiers:
|
||
low, high = tier.masks(n_tot, t)
|
||
if low or high:
|
||
target, arm = tier.target, ("low" if low else "high")
|
||
break
|
||
de = design_effect(rho, t)
|
||
rows.append({
|
||
"N_Total": n_tot, "N_Per_Class": n_cls, "T": t, "Budget": n_tot * t,
|
||
"Tier": target, "Arm": arm, "Design_Effect": de, "N_Eff_Per_Class": n_cls * t / de,
|
||
"Pred": predicted_accuracy(comp, n_cls, min(t, t_test), t_test),
|
||
"Obs": float(r.Part_Bal_Accuracy)})
|
||
cells = pd.DataFrame(rows).sort_values(["Budget", "N_Total"]).reset_index(drop=True)
|
||
|
||
tier_rows = []
|
||
for tier in tiers:
|
||
sub = cells[cells["Tier"] == tier.target]
|
||
lo, hi = sub[sub["Arm"] == "low"], sub[sub["Arm"] == "high"]
|
||
if lo.empty or hi.empty:
|
||
continue
|
||
tier_rows.append({
|
||
"Target": tier.target,
|
||
"Obs_Low": lo["Obs"].mean(), "Obs_High": hi["Obs"].mean(),
|
||
"Pred_Low": lo["Pred"].mean(), "Pred_High": hi["Pred"].mean(),
|
||
"Obs_Diff": hi["Obs"].mean() - lo["Obs"].mean(),
|
||
"Pred_Diff": hi["Pred"].mean() - lo["Pred"].mean()})
|
||
return cells, pd.DataFrame(tier_rows)
|
||
|
||
|
||
def icc_reading(rho):
|
||
"""Plain-language reading of an intraclass correlation for the N-vs-T question."""
|
||
if not np.isfinite(rho):
|
||
return "The ICC could not be estimated."
|
||
if rho >= 0.5:
|
||
return ("**High ICC.** Most of what varies is a stable property of the person. A second, "
|
||
"third or tenth trial from the same person mostly repeats what is already known, so "
|
||
"spending the budget on **more people** should pay off clearly.")
|
||
if rho >= 0.2:
|
||
return ("**Moderate ICC.** A person's trials share a stable part but also carry real "
|
||
"trial-to-trial noise. The first few trials per person are valuable and then returns "
|
||
"fade, so **more people beats more trials once each person is past the saturation "
|
||
"point Tsat**, and the gap should shrink as the total budget grows.")
|
||
return ("**Low ICC.** Person-to-person differences are small next to trial-to-trial noise, so "
|
||
"trials are close to interchangeable and it should matter little whether they come from "
|
||
"few or many people. Expect little or no diversity advantage.")
|
||
|
||
|
||
def theory_markdown(run, metrics_str, comp, feat, tier_tbl, cells):
|
||
"""Plain-language variance-decomposition report: between-person variability, within-person
|
||
variability, and what they imply for splitting a fixed budget between people and trials."""
|
||
s = run["settings"]
|
||
fixed = bool(s.fixed_metrics)
|
||
rho = comp["rho"]
|
||
sb, sw = math.sqrt(comp["s2b"]), math.sqrt(comp["s2w"])
|
||
pts = lambda v: f"{v * 100:+.1f} points"
|
||
L = [
|
||
"# Variance decomposition: people versus trials",
|
||
f"*Condition **{run['condition']}** · metric set **{metrics_str}** · numbers are from your "
|
||
f"latest run.*",
|
||
"",
|
||
"## The framework",
|
||
"Every measurement is split into three parts: a **class effect** (the Class A versus Class B "
|
||
"gap), a **person effect** (how much people differ from each other within a class, "
|
||
"between-subject variability, σb²) and **trial noise** (how much one person varies from trial "
|
||
"to trial, within-subject variability, σw²). How a fixed budget of N people × T trials should "
|
||
"be allocated depends only on how big the person part is compared with the noise part. That "
|
||
"ratio is the intraclass correlation, ICC = σb² / (σb² + σw²).",
|
||
"",
|
||
"## What the numbers are",
|
||
f"Estimated on the model's out-of-sample score (every person scored by a model that never saw "
|
||
f"them; {comp['n_part']} people, {comp['n_trials']} trials, {comp['mean_trials']:.1f} trials "
|
||
f"each on average):",
|
||
f"- Class gap between group means: **{abs(comp['delta']):.3g}** score units "
|
||
f"({comp['d_between']:.1f} person-to-person standard deviations).",
|
||
f"- Between-person spread (σb): **{sb:.3g}**. Within-person, trial-to-trial spread (σw): "
|
||
f"**{sw:.3g}**.",
|
||
f"- Intraclass correlation **ICC = {rho:.2f}**: {rho * 100:.0f}% of the variation within a class "
|
||
f"is stable per person, {(1 - rho) * 100:.0f}% is trial noise.",
|
||
icc_reading(rho),
|
||
]
|
||
if len(feat):
|
||
q1, med, q3 = feat["rho"].quantile([0.25, 0.5, 0.75])
|
||
L.append(f"Across the {len(feat)} individual features the median ICC is {med:.2f} "
|
||
f"(middle half {q1:.2f} to {q3:.2f}).")
|
||
|
||
L += ["", "## What that means for allocating a fixed budget"]
|
||
if rho > 1e-6:
|
||
eff = ", ".join(f"T={t}: {t / design_effect(rho, t):.1f} of {t}" for t in (3, 6, 12, 18))
|
||
L += [
|
||
f"- **Effective trials.** N people × T trials are worth N·T / (1 + (T−1)·ICC) independent "
|
||
f"trials. With ICC = {rho:.2f}, one person's trials are worth {eff}. Even with unlimited "
|
||
f"trials one person is worth at most {1 / rho:.1f} independent trials.",
|
||
f"- **Saturation point.** Trial noise per person (σw²/T) falls below the stable person "
|
||
f"spread (σb²) at Tsat = (1 − ICC) / ICC = **{comp['t_sat']:.1f} trials**. Beyond that, more "
|
||
f"trials from the same person add little.",
|
||
]
|
||
else:
|
||
L.append("- The ICC is essentially zero, so every trial is worth a full independent trial and "
|
||
"there is no saturation point.")
|
||
L.append(f"- **Error of a class mean** from N people with T trials each is σb²/N + σw²/(N·T). At a "
|
||
f"fixed budget b = N·T per class the second term is constant, so only the first changes, and "
|
||
f"only more people shrinks it. The prize for diversity is therefore set by σb² "
|
||
f"(here {comp['s2b']:.3g}) relative to the budget.")
|
||
|
||
L += ["", "## Does the theory match your grid?"]
|
||
if len(tier_tbl):
|
||
agree = 0
|
||
for r in tier_tbl.itertuples(index=False):
|
||
agree += int(np.sign(round(r.Obs_Diff, 3)) == np.sign(round(r.Pred_Diff, 3)))
|
||
L.append(f"- **~{r.Target} trials:** the variance model predicts the high-N arm to be "
|
||
f"{pts(r.Pred_Diff)} versus the low-N arm; you observed {pts(r.Obs_Diff)}.")
|
||
L.append(f"The predicted and observed differences point the same way at {agree} of "
|
||
f"{len(tier_tbl)} budgets.")
|
||
offset = float((cells["Obs"] - cells["Pred"]).mean()) if len(cells) else float("nan")
|
||
L.append(f"Across all cells, observed accuracy is {pts(offset)} relative to the prediction on "
|
||
f"average. The model treats the classifier as a one-dimensional score and ignores the "
|
||
f"extra error of estimating the weights from small training sets, so read it for "
|
||
f"**direction and plateau, not absolute accuracy**."
|
||
+ ("" if fixed else " Observed numbers here are best-of-sweep and therefore optimistic; "
|
||
"re-run with a fixed metric set for a clean comparison."))
|
||
else:
|
||
L.append("Predicted-versus-observed needs balanced results in both arms of at least one budget.")
|
||
return "\n\n".join(L)
|
||
|
||
|
||
def build_theory(run, metrics_str, summary_df):
|
||
"""Variance components, predicted-vs-observed tables and the plain-language report."""
|
||
bundle = train_final_model(run["df"], run["condition"], run["classes"], metrics_str,
|
||
run["feature_cols"], run["chromos"])
|
||
cols = bundle["feature_columns"]
|
||
rows = _labelled_rows(run["df"], run["condition"], run["classes"])
|
||
groups = rows["Participant"].to_numpy()
|
||
y = rows["Participant"].map(run["classes"]).to_numpy()
|
||
X = rows[cols].replace([np.inf, -np.inf], np.nan).to_numpy(float)
|
||
|
||
comp = variance_components(lopo_trial_scores(X, y, groups), groups, y)
|
||
if comp is None:
|
||
raise ValueError("Too few labelled participants/trials to estimate variance components.")
|
||
|
||
feat_rows = []
|
||
for j, col in enumerate(cols):
|
||
fc = variance_components(X[:, j], groups, y)
|
||
if fc is None:
|
||
continue
|
||
roi, chromo, metric = parse_feature_key(col)
|
||
feat_rows.append({"Feature": col, "ROI": roi, "Chromophore": chromo, "Metric": metric,
|
||
"rho": fc["rho"], "s2b": fc["s2b"], "s2w": fc["s2w"], "delta": fc["delta"]})
|
||
feat = pd.DataFrame(feat_rows)
|
||
|
||
cells, tier_tbl = theory_tables(comp, summary_df, run["tiers"])
|
||
md = theory_markdown(run, metrics_str, comp, feat, tier_tbl, cells)
|
||
return {"comp": comp, "feat": feat, "cells": cells, "tiers": tier_tbl,
|
||
"markdown": md, "metrics": metrics_str}
|
||
|
||
|
||
# ==========================================================================
|
||
# Variance / theory widget (tab 4)
|
||
# ==========================================================================
|
||
class VarianceTheoryWidget(QWidget):
|
||
"""Tab 4: variance decomposition, theory-vs-grid comparison and the plain-language report."""
|
||
|
||
VIEWS = ["Variance components: class, person, trial",
|
||
"Fixed-budget curves: theory vs your grid",
|
||
"Predicted vs observed per (N, T) cell",
|
||
"Effective trials per person vs T",
|
||
"Learning curve (needs a single-T run)"]
|
||
CELL_COLS = [("Budget", "Budget", "int"), ("N / class", "N_Per_Class", "num"), ("T", "T", "int"),
|
||
("Arm", "Arm", "text"), ("Design effect", "Design_Effect", "num"),
|
||
("Eff. N / class", "N_Eff_Per_Class", "num"), ("Predicted %", "Pred", "pct"),
|
||
("Observed %", "Obs", "pct"), ("Obs − Pred (pts)", "Diff", "pts")]
|
||
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
self.run = None
|
||
self.summary = None
|
||
self.res = None
|
||
self._build_ui()
|
||
|
||
def _build_ui(self):
|
||
root = QVBoxLayout(self)
|
||
top = QHBoxLayout()
|
||
top.addWidget(QLabel("Metric set:"))
|
||
self.metric_combo = QComboBox()
|
||
self.metric_combo.currentIndexChanged.connect(self.on_metric_changed)
|
||
top.addWidget(self.metric_combo, 1)
|
||
self.save_btn = QPushButton("Save report (.md)")
|
||
self.save_btn.setEnabled(False)
|
||
self.save_btn.clicked.connect(self.save_report)
|
||
top.addWidget(self.save_btn)
|
||
root.addLayout(top)
|
||
|
||
self.status_label = QLabel("Run the grid experiment to load the variance analysis here.")
|
||
self.status_label.setWordWrap(True)
|
||
root.addWidget(self.status_label)
|
||
|
||
splitter = QSplitter(Qt.Horizontal)
|
||
self.report = QTextEdit()
|
||
self.report.setReadOnly(True)
|
||
splitter.addWidget(self.report)
|
||
|
||
right = QWidget()
|
||
rl = QVBoxLayout(right)
|
||
rl.setContentsMargins(0, 0, 0, 0)
|
||
row = QHBoxLayout()
|
||
row.addWidget(QLabel("View:"))
|
||
self.view_combo = QComboBox()
|
||
self.view_combo.addItems(self.VIEWS)
|
||
self.view_combo.currentIndexChanged.connect(lambda _i: self.draw())
|
||
row.addWidget(self.view_combo, 1)
|
||
self.popout_btn = QPushButton()
|
||
row.addWidget(self.popout_btn)
|
||
rl.addLayout(row)
|
||
|
||
self.plot_panel = QWidget()
|
||
panel_l = QVBoxLayout(self.plot_panel)
|
||
panel_l.setContentsMargins(0, 0, 0, 0)
|
||
if MPL_OK:
|
||
self.fig = Figure(figsize=(6, 4.5), dpi=100)
|
||
self.canvas = FigureCanvasQTAgg(self.fig)
|
||
self.toolbar = NavigationToolbar2QT(self.canvas, self)
|
||
panel_l.addWidget(self.toolbar)
|
||
panel_l.addWidget(self.canvas, 1)
|
||
else:
|
||
self.fig = self.canvas = None
|
||
panel_l.addWidget(QLabel("matplotlib is not installed, so plots are unavailable "
|
||
"(the report and table still work)."), 1)
|
||
rl.addWidget(self.plot_panel, 3)
|
||
self.table = QTableWidget()
|
||
self.table.setColumnCount(len(self.CELL_COLS))
|
||
self.table.setHorizontalHeaderLabels([c[0] for c in self.CELL_COLS])
|
||
self.table.setEditTriggers(QTableWidget.NoEditTriggers)
|
||
rl.addWidget(self.table, 2)
|
||
splitter.addWidget(right)
|
||
self.splitter = splitter
|
||
splitter.setSizes([650, 700])
|
||
root.addWidget(splitter, 1)
|
||
# popped out: the table takes the plot's space (placeholder stays one slim line)
|
||
self.popout = PopOutController(
|
||
self.plot_panel, rl, self.popout_btn, "Variance & Theory - plot", stretch=3,
|
||
placeholder_stretch=0,
|
||
on_change=lambda popped: self.splitter.setSizes([480, 870] if popped else [650, 700]))
|
||
|
||
def close_popout(self):
|
||
self.popout.close()
|
||
|
||
# ------------------------------------------------------------- lifecycle
|
||
def clear(self):
|
||
self.run = self.summary = self.res = None
|
||
self.metric_combo.blockSignals(True)
|
||
self.metric_combo.clear()
|
||
self.metric_combo.blockSignals(False)
|
||
self.report.clear()
|
||
self.table.setRowCount(0)
|
||
self.save_btn.setEnabled(False)
|
||
self.status_label.setText("Run the grid experiment to load the variance analysis here.")
|
||
self.draw()
|
||
|
||
def set_context(self, run, board, summary_df):
|
||
self.run, self.summary = run, summary_df
|
||
top = board.head(10).reset_index(drop=True)
|
||
self.metric_combo.blockSignals(True)
|
||
self.metric_combo.clear()
|
||
for i, row in enumerate(top.itertuples(index=False)):
|
||
tag = " ★ best" if i == 0 else ""
|
||
self.metric_combo.addItem(f"{row.Metrics}{tag}", row.Metrics)
|
||
self.metric_combo.blockSignals(False)
|
||
self.recompute()
|
||
|
||
def on_metric_changed(self, _i):
|
||
if self.run is not None:
|
||
self.recompute()
|
||
|
||
def recompute(self):
|
||
metrics = self.metric_combo.currentData()
|
||
if self.run is None or not metrics:
|
||
return
|
||
QApplication.setOverrideCursor(Qt.WaitCursor)
|
||
err = None
|
||
try:
|
||
self.res = build_theory(self.run, metrics, self.summary)
|
||
except Exception as e:
|
||
self.res, err = None, f"{e}"
|
||
finally:
|
||
QApplication.restoreOverrideCursor()
|
||
if self.res is None:
|
||
self.report.setPlainText(f"Could not compute the variance analysis:\n{err}")
|
||
self.table.setRowCount(0)
|
||
self.save_btn.setEnabled(False)
|
||
self.status_label.setText("Variance analysis failed.")
|
||
else:
|
||
c = self.res["comp"]
|
||
self.report.setMarkdown(self.res["markdown"])
|
||
self.populate_table()
|
||
self.save_btn.setEnabled(True)
|
||
self.status_label.setText(
|
||
f"[{metrics}] | ICC {c['rho']:.2f} | person SD {math.sqrt(c['s2b']):.3g} | "
|
||
f"trial SD {math.sqrt(c['s2w']):.3g} | saturation at about {c['t_sat']:.1f} trials/person")
|
||
self.draw()
|
||
|
||
def save_report(self):
|
||
if self.res is None:
|
||
return
|
||
path, _ = QFileDialog.getSaveFileName(
|
||
self, "Save plain-language report", "variance_theory_report.md", "Markdown (*.md)")
|
||
if not path:
|
||
return
|
||
try:
|
||
Path(path).write_text(self.res["markdown"], encoding="utf-8")
|
||
QMessageBox.information(self, "Saved", f"Report saved to:\n{path}")
|
||
except Exception as e:
|
||
QMessageBox.critical(self, "Save Error", f"Failed to save report:\n{e}")
|
||
|
||
# ----------------------------------------------------------------- table
|
||
def populate_table(self):
|
||
cells = self.res["cells"].copy()
|
||
t = self.table
|
||
t.setSortingEnabled(False)
|
||
t.setRowCount(len(cells))
|
||
if len(cells):
|
||
cells["Diff"] = cells["Obs"] - cells["Pred"]
|
||
for r, rec in enumerate(cells.to_dict("records")):
|
||
for c, (_h, key, kind) in enumerate(self.CELL_COLS):
|
||
v = rec[key]
|
||
if kind == "text":
|
||
blank = v is None or (isinstance(v, float) and np.isnan(v))
|
||
item = QTableWidgetItem("" if blank else str(v))
|
||
item.setFlags(item.flags() & ~Qt.ItemIsEditable)
|
||
elif kind == "int":
|
||
item = NumItem(v, f"{int(v)}")
|
||
elif kind == "pct":
|
||
item = NumItem(v, f"{v * 100:.1f}")
|
||
elif kind == "pts":
|
||
item = NumItem(v, f"{v * 100:+.1f}")
|
||
else:
|
||
item = NumItem(v, f"{v:.2f}")
|
||
t.setItem(r, c, item)
|
||
t.setSortingEnabled(True)
|
||
t.resizeColumnsToContents()
|
||
|
||
# ------------------------------------------------------------------ plots
|
||
def draw(self):
|
||
if not MPL_OK or self.canvas is None:
|
||
return
|
||
fig = self.fig
|
||
fig.clear()
|
||
if self.res is None:
|
||
ax = fig.add_subplot(111)
|
||
ax.axis("off")
|
||
ax.text(0.5, 0.5, "Run the grid experiment to load the variance analysis.",
|
||
ha="center", va="center", fontsize=11)
|
||
else:
|
||
fns = [self._draw_components, self._draw_curves, self._draw_scatter, self._draw_effective,
|
||
self._draw_learning]
|
||
try:
|
||
fns[self.view_combo.currentIndex()](fig)
|
||
fig.tight_layout()
|
||
except Exception as e:
|
||
fig.clear()
|
||
ax = fig.add_subplot(111)
|
||
ax.axis("off")
|
||
ax.text(0.5, 0.5, f"Could not draw this view:\n{e}", ha="center", va="center")
|
||
self.canvas.draw_idle()
|
||
|
||
def _draw_components(self, fig):
|
||
comp, feat = self.res["comp"], self.res["feat"]
|
||
ax1, ax2 = fig.subplots(2, 1, gridspec_kw={"height_ratios": [1, 2]})
|
||
parts = [("Class gap (A vs B)", comp["share_c"], "#2e86c1"),
|
||
("Person-to-person", comp["share_b"], "#d98c1f"),
|
||
("Trial-to-trial", comp["share_w"], "#7f8c8d")]
|
||
left = 0.0
|
||
for label, share, colour in parts:
|
||
ax1.barh([0], [share * 100], left=left, color=colour, label=f"{label} {share * 100:.0f}%")
|
||
left += share * 100
|
||
ax1.set_xlim(0, 100)
|
||
ax1.set_yticks([])
|
||
ax1.set_xlabel("% of total variance of the out-of-sample model score")
|
||
ax1.legend(fontsize=7, ncol=3, loc="lower center", bbox_to_anchor=(0.5, 1.02))
|
||
if len(feat):
|
||
ax2.hist(feat["rho"].dropna(), bins=np.linspace(0, 1, 21), color="#b8c4d0", edgecolor="white")
|
||
ax2.axvline(feat["rho"].median(), color="black", ls=":", label=f"median feature {feat['rho'].median():.2f}")
|
||
ax2.axvline(comp["rho"], color="#c0392b", lw=2, label=f"model score {comp['rho']:.2f}")
|
||
ax2.set_xlim(0, 1)
|
||
ax2.set_xlabel("ICC (0 = a person's trials are interchangeable, 1 = a person is a single fixed value)")
|
||
ax2.set_ylabel("Number of features")
|
||
ax2.legend(fontsize=8)
|
||
|
||
def _draw_curves(self, fig):
|
||
comp, cells = self.res["comp"], self.res["cells"]
|
||
tiers = self.run["tiers"]
|
||
axes = fig.subplots(1, len(tiers), sharey=True, squeeze=False)[0]
|
||
t_test = comp["mean_trials"]
|
||
n_grid = np.arange(1, 21)
|
||
for ax, tier in zip(axes, tiers):
|
||
pred = [predicted_accuracy(comp, n, min(tier.target / (2 * n), t_test), t_test) for n in n_grid]
|
||
ax.plot(n_grid, np.array(pred) * 100, color="black", lw=1.6, label="theory")
|
||
if len(cells):
|
||
sub = cells[cells["Tier"] == tier.target]
|
||
for arm in ("low", "high"):
|
||
s = sub[sub["Arm"] == arm]
|
||
if len(s):
|
||
ax.scatter(s["N_Per_Class"], s["Obs"] * 100, s=45, color=ARM_COLOURS[arm],
|
||
edgecolor="white", zorder=3,
|
||
label=f"observed, {'low-N / high-T' if arm == 'low' else 'high-N / low-T'}")
|
||
ax.set_title(f"~{tier.target} trials", fontsize=10)
|
||
ax.set_xlabel("Training people per class (T = budget / N)")
|
||
axes[0].set_ylabel("Held-out participant accuracy (%)")
|
||
axes[0].legend(fontsize=7, loc="lower right")
|
||
|
||
def _draw_scatter(self, fig):
|
||
cells = self.res["cells"]
|
||
ax = fig.add_subplot(111)
|
||
if cells.empty:
|
||
ax.axis("off")
|
||
ax.text(0.5, 0.5, "No balanced cells to compare.", ha="center", va="center")
|
||
return
|
||
for arm in ("low", "high", None):
|
||
s = cells[cells["Arm"].isna()] if arm is None else cells[cells["Arm"] == arm]
|
||
if len(s):
|
||
ax.scatter(s["Pred"] * 100, s["Obs"] * 100, s=50, color=ARM_COLOURS[arm],
|
||
edgecolor="white", zorder=3,
|
||
label={"low": "low-N / high-T", "high": "high-N / low-T", None: "other"}[arm])
|
||
for r in cells.itertuples(index=False):
|
||
ax.annotate(f"{int(r.N_Total)}x{int(r.T)}", (r.Pred * 100, r.Obs * 100),
|
||
textcoords="offset points", xytext=(4, 4), fontsize=7)
|
||
lo = min(cells["Pred"].min(), cells["Obs"].min()) * 100 - 1
|
||
hi = max(cells["Pred"].max(), cells["Obs"].max()) * 100 + 1
|
||
ax.plot([lo, hi], [lo, hi], color="#999999", ls="--", label="perfect agreement")
|
||
ax.set_xlabel("Predicted accuracy (%) from the variance model")
|
||
ax.set_ylabel("Observed accuracy (%) in your grid")
|
||
ax.set_title("Each point is one (N total x T) cell")
|
||
ax.legend(fontsize=8)
|
||
|
||
def _draw_learning(self, fig):
|
||
ax = fig.add_subplot(111)
|
||
res = learning_curve_table(self.summary) if self.summary is not None else None
|
||
if res is None:
|
||
ax.axis("off")
|
||
ax.text(0.5, 0.5, "This view needs a single-T learning-curve run.\nUse 'Learning-curve preset' "
|
||
"on the setup tab (ideally with a fixed metric set).",
|
||
ha="center", va="center", fontsize=10)
|
||
return
|
||
tbl, t_val = res
|
||
x = tbl["N_Per_Class"].to_numpy()
|
||
ax.errorbar(x, tbl["Mean"] * 100, yerr=tbl["SE"] * 100, fmt="o", color="#2e86c1", capsize=3,
|
||
label="observed (mean ± SE over seeds)")
|
||
xs = np.linspace(x.min(), x.max() * 1.6, 120)
|
||
a, b = learning_curve_fit(tbl)
|
||
if b > 0:
|
||
ax.plot(xs, (a - b / xs) * 100, color="black", lw=1.5,
|
||
label=f"fit {a * 100:.1f}% − {b * 100:.1f}% / N")
|
||
ax.axhline(a * 100, color="#999999", ls=":", label=f"asymptote {a * 100:.1f}%")
|
||
comp = self.res["comp"]
|
||
t_test = comp["mean_trials"]
|
||
ax.plot(xs, [predicted_accuracy(comp, n, min(t_val, t_test), t_test) * 100 for n in xs],
|
||
color="#c0392b", ls="--", label="variance-model prediction")
|
||
ax.set_xlabel("Training people per class")
|
||
ax.set_ylabel("Held-out participant accuracy (%)")
|
||
ax.set_title(f"Learning curve at T = {t_val} trials per person")
|
||
ax.legend(fontsize=8)
|
||
|
||
def _draw_effective(self, fig):
|
||
comp = self.res["comp"]
|
||
rho = comp["rho"]
|
||
ax = fig.add_subplot(111)
|
||
t = np.arange(1, 31)
|
||
ax.plot(t, t, color="#999999", ls=":", label="if trials were independent")
|
||
for r, ls in ((0.1, "--"), (0.3, "--"), (0.6, "--")):
|
||
ax.plot(t, t / design_effect(r, t), ls=ls, lw=1, label=f"ICC {r}")
|
||
ax.plot(t, t / design_effect(rho, t), color="#c0392b", lw=2.5, label=f"this model, ICC {rho:.2f}")
|
||
if np.isfinite(comp["t_sat"]) and comp["t_sat"] < 30:
|
||
ax.axvline(comp["t_sat"], color="#c0392b", ls=":", lw=1)
|
||
ax.text(comp["t_sat"], ax.get_ylim()[1] * 0.02, f" Tsat = {comp['t_sat']:.1f}", color="#c0392b",
|
||
fontsize=8, va="bottom")
|
||
ax.set_xlabel("Trials per person (T)")
|
||
ax.set_ylabel("Effective independent trials from one person")
|
||
ax.set_title("Diminishing returns of extra trials from the same person")
|
||
ax.legend(fontsize=8)
|
||
|
||
|
||
# ==========================================================================
|
||
# GUI
|
||
# ==========================================================================
|
||
class DataSplitterGUI(QWidget):
|
||
"""Assign participants to classes and launch the N vs T grid experiment."""
|
||
|
||
def __init__(self, parent: QWidget | None = None, main_window: QWidget | None = None) -> None:
|
||
super().__init__(parent)
|
||
self.setWindowTitle("fNIRS ML Builder - N vs T Grid Experiment")
|
||
self.resize(1350, 950)
|
||
|
||
self.main_window = main_window
|
||
self.df = None
|
||
self.all_subjects = []
|
||
self.class_assign = {} # subject -> UNASSIGNED / CLASS_NEG / CLASS_POS
|
||
self.chromo_checkboxes = {}
|
||
self.grid_worker = None
|
||
self.last_summary = None # summary DataFrame of the latest run (partial if cancelled)
|
||
self.last_run = None # snapshot of the settings that produced last_summary
|
||
self.last_pivots = (None, None) # (accuracy pivot, total-trials pivot) of the latest run
|
||
self._progress_block = None # log block index of the in-place progress line
|
||
|
||
self.init_ui()
|
||
self.pull_data_from_main_window(silent=True)
|
||
|
||
# ------------------------------------------------------------------ UI
|
||
@staticmethod
|
||
def _spin(lo, hi, value, tip=""):
|
||
box = QSpinBox()
|
||
box.setRange(lo, hi)
|
||
box.setValue(value)
|
||
if tip:
|
||
box.setToolTip(tip)
|
||
return box
|
||
|
||
@staticmethod
|
||
def _group(title):
|
||
"""A titled box with a roomy grid inside."""
|
||
box = QGroupBox(title)
|
||
lay = QGridLayout(box)
|
||
lay.setHorizontalSpacing(8)
|
||
lay.setVerticalSpacing(6)
|
||
return box, lay
|
||
|
||
@staticmethod
|
||
def _hint(text):
|
||
"""Small grey explanatory text."""
|
||
label = QLabel(text)
|
||
label.setWordWrap(True)
|
||
label.setStyleSheet("color: gray; font-size: 11px;")
|
||
return label
|
||
|
||
def _build_settings_panel(self):
|
||
"""The run settings, grouped by what they decide: design, metrics, sampling, checks.
|
||
|
||
Every widget keeps its attribute name; only the layout is organised differently.
|
||
"""
|
||
d = GridSettings()
|
||
|
||
self.spin_n_min = self._spin(1, 100, d.n_min, "Fewest training participants per class.")
|
||
self.spin_n_max = self._spin(1, 100, d.n_max, "Most training participants per class.")
|
||
self.spin_t_min = self._spin(1, 500, d.t_min, "Fewest trials per training participant.")
|
||
self.spin_t_max = self._spin(1, 500, d.t_max, "Most trials per training participant.")
|
||
self.spin_test_k = self._spin(1, 50, d.test_k, "Held-out participants per class (same for every cell).")
|
||
self.spin_seeds = self._spin(1, 1_000_000, d.n_seeds, "Number of random train/test splits (Monte Carlo seeds).")
|
||
self.spin_seed_start = self._spin(0, 1_000_000, d.seed_start, "First seed value (change to draw a fresh set of splits).")
|
||
self.spin_workers = self._spin(1, max(64, DEFAULT_WORKERS), d.workers, "Worker processes.")
|
||
|
||
self.spin_tol = QDoubleSpinBox()
|
||
self.spin_tol.setRange(0.5, 50.0)
|
||
self.spin_tol.setSingleStep(0.5)
|
||
self.spin_tol.setDecimals(1)
|
||
self.spin_tol.setSuffix(" %")
|
||
self.spin_tol.setValue(d.budget_tolerance * 100)
|
||
self.spin_tol.setToolTip(
|
||
"How close N x T must be to each target budget (~72, ~96, ~144 total trials).\n"
|
||
"Smaller = tighter matching between the low-N and high-N arms.")
|
||
|
||
self.spin_combo = self._spin(
|
||
0, len(METRIC_REGISTRY), d.top_k_expand,
|
||
"After all singles and pairs, take this many best single metrics and test every\n"
|
||
"subset of them with 3 or more metrics. 5 adds 16 sets; 0 disables.")
|
||
self.chk_unbalanced = QCheckBox("Also run unbalanced sets (all remaining Class B)")
|
||
self.chk_unbalanced.setChecked(d.include_unbalanced)
|
||
self.chk_unbalanced.setToolTip("The pivots and iso-budget report always use balanced runs only.")
|
||
self.chk_budget_only = QCheckBox("Budget-only: run just the (N, T) cells of the three budget groups")
|
||
self.chk_budget_only.setChecked(d.budget_only)
|
||
self.chk_budget_only.setToolTip(
|
||
"Checked: skip every (N, T) cell that is not part of a budget group.\n"
|
||
"Unchecked: run the full N x T table (including cells never used in the report).")
|
||
|
||
# Two-stage design: stage 1 sweeps metric sets (exploratory), stage 2 re-runs the grid
|
||
# with ONE pre-specified set so nothing about the held-out people picks the features.
|
||
self.chk_fixed = QCheckBox("Stage 2: use ONLY this metric set (no sweep)")
|
||
self.chk_fixed.setToolTip(
|
||
"Evaluate ONLY the metric set typed here (comma-separated, e.g. RMS or RMS, TTP).\n"
|
||
"No selection happens on the held-out participants, so accuracies are free of\n"
|
||
"selection bias, and the run is ~70x cheaper, so many more seeds are practical.\n"
|
||
f"Valid names: {', '.join(METRIC_REGISTRY)}")
|
||
self.edit_fixed = QLineEdit("RMS")
|
||
self.edit_fixed.setEnabled(False)
|
||
self.btn_use_best = QPushButton("Use best from last run")
|
||
self.btn_use_best.setToolTip("Fill in the top-ranked metric set of the latest sweep.")
|
||
self.btn_use_best.setEnabled(False)
|
||
self.btn_use_best.clicked.connect(self.use_best_as_fixed)
|
||
|
||
self.chk_null = QCheckBox("Null run: shuffle class labels (accuracy should fall to ~50%)")
|
||
self.chk_null.setToolTip(
|
||
"Each seed gives the Class A / Class B labels to randomly chosen people, then runs the\n"
|
||
"whole pipeline as usual. Real signal disappears, so any accuracy above 50% is bias in the\n"
|
||
"pipeline itself (e.g. picking the best of many metric sets). Use the SAME settings as the\n"
|
||
"real run; the log then compares the two. Tabs and exports keep showing your last real run.")
|
||
self.btn_curve = QPushButton("Learning-curve preset")
|
||
self.btn_curve.setToolTip(
|
||
"Sets one trials-per-person value (5), turns budget-only OFF and spans training people\n"
|
||
"from 2 per class up to the most your class sizes allow, to show where accuracy levels off.\n"
|
||
"Best combined with 'Stage 2: use ONLY this metric set'.")
|
||
self.btn_curve.clicked.connect(self.apply_learning_curve_preset)
|
||
|
||
self.estimate_label = QLabel("")
|
||
self.estimate_label.setWordWrap(True)
|
||
self.estimate_label.setStyleSheet("padding: 6px; border: 1px solid gray; border-radius: 4px;")
|
||
|
||
panel = QWidget()
|
||
col = QVBoxLayout(panel)
|
||
col.setContentsMargins(0, 0, 0, 0)
|
||
|
||
# --- Design ---
|
||
design, g = self._group("3. Design: what is compared")
|
||
g.addWidget(QLabel("Training people per class"), 0, 0)
|
||
g.addWidget(self.spin_n_min, 0, 1)
|
||
g.addWidget(QLabel("to"), 0, 2)
|
||
g.addWidget(self.spin_n_max, 0, 3)
|
||
g.addWidget(QLabel("Trials per person (T)"), 1, 0)
|
||
g.addWidget(self.spin_t_min, 1, 1)
|
||
g.addWidget(QLabel("to"), 1, 2)
|
||
g.addWidget(self.spin_t_max, 1, 3)
|
||
g.addWidget(QLabel("Held-out people per class"), 2, 0)
|
||
g.addWidget(self.spin_test_k, 2, 1)
|
||
g.addWidget(self.chk_budget_only, 3, 0, 1, 5)
|
||
g.addWidget(QLabel("Budget tolerance (±)"), 4, 0)
|
||
g.addWidget(self.spin_tol, 4, 1)
|
||
g.addWidget(self.chk_unbalanced, 5, 0, 1, 5)
|
||
g.addWidget(self.btn_curve, 6, 0, 1, 5)
|
||
g.setColumnStretch(4, 1)
|
||
col.addWidget(design)
|
||
|
||
# --- Metrics ---
|
||
metrics, m = self._group("4. Metrics: what the classifier sees")
|
||
m.addWidget(QLabel("<b>Stage 1: sweep</b> (exploratory)"), 0, 0, 1, 3)
|
||
m.addWidget(QLabel("Expand the top-K single metrics (K)"), 1, 0, 1, 2)
|
||
m.addWidget(self.spin_combo, 1, 2)
|
||
m.addWidget(QLabel("<b>Stage 2: fixed metric set</b> (confirmatory)"), 2, 0, 1, 3)
|
||
m.addWidget(self.chk_fixed, 3, 0, 1, 3)
|
||
m.addWidget(self.edit_fixed, 4, 0, 1, 2)
|
||
m.addWidget(self.btn_use_best, 4, 2)
|
||
m.addWidget(self._hint("Stage 1 tries every single metric, every pair and the subsets of the top K, "
|
||
"and keeps the best per split, so its accuracies are optimistic. Stage 2 "
|
||
"evaluates only the set you name."), 5, 0, 1, 3)
|
||
m.setColumnStretch(1, 1)
|
||
col.addWidget(metrics)
|
||
|
||
# --- Sampling and compute ---
|
||
sampling, s = self._group("5. Sampling and compute")
|
||
s.addWidget(QLabel("Seeds (random splits)"), 0, 0)
|
||
s.addWidget(self.spin_seeds, 0, 1)
|
||
s.addWidget(QLabel("First seed"), 1, 0)
|
||
s.addWidget(self.spin_seed_start, 1, 1)
|
||
s.addWidget(QLabel("Worker processes"), 2, 0)
|
||
s.addWidget(self.spin_workers, 2, 1)
|
||
s.setColumnStretch(2, 1)
|
||
col.addWidget(sampling)
|
||
|
||
# --- Checks ---
|
||
checks, c = self._group("6. Sanity check")
|
||
c.addWidget(self.chk_null, 0, 0)
|
||
col.addWidget(checks)
|
||
col.addStretch()
|
||
|
||
for w in (self.spin_n_min, self.spin_n_max, self.spin_t_min, self.spin_t_max,
|
||
self.spin_test_k, self.spin_seeds, self.spin_seed_start, self.spin_combo):
|
||
w.valueChanged.connect(self.update_estimate)
|
||
self.spin_tol.valueChanged.connect(self.update_estimate)
|
||
for chk in (self.chk_unbalanced, self.chk_budget_only, self.chk_fixed):
|
||
chk.toggled.connect(self.update_estimate)
|
||
self.chk_fixed.toggled.connect(self.edit_fixed.setEnabled)
|
||
self.chk_fixed.toggled.connect(lambda on: self.spin_combo.setEnabled(not on))
|
||
self.edit_fixed.textChanged.connect(self.update_estimate)
|
||
return panel
|
||
|
||
def init_ui(self):
|
||
main_layout = QVBoxLayout(self)
|
||
self.tabs = QTabWidget()
|
||
main_layout.addWidget(self.tabs)
|
||
|
||
# ---- Tab 1: setup ----
|
||
def button_css(color):
|
||
# A plain background-color stylesheet hides Qt's disabled look, so state it explicitly
|
||
return (f"QPushButton {{ font-weight: bold; color: white; padding: 10px 18px; "
|
||
f"font-size: 13px; background-color: {color}; }} "
|
||
f"QPushButton:disabled {{ background-color: #8c8c8c; color: #d8d8d8; }}")
|
||
|
||
self.export_btn = QPushButton("Export Best Model (.joblib)")
|
||
self.export_btn.setStyleSheet(button_css("#3a6ea5"))
|
||
self.export_btn.setEnabled(False)
|
||
self.export_btn.clicked.connect(self.export_best_model)
|
||
|
||
self.export_xlsx_btn = QPushButton("Export Results (.xlsx)")
|
||
self.export_xlsx_btn.setStyleSheet(button_css("#7a5a9e"))
|
||
self.export_xlsx_btn.setEnabled(False)
|
||
self.export_xlsx_btn.clicked.connect(self.export_results_excel)
|
||
|
||
self.cancel_btn = QPushButton("Cancel")
|
||
self.cancel_btn.setStyleSheet(button_css("#a33a3a"))
|
||
self.cancel_btn.setEnabled(False)
|
||
self.cancel_btn.clicked.connect(self.cancel_grid_experiment)
|
||
|
||
self.grid_exp_btn = QPushButton("Run N vs T Grid Experiment")
|
||
self.grid_exp_btn.setStyleSheet(button_css("#4a7c59"))
|
||
self.grid_exp_btn.clicked.connect(self.start_grid_experiment)
|
||
|
||
tab_setup = QWidget()
|
||
setup = QVBoxLayout(tab_setup)
|
||
|
||
setup.addWidget(self._hint(
|
||
"Workflow: load data and assign classes (left) → run a Stage 1 sweep to find the best "
|
||
"metric set → press 'Use best from last run' and run Stage 2 with many seeds → use the "
|
||
"learning-curve preset and a null run as checks. Results and exports are on the other tabs."))
|
||
|
||
# ---- Row 1: data, condition, chromophores ----
|
||
top = QGroupBox("1. Data and features")
|
||
top_l = QGridLayout(top)
|
||
self.data_info_label = QLabel("No data loaded.")
|
||
self.data_info_label.setWordWrap(True)
|
||
top_l.addWidget(self.data_info_label, 0, 0, 1, 3)
|
||
reload_btn = QPushButton("Reload from App Data")
|
||
reload_btn.clicked.connect(lambda: self.pull_data_from_main_window(silent=False))
|
||
top_l.addWidget(reload_btn, 0, 3)
|
||
top_l.addWidget(QLabel("Condition:"), 1, 0)
|
||
self.condition_combo = QComboBox()
|
||
top_l.addWidget(self.condition_combo, 1, 1)
|
||
top_l.addWidget(QLabel("Chromophores:"), 1, 2, Qt.AlignRight)
|
||
self.chromo_container_layout = QHBoxLayout()
|
||
top_l.addLayout(self.chromo_container_layout, 1, 3)
|
||
top_l.setColumnStretch(0, 1)
|
||
setup.addWidget(top)
|
||
|
||
# ---- Row 2: participants (left) | run settings (right) ----
|
||
split = QSplitter(Qt.Horizontal)
|
||
|
||
people = QGroupBox("2. Participants and classes")
|
||
people_l = QVBoxLayout(people)
|
||
assign = QGridLayout()
|
||
assign.addWidget(QLabel("Class A if path contains:"), 0, 0)
|
||
self.class_a_edit = QLineEdit("CVI")
|
||
assign.addWidget(self.class_a_edit, 0, 1)
|
||
assign.addWidget(QLabel("Class B if path contains:"), 0, 2)
|
||
self.class_b_edit = QLineEdit("Control")
|
||
assign.addWidget(self.class_b_edit, 0, 3)
|
||
assign_btn = QPushButton("Assign Classes from Path")
|
||
assign_btn.clicked.connect(self.assign_classes_from_path)
|
||
assign.addWidget(assign_btn, 0, 4)
|
||
people_l.addLayout(assign)
|
||
self.assign_summary_label = QLabel("")
|
||
people_l.addWidget(self.assign_summary_label)
|
||
self.table = QTableWidget()
|
||
self.table.setColumnCount(2)
|
||
self.table.setHorizontalHeaderLabels(["Subject Absolute File Path", "True Class"])
|
||
header = self.table.horizontalHeader()
|
||
header.setSectionResizeMode(0, QHeaderView.Stretch)
|
||
header.setSectionResizeMode(1, QHeaderView.Interactive)
|
||
self.table.setColumnWidth(1, 150)
|
||
people_l.addWidget(self.table, 1)
|
||
split.addWidget(people)
|
||
|
||
settings_scroll = QScrollArea()
|
||
settings_scroll.setWidgetResizable(True)
|
||
settings_scroll.setFrameShape(QFrame.NoFrame)
|
||
settings_scroll.setWidget(self._build_settings_panel())
|
||
|
||
run_col = QWidget()
|
||
run_l = QVBoxLayout(run_col)
|
||
run_l.setContentsMargins(0, 0, 0, 0)
|
||
run_l.addWidget(settings_scroll, 1)
|
||
run_l.addWidget(self.estimate_label)
|
||
run_l.addWidget(self.grid_exp_btn)
|
||
run_col.setMinimumWidth(470)
|
||
split.addWidget(run_col)
|
||
split.setStretchFactor(0, 3)
|
||
split.setStretchFactor(1, 2)
|
||
split.setSizes([760, 560])
|
||
setup.addWidget(split, 1)
|
||
|
||
# ---- Tab 2: logs ----
|
||
tab_logs = QWidget()
|
||
logs_l = QVBoxLayout(tab_logs)
|
||
logs_l.addWidget(QLabel("<b>Real-time sweep logs:</b>"))
|
||
self.log_text = QTextEdit()
|
||
self.log_text.setReadOnly(True)
|
||
self.log_text.setStyleSheet(
|
||
"background-color: #1e1e1e; color: #00ff00; "
|
||
"font-family: Consolas, monospace; font-size: 12px;")
|
||
logs_l.addWidget(self.log_text)
|
||
|
||
# Run controls live next to the log so cancelling/exporting never needs a tab switch
|
||
log_btn_row = QHBoxLayout()
|
||
log_btn_row.addStretch()
|
||
log_btn_row.addWidget(self.export_xlsx_btn)
|
||
log_btn_row.addWidget(self.export_btn)
|
||
log_btn_row.addWidget(self.cancel_btn)
|
||
logs_l.addLayout(log_btn_row)
|
||
|
||
self.tabs.addTab(tab_setup, "Experiment Setup")
|
||
self.tabs.addTab(tab_logs, "Processing Logs")
|
||
|
||
self.explain_tab = ModelExplainWidget()
|
||
self.tabs.addTab(self.explain_tab, "Model Explanation")
|
||
|
||
self.theory_tab = VarianceTheoryWidget()
|
||
self.tabs.addTab(self.theory_tab, "Variance & Theory")
|
||
|
||
self.update_estimate()
|
||
|
||
def closeEvent(self, event):
|
||
"""Don't leave worker processes running if the window is closed mid-run."""
|
||
if self.grid_worker is not None and self.grid_worker.isRunning():
|
||
self.grid_worker.cancel()
|
||
self.grid_worker.wait(10000)
|
||
self.explain_tab.shutdown()
|
||
self.explain_tab.close_popout()
|
||
self.theory_tab.close_popout()
|
||
super().closeEvent(event)
|
||
|
||
# ------------------------------------------------------------ settings
|
||
def read_settings(self):
|
||
"""GridSettings from the widgets, or None (after warning) if they are inconsistent."""
|
||
if self.spin_n_min.value() > self.spin_n_max.value():
|
||
QMessageBox.warning(self, "Configuration Error",
|
||
"Minimum training participants per class exceeds the maximum.")
|
||
return None
|
||
if self.spin_t_min.value() > self.spin_t_max.value():
|
||
QMessageBox.warning(self, "Configuration Error",
|
||
"Minimum trials per participant exceeds the maximum.")
|
||
return None
|
||
if self.chk_fixed.isChecked():
|
||
fixed, bad = parse_metric_list(self.edit_fixed.text())
|
||
if bad or not fixed:
|
||
QMessageBox.warning(
|
||
self, "Configuration Error",
|
||
(f"Unknown metric name(s): {', '.join(bad)}.\n" if bad else "No metric given.\n")
|
||
+ f"Valid names: {', '.join(METRIC_REGISTRY)}")
|
||
return None
|
||
return self._settings_unchecked()
|
||
|
||
def _settings_unchecked(self):
|
||
return GridSettings(
|
||
n_min=self.spin_n_min.value(), n_max=self.spin_n_max.value(),
|
||
t_min=self.spin_t_min.value(), t_max=self.spin_t_max.value(),
|
||
test_k=self.spin_test_k.value(),
|
||
n_seeds=self.spin_seeds.value(), seed_start=self.spin_seed_start.value(),
|
||
include_unbalanced=self.chk_unbalanced.isChecked(),
|
||
budget_only=self.chk_budget_only.isChecked(),
|
||
budget_tolerance=self.spin_tol.value() / 100.0,
|
||
top_k_expand=self.spin_combo.value(),
|
||
fixed_metrics=(parse_metric_list(self.edit_fixed.text())[0]
|
||
if self.chk_fixed.isChecked() else ()),
|
||
shuffle_labels=self.chk_null.isChecked(),
|
||
workers=self.spin_workers.value(),
|
||
)
|
||
|
||
def use_best_as_fixed(self):
|
||
"""Stage 1 -> stage 2: copy the best metric set of the latest sweep into the fixed field."""
|
||
if self.last_summary is None:
|
||
QMessageBox.warning(self, "No results", "Run a sweep first.")
|
||
return
|
||
best = rank_metric_sets(self.last_summary).iloc[0]["Metrics"]
|
||
self.edit_fixed.setText(str(best))
|
||
self.chk_fixed.setChecked(True)
|
||
|
||
def apply_learning_curve_preset(self):
|
||
"""Settings for a single-T learning curve: N from 1 up to the most the class sizes allow."""
|
||
values = list(self.class_assign.values())
|
||
n_a, n_b = values.count(CLASS_NEG), values.count(CLASS_POS)
|
||
max_n = min(n_a, n_b) - self.spin_test_k.value()
|
||
if max_n < 4:
|
||
QMessageBox.warning(self, "Learning curve",
|
||
"Assign participants to both classes first (each class needs at least "
|
||
"4 more people than the held-out count).")
|
||
return
|
||
self.chk_budget_only.setChecked(False)
|
||
self.chk_unbalanced.setChecked(False)
|
||
self.spin_n_min.setValue(2)
|
||
self.spin_n_max.setValue(max_n)
|
||
self.spin_t_min.setValue(5)
|
||
self.spin_t_max.setValue(5)
|
||
self.append_log(f"Learning-curve preset: 2 to {max_n} training people per class at T = 5, "
|
||
f"budget-only off. Tick 'Stage 2' for an unbiased curve.")
|
||
|
||
def update_estimate(self, *_):
|
||
"""Live size of the planned run (configurations x metric sets per configuration)."""
|
||
if not hasattr(self, "estimate_label"):
|
||
return
|
||
values = list(self.class_assign.values())
|
||
n_a, n_b = values.count(CLASS_NEG), values.count(CLASS_POS)
|
||
if n_a == 0 or n_b == 0:
|
||
self.estimate_label.setText("Assign participants to Class A and Class B to see the run-size estimate.")
|
||
return
|
||
s = self._settings_unchecked()
|
||
if s.n_min > s.n_max or s.t_min > s.t_max:
|
||
self.estimate_label.setText("<b>Min exceeds max in a range.</b>")
|
||
return
|
||
configs, sets_each = estimate_workload(s, n_a, n_b)
|
||
if configs == 0:
|
||
self.estimate_label.setText(
|
||
"<b>No configurations match these settings</b> (check class sizes vs. held-out count, "
|
||
"N/T ranges and budget-only mode).")
|
||
return
|
||
mode = "budget cells only" if s.budget_only else "full N x T table"
|
||
warn = ""
|
||
if self.chk_fixed.isChecked() and (not s.fixed_metrics
|
||
or parse_metric_list(self.edit_fixed.text())[1]):
|
||
warn = " <b>(Fixed metric field is empty or has unknown names.)</b>"
|
||
self.estimate_label.setText(
|
||
f"<b>Planned run:</b> {configs:,} configurations ({mode}) × {sets_each} metric sets each "
|
||
f"≈ {configs * sets_each:,} LDA fits.{warn}")
|
||
|
||
# ------------------------------------------------------------- data in
|
||
def pull_data_from_main_window(self, silent=False):
|
||
def warn(msg):
|
||
self.data_info_label.setText(msg)
|
||
if not silent:
|
||
QMessageBox.warning(self, "Warning", msg)
|
||
|
||
if not self.main_window or not hasattr(self.main_window, "fir_feature_dict"):
|
||
warn("No main window / fir_feature_dict available.")
|
||
return
|
||
fir_dict = self.main_window.fir_feature_dict
|
||
if not isinstance(fir_dict, dict) or not fir_dict:
|
||
warn("main_window.fir_feature_dict is currently empty.")
|
||
return
|
||
df, kind, info = parse_fir_dict(fir_dict)
|
||
if df.empty:
|
||
warn("Could not extract valid features from fir_feature_dict.")
|
||
return
|
||
if info["dropped"] and not silent:
|
||
QMessageBox.warning(
|
||
self, "Mixed data",
|
||
f"{len(info['dropped'])} file(s) have no trial-specific features (Step 21.5 did not run "
|
||
f"for them) and were excluded from this trial-wise dataset.")
|
||
self.load_from_dataframe(df, kind)
|
||
|
||
def load_from_dataframe(self, df: pd.DataFrame, kind: str):
|
||
try:
|
||
self.df = df
|
||
self.all_subjects = sorted(df["Participant"].astype(str).unique().tolist())
|
||
for s in self.all_subjects:
|
||
self.class_assign.setdefault(s, UNASSIGNED)
|
||
|
||
self.condition_combo.clear()
|
||
self.condition_combo.addItems(sorted(df["Condition"].unique().tolist()))
|
||
|
||
while self.chromo_container_layout.count():
|
||
item = self.chromo_container_layout.takeAt(0)
|
||
if item.widget():
|
||
item.widget().deleteLater()
|
||
self.chromo_checkboxes = {}
|
||
detected = self.detect_chromophores()
|
||
for ch in detected:
|
||
chk = QCheckBox(ch)
|
||
chk.setChecked(True)
|
||
self.chromo_checkboxes[ch] = chk
|
||
self.chromo_container_layout.addWidget(chk)
|
||
|
||
if kind == KIND_TRIAL:
|
||
counts = df[df["Kind"] == KIND_TRIAL].groupby(["Participant", "Condition"]).size()
|
||
kind_txt = (f"trial-wise features (median {int(counts.median())} trials per "
|
||
f"participant and condition)")
|
||
else:
|
||
kind_txt = "whole-participant features (no trial-specific features; Step 21.5 not run)"
|
||
chromo_txt = ("" if detected else
|
||
" | <b>No chromophore tags (hbo/hbr/hbt/...) found in the feature names</b>")
|
||
self.data_info_label.setText(
|
||
f"<b>Loaded {len(self.all_subjects)} participants</b> | {kind_txt}{chromo_txt}")
|
||
|
||
self.populate_table()
|
||
except Exception as e:
|
||
QMessageBox.critical(self, "Error", f"Failed to process pulled data from main window:\n{e}")
|
||
|
||
def feature_columns(self):
|
||
return [c for c in self.df.columns if c not in META_COLS]
|
||
|
||
def detect_chromophores(self):
|
||
"""Chromophores actually present in the feature names (nothing is invented)."""
|
||
return detect_chromophores_in(self.feature_columns()) if self.df is not None else []
|
||
|
||
# ------------------------------------------------------ class assignment
|
||
def assigned_classes(self):
|
||
return {s: c for s, c in self.class_assign.items() if c in (CLASS_NEG, CLASS_POS)}
|
||
|
||
def selected_chromophores(self):
|
||
return [ch for ch, chk in self.chromo_checkboxes.items() if chk.isChecked()]
|
||
|
||
def assign_classes_from_path(self):
|
||
a_kw = self.class_a_edit.text().strip().lower()
|
||
b_kw = self.class_b_edit.text().strip().lower()
|
||
n_a = n_b = n_u = 0
|
||
for sub in self.all_subjects:
|
||
low = sub.lower()
|
||
in_a = bool(a_kw) and a_kw in low
|
||
in_b = bool(b_kw) and b_kw in low
|
||
if in_a and not in_b:
|
||
self.class_assign[sub] = CLASS_NEG
|
||
n_a += 1
|
||
elif in_b and not in_a:
|
||
self.class_assign[sub] = CLASS_POS
|
||
n_b += 1
|
||
else:
|
||
self.class_assign[sub] = UNASSIGNED
|
||
n_u += 1
|
||
self.append_log(f"Assigned classes from path: Class A={n_a}, Class B={n_b}, unassigned={n_u}")
|
||
self.populate_table()
|
||
|
||
def refresh_summary(self):
|
||
values = [self.class_assign.get(s, UNASSIGNED) for s in self.all_subjects]
|
||
self.assign_summary_label.setText(
|
||
f"Class A={values.count(CLASS_NEG)} | Class B={values.count(CLASS_POS)} | "
|
||
f"Unassigned={values.count(UNASSIGNED)}")
|
||
self.update_estimate()
|
||
|
||
def populate_table(self):
|
||
if self.df is None or self.df.empty:
|
||
return
|
||
|
||
self.table.setRowCount(len(self.all_subjects))
|
||
for row, sub in enumerate(self.all_subjects):
|
||
item = QTableWidgetItem(sub)
|
||
item.setFlags(item.flags() & ~Qt.ItemIsEditable)
|
||
self.table.setItem(row, 0, item)
|
||
|
||
combo = QComboBox()
|
||
combo.addItems([UNASSIGNED, CLASS_NEG, CLASS_POS])
|
||
combo.setCurrentText(self.class_assign.get(sub, UNASSIGNED))
|
||
combo.currentTextChanged.connect(
|
||
lambda value, subject=sub: self.on_class_changed(subject, value))
|
||
self.table.setCellWidget(row, 1, combo)
|
||
|
||
self.refresh_summary()
|
||
|
||
def on_class_changed(self, sub, text):
|
||
self.class_assign[sub] = text
|
||
self.refresh_summary()
|
||
|
||
# ---------------------------------------------------------------- run
|
||
def start_grid_experiment(self):
|
||
if self.df is None:
|
||
QMessageBox.warning(self, "No data", "No data loaded.")
|
||
return
|
||
|
||
assigned = self.assigned_classes()
|
||
n_a = sum(c == CLASS_NEG for c in assigned.values())
|
||
n_b = sum(c == CLASS_POS for c in assigned.values())
|
||
if n_a < MIN_PER_CLASS or n_b < MIN_PER_CLASS:
|
||
QMessageBox.warning(
|
||
self, "Configuration Error",
|
||
f"Please assign at least {MIN_PER_CLASS} participants to Class A and "
|
||
f"{MIN_PER_CLASS} to Class B.")
|
||
return
|
||
|
||
settings = self.read_settings()
|
||
if settings is None:
|
||
return
|
||
if min(n_a, n_b) < settings.test_k + settings.n_min:
|
||
QMessageBox.warning(
|
||
self, "Configuration Error",
|
||
f"With {settings.test_k} held-out per class and at least {settings.n_min} training "
|
||
f"participants per class, each class needs {settings.test_k + settings.n_min} participants "
|
||
f"(Class A={n_a}, Class B={n_b}).")
|
||
return
|
||
configs, sets_each = estimate_workload(settings, n_a, n_b)
|
||
if configs == 0:
|
||
QMessageBox.warning(
|
||
self, "Configuration Error",
|
||
"No (N, T) configurations match these settings. Widen the N/T ranges, loosen the budget "
|
||
"tolerance, or turn off budget-only mode.")
|
||
return
|
||
|
||
chromos = self.selected_chromophores()
|
||
if not chromos:
|
||
QMessageBox.warning(self, "Configuration Error", "Please select at least one chromophore.")
|
||
return
|
||
|
||
feature_cols = self.feature_columns()
|
||
per_chromo, total_cols, empty_metrics = describe_feature_columns(feature_cols, chromos)
|
||
no_cols = [ch for ch, n in per_chromo.items() if n == 0]
|
||
if no_cols:
|
||
QMessageBox.warning(
|
||
self, "Configuration Error",
|
||
f"No feature columns match chromophore(s): {', '.join(no_cols)}.\n"
|
||
f"Uncheck them, or check the feature names (e.g. {feature_cols[:2]}).")
|
||
return
|
||
if settings.fixed_metrics and not select_columns(feature_cols, settings.fixed_metrics, chromos):
|
||
QMessageBox.warning(
|
||
self, "Configuration Error",
|
||
f"The fixed metric set [{', '.join(settings.fixed_metrics)}] matches no feature columns "
|
||
f"for the selected chromophores.")
|
||
return
|
||
|
||
condition = self.condition_combo.currentText()
|
||
run_info = {
|
||
"df": self.df,
|
||
"classes": assigned,
|
||
"condition": condition,
|
||
"chromos": chromos,
|
||
"feature_cols": feature_cols,
|
||
"settings": settings,
|
||
"tiers": settings.tiers,
|
||
"class_hints": {CLASS_NEG: self.class_a_edit.text().strip(),
|
||
CLASS_POS: self.class_b_edit.text().strip()},
|
||
}
|
||
if not settings.shuffle_labels: # a null run never replaces the real results or tabs
|
||
self.last_summary = None
|
||
self.last_pivots = (None, None)
|
||
self.explain_tab.clear()
|
||
self.theory_tab.clear()
|
||
self.last_run = run_info
|
||
|
||
self.tabs.setCurrentIndex(1)
|
||
self.log_text.clear()
|
||
self._progress_block = None
|
||
self.append_log(f"Condition: {condition} | Chromophores used: {', '.join(chromos)}")
|
||
self.append_log("Feature columns per chromophore (all metrics): "
|
||
+ ", ".join(f"{ch}={n}" for ch, n in per_chromo.items())
|
||
+ f" | total used together: {total_cols}")
|
||
if empty_metrics:
|
||
self.append_log(f"Metrics with no matching columns (skipped): {', '.join(empty_metrics)}")
|
||
self.append_log("Selected chromophores are pooled into one feature set; "
|
||
"the sweep varies metrics, not chromophores.")
|
||
self.append_log(
|
||
f"Grid: train/class {settings.n_min}-{settings.n_max} | trials {settings.t_min}-{settings.t_max} | "
|
||
f"held-out {settings.test_k}/class | seeds {settings.seed_start}..."
|
||
f"{settings.seed_start + settings.n_seeds - 1} | "
|
||
f"{'budget cells only' if settings.budget_only else 'full N x T table'} "
|
||
f"(tolerance ±{settings.budget_tolerance * 100:g}%) | "
|
||
f"{'balanced + unbalanced' if settings.include_unbalanced else 'balanced only'}")
|
||
if settings.fixed_metrics:
|
||
fixed_cols = select_columns(feature_cols, settings.fixed_metrics, chromos)
|
||
self.append_log(
|
||
f"STAGE 2: metric set fixed in advance = [{', '.join(settings.fixed_metrics)}] "
|
||
f"({len(fixed_cols)} features). No selection on held-out participants | "
|
||
f"{configs:,} configurations planned")
|
||
else:
|
||
self.append_log(
|
||
f"STAGE 1 (exploratory sweep): all singles + all pairs + every 3+ subset of the top "
|
||
f"{settings.top_k_expand} singles = {sets_each} sets per configuration | "
|
||
f"{configs:,} configurations planned")
|
||
|
||
if settings.shuffle_labels:
|
||
self.append_log("NULL RUN: class labels are shuffled across participants in every seed. "
|
||
"Your last real results stay in the tabs and exports.")
|
||
self.set_running(True)
|
||
self.grid_worker = DiversityDensityGridWorker(
|
||
self.df, assigned, feature_cols, chromos, condition, settings)
|
||
self.grid_worker.log_signal.connect(self.append_log)
|
||
self.grid_worker.progress_log_signal.connect(self.update_progress_log)
|
||
self.grid_worker.finished_signal.connect(self.on_grid_finished)
|
||
self.grid_worker.start()
|
||
|
||
def set_running(self, running):
|
||
self.grid_exp_btn.setEnabled(not running)
|
||
self.cancel_btn.setEnabled(running)
|
||
if running:
|
||
self.export_btn.setEnabled(False)
|
||
self.export_xlsx_btn.setEnabled(False)
|
||
|
||
def cancel_grid_experiment(self):
|
||
if self.grid_worker is not None and self.grid_worker.isRunning():
|
||
self.cancel_btn.setEnabled(False)
|
||
self.append_log("\nCancel requested - stopping all workers...")
|
||
self.grid_worker.cancel()
|
||
|
||
def _finish_null_run(self, summary_df, cancelled):
|
||
"""Report a label-shuffle run and compare it with the latest real run."""
|
||
rule = "=" * 65
|
||
self.append_log(f"\n{rule}")
|
||
self.append_log(f"=== {'PARTIAL ' if cancelled else ''}NULL RUN (SHUFFLED LABELS) ===")
|
||
self.append_log(rule)
|
||
real_settings = self.last_run["settings"] if self.last_run else None
|
||
try:
|
||
for line in null_report_lines(summary_df, self.grid_worker.settings,
|
||
self.last_summary, real_settings):
|
||
self.append_log(line)
|
||
except Exception as e:
|
||
self.append_log(f"Note: could not complete the null report: {e}")
|
||
self.append_log(f"{rule}\nThe tabs and exports still show your last real run.\n")
|
||
|
||
def on_grid_finished(self, summary_df):
|
||
self.set_running(False)
|
||
cancelled = bool(self.grid_worker and self.grid_worker.was_cancelled)
|
||
# Exports belong to the last REAL run; a null run must not leave them greyed out
|
||
self.export_btn.setEnabled(self.last_summary is not None)
|
||
self.export_xlsx_btn.setEnabled(self.last_summary is not None)
|
||
|
||
if summary_df.empty:
|
||
msg = ("The run was cancelled before any configuration finished."
|
||
if cancelled else "The grid sweep completed, but no results were generated.")
|
||
QMessageBox.warning(self, "Experiment Finished", msg)
|
||
return
|
||
|
||
if self.grid_worker is not None and self.grid_worker.settings.shuffle_labels:
|
||
self._finish_null_run(summary_df, cancelled)
|
||
return
|
||
|
||
tiers = self.last_run["tiers"] if self.last_run else DEFAULT_TIERS
|
||
summary_df["Total_Trials"] = summary_df["N_Train_Subjects"] * summary_df["T_Trials_Per_Subject"]
|
||
self.last_summary = summary_df
|
||
self.export_btn.setEnabled(True)
|
||
self.export_xlsx_btn.setEnabled(True)
|
||
self.btn_use_best.setEnabled(True)
|
||
|
||
title = "PARTIAL RESULTS (CANCELLED) - " if cancelled else ""
|
||
rule = "=" * 65
|
||
self.append_log(f"\n{rule}")
|
||
self.append_log(f"=== {title}ISO-BUDGET STRATEGY & STATISTICAL REPORT (MATCHED TRIAL VOLUMES) ===")
|
||
self.append_log(rule)
|
||
|
||
pivot_acc = pivot_trials = None
|
||
try:
|
||
for line in leaderboard_lines(rank_metric_sets(summary_df)):
|
||
self.append_log(line)
|
||
classes = self.last_run["classes"] if self.last_run else {}
|
||
for line in problem_participant_lines(problem_participants(summary_df, classes)):
|
||
self.append_log(line)
|
||
pivot_acc, pivot_trials = build_pivots(summary_df)
|
||
self.last_pivots = (pivot_acc, pivot_trials)
|
||
self.append_log("\n[Balanced Runs Only - Mean Held-Out Participant Balanced Accuracy]:")
|
||
self.append_log(str(pivot_acc.round(4)))
|
||
fixed_run = bool(self.last_run and self.last_run["settings"].fixed_metrics)
|
||
for line in iso_budget_report(summary_df, tiers, fixed=fixed_run):
|
||
self.append_log(line)
|
||
for line in learning_curve_lines(summary_df):
|
||
self.append_log(line)
|
||
self.append_log(f"{rule}\n")
|
||
except Exception as e:
|
||
self.append_log(f"Note: Could not complete the report: {e}")
|
||
try:
|
||
self.explain_tab.set_context(self.last_run, rank_metric_sets(summary_df))
|
||
self.append_log("Model Explanation tab loaded with the best metric set.")
|
||
except Exception as e:
|
||
self.append_log(f"Note: could not build the model explanation: {e}\n{traceback.format_exc()}")
|
||
try:
|
||
self.theory_tab.set_context(self.last_run, rank_metric_sets(summary_df), summary_df)
|
||
self.append_log("Variance & Theory tab loaded.")
|
||
except Exception as e:
|
||
self.append_log(f"Note: could not build the variance analysis: {e}\n{traceback.format_exc()}")
|
||
self.append_log("Use 'Export Results (.xlsx)' to save the workbook "
|
||
"(it stays available until the next run).")
|
||
|
||
def export_results_excel(self):
|
||
"""Save the latest run's grid results and pivots as a formatted Excel workbook."""
|
||
if self.last_summary is None:
|
||
QMessageBox.warning(self, "No results", "Run the grid experiment first.")
|
||
return
|
||
path, _ = QFileDialog.getSaveFileName(
|
||
self, "Save N vs T Experiment Results",
|
||
"diversity_density_experiment_results.xlsx", "Excel Files (*.xlsx)")
|
||
if not path:
|
||
return
|
||
tiers = self.last_run["tiers"] if self.last_run else DEFAULT_TIERS
|
||
pivot_acc, pivot_trials = self.last_pivots
|
||
try:
|
||
export_results_xlsx(path, self.last_summary, pivot_acc, pivot_trials, tiers)
|
||
self.append_log(f">>> Results exported to Excel workbook with permutation highlighting:\n{path}")
|
||
QMessageBox.information(self, "Success", f"Experiment results saved to Excel:\n{path}")
|
||
except Exception as e:
|
||
QMessageBox.critical(self, "Save Error", f"Failed to save Excel file:\n{e}")
|
||
|
||
# --------------------------------------------------------- model export
|
||
def export_best_model(self):
|
||
"""Train the best-ranked metric set on all labelled participants and save it."""
|
||
if self.last_summary is None or self.last_run is None:
|
||
QMessageBox.warning(self, "No results", "Run the grid experiment first.")
|
||
return
|
||
|
||
board = rank_metric_sets(self.last_summary)
|
||
best = board.iloc[0]
|
||
run = self.last_run
|
||
|
||
path, _ = QFileDialog.getSaveFileName(
|
||
self, "Export Best Model", "best_model.joblib", "Joblib Files (*.joblib)")
|
||
if not path:
|
||
return
|
||
try:
|
||
bundle = train_final_model(
|
||
run["df"], run["condition"], run["classes"], best["Metrics"],
|
||
run["feature_cols"], run["chromos"], class_hints=run["class_hints"],
|
||
selection={"Top1_Count": int(best["Top1_Count"]),
|
||
"Top3_Count": int(best["Top3_Count"]),
|
||
"Mean_Part_Bal_Acc": float(best["Mean_Part_Bal_Acc"]),
|
||
"Grid_Rows": int(len(self.last_summary))})
|
||
joblib.dump(bundle, path)
|
||
except Exception as e:
|
||
QMessageBox.critical(self, "Export Error", f"Failed to export model:\n{e}")
|
||
return
|
||
|
||
n = bundle["n_train_participants"]
|
||
self.append_log(
|
||
f">>> Exported best model [{best['Metrics']}] ({len(bundle['feature_columns'])} features, "
|
||
f"{', '.join(run['chromos'])}) trained on {n[CLASS_NEG]} Class A + {n[CLASS_POS]} Class B "
|
||
f"participants ({bundle['n_train_rows']} rows):\n{path}")
|
||
QMessageBox.information(
|
||
self, "Model exported",
|
||
f"Saved model using metrics: {best['Metrics']}\n\n{path}\n\n"
|
||
f"Load with load_model_bundle() and apply with classify_participant().")
|
||
|
||
# ---------------------------------------------------------------- logs
|
||
def update_progress_log(self, text):
|
||
"""Keep a single progress line in the log, updated in place.
|
||
|
||
The line is tracked by block index rather than assumed to be last, so other log
|
||
messages (e.g. "Cancel requested") or late progress updates never create a duplicate.
|
||
"""
|
||
doc = self.log_text.document()
|
||
block = (doc.findBlockByNumber(self._progress_block)
|
||
if self._progress_block is not None else None)
|
||
if block is not None and block.isValid() and block.text().startswith("Progress:"):
|
||
cursor = QTextCursor(block)
|
||
cursor.movePosition(QTextCursor.StartOfBlock)
|
||
cursor.movePosition(QTextCursor.EndOfBlock, QTextCursor.KeepAnchor)
|
||
cursor.removeSelectedText()
|
||
cursor.insertText(text)
|
||
else:
|
||
self.log_text.append(text)
|
||
self._progress_block = doc.blockCount() - 1
|
||
|
||
def append_log(self, text: str):
|
||
self.log_text.append(text)
|
||
bar = self.log_text.verticalScrollBar()
|
||
bar.setValue(bar.maximum())
|
||
|
||
|
||
# ==========================================================================
|
||
# Tool 2: apply a saved model to newly collected data
|
||
# ==========================================================================
|
||
class ModelInferenceGUI(QWidget):
|
||
"""Load an exported model bundle and classify the data currently in the main window.
|
||
|
||
Uses the same parser as training (parse_fir_dict), so feature names and conditions match
|
||
exactly, and the same decision rule: a participant's class is the sign of their mean
|
||
trial score. Model files are pickles - only open bundles you created or trust.
|
||
"""
|
||
|
||
PART_HEADERS = ["Subject File", "Trials", "Mean Score", f"% Trials {CLASS_POS}",
|
||
"Missing Features", "Predicted Class"]
|
||
TRIAL_HEADERS = ["Subject File", "Trial", "Score", "Predicted Class"]
|
||
|
||
def __init__(self, parent: QWidget | None = None, main_window: QWidget | None = None) -> None:
|
||
super().__init__(parent)
|
||
self.setWindowTitle("fNIRS Model Inference - classify new participants")
|
||
self.resize(1100, 700)
|
||
|
||
self.main_window = main_window
|
||
self.bundle = None
|
||
self.trials_df = None
|
||
self.participants_df = None
|
||
|
||
self.init_ui()
|
||
|
||
# ------------------------------------------------------------------ UI
|
||
@staticmethod
|
||
def _make_table(headers):
|
||
table = QTableWidget()
|
||
table.setColumnCount(len(headers))
|
||
table.setHorizontalHeaderLabels(headers)
|
||
header = table.horizontalHeader()
|
||
header.setSectionResizeMode(0, QHeaderView.Stretch)
|
||
for col in range(1, len(headers)):
|
||
header.setSectionResizeMode(col, QHeaderView.Interactive)
|
||
table.setColumnWidth(col, 140)
|
||
return table
|
||
|
||
def init_ui(self):
|
||
layout = QVBoxLayout(self)
|
||
|
||
model_box = QGroupBox("1. Load trained model (.joblib)")
|
||
model_l = QHBoxLayout(model_box)
|
||
load_btn = QPushButton("Browse Model...")
|
||
load_btn.clicked.connect(self.browse_model)
|
||
model_l.addWidget(load_btn)
|
||
self.model_info_label = QLabel("No model loaded.")
|
||
self.model_info_label.setWordWrap(True)
|
||
model_l.addWidget(self.model_info_label, 1)
|
||
layout.addWidget(model_box)
|
||
|
||
run_box = QGroupBox("2. Classify the data currently loaded in the main window")
|
||
run_l = QHBoxLayout(run_box)
|
||
self.run_btn = QPushButton("Run Predictions")
|
||
self.run_btn.setStyleSheet(
|
||
"font-weight: bold; background-color: #2b5b84; color: white; padding: 8px 16px;")
|
||
self.run_btn.clicked.connect(self.run_inference)
|
||
run_l.addWidget(self.run_btn)
|
||
run_l.addStretch()
|
||
self.export_parts_btn = QPushButton("Export Participants CSV...")
|
||
self.export_parts_btn.setEnabled(False)
|
||
self.export_parts_btn.clicked.connect(lambda: self.export_csv("participants"))
|
||
run_l.addWidget(self.export_parts_btn)
|
||
self.export_trials_btn = QPushButton("Export Trials CSV...")
|
||
self.export_trials_btn.setEnabled(False)
|
||
self.export_trials_btn.clicked.connect(lambda: self.export_csv("trials"))
|
||
run_l.addWidget(self.export_trials_btn)
|
||
layout.addWidget(run_box)
|
||
|
||
self.summary_label = QLabel("")
|
||
self.summary_label.setWordWrap(True)
|
||
layout.addWidget(self.summary_label)
|
||
|
||
self.result_tabs = QTabWidget()
|
||
self.part_table = self._make_table(self.PART_HEADERS)
|
||
self.part_table.setSelectionBehavior(QTableWidget.SelectRows)
|
||
self.part_table.setSelectionMode(QTableWidget.SingleSelection)
|
||
# Clicking (or pressing Enter on) a participant jumps to just their trials
|
||
self.part_table.cellClicked.connect(self.on_participant_chosen)
|
||
self.part_table.cellActivated.connect(self.on_participant_chosen)
|
||
|
||
trials_page = QWidget()
|
||
trials_l = QVBoxLayout(trials_page)
|
||
trials_l.setContentsMargins(0, 0, 0, 0)
|
||
filter_row = QHBoxLayout()
|
||
self.trial_filter_label = QLabel("")
|
||
filter_row.addWidget(self.trial_filter_label, 1)
|
||
self.show_all_btn = QPushButton("Show all trials")
|
||
self.show_all_btn.clicked.connect(lambda: self.show_trials(None))
|
||
filter_row.addWidget(self.show_all_btn)
|
||
trials_l.addLayout(filter_row)
|
||
self.trial_table = self._make_table(self.TRIAL_HEADERS)
|
||
trials_l.addWidget(self.trial_table, 1)
|
||
|
||
self.trial_filter = None # participant path currently shown on the Trials tab (None = all)
|
||
self.result_tabs.addTab(self.part_table, "Participants")
|
||
self.result_tabs.addTab(trials_page, "Trials")
|
||
layout.addWidget(self.result_tabs, 1)
|
||
self.update_trial_filter_label()
|
||
|
||
note = QLabel(
|
||
f"A participant is {CLASS_POS} when the mean of their trial scores is above 0, otherwise "
|
||
f"{CLASS_NEG}. Scores are uncalibrated LDA decision scores, not probabilities. "
|
||
f"'Missing Features' is the share of the model's feature values that were absent "
|
||
f"and filled with the training median.")
|
||
note.setWordWrap(True)
|
||
layout.addWidget(note)
|
||
|
||
# --------------------------------------------------------------- model
|
||
def browse_model(self):
|
||
path, _ = QFileDialog.getOpenFileName(self, "Open Model Bundle", "", "Joblib Files (*.joblib)")
|
||
if not path:
|
||
return
|
||
try:
|
||
bundle = load_model_bundle(path)
|
||
except Exception as e:
|
||
QMessageBox.critical(self, "Error", f"Failed to load model file:\n{e}")
|
||
return
|
||
|
||
self.bundle = bundle
|
||
self.clear_results()
|
||
|
||
esc = html.escape
|
||
n = bundle.get("n_train_participants", {})
|
||
neg, pos = bundle["class_neg"], bundle["class_pos"]
|
||
sel = bundle.get("selection", {})
|
||
acc_txt = (f" | <b>Grid estimate (bal. acc):</b> {sel['Mean_Part_Bal_Acc'] * 100:.1f}%"
|
||
if "Mean_Part_Bal_Acc" in sel else "")
|
||
self.model_info_label.setText(
|
||
f"<b>Loaded:</b> {esc(Path(path).name)} | <b>Condition:</b> {esc(str(bundle['condition']))} | "
|
||
f"<b>Metrics:</b> {esc(', '.join(bundle['metrics']))} | "
|
||
f"<b>Chromophores:</b> {esc(', '.join(bundle['chromophores']))} | "
|
||
f"<b>Features:</b> {len(bundle['feature_columns'])} | "
|
||
f"<b>Trained on:</b> {n.get(neg, '?')} {neg} + {n.get(pos, '?')} {pos} participants"
|
||
f"{acc_txt}")
|
||
|
||
# ------------------------------------------------------------- results
|
||
def update_trial_filter_label(self):
|
||
if self.trial_filter is None:
|
||
n = 0 if self.trials_df is None else len(self.trials_df)
|
||
self.trial_filter_label.setText(
|
||
f"Showing all trials ({n}). Click a participant on the Participants tab to see only theirs.")
|
||
else:
|
||
n = int((self.trials_df["Participant"] == self.trial_filter).sum())
|
||
self.trial_filter_label.setText(
|
||
f"<b>Showing {n} trial(s) for:</b> {html.escape(self.trial_filter)}")
|
||
self.show_all_btn.setVisible(self.trial_filter is not None)
|
||
|
||
def show_trials(self, participant=None):
|
||
"""Fill the Trials tab with every trial, or only one participant's (without switching tabs)."""
|
||
self.trial_filter = participant
|
||
trials = self.trials_df
|
||
if trials is None:
|
||
self.trial_table.setRowCount(0)
|
||
else:
|
||
if participant is not None:
|
||
trials = trials[trials["Participant"] == participant]
|
||
self._fill_table(self.trial_table, [
|
||
[r.Participant, r.Trial, f"{r.Score:+.3f}", r.Predicted_Class]
|
||
for r in trials.itertuples(index=False)], class_col=3)
|
||
self.result_tabs.setTabText(1, "Trials" if participant is None else f"Trials ({len(trials)})")
|
||
self.update_trial_filter_label()
|
||
|
||
def on_participant_chosen(self, row, _column=0):
|
||
"""Participants tab -> Trials tab, filtered to the chosen participant."""
|
||
item = self.part_table.item(row, 0)
|
||
if item is None or self.trials_df is None:
|
||
return
|
||
self.show_trials(item.text())
|
||
self.result_tabs.setCurrentIndex(1)
|
||
|
||
def clear_results(self):
|
||
self.trials_df = self.participants_df = None
|
||
self.part_table.setRowCount(0)
|
||
self.show_trials(None)
|
||
self.summary_label.setText("")
|
||
self.export_parts_btn.setEnabled(False)
|
||
self.export_trials_btn.setEnabled(False)
|
||
|
||
def _fill_table(self, table, rows, class_col):
|
||
"""Fill a table with text rows and colour the predicted-class column."""
|
||
neg_colour, pos_colour = QColor(200, 225, 255), QColor(255, 240, 200)
|
||
table.setRowCount(len(rows))
|
||
for r, values in enumerate(rows):
|
||
for c, text in enumerate(values):
|
||
item = QTableWidgetItem(str(text))
|
||
item.setFlags(item.flags() & ~Qt.ItemIsEditable)
|
||
if c == class_col:
|
||
is_neg = text == self.bundle["class_neg"]
|
||
item.setBackground(neg_colour if is_neg else pos_colour)
|
||
item.setForeground(QColor(0, 0, 0))
|
||
table.setItem(r, c, item)
|
||
|
||
def run_inference(self):
|
||
if self.bundle is None:
|
||
QMessageBox.warning(self, "Warning", "Please load a trained model (.joblib) first.")
|
||
return
|
||
|
||
fir_dict = getattr(self.main_window, "fir_feature_dict", None)
|
||
if not isinstance(fir_dict, dict) or not fir_dict:
|
||
QMessageBox.warning(self, "Warning",
|
||
"No data: main_window.fir_feature_dict is missing or empty.")
|
||
return
|
||
|
||
df, kind, info = parse_fir_dict(fir_dict)
|
||
if df.empty:
|
||
QMessageBox.warning(self, "Warning", "Could not extract valid features from the main window data.")
|
||
return
|
||
|
||
condition = self.bundle["condition"]
|
||
rows = df[df["Condition"] == condition]
|
||
if rows.empty:
|
||
available = ", ".join(sorted(df["Condition"].unique())) or "none"
|
||
QMessageBox.warning(
|
||
self, "Warning",
|
||
f"No data for the model's condition '{condition}'.\nConditions in the main window: {available}")
|
||
return
|
||
|
||
missing = missing_feature_columns(self.bundle, rows)
|
||
if missing:
|
||
QMessageBox.warning(
|
||
self, "Features missing",
|
||
f"The data lacks {len(missing)} of the model's {len(self.bundle['feature_columns'])} "
|
||
f"feature columns, e.g.:\n" + "\n".join(missing[:5]) +
|
||
"\n\nCheck that the same FIR/feature steps were run as when the model was trained.")
|
||
return
|
||
|
||
try:
|
||
trials, parts = classify_dataframe(self.bundle, rows)
|
||
except Exception as e:
|
||
QMessageBox.critical(self, "Prediction Error", f"Failed during model prediction:\n{e}")
|
||
return
|
||
|
||
self.trials_df, self.participants_df = trials, parts
|
||
|
||
self._fill_table(self.part_table, [
|
||
[r.Participant, r.N_Trials, f"{r.Mean_Score:+.3f}", f"{r.Frac_Trials_Class_B * 100:.0f}%",
|
||
f"{r.Missing_Frac * 100:.1f}%", r.Predicted_Class]
|
||
for r in parts.itertuples(index=False)], class_col=5)
|
||
self.show_trials(None)
|
||
self.result_tabs.setCurrentIndex(0)
|
||
|
||
neg, pos = self.bundle["class_neg"], self.bundle["class_pos"]
|
||
n_pos = int((parts["Predicted_Class"] == pos).sum())
|
||
text = (f"<b>{len(parts)} participants</b> ({len(parts) - n_pos} {neg}, {n_pos} {pos}) | "
|
||
f"{len(trials)} trials | condition: {html.escape(str(condition))}")
|
||
if info["dropped"]:
|
||
text += (f" | <b>{len(info['dropped'])} file(s) skipped</b> (no trial-specific features)")
|
||
self.summary_label.setText(text)
|
||
self.export_parts_btn.setEnabled(True)
|
||
self.export_trials_btn.setEnabled(True)
|
||
|
||
def export_csv(self, which):
|
||
df = self.participants_df if which == "participants" else self.trials_df
|
||
if df is None or df.empty:
|
||
QMessageBox.warning(self, "Warning", "No prediction results available to export.")
|
||
return
|
||
path, _ = QFileDialog.getSaveFileName(
|
||
self, f"Export {which.title()}", f"model_predictions_{which}.csv", "CSV Files (*.csv)")
|
||
if not path:
|
||
return
|
||
try:
|
||
out = df.copy()
|
||
if "Missing_Frac" in out.columns:
|
||
out = out.rename(columns={"Missing_Frac": "Missing_Features_Frac"})
|
||
out.to_csv(path, index=False)
|
||
QMessageBox.information(self, "Success", f"Exported to:\n{path}")
|
||
except Exception as e:
|
||
QMessageBox.critical(self, "Error", f"Failed to export CSV:\n{e}")
|
||
|
||
|
||
# ==========================================================================
|
||
# Plugin entry point
|
||
# ==========================================================================
|
||
class Plugin:
|
||
"""Plugin entry point contract loaded by PluginManager."""
|
||
|
||
def __init__(self, main_window: QWidget) -> None:
|
||
self.main_window = main_window
|
||
self.name = "ML Channel Builder"
|
||
self.widget_instance: DataSplitterGUI | None = None
|
||
self.widget_instance2: ModelInferenceGUI | None = None
|
||
|
||
def register_menu(self, plugin_menu: QMenu) -> None:
|
||
open_action = QAction("Open ML Tool", self.main_window)
|
||
open_action.triggered.connect(self.show_widget)
|
||
open_action2 = QAction("Open ML Tool 2 (Model Inference)", self.main_window)
|
||
open_action2.triggered.connect(self.show_widget2)
|
||
about_action = QAction("About ML Builder", self.main_window)
|
||
about_action.triggered.connect(self.show_about)
|
||
plugin_menu.addAction(open_action)
|
||
plugin_menu.addAction(open_action2)
|
||
plugin_menu.addAction(about_action)
|
||
|
||
def show_widget(self) -> None:
|
||
if self.widget_instance is None or not self.widget_instance.isVisible():
|
||
self.widget_instance = DataSplitterGUI(main_window=self.main_window)
|
||
self.widget_instance.show()
|
||
else:
|
||
self.widget_instance.raise_()
|
||
self.widget_instance.activateWindow()
|
||
|
||
def show_widget2(self) -> None:
|
||
if self.widget_instance2 is None or not self.widget_instance2.isVisible():
|
||
self.widget_instance2 = ModelInferenceGUI(main_window=self.main_window)
|
||
self.widget_instance2.show()
|
||
else:
|
||
self.widget_instance2.raise_()
|
||
self.widget_instance2.activateWindow()
|
||
|
||
def show_about(self) -> None:
|
||
title = "ML Channel Builder"
|
||
version = "2.1.0"
|
||
author = "User"
|
||
description = ("fNIRS LDA classification: N-vs-T diversity/density grid experiment "
|
||
"with metric-combination sweeping and best-model export, plus a "
|
||
"model inference tool for classifying newly collected data.")
|
||
|
||
json_path = Path(__file__).resolve().parent / "plugin.json"
|
||
if json_path.is_file():
|
||
try:
|
||
with open(json_path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
if isinstance(data, dict):
|
||
title = data.get("name", title)
|
||
version = data.get("version", version)
|
||
author = data.get("author", author)
|
||
description = data.get("description", description)
|
||
except Exception:
|
||
pass
|
||
|
||
QMessageBox.about(
|
||
self.main_window, f"About {title}",
|
||
f"<b>{title}</b><br><b>Version:</b> {version} | "
|
||
f"<b>Author:</b> {author}<br><br>{description}") |