"""Public helper module. Conceptually, this module provides the low-level one-bin decoding kernels used by higher-level decoding workflows. It exists as a separate unit so classifier setup, balancing, and cross- validation behavior stay centralized instead of being duplicated across multiple decoding pipelines. It connects normalized trial matrices to decoding pipeline orchestration. """ from __future__ import annotations from typing import Optional, Tuple import numpy as np from .stats_utils import downsample_balance, zscore_cols __all__ = [ "decode_one_bin_svm", "decode_one_bin_svm_matlab_compat", ] def decode_one_bin_svm( X1: np.ndarray, X2: np.ndarray, mintrial: int = 5, balance_method: str = "downsample", zscoring: bool = True, run_seed: int = 0, rng_obj: Optional[np.random.Generator] = None, ) -> Tuple[float, float]: """Decode one time bin with linear SVM and stratified cross-validation. Parameters ---------- X1, X2: Condition matrices with shape `(trials, features)`. mintrial: Minimum number of trials required in each condition. balance_method: Class-balancing strategy. Supported values are `downsample`, `no`, and `smote` (currently routed through the same balancing path as `downsample`). zscoring: Whether to z-score features before fitting. run_seed: Seed used when a local RNG must be created. rng_obj: Optional RNG instance to reuse across repeated decoding calls. Returns ------- float, float Mean classification accuracy and shuffled-label control accuracy in percent. """ X1 = np.asarray(X1, dtype=np.float32) X2 = np.asarray(X2, dtype=np.float32) if (X1.shape[0] < int(mintrial)) or (X2.shape[0] < int(mintrial)): return np.nan, np.nan if rng_obj is None: rng_obj = np.random.default_rng(run_seed) if balance_method == "downsample": X, y = downsample_balance(X1, X2, rng_obj=rng_obj) elif balance_method == "no": X = np.vstack([X1, X2]) y = np.concatenate([np.ones(X1.shape[0]), -np.ones(X2.shape[0])]) elif balance_method == "smote": X, y = downsample_balance(X1, X2, rng_obj=rng_obj) else: raise ValueError("Unsupported balance_method") if zscoring: X = zscore_cols(X) # Remove feature columns that are entirely zero or non-finite after # normalization so the classifier sees only informative dimensions. keep = np.any(np.isfinite(X) & (X != 0), axis=0) X = X[:, keep] if X.shape[1] == 0: return np.nan, np.nan X = np.nan_to_num(X, nan=0.0, posinf=0.0, neginf=0.0) from sklearn.model_selection import StratifiedKFold from sklearn.svm import SVC c1 = np.sum(y == 1) c2 = np.sum(y == -1) min_class = int(min(c1, c2)) if min_class < 2: return np.nan, np.nan n_splits = min(5, min_class) if n_splits < 2: return np.nan, np.nan skf = StratifiedKFold(n_splits=n_splits, shuffle=True, random_state=run_seed) acc = [] acc_sh = [] # Evaluate the same held-out folds against both the true and shuffled test # labels so the control baseline stays tied to the exact split structure. for tr_idx, te_idx in skf.split(X, y): model = SVC(kernel="linear", C=1.0) model.fit(X[tr_idx, :], y[tr_idx]) pred = model.predict(X[te_idx, :]) yy = y[te_idx] acc.append(100.0 * np.mean(pred == yy)) yy_sh = yy.copy() rng_obj.shuffle(yy_sh) acc_sh.append(100.0 * np.mean(pred == yy_sh)) return float(np.mean(acc)), float(np.mean(acc_sh)) def decode_one_bin_svm_matlab_compat( X1: np.ndarray, X2: np.ndarray, mintrial: int = 5, balance_method: str = "downsample", zscoring: bool = True, run_seed: int = 0, rng_obj: Optional[np.random.Generator] = None, ) -> Tuple[float, float]: """Decode one time bin with the stricter compatibility path. Parameters ---------- X1, X2: Condition matrices expected in the orientation used by this decoding path. mintrial: Minimum number of trials required in each condition. balance_method: Class-balancing strategy. Supported values are `downsample`, `no`, and `smote`. zscoring: Whether to z-score features before fitting. run_seed: Seed used when a local RNG must be created. rng_obj: Optional RNG instance to reuse across repeated decoding calls. Returns ------- float, float Mean classification accuracy and shuffled-label control accuracy in percent. """ X1 = np.asarray(X1, dtype=np.float32) X2 = np.asarray(X2, dtype=np.float32) mintrial = int(mintrial) if (X1.shape[1] < mintrial) or (X2.shape[1] < mintrial): return np.nan, np.nan if rng_obj is None: rng_obj = np.random.default_rng(run_seed) if balance_method == "downsample": X, y = downsample_balance(X1, X2, rng_obj=rng_obj) elif balance_method == "no": X = np.vstack([X1, X2]) y = np.concatenate([np.ones(X1.shape[0]), -np.ones(X2.shape[0])]) elif balance_method == "smote": X, y = downsample_balance(X1, X2, rng_obj=rng_obj) else: raise ValueError("Unsupported balance_method") if zscoring: X = zscore_cols(X) # This path keeps the historical all-zero column filter rather than the # broader finite-and-nonzero check used by the generic decoder. col_zero = ~np.any(X, axis=0) X = X[:, ~col_zero] if X.shape[1] == 0: return np.nan, np.nan from sklearn.model_selection import StratifiedKFold from sklearn.svm import NuSVC def _run_cv(skf_obj, model_factory): acc = [] acc_sh = [] for tr_idx, te_idx in skf_obj.split(X, y): model = model_factory() try: model.fit(X[tr_idx, :].astype(np.float64, copy=False), y[tr_idx]) except ValueError as ex: msg = str(ex) if "dual coefficients or intercepts are not finite" in msg: return np.nan, np.nan raise pred = model.predict(X[te_idx, :].astype(np.float64, copy=False)) yy = y[te_idx] acc.append(100.0 * np.mean(pred == yy)) yy_sh = yy.copy() yy_sh = yy_sh[rng_obj.permutation(yy_sh.size)] acc_sh.append(100.0 * np.mean(pred == yy_sh)) return float(np.mean(acc)), float(np.mean(acc_sh)) # Derive the split seed from the caller RNG so repeated notebook runs can # stay reproducible while still varying across decode calls. split_seed = int(rng_obj.integers(0, 2**31 - 1)) skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=split_seed) model_factory = lambda: NuSVC(nu=0.5, kernel="linear") try: return _run_cv(skf, model_factory) except ValueError as ex: msg = str(ex) if "dual coefficients or intercepts are not finite" not in msg: raise skf_retry = StratifiedKFold(n_splits=5, shuffle=True, random_state=0) return _run_cv(skf_retry, model_factory)