From 2b06658c2158f2c79d386aef40a24586070c5102 Mon Sep 17 00:00:00 2001 From: Bru Date: Fri, 19 Jun 2026 20:33:49 +0200 Subject: [PATCH 01/15] feat(evaluations): cross-subject transfer learning via CrossSubjectEvaluation Run target-aware transfer protocols through the existing CrossSubjectEvaluation, instead of a dedicated evaluation engine or a separate transfer module. - CrossSubjectEvaluation gains calibration_size (+ calibration_labeled): when > 0 it wraps its CrossSubjectSplitter in TransferSplitter, so each fold yields (train, calibration, test). - base.py consumes the split with `train, *cal, test` (parallel + serial paths); the held-out calibration slice is routed RAW to the pipeline steps that request it via sklearn metadata routing (subjects, X_target_unlabeled / X_target_labeled). Plain pipelines request nothing and are unaffected -- the estimator owns the target representation, so there is no transform-through-steps. - TransferSplitter is the single generic transfer splitter (subject / session / dataset). Removes moabb/evaluations/transfer.py and CrossSubjectTransferSplitter. Usage: CrossSubjectEvaluation(..., calibration_size=0.2).process(pipelines) --- moabb/evaluations/__init__.py | 1 + moabb/evaluations/base.py | 53 ++++++++++++++-- moabb/evaluations/evaluations.py | 59 ++++++++++++++++-- moabb/evaluations/splitters.py | 101 +++++++++++++++++++++++++++++++ moabb/tests/test_evaluations.py | 64 ++++++++++++++++++++ moabb/tests/test_splits.py | 43 +++++++++++++ 6 files changed, 310 insertions(+), 11 deletions(-) diff --git a/moabb/evaluations/__init__.py b/moabb/evaluations/__init__.py index ee17fa525..d75de227a 100644 --- a/moabb/evaluations/__init__.py +++ b/moabb/evaluations/__init__.py @@ -15,6 +15,7 @@ CrossSessionSplitter, CrossSubjectSplitter, LearningCurveSplitter, + TransferSplitter, WithinSessionSplitter, WithinSubjectSplitter, ) diff --git a/moabb/evaluations/base.py b/moabb/evaluations/base.py index e572f5f11..d6fff395d 100644 --- a/moabb/evaluations/base.py +++ b/moabb/evaluations/base.py @@ -12,9 +12,11 @@ import numpy as np import pandas as pd from joblib import Parallel, delayed +from sklearn import config_context from sklearn.base import BaseEstimator, clone from sklearn.model_selection import StratifiedKFold from sklearn.preprocessing import LabelEncoder +from sklearn.utils.metadata_routing import get_routing_for_object from moabb.analysis import Results from moabb.datasets.base import ( # noqa: F401 - CacheConfig used in type hints @@ -44,6 +46,25 @@ log = logging.getLogger(__name__) + +def _route_transfer_metadata(estimator, subjects, calib=None): + """Keep only the transfer metadata an estimator requests at ``fit``. + + ``subjects`` is the per-trial source-subject array; ``calib`` is an optional + dict carrying the (raw) calibration slice, e.g. + ``{"X_target_unlabeled": X[calib_idx]}``. Steps opt in via + ``set_fit_request(...)``; plain pipelines request nothing, so this returns + ``{}`` and the fit is unchanged. The framework passes the slice raw -- the + estimator owns the target representation. + """ + candidate = {"subjects": subjects} + if calib: + candidate.update(calib) + with config_context(enable_metadata_routing=True): + kept = get_routing_for_object(estimator).consumes("fit", set(candidate)) + return {k: v for k, v in candidate.items() if k in kept} + + # Making the optuna soft dependency @@ -100,6 +121,7 @@ def _evaluate_fold( session, cv_ind, split_metadata=None, + calib_idx=None, ): """Evaluate a single CV fold. Pure function, no shared mutable state. @@ -174,6 +196,18 @@ def _evaluate_fold( tracker = emissions_obj.create_tracker() tracker.start() + # Optional transfer-learning calibration slice (raw), routed to the steps + # that request it. Empty for ordinary evaluations -> plain fit. + calib = None + if calib_idx is not None and len(calib_idx): + if config.get("calibration_labeled", False): + y_calib = y[calib_idx] if mne_labels else le.transform(y[calib_idx]) + calib = {"X_target_labeled": X[calib_idx], "y_target_labeled": y_calib} + else: + calib = {"X_target_unlabeled": X[calib_idx]} + subjects_train = metadata["subject"].to_numpy()[train_idx] + fit_params = _route_transfer_metadata(cvclf, subjects_train, calib) + # Fit model task_name = None emissions = math.nan @@ -181,7 +215,8 @@ def _evaluate_fold( task_name = str(uuid4()) tracker.start_task(task_name) t_start = perf_counter() - cvclf.fit(X[train_idx], y_train) + with config_context(enable_metadata_routing=True): + cvclf.fit(X[train_idx], y_train, **fit_params) duration = perf_counter() - t_start if tracker is not None: emissions_data = tracker.stop_task() @@ -565,7 +600,7 @@ def _build_scored_result( res["score"] = self.error_score return res - def _fit_cv(self, model, X_train, y_train, tracker=None): + def _fit_cv(self, model, X_train, y_train, tracker=None, fit_params=None): """Fit a model for a CV fold with optional CodeCarbon tracking.""" task_name = None emissions = math.nan @@ -573,7 +608,8 @@ def _fit_cv(self, model, X_train, y_train, tracker=None): task_name = str(uuid4()) tracker.start_task(task_name) t_start = perf_counter() - model.fit(X_train, y_train) + with config_context(enable_metadata_routing=True): + model.fit(X_train, y_train, **(fit_params or {})) duration = perf_counter() - t_start if tracker is not None: emissions_data = tracker.stop_task() @@ -668,19 +704,23 @@ def _build_eval_config(self, param_grid): ), "score_per_session": self._score_per_session, "param_grid": None, # overridden per-task below if needed + "calibration_labeled": getattr(self, "calibration_labeled", False), } @staticmethod def _preview_splits(splitter, y, metadata): """Materialize folds up front with optional splitter metadata.""" preview = [] - for cv_ind, (train_idx, test_idx) in enumerate(splitter.split(y, metadata)): + # ``*cal`` absorbs the optional calibration slice from a transfer + # splitter; a plain 2-tuple splitter gives cal == [] (no calibration). + for cv_ind, (train_idx, *cal, test_idx) in enumerate(splitter.split(y, metadata)): + calib_idx = cal[0] if cal else train_idx[:0] split_metadata = None if hasattr(splitter, "get_metadata"): split_metadata = splitter.get_metadata() if split_metadata is not None: split_metadata = dict(split_metadata) - preview.append((cv_ind, train_idx, test_idx, split_metadata)) + preview.append((cv_ind, train_idx, calib_idx, test_idx, split_metadata)) return preview def _build_task_list( @@ -691,7 +731,7 @@ def _build_task_list( config = self._build_eval_config(param_grid) fold_preview = self._preview_splits(splitter, y, metadata) - for cv_ind, train_idx, test_idx, split_meta in fold_preview: + for cv_ind, train_idx, calib_idx, test_idx, split_meta in fold_preview: test_meta = metadata.iloc[test_idx] subject = test_meta["subject"].iloc[0] @@ -719,6 +759,7 @@ def _build_task_list( "session": session, "cv_ind": cv_ind, "split_metadata": split_meta, + "calib_idx": calib_idx, } ) return tasks diff --git a/moabb/evaluations/evaluations.py b/moabb/evaluations/evaluations.py index 410b7dc9b..1ce78d4fc 100644 --- a/moabb/evaluations/evaluations.py +++ b/moabb/evaluations/evaluations.py @@ -7,10 +7,11 @@ from sklearn.preprocessing import LabelEncoder from tqdm import tqdm -from moabb.evaluations.base import BaseEvaluation +from moabb.evaluations.base import BaseEvaluation, _route_transfer_metadata from moabb.evaluations.splitters import ( CrossSessionSplitter, CrossSubjectSplitter, + TransferSplitter, WithinSessionSplitter, WithinSubjectSplitter, ) @@ -455,6 +456,19 @@ class CrossSubjectEvaluation(BaseEvaluation): n_splits : int or None Number of splits for cross-validation. If None, the number of splits is equal to the number of subjects. Defaults to ``None``. + calibration_size : float + Transfer-learning calibration. Fraction of each held-out target subject + made available for adaptation, in ``[0, 1]``. ``0.0`` (default) is the + ordinary cross-subject split. When ``> 0`` the splitter yields a + ``(train, calibration, test)`` fold and the calibration slice is routed + (raw) to the pipeline steps that request it via ``set_fit_request`` -- + ``subjects`` plus ``X_target_unlabeled`` (or ``X_target_labeled`` / + ``y_target_labeled`` if ``calibration_labeled``). Steps that request + nothing are unaffected. + calibration_labeled : bool + If True, the calibration slice is offered with labels + (``X_target_labeled`` / ``y_target_labeled``); otherwise unlabeled + (``X_target_unlabeled``). Defaults to ``False``. Notes ----- @@ -466,8 +480,23 @@ class CrossSubjectEvaluation(BaseEvaluation): _score_per_session = True _needs_all_subjects = True + def __init__( + self, + *args, + calibration_size: float = 0.0, + calibration_labeled: bool = False, + **kwargs, + ): + super().__init__(*args, **kwargs) + self.calibration_size = calibration_size + self.calibration_labeled = calibration_labeled + def _create_splitter(self): - """Create the CrossSubjectSplitter for parallel evaluation.""" + """Create the CrossSubjectSplitter for parallel evaluation. + + Wrapped in a :class:`~moabb.evaluations.splitters.TransferSplitter` when + ``calibration_size > 0`` so each fold yields a calibration slice. + """ if self.n_splits is None: default_class = LeaveOneGroupOut default_kwargs = {} @@ -476,9 +505,12 @@ def _create_splitter(self): default_kwargs = {"n_splits": self.n_splits} cv_class, cv_kwargs = self._resolve_cv(default_class, default_kwargs) - return CrossSubjectSplitter( + splitter = CrossSubjectSplitter( cv_class=cv_class, random_state=self.random_state, **cv_kwargs ) + if self.calibration_size > 0: + return TransferSplitter(splitter, calibration_size=self.calibration_size) + return splitter # flake8: noqa: C901 def evaluate( @@ -533,13 +565,15 @@ def evaluate( tracker.start() # Progressbar at subject level - for cv_ind, (train, test) in enumerate( + # ``*cal`` absorbs the optional calibration slice from a transfer split. + for cv_ind, (train, *cal, test) in enumerate( tqdm( self.cv.split(y, metadata), total=n_subjects, desc=f"{dataset.code}-CrossSubject", ) ): + calib = cal[0] if cal else train[:0] subject = groups[test[0]] # now we can check if this subject has results run_pipes = self.results.not_yet_computed( @@ -552,8 +586,23 @@ def evaluate( ) cvclf = clone(clf) + calib_md = None + if len(calib): + if self.calibration_labeled: + calib_md = { + "X_target_labeled": X[calib], + "y_target_labeled": y[calib], + } + else: + calib_md = {"X_target_unlabeled": X[calib]} + fit_params = _route_transfer_metadata(cvclf, groups[train], calib_md) + duration, emissions, task_name = self._fit_cv( - cvclf, X[train], y[train], tracker if _carbonfootprint else None + cvclf, + X[train], + y[train], + tracker if _carbonfootprint else None, + fit_params=fit_params, ) self._maybe_save_model_cv( cvclf, dataset, subject, "", name, cv_ind, eval_type="CrossSubject" diff --git a/moabb/evaluations/splitters.py b/moabb/evaluations/splitters.py index ab3a7a77f..5fc253539 100644 --- a/moabb/evaluations/splitters.py +++ b/moabb/evaluations/splitters.py @@ -591,6 +591,107 @@ def get_metadata(self): return self._last_split_metadata +def _split_target_fraction( + target_idx: np.ndarray, calibration_size: float +) -> tuple[np.ndarray, np.ndarray]: + """Slice a held-out target subject's trials into (calibration, test). + + The first ``calibration_size`` fraction of ``target_idx`` (in trial order) is + the calibration/adaptation slice; the rest is evaluated. ``0.0`` yields no + calibration; ``1.0`` is transductive (calibration and test are identical). + For any fraction strictly between 0 and 1, at least one trial is kept on each + side. + """ + target_idx = np.asarray(target_idx, dtype=int) + if len(target_idx) == 0: + raise ValueError("Empty held-out target index.") + if not 0.0 <= calibration_size <= 1.0: + raise ValueError(f"calibration_size must be in [0, 1]. Got {calibration_size!r}.") + + if calibration_size == 0.0: + return np.array([], dtype=int), target_idx + if calibration_size == 1.0: + return target_idx, target_idx + + n_calib = int(round(calibration_size * len(target_idx))) + n_calib = min(max(n_calib, 1), len(target_idx) - 1) + return target_idx[:n_calib], target_idx[n_calib:] + + +class TransferSplitter(BaseCrossValidator): + """Add an optional calibration/adaptation slice to any leave-one-group-out split. + + This is the generic transfer-learning split. It wraps a base moabb splitter + (:class:`CrossSubjectSplitter`, :class:`CrossSessionSplitter`, + :class:`CrossDatasetSplitter`, ...) and, for each ``(train, test)`` fold the + base produces, carves the first ``calibration_size`` fraction off the held-out + *test* group as a calibration slice, yielding **three** index arrays:: + + train everything the base splitter put in train (the sources) + calibration the first ``calibration_size`` fraction of the held-out + group -- the adaptation/calibration slice (may be empty) + test the remaining held-out trials, to be scored + + The same mechanism therefore expresses subject-transfer, session-transfer and + dataset-transfer. A split only decides *which trials go where*; whether the + calibration slice is consumed with or without labels, and how the test block is + predicted, are estimator/scoring concerns and stay out of scope here. + + Consume it uniformly -- the ``*cal`` absorbs the (optional) calibration slice, + so the same loop also accepts a plain 2-tuple splitter:: + + for train, *cal, test in splitter.split(y, metadata): + calib = cal[0] if cal else test[:0] + + Parameters + ---------- + base_splitter : BaseCrossValidator + A moabb splitter exposing ``split(y, metadata) -> (train, test)`` and + ``get_n_splits(metadata)`` (typically a leave-one-group-out splitter). + calibration_size : float + Fraction of each held-out group reserved for calibration/adaptation, in + ``[0, 1]``. ``0.0`` keeps the base geometry with an empty calibration slice; + ``1.0`` is fully transductive (calibration and test are identical). For any + value strictly between 0 and 1, at least one trial is kept on each side. + Defaults to ``0.0``. + + Yields + ------ + train : ndarray + Source training indices. + calibration : ndarray + Held-out calibration/adaptation indices (empty when ``calibration_size`` is + ``0.0``; equal to ``test`` when ``1.0``). + test : ndarray + Held-out evaluation indices. + """ + + def __init__(self, base_splitter: BaseCrossValidator, calibration_size: float = 0.0): + if not 0.0 <= calibration_size <= 1.0: + raise ValueError( + f"calibration_size must be in [0, 1]. Got {calibration_size!r}." + ) + self.base_splitter = base_splitter + self.calibration_size = calibration_size + + def get_n_splits(self, metadata): + """Return the number of splits (delegated to the base splitter).""" + return self.base_splitter.get_n_splits(metadata) + + def split(self, y, metadata): + # ``*_`` tolerates a base that yields more than (train, test); the held-out + # group to split is always the last element. + for train_idx, *_, group_idx in self.base_splitter.split(y, metadata): + calibration_idx, test_idx = _split_target_fraction( + group_idx, self.calibration_size + ) + yield train_idx, calibration_idx, test_idx + + def get_metadata(self): + """Return metadata for the most recent split (from the base splitter).""" + return _splitter_metadata(self.base_splitter) + + class CrossDatasetSplitter(BaseCrossValidator): """Data splitter for leave-dataset-out style evaluation. diff --git a/moabb/tests/test_evaluations.py b/moabb/tests/test_evaluations.py index 2d6378819..31faac132 100644 --- a/moabb/tests/test_evaluations.py +++ b/moabb/tests/test_evaluations.py @@ -1034,3 +1034,67 @@ def test_error_folds_multi_metric_use_fallback(self): np.testing.assert_almost_equal(res["score_accuracy"], 0.4) # score_f1: avg(0.7, 0.0) = 0.35 np.testing.assert_almost_equal(res["score_f1"], 0.35) + + +# Transfer-learning calibration through the stock CrossSubjectEvaluation. +_TRANSFER_CAPTURE = [] + + +class _TransferRecorder(sklearn.base.TransformerMixin, sklearn.base.BaseEstimator): + """A target-aware step that records the transfer metadata it is routed.""" + + def fit(self, X, y=None, subjects=None, X_target_unlabeled=None): + _TRANSFER_CAPTURE.append( + { + "n_subjects": 0 if subjects is None else len(subjects), + "n_target": 0 if X_target_unlabeled is None else len(X_target_unlabeled), + } + ) + return self + + def transform(self, X): + return X + + +def test_cross_subject_calibration_routes_to_estimator(): + """calibration_size>0 routes subjects + the (raw) calibration slice to the + steps that request it, via the unchanged CrossSubjectEvaluation.""" + from sklearn import config_context + + _TRANSFER_CAPTURE.clear() + with config_context(enable_metadata_routing=True): + step = _TransferRecorder().set_fit_request(subjects=True, X_target_unlabeled=True) + pipe = make_pipeline(Covariances("oas"), step, CSP(8), LDA()) + ds = FakeDataset(["left_hand", "right_hand"], n_subjects=3, n_sessions=2, seed=9) + + evaluation = ev.CrossSubjectEvaluation( + paradigm=FakeImageryParadigm(), + datasets=[ds], + calibration_size=0.5, + overwrite=True, + n_jobs=1, + suffix="calibroute", + ) + results = evaluation.process(pipelines=OrderedDict([("T", pipe)])) + + assert len(results) > 0 + assert _TRANSFER_CAPTURE, "transfer step was never fitted" + assert all(c["n_subjects"] > 0 for c in _TRANSFER_CAPTURE) + assert all(c["n_target"] > 0 for c in _TRANSFER_CAPTURE) + + +def test_cross_subject_calibration_leaves_plain_pipeline_unaffected(): + """A plain pipeline runs through calibration_size>0 unchanged (calib ignored).""" + pipe = make_pipeline(Covariances("oas"), CSP(8), LDA()) + ds = FakeDataset(["left_hand", "right_hand"], n_subjects=3, n_sessions=2, seed=9) + + evaluation = ev.CrossSubjectEvaluation( + paradigm=FakeImageryParadigm(), + datasets=[ds], + calibration_size=0.5, + overwrite=True, + n_jobs=1, + suffix="calibplain", + ) + results = evaluation.process(pipelines=OrderedDict([("P", pipe)])) + assert len(results) > 0 diff --git a/moabb/tests/test_splits.py b/moabb/tests/test_splits.py index 8b6464cd5..c176c016c 100644 --- a/moabb/tests/test_splits.py +++ b/moabb/tests/test_splits.py @@ -23,6 +23,7 @@ CrossSessionSplitter, CrossSubjectSplitter, LearningCurveSplitter, + TransferSplitter, WithinSessionSplitter, WithinSubjectSplitter, ) @@ -673,3 +674,45 @@ def test_cross_dataset_requires_group_column(data): splitter = CrossDatasetSplitter(group_column="does_not_exist") with pytest.raises(ValueError): list(splitter.split(y, metadata)) + + +@pytest.mark.parametrize("calibration_size", [0.0, 0.3, 1.0]) +def test_transfer_splitter_generalizes_across_splits(calibration_size, data): + """TransferSplitter adds calibration to subject/session/dataset splits alike, + consumed uniformly with ``train, *cal, test``.""" + _, y, metadata = data + dataset_meta = _metadata_with_dataset_column(metadata) + cases = { + "subject": (CrossSubjectSplitter(), metadata), + "session": (CrossSessionSplitter(), metadata), + "dataset": (CrossDatasetSplitter(), dataset_meta), + } + + for _name, (base, meta) in cases.items(): + wrapped = TransferSplitter(base, calibration_size=calibration_size) + base_folds = list(base.split(y, meta)) + xfer_folds = list(wrapped.split(y, meta)) + + assert len(xfer_folds) == len(base_folds) == wrapped.get_n_splits(meta) + + for (b_train, b_test), fold in zip(base_folds, xfer_folds): + train, *cal, test = fold # generic consumption + calib = cal[0] if cal else test[:0] + + assert np.array_equal(train, b_train) + assert np.intersect1d(train, b_test).size == 0 + + if calibration_size == 0.0: + assert calib.size == 0 + assert np.array_equal(test, b_test) + elif calibration_size == 1.0: + assert np.array_equal(calib, b_test) + assert np.array_equal(test, b_test) + else: + assert calib.size >= 1 and test.size >= 1 + assert np.array_equal(np.concatenate([calib, test]), b_test) + + +def test_transfer_splitter_invalid_calibration_size(): + with pytest.raises(ValueError): + TransferSplitter(CrossSubjectSplitter(), calibration_size=1.5) From e0f95de6be613122d03f363c2df92124a5aed420 Mon Sep 17 00:00:00 2001 From: Bru Date: Fri, 19 Jun 2026 22:44:21 +0200 Subject: [PATCH 02/15] refactor(evaluations): configure transfer via cv_kwargs, expose cv_class - calibration_size / calibration_labeled now ride cv_kwargs instead of bespoke CrossSubjectEvaluation __init__ params: read from self.cv_kwargs and stripped before the inner CV. No __init__ override. - Expose cv_class as a documented option (like WithinSessionEvaluation); it composes with calibration. - base.py / serial evaluate() read calibration_labeled from cv_kwargs. Numerically identical to CrossSubjectTargetAwareEvaluation.process() on BNCI2014_004 (max abs score diff 0.0) when the target-aware estimator covs the raw target and declares matching fit/transform metadata requests. Usage: CrossSubjectEvaluation(..., cv_kwargs={"calibration_size": 0.2}).process(pipelines) --- moabb/evaluations/base.py | 2 +- moabb/evaluations/evaluations.py | 58 ++++++++++++++++---------------- moabb/tests/test_evaluations.py | 32 ++++++++++++++++-- 3 files changed, 60 insertions(+), 32 deletions(-) diff --git a/moabb/evaluations/base.py b/moabb/evaluations/base.py index d6fff395d..0f69164a3 100644 --- a/moabb/evaluations/base.py +++ b/moabb/evaluations/base.py @@ -704,7 +704,7 @@ def _build_eval_config(self, param_grid): ), "score_per_session": self._score_per_session, "param_grid": None, # overridden per-task below if needed - "calibration_labeled": getattr(self, "calibration_labeled", False), + "calibration_labeled": self.cv_kwargs.get("calibration_labeled", False), } @staticmethod diff --git a/moabb/evaluations/evaluations.py b/moabb/evaluations/evaluations.py index 1ce78d4fc..281769a25 100644 --- a/moabb/evaluations/evaluations.py +++ b/moabb/evaluations/evaluations.py @@ -456,19 +456,24 @@ class CrossSubjectEvaluation(BaseEvaluation): n_splits : int or None Number of splits for cross-validation. If None, the number of splits is equal to the number of subjects. Defaults to ``None``. - calibration_size : float - Transfer-learning calibration. Fraction of each held-out target subject - made available for adaptation, in ``[0, 1]``. ``0.0`` (default) is the - ordinary cross-subject split. When ``> 0`` the splitter yields a - ``(train, calibration, test)`` fold and the calibration slice is routed - (raw) to the pipeline steps that request it via ``set_fit_request`` -- - ``subjects`` plus ``X_target_unlabeled`` (or ``X_target_labeled`` / - ``y_target_labeled`` if ``calibration_labeled``). Steps that request - nothing are unaffected. - calibration_labeled : bool - If True, the calibration slice is offered with labels - (``X_target_labeled`` / ``y_target_labeled``); otherwise unlabeled - (``X_target_unlabeled``). Defaults to ``False``. + cv_class : type or None + Cross-validation strategy used to hold out subjects (e.g. + ``LeaveOneGroupOut``, ``GroupShuffleSplit``, ``GroupKFold``). Defaults to + ``None`` (``LeaveOneGroupOut``, or ``GroupKFold`` when ``n_splits`` is set). + cv_kwargs : dict + Keyword arguments for ``cv_class``. Two extra keys enable transfer learning: + + - ``calibration_size`` (float, default ``0.0``): fraction of each held-out + target subject made available for adaptation, in ``[0, 1]``. ``0.0`` is + the ordinary cross-subject split; when ``> 0`` each fold becomes + ``(train, calibration, test)`` and the calibration slice is routed + (raw) to the pipeline steps that request it via ``set_fit_request`` -- + ``subjects`` plus ``X_target_unlabeled`` (or ``X_target_labeled`` / + ``y_target_labeled`` when ``calibration_labeled``). Steps that request + nothing are unaffected. + - ``calibration_labeled`` (bool, default ``False``): offer the + calibration slice with labels (``X_target_labeled`` / + ``y_target_labeled``) instead of unlabeled (``X_target_unlabeled``). Notes ----- @@ -480,23 +485,15 @@ class CrossSubjectEvaluation(BaseEvaluation): _score_per_session = True _needs_all_subjects = True - def __init__( - self, - *args, - calibration_size: float = 0.0, - calibration_labeled: bool = False, - **kwargs, - ): - super().__init__(*args, **kwargs) - self.calibration_size = calibration_size - self.calibration_labeled = calibration_labeled - def _create_splitter(self): """Create the CrossSubjectSplitter for parallel evaluation. - Wrapped in a :class:`~moabb.evaluations.splitters.TransferSplitter` when - ``calibration_size > 0`` so each fold yields a calibration slice. + ``cv_kwargs["calibration_size"] > 0`` wraps it in a + :class:`~moabb.evaluations.splitters.TransferSplitter`, so each fold + yields a ``(train, calibration, test)`` transfer split. """ + calibration_size = self.cv_kwargs.get("calibration_size", 0.0) + if self.n_splits is None: default_class = LeaveOneGroupOut default_kwargs = {} @@ -505,11 +502,14 @@ def _create_splitter(self): default_kwargs = {"n_splits": self.n_splits} cv_class, cv_kwargs = self._resolve_cv(default_class, default_kwargs) + # calibration_* configure the transfer wrapper, not the inner CV. + cv_kwargs.pop("calibration_size", None) + cv_kwargs.pop("calibration_labeled", None) splitter = CrossSubjectSplitter( cv_class=cv_class, random_state=self.random_state, **cv_kwargs ) - if self.calibration_size > 0: - return TransferSplitter(splitter, calibration_size=self.calibration_size) + if calibration_size > 0: + return TransferSplitter(splitter, calibration_size=calibration_size) return splitter # flake8: noqa: C901 @@ -588,7 +588,7 @@ def evaluate( calib_md = None if len(calib): - if self.calibration_labeled: + if self.cv_kwargs.get("calibration_labeled", False): calib_md = { "X_target_labeled": X[calib], "y_target_labeled": y[calib], diff --git a/moabb/tests/test_evaluations.py b/moabb/tests/test_evaluations.py index 31faac132..56d0fe394 100644 --- a/moabb/tests/test_evaluations.py +++ b/moabb/tests/test_evaluations.py @@ -1070,7 +1070,7 @@ def test_cross_subject_calibration_routes_to_estimator(): evaluation = ev.CrossSubjectEvaluation( paradigm=FakeImageryParadigm(), datasets=[ds], - calibration_size=0.5, + cv_kwargs={"calibration_size": 0.5}, overwrite=True, n_jobs=1, suffix="calibroute", @@ -1091,10 +1091,38 @@ def test_cross_subject_calibration_leaves_plain_pipeline_unaffected(): evaluation = ev.CrossSubjectEvaluation( paradigm=FakeImageryParadigm(), datasets=[ds], - calibration_size=0.5, + cv_kwargs={"calibration_size": 0.5}, overwrite=True, n_jobs=1, suffix="calibplain", ) results = evaluation.process(pipelines=OrderedDict([("P", pipe)])) assert len(results) > 0 + + +def test_cross_subject_calibration_with_custom_cv(): + """cv_class is exposed (like WithinSession) and composes with calibration.""" + from sklearn import config_context + from sklearn.model_selection import GroupShuffleSplit + + _TRANSFER_CAPTURE.clear() + with config_context(enable_metadata_routing=True): + step = _TransferRecorder().set_fit_request(subjects=True, X_target_unlabeled=True) + pipe = make_pipeline(Covariances("oas"), step, CSP(8), LDA()) + ds = FakeDataset(["left_hand", "right_hand"], n_subjects=5, n_sessions=2, seed=9) + + evaluation = ev.CrossSubjectEvaluation( + paradigm=FakeImageryParadigm(), + datasets=[ds], + cv_class=GroupShuffleSplit, + cv_kwargs={"calibration_size": 0.5, "n_splits": 2}, + random_state=0, + overwrite=True, + n_jobs=1, + suffix="calibcv", + ) + results = evaluation.process(pipelines=OrderedDict([("T", pipe)])) + + assert len(results) > 0 + assert _TRANSFER_CAPTURE, "transfer step was never fitted" + assert all(c["n_subjects"] > 0 and c["n_target"] > 0 for c in _TRANSFER_CAPTURE) From 08cfbff3d027f972fa3ad50a59d6909a331a5b0d Mon Sep 17 00:00:00 2001 From: Bru Date: Fri, 19 Jun 2026 23:01:54 +0200 Subject: [PATCH 03/15] refactor(evaluations): make calibration native to CrossSubjectSplitter Move the transfer calibration into the splitter so CrossSubjectEvaluation's _create_splitter is the plain original (no .get / .pop / wrapper). - CrossSubjectSplitter gains a calibration_size param: yields (train, calib, test) when > 0, otherwise the usual (train, test). Removes TransferSplitter. - _resolve_cv now always merges self.cv_kwargs over the defaults (a latent fix), so calibration_size flows via cv_kwargs with the default cv_class too. - Drop calibration_labeled: _evaluate_fold offers all transfer kwargs (subjects, X_target_unlabeled, X_target_labeled, y_target_labeled) and metadata routing (consumes) keeps only what the estimator requested, so the estimator's set_fit_request decides labeled vs unlabeled. Numerically identical to CrossSubjectTargetAwareEvaluation.process() on BNCI2014_004 (max abs score diff 0.0). --- moabb/evaluations/__init__.py | 1 - moabb/evaluations/base.py | 33 +++++----- moabb/evaluations/evaluations.py | 49 +++++---------- moabb/evaluations/splitters.py | 102 ++++++++----------------------- moabb/tests/test_splits.py | 62 ++++++++----------- 5 files changed, 88 insertions(+), 159 deletions(-) diff --git a/moabb/evaluations/__init__.py b/moabb/evaluations/__init__.py index d75de227a..ee17fa525 100644 --- a/moabb/evaluations/__init__.py +++ b/moabb/evaluations/__init__.py @@ -15,7 +15,6 @@ CrossSessionSplitter, CrossSubjectSplitter, LearningCurveSplitter, - TransferSplitter, WithinSessionSplitter, WithinSubjectSplitter, ) diff --git a/moabb/evaluations/base.py b/moabb/evaluations/base.py index 0f69164a3..c11c4009c 100644 --- a/moabb/evaluations/base.py +++ b/moabb/evaluations/base.py @@ -196,15 +196,18 @@ def _evaluate_fold( tracker = emissions_obj.create_tracker() tracker.start() - # Optional transfer-learning calibration slice (raw), routed to the steps - # that request it. Empty for ordinary evaluations -> plain fit. + # Optional transfer-learning calibration slice (raw). Offer it under every + # name; _route_transfer_metadata keeps only what the estimator requests, so + # the estimator's set_fit_request decides labeled vs unlabeled. Empty for + # ordinary evaluations -> plain fit. calib = None if calib_idx is not None and len(calib_idx): - if config.get("calibration_labeled", False): - y_calib = y[calib_idx] if mne_labels else le.transform(y[calib_idx]) - calib = {"X_target_labeled": X[calib_idx], "y_target_labeled": y_calib} - else: - calib = {"X_target_unlabeled": X[calib_idx]} + y_calib = y[calib_idx] if mne_labels else le.transform(y[calib_idx]) + calib = { + "X_target_unlabeled": X[calib_idx], + "X_target_labeled": X[calib_idx], + "y_target_labeled": y_calib, + } subjects_train = metadata["subject"].to_numpy()[train_idx] fit_params = _route_transfer_metadata(cvclf, subjects_train, calib) @@ -498,13 +501,14 @@ def __init__( ) def _resolve_cv(self, default_class, default_kwargs=None): - """Resolve the cross-validation class and kwargs for a splitter.""" - if self.cv_class is None: - cv_class = default_class - cv_kwargs = {} if default_kwargs is None else dict(default_kwargs) - else: - cv_class = self.cv_class - cv_kwargs = dict(self.cv_kwargs) + """Resolve the cross-validation class and kwargs for a splitter. + + ``self.cv_kwargs`` always overrides the defaults, whether or not a custom + ``cv_class`` is set -- so splitter options (e.g. ``calibration_size``) + passed via ``cv_kwargs`` are honored with the default ``cv_class`` too. + """ + cv_class = default_class if self.cv_class is None else self.cv_class + cv_kwargs = {**(default_kwargs or {}), **self.cv_kwargs} return cv_class, cv_kwargs def _load_data( @@ -704,7 +708,6 @@ def _build_eval_config(self, param_grid): ), "score_per_session": self._score_per_session, "param_grid": None, # overridden per-task below if needed - "calibration_labeled": self.cv_kwargs.get("calibration_labeled", False), } @staticmethod diff --git a/moabb/evaluations/evaluations.py b/moabb/evaluations/evaluations.py index 281769a25..0463646fb 100644 --- a/moabb/evaluations/evaluations.py +++ b/moabb/evaluations/evaluations.py @@ -11,7 +11,6 @@ from moabb.evaluations.splitters import ( CrossSessionSplitter, CrossSubjectSplitter, - TransferSplitter, WithinSessionSplitter, WithinSubjectSplitter, ) @@ -461,19 +460,13 @@ class CrossSubjectEvaluation(BaseEvaluation): ``LeaveOneGroupOut``, ``GroupShuffleSplit``, ``GroupKFold``). Defaults to ``None`` (``LeaveOneGroupOut``, or ``GroupKFold`` when ``n_splits`` is set). cv_kwargs : dict - Keyword arguments for ``cv_class``. Two extra keys enable transfer learning: - - - ``calibration_size`` (float, default ``0.0``): fraction of each held-out - target subject made available for adaptation, in ``[0, 1]``. ``0.0`` is - the ordinary cross-subject split; when ``> 0`` each fold becomes - ``(train, calibration, test)`` and the calibration slice is routed - (raw) to the pipeline steps that request it via ``set_fit_request`` -- - ``subjects`` plus ``X_target_unlabeled`` (or ``X_target_labeled`` / - ``y_target_labeled`` when ``calibration_labeled``). Steps that request - nothing are unaffected. - - ``calibration_labeled`` (bool, default ``False``): offer the - calibration slice with labels (``X_target_labeled`` / - ``y_target_labeled``) instead of unlabeled (``X_target_unlabeled``). + Keyword arguments for ``cv_class``. ``calibration_size`` (float in + ``[0, 1]``, default ``0.0``) enables transfer learning: when ``> 0`` each + fold becomes ``(train, calibration, test)`` and the held-out calibration + slice is routed (raw) to the pipeline steps that request it via + ``set_fit_request`` -- ``subjects`` and ``X_target_unlabeled`` / + ``X_target_labeled`` / ``y_target_labeled`` (the estimator's requests + decide labeled vs unlabeled). Steps that request nothing are unaffected. Notes ----- @@ -488,12 +481,10 @@ class CrossSubjectEvaluation(BaseEvaluation): def _create_splitter(self): """Create the CrossSubjectSplitter for parallel evaluation. - ``cv_kwargs["calibration_size"] > 0`` wraps it in a - :class:`~moabb.evaluations.splitters.TransferSplitter`, so each fold - yields a ``(train, calibration, test)`` transfer split. + ``calibration_size`` passed via ``cv_kwargs`` turns each fold into a + ``(train, calibration, test)`` transfer split (see + :class:`~moabb.evaluations.splitters.CrossSubjectSplitter`). """ - calibration_size = self.cv_kwargs.get("calibration_size", 0.0) - if self.n_splits is None: default_class = LeaveOneGroupOut default_kwargs = {} @@ -502,15 +493,9 @@ def _create_splitter(self): default_kwargs = {"n_splits": self.n_splits} cv_class, cv_kwargs = self._resolve_cv(default_class, default_kwargs) - # calibration_* configure the transfer wrapper, not the inner CV. - cv_kwargs.pop("calibration_size", None) - cv_kwargs.pop("calibration_labeled", None) - splitter = CrossSubjectSplitter( + return CrossSubjectSplitter( cv_class=cv_class, random_state=self.random_state, **cv_kwargs ) - if calibration_size > 0: - return TransferSplitter(splitter, calibration_size=calibration_size) - return splitter # flake8: noqa: C901 def evaluate( @@ -588,13 +573,11 @@ def evaluate( calib_md = None if len(calib): - if self.cv_kwargs.get("calibration_labeled", False): - calib_md = { - "X_target_labeled": X[calib], - "y_target_labeled": y[calib], - } - else: - calib_md = {"X_target_unlabeled": X[calib]} + calib_md = { + "X_target_unlabeled": X[calib], + "X_target_labeled": X[calib], + "y_target_labeled": y[calib], + } fit_params = _route_transfer_metadata(cvclf, groups[train], calib_md) duration, emissions, task_name = self._fit_cv( diff --git a/moabb/evaluations/splitters.py b/moabb/evaluations/splitters.py index 5fc253539..35931ed9a 100644 --- a/moabb/evaluations/splitters.py +++ b/moabb/evaluations/splitters.py @@ -507,6 +507,13 @@ class CrossSubjectSplitter(BaseCrossValidator): Cross-validation strategy for splitting the subjects between train and test sets. By default, use LeaveOneGroupOut, which keeps one subject as a test. Defaults to ``LeaveOneGroupOut``. + calibration_size : float + Transfer-learning calibration, in ``[0, 1]``. ``0.0`` (default) is the + ordinary cross-subject split, yielding ``(train, test)``. When ``> 0``, + the first ``calibration_size`` fraction of each held-out subject's trials + is set aside for adaptation and the splitter yields + ``(train, calibration, test)`` instead; ``1.0`` is transductive + (calibration equals test). random_state : int or None Controls the randomness of the cross-validation. Pass an int for reproducible output across multiple calls. @@ -519,6 +526,9 @@ class CrossSubjectSplitter(BaseCrossValidator): train : ndarray The training set indices for that split. + calibration : ndarray + The held-out calibration indices. Only yielded when ``calibration_size > 0``. + test : ndarray The testing set indices for that split. """ @@ -526,10 +536,16 @@ class CrossSubjectSplitter(BaseCrossValidator): def __init__( self, cv_class: type[BaseCrossValidator] = LeaveOneGroupOut, + calibration_size: float = 0.0, random_state: int = None, **cv_kwargs, ): + if not 0.0 <= calibration_size <= 1.0: + raise ValueError( + f"calibration_size must be in [0, 1]. Got {calibration_size!r}." + ) self.cv_class = cv_class + self.calibration_size = calibration_size self.cv_kwargs = cv_kwargs self._cv_kwargs = dict(**cv_kwargs) @@ -584,7 +600,17 @@ def split(self, y, metadata): for train_session_idx, test_session_idx in splitter.split(**split_kwargs): self._last_split_metadata = _splitter_metadata(splitter) - yield all_index[train_session_idx], all_index[test_session_idx] + train_idx = all_index[train_session_idx] + group_idx = all_index[test_session_idx] + if self.calibration_size > 0: + # Transfer learning: carve a calibration slice off the held-out + # subject -> (train, calibration, test). + calib_idx, test_idx = _split_target_fraction( + group_idx, self.calibration_size + ) + yield train_idx, calib_idx, test_idx + else: + yield train_idx, group_idx def get_metadata(self): """Return metadata for the most recent split.""" @@ -618,80 +644,6 @@ def _split_target_fraction( return target_idx[:n_calib], target_idx[n_calib:] -class TransferSplitter(BaseCrossValidator): - """Add an optional calibration/adaptation slice to any leave-one-group-out split. - - This is the generic transfer-learning split. It wraps a base moabb splitter - (:class:`CrossSubjectSplitter`, :class:`CrossSessionSplitter`, - :class:`CrossDatasetSplitter`, ...) and, for each ``(train, test)`` fold the - base produces, carves the first ``calibration_size`` fraction off the held-out - *test* group as a calibration slice, yielding **three** index arrays:: - - train everything the base splitter put in train (the sources) - calibration the first ``calibration_size`` fraction of the held-out - group -- the adaptation/calibration slice (may be empty) - test the remaining held-out trials, to be scored - - The same mechanism therefore expresses subject-transfer, session-transfer and - dataset-transfer. A split only decides *which trials go where*; whether the - calibration slice is consumed with or without labels, and how the test block is - predicted, are estimator/scoring concerns and stay out of scope here. - - Consume it uniformly -- the ``*cal`` absorbs the (optional) calibration slice, - so the same loop also accepts a plain 2-tuple splitter:: - - for train, *cal, test in splitter.split(y, metadata): - calib = cal[0] if cal else test[:0] - - Parameters - ---------- - base_splitter : BaseCrossValidator - A moabb splitter exposing ``split(y, metadata) -> (train, test)`` and - ``get_n_splits(metadata)`` (typically a leave-one-group-out splitter). - calibration_size : float - Fraction of each held-out group reserved for calibration/adaptation, in - ``[0, 1]``. ``0.0`` keeps the base geometry with an empty calibration slice; - ``1.0`` is fully transductive (calibration and test are identical). For any - value strictly between 0 and 1, at least one trial is kept on each side. - Defaults to ``0.0``. - - Yields - ------ - train : ndarray - Source training indices. - calibration : ndarray - Held-out calibration/adaptation indices (empty when ``calibration_size`` is - ``0.0``; equal to ``test`` when ``1.0``). - test : ndarray - Held-out evaluation indices. - """ - - def __init__(self, base_splitter: BaseCrossValidator, calibration_size: float = 0.0): - if not 0.0 <= calibration_size <= 1.0: - raise ValueError( - f"calibration_size must be in [0, 1]. Got {calibration_size!r}." - ) - self.base_splitter = base_splitter - self.calibration_size = calibration_size - - def get_n_splits(self, metadata): - """Return the number of splits (delegated to the base splitter).""" - return self.base_splitter.get_n_splits(metadata) - - def split(self, y, metadata): - # ``*_`` tolerates a base that yields more than (train, test); the held-out - # group to split is always the last element. - for train_idx, *_, group_idx in self.base_splitter.split(y, metadata): - calibration_idx, test_idx = _split_target_fraction( - group_idx, self.calibration_size - ) - yield train_idx, calibration_idx, test_idx - - def get_metadata(self): - """Return metadata for the most recent split (from the base splitter).""" - return _splitter_metadata(self.base_splitter) - - class CrossDatasetSplitter(BaseCrossValidator): """Data splitter for leave-dataset-out style evaluation. diff --git a/moabb/tests/test_splits.py b/moabb/tests/test_splits.py index c176c016c..d78c8751f 100644 --- a/moabb/tests/test_splits.py +++ b/moabb/tests/test_splits.py @@ -23,7 +23,6 @@ CrossSessionSplitter, CrossSubjectSplitter, LearningCurveSplitter, - TransferSplitter, WithinSessionSplitter, WithinSubjectSplitter, ) @@ -677,42 +676,35 @@ def test_cross_dataset_requires_group_column(data): @pytest.mark.parametrize("calibration_size", [0.0, 0.3, 1.0]) -def test_transfer_splitter_generalizes_across_splits(calibration_size, data): - """TransferSplitter adds calibration to subject/session/dataset splits alike, +def test_cross_subject_calibration(calibration_size, data): + """CrossSubjectSplitter carves a calibration slice off the held-out subject, consumed uniformly with ``train, *cal, test``.""" _, y, metadata = data - dataset_meta = _metadata_with_dataset_column(metadata) - cases = { - "subject": (CrossSubjectSplitter(), metadata), - "session": (CrossSessionSplitter(), metadata), - "dataset": (CrossDatasetSplitter(), dataset_meta), - } - - for _name, (base, meta) in cases.items(): - wrapped = TransferSplitter(base, calibration_size=calibration_size) - base_folds = list(base.split(y, meta)) - xfer_folds = list(wrapped.split(y, meta)) - - assert len(xfer_folds) == len(base_folds) == wrapped.get_n_splits(meta) - - for (b_train, b_test), fold in zip(base_folds, xfer_folds): - train, *cal, test = fold # generic consumption - calib = cal[0] if cal else test[:0] - - assert np.array_equal(train, b_train) - assert np.intersect1d(train, b_test).size == 0 - - if calibration_size == 0.0: - assert calib.size == 0 - assert np.array_equal(test, b_test) - elif calibration_size == 1.0: - assert np.array_equal(calib, b_test) - assert np.array_equal(test, b_test) - else: - assert calib.size >= 1 and test.size >= 1 - assert np.array_equal(np.concatenate([calib, test]), b_test) + base = CrossSubjectSplitter() + cal_split = CrossSubjectSplitter(calibration_size=calibration_size) + + base_folds = list(base.split(y, metadata)) + cal_folds = list(cal_split.split(y, metadata)) + assert len(cal_folds) == len(base_folds) == cal_split.get_n_splits(metadata) + + for (b_train, b_test), fold in zip(base_folds, cal_folds): + train, *cal, test = fold # generic consumption: 2- or 3-tuple + calib = cal[0] if cal else test[:0] + + assert np.array_equal(train, b_train) + assert np.intersect1d(train, b_test).size == 0 + + if calibration_size == 0.0: + assert calib.size == 0 + assert np.array_equal(test, b_test) + elif calibration_size == 1.0: + assert np.array_equal(calib, b_test) + assert np.array_equal(test, b_test) + else: + assert calib.size >= 1 and test.size >= 1 + assert np.array_equal(np.concatenate([calib, test]), b_test) -def test_transfer_splitter_invalid_calibration_size(): +def test_cross_subject_calibration_invalid_size(): with pytest.raises(ValueError): - TransferSplitter(CrossSubjectSplitter(), calibration_size=1.5) + CrossSubjectSplitter(calibration_size=1.5) From 56b1a6a7fd8efde7cf36f782b819dcb9fa05f27e Mon Sep 17 00:00:00 2001 From: Anton Date: Mon, 22 Jun 2026 17:03:40 +0200 Subject: [PATCH 04/15] - Added named cross-subject protocols via `CsMode`. - Added separate prediction modes: blockwise vs trialwise. - Enabled strict one-trial-at-a-time testing. - Added unlabeled target adaptation with 20%, 50%, or 100% target data. - Added labeled target calibration with 20% or 50% target data. - Fixed unsafe 100% labeled target usage by forbidding it. - Ensured labels are routed only when the protocol allows labeled calibration. - Ensured unlabeled protocols never pass target labels. - Updated the splitter to produce optional `(train, calibration, test)` folds. - Made protocols easier to report clearly in benchmarks and papers. --- moabb/evaluations/base.py | 167 ++++++++++++++++++++++++---- moabb/evaluations/evaluations.py | 152 ++++++++++++++++++++------ moabb/evaluations/protocols.py | 83 ++++++++++++++ moabb/evaluations/splitters.py | 182 +++++++++++++++++++++---------- 4 files changed, 468 insertions(+), 116 deletions(-) create mode 100644 moabb/evaluations/protocols.py diff --git a/moabb/evaluations/base.py b/moabb/evaluations/base.py index c11c4009c..9ae3e35d2 100644 --- a/moabb/evaluations/base.py +++ b/moabb/evaluations/base.py @@ -13,7 +13,7 @@ import pandas as pd from joblib import Parallel, delayed from sklearn import config_context -from sklearn.base import BaseEstimator, clone +from sklearn.base import BaseEstimator, ClassifierMixin, clone from sklearn.model_selection import StratifiedKFold from sklearn.preprocessing import LabelEncoder from sklearn.utils.metadata_routing import get_routing_for_object @@ -40,30 +40,138 @@ ) from moabb.paradigms.base import BaseParadigm from moabb.utils import verbose - +from moabb.evaluations.protocols import PredictMode search_methods, optuna_available = check_search_available() log = logging.getLogger(__name__) -def _route_transfer_metadata(estimator, subjects, calib=None): - """Keep only the transfer metadata an estimator requests at ``fit``. +def _route_transfer_metadata( + estimator, + subjects, + calib=None, + calibration_labeled=False, +): + """Keep only protocol-allowed transfer metadata requested at ``fit``. + + ``subjects`` is the per-trial source-subject array. + + ``calib`` is an optional dict with raw calibration data: + ``{"X": X[calib_idx], "y": y_calib}``. + + If ``calibration_labeled`` is False, calibration can only be routed as + ``X_target_unlabeled``. + + If ``calibration_labeled`` is True, calibration can be routed as + ``X_target_labeled`` and ``y_target_labeled``. - ``subjects`` is the per-trial source-subject array; ``calib`` is an optional - dict carrying the (raw) calibration slice, e.g. - ``{"X_target_unlabeled": X[calib_idx]}``. Steps opt in via - ``set_fit_request(...)``; plain pipelines request nothing, so this returns - ``{}`` and the fit is unchanged. The framework passes the slice raw -- the - estimator owns the target representation. + Steps opt in via ``set_fit_request(...)``. Plain pipelines request nothing, + so this returns ``{}`` and the fit is unchanged. """ candidate = {"subjects": subjects} - if calib: - candidate.update(calib) + + if calib is not None: + if calibration_labeled: + candidate["X_target_labeled"] = calib["X"] + candidate["y_target_labeled"] = calib["y"] + else: + candidate["X_target_unlabeled"] = calib["X"] + with config_context(enable_metadata_routing=True): kept = get_routing_for_object(estimator).consumes("fit", set(candidate)) + return {k: v for k, v in candidate.items() if k in kept} +class TrialwisePredictWrapper(ClassifierMixin, BaseEstimator): + """Wrap a fitted estimator and force one-trial-at-a-time prediction.""" + + _estimator_type = "classifier" + + def __init__(self, fitted_estimator): + self.fitted_estimator = fitted_estimator + self.classes_ = self._get_classes(fitted_estimator) + + def fit(self, X, y=None): + raise RuntimeError( + "TrialwisePredictWrapper wraps an already fitted estimator." + ) + + def predict(self, X): + return np.asarray( + [ + self.fitted_estimator.predict(X[i : i + 1])[0] + for i in range(len(X)) + ] + ) + + def predict_proba(self, X): + if not hasattr(self.fitted_estimator, "predict_proba"): + raise AttributeError( + "Wrapped estimator does not provide predict_proba." + ) + + return np.vstack( + [ + np.asarray( + self.fitted_estimator.predict_proba(X[i : i + 1]) + )[0] + for i in range(len(X)) + ] + ) + + def decision_function(self, X): + if not hasattr(self.fitted_estimator, "decision_function"): + raise AttributeError( + "Wrapped estimator does not provide decision_function." + ) + + rows = [] + + for i in range(len(X)): + out = np.asarray( + self.fitted_estimator.decision_function(X[i : i + 1]) + ) + + if out.ndim == 0: + rows.append(out.item()) + elif out.shape[0] == 1: + rows.append(out[0]) + else: + rows.append(out) + + out = np.asarray(rows) + + if out.ndim == 2 and out.shape[1] == 1: + return out.ravel() + + return out + + @staticmethod + def _get_classes(estimator): + if hasattr(estimator, "classes_"): + return estimator.classes_ + + if hasattr(estimator, "steps"): + final_estimator = estimator.steps[-1][1] + if hasattr(final_estimator, "classes_"): + return final_estimator.classes_ + + raise AttributeError( + "Wrapped estimator does not expose classes_." + ) + + +def _wrap_predict_mode(estimator, predict_mode): + predict_mode = PredictMode(predict_mode) + + if predict_mode == PredictMode.BLOCKWISE: + return estimator + + if predict_mode == PredictMode.TRIALWISE: + return TrialwisePredictWrapper(estimator) + + raise ValueError(f"Unknown predict_mode={predict_mode!r}.") # Making the optuna soft dependency @@ -162,11 +270,15 @@ def _evaluate_fold( score_per_session = config["score_per_session"] mne_labels = config["mne_labels"] codecarbon_config = config["codecarbon_config"] + predict_mode = config.get("predict_mode", PredictMode.BLOCKWISE.value) # Label encode per fold (matching old per-session/per-subject scoping) if not mne_labels: le = LabelEncoder() - combined_y = np.concatenate([y[train_idx], y[test_idx]]) + if calib_idx is not None and len(calib_idx): + combined_y = np.concatenate([y[train_idx], y[test_idx], y[calib_idx]]) + else: + combined_y = np.concatenate([y[train_idx], y[test_idx]]) le.fit(combined_y) y_train = le.transform(y[train_idx]) y_test = le.transform(y[test_idx]) @@ -196,20 +308,27 @@ def _evaluate_fold( tracker = emissions_obj.create_tracker() tracker.start() - # Optional transfer-learning calibration slice (raw). Offer it under every - # name; _route_transfer_metadata keeps only what the estimator requests, so - # the estimator's set_fit_request decides labeled vs unlabeled. Empty for - # ordinary evaluations -> plain fit. + # Optional transfer-learning calibration slice (raw). The protocol decides + # whether it is offered as unlabeled or labeled target metadata. calib = None if calib_idx is not None and len(calib_idx): y_calib = y[calib_idx] if mne_labels else le.transform(y[calib_idx]) calib = { - "X_target_unlabeled": X[calib_idx], - "X_target_labeled": X[calib_idx], - "y_target_labeled": y_calib, + "X": X[calib_idx], + "y": y_calib, } + + calibration_labeled = False + if split_metadata is not None: + calibration_labeled = bool(split_metadata.get("calibration_labeled", False)) + subjects_train = metadata["subject"].to_numpy()[train_idx] - fit_params = _route_transfer_metadata(cvclf, subjects_train, calib) + fit_params = _route_transfer_metadata( + cvclf, + subjects_train, + calib=calib, + calibration_labeled=calibration_labeled, + ) # Fit model task_name = None @@ -244,7 +363,8 @@ def _evaluate_fold( ) _save_model_cv(model=cvclf, save_path=model_save_path, cv_index=str(cv_ind)) - scorer = _create_scorer(cvclf, scoring) + score_estimator = _wrap_predict_mode(cvclf, predict_mode) + scorer = _create_scorer(score_estimator, scoring) # Build score groups: per-session or full test set if score_per_session: @@ -260,7 +380,7 @@ def _evaluate_fold( for group_idx, group_y, group_session in score_groups: is_error = False try: - score = scorer(cvclf, X[group_idx], group_y) + score = scorer(score_estimator, X[group_idx], group_y) except ValueError as err: if error_score == "raise": raise err @@ -707,6 +827,7 @@ def _build_eval_config(self, param_grid): self.emissions.codecarbon_config if _carbonfootprint else None ), "score_per_session": self._score_per_session, + "predict_mode": getattr(self, "predict_mode", PredictMode.BLOCKWISE.value), "param_grid": None, # overridden per-task below if needed } diff --git a/moabb/evaluations/evaluations.py b/moabb/evaluations/evaluations.py index 0463646fb..940dab8bd 100644 --- a/moabb/evaluations/evaluations.py +++ b/moabb/evaluations/evaluations.py @@ -7,13 +7,22 @@ from sklearn.preprocessing import LabelEncoder from tqdm import tqdm -from moabb.evaluations.base import BaseEvaluation, _route_transfer_metadata +from moabb.evaluations.base import ( + BaseEvaluation, + _route_transfer_metadata, + _wrap_predict_mode, +) from moabb.evaluations.splitters import ( CrossSessionSplitter, CrossSubjectSplitter, WithinSessionSplitter, WithinSubjectSplitter, ) +from moabb.evaluations.protocols import ( + PredictMode, + resolve_cs_mode, + validate_transfer_protocol, +) if TYPE_CHECKING: @@ -464,9 +473,9 @@ class CrossSubjectEvaluation(BaseEvaluation): ``[0, 1]``, default ``0.0``) enables transfer learning: when ``> 0`` each fold becomes ``(train, calibration, test)`` and the held-out calibration slice is routed (raw) to the pipeline steps that request it via - ``set_fit_request`` -- ``subjects`` and ``X_target_unlabeled`` / - ``X_target_labeled`` / ``y_target_labeled`` (the estimator's requests - decide labeled vs unlabeled). Steps that request nothing are unaffected. + ``set_fit_request`` -- ``subjects`` plus either ``X_target_unlabeled`` + or ``X_target_labeled`` / ``y_target_labeled``, depending on + ``calibration_labeled``. Notes ----- @@ -478,12 +487,47 @@ class CrossSubjectEvaluation(BaseEvaluation): _score_per_session = True _needs_all_subjects = True + def __init__( + self, + *args, + cs_mode=None, + predict_mode=PredictMode.BLOCKWISE.value, + **kwargs, + ): + cv_kwargs = dict(kwargs.get("cv_kwargs") or {}) + + if cs_mode is not None: + if "calibration_size" in cv_kwargs or "calibration_labeled" in cv_kwargs: + raise ValueError( + "Pass either cs_mode or calibration_size/calibration_labeled, " + "not both." + ) + + if PredictMode(predict_mode) != PredictMode.BLOCKWISE: + raise ValueError("Pass either cs_mode or predict_mode, not both.") + + params = resolve_cs_mode(cs_mode) + cv_kwargs["calibration_size"] = params["calibration_size"] + cv_kwargs["calibration_labeled"] = params["calibration_labeled"] + predict_mode = params["predict_mode"] + + validate_transfer_protocol( + cv_kwargs.get("calibration_size", 0.0), + cv_kwargs.get("calibration_labeled", False), + ) + + self.predict_mode = PredictMode(predict_mode).value + kwargs["cv_kwargs"] = cv_kwargs + + super().__init__(*args, **kwargs) + def _create_splitter(self): """Create the CrossSubjectSplitter for parallel evaluation. - ``calibration_size`` passed via ``cv_kwargs`` turns each fold into a - ``(train, calibration, test)`` transfer split (see - :class:`~moabb.evaluations.splitters.CrossSubjectSplitter`). + ``calibration_size`` and ``calibration_labeled`` passed via + ``cv_kwargs`` turn each fold into a transfer split: + + ``(train, calibration, test)``. """ if self.n_splits is None: default_class = LeaveOneGroupOut @@ -494,7 +538,9 @@ def _create_splitter(self): cv_class, cv_kwargs = self._resolve_cv(default_class, default_kwargs) return CrossSubjectSplitter( - cv_class=cv_class, random_state=self.random_state, **cv_kwargs + cv_class=cv_class, + random_state=self.random_state, + **cv_kwargs, ) # flake8: noqa: C901 @@ -509,48 +555,54 @@ def evaluate( if not self.is_valid(dataset): reason = self._get_incompatibility_reason(dataset) raise AssertionError( - f"Dataset '{dataset.code}' is not appropriate for {self.__class__.__name__}: {reason}" + f"Dataset '{dataset.code}' is not appropriate for " + f"{self.__class__.__name__}: {reason}" ) - # this is a bit awkward, but we need to check if at least one pipe - # have to be run before loading the data. If at least one pipeline - # need to be run, we have to load all the data. - # we might need a better granularity, if we query the DB + run_pipes = {} for subject in dataset.subject_list: run_pipes.update( self.results.not_yet_computed( - pipelines, dataset, subject, process_pipeline + pipelines, + dataset, + subject, + process_pipeline, ) ) + if len(run_pipes) == 0: return X, y, metadata = self._load_data( - dataset, run_pipes, process_pipeline, postprocess_pipeline + dataset, + run_pipes, + process_pipeline, + postprocess_pipeline, ) + le = LabelEncoder() y = y if self.mne_labels else le.fit_transform(y) - # extract metadata groups = metadata.subject.values sessions = metadata.session.values n_subjects = len(dataset.subject_list) nchan = self._get_nchan(X) - # perform leave one subject out CV self.cv = self._create_splitter() + if self.n_splits is not None and self.cv_class is None: n_subjects = self.n_splits - inner_cv = StratifiedKFold(3, shuffle=True, random_state=self.random_state) + inner_cv = StratifiedKFold( + 3, + shuffle=True, + random_state=self.random_state, + ) if _carbonfootprint: - # Initialise CodeCarbon per cross-validation tracker = self.emissions.create_tracker() tracker.start() - # Progressbar at subject level - # ``*cal`` absorbs the optional calibration slice from a transfer split. for cv_ind, (train, *cal, test) in enumerate( tqdm( self.cv.split(y, metadata), @@ -560,25 +612,48 @@ def evaluate( ): calib = cal[0] if cal else train[:0] subject = groups[test[0]] - # now we can check if this subject has results + + split_metadata = None + if hasattr(self.cv, "get_metadata"): + split_metadata = self.cv.get_metadata() + if split_metadata is not None: + split_metadata = dict(split_metadata) + run_pipes = self.results.not_yet_computed( - pipelines, dataset, subject, process_pipeline + pipelines, + dataset, + subject, + process_pipeline, ) - # iterate over pipelines + for name, clf in run_pipes.items(): clf = self._grid_search( - param_grid=param_grid, name=name, grid_clf=clf, inner_cv=inner_cv + param_grid=param_grid, + name=name, + grid_clf=clf, + inner_cv=inner_cv, ) cvclf = clone(clf) calib_md = None if len(calib): calib_md = { - "X_target_unlabeled": X[calib], - "X_target_labeled": X[calib], - "y_target_labeled": y[calib], + "X": X[calib], + "y": y[calib], } - fit_params = _route_transfer_metadata(cvclf, groups[train], calib_md) + + calibration_labeled = False + if split_metadata is not None: + calibration_labeled = bool( + split_metadata.get("calibration_labeled", False) + ) + + fit_params = _route_transfer_metadata( + cvclf, + groups[train], + calib=calib_md, + calibration_labeled=calibration_labeled, + ) duration, emissions, task_name = self._fit_cv( cvclf, @@ -587,14 +662,20 @@ def evaluate( tracker if _carbonfootprint else None, fit_params=fit_params, ) + self._maybe_save_model_cv( - cvclf, dataset, subject, "", name, cv_ind, eval_type="CrossSubject" + cvclf, + dataset, + subject, + "", + name, + cv_ind, + eval_type="CrossSubject", ) - # Create scorer once per pipeline - scorer = _create_scorer(cvclf, self.paradigm.scoring) + score_estimator = _wrap_predict_mode(cvclf, self.predict_mode) + scorer = _create_scorer(score_estimator, self.paradigm.scoring) - # Evaluate on each session for session in np.unique(sessions[test]): ix = sessions[test] == session @@ -607,9 +688,10 @@ def evaluate( nchan, duration, scorer, - cvclf, + score_estimator, X[test[ix]], y[test[ix]], + split_metadata=split_metadata, ) if _carbonfootprint: @@ -626,11 +708,13 @@ def is_valid(self, dataset: "BaseDataset") -> bool: def _get_incompatibility_reason(self, dataset): """Get specific reason for dataset incompatibility.""" n_subjects = len(dataset.subject_list) + if n_subjects <= 1: return ( f"dataset has only {n_subjects} subject(s), " f"but {self.__class__.__name__} requires at least 2 subjects" ) + return "requirements not met" diff --git a/moabb/evaluations/protocols.py b/moabb/evaluations/protocols.py new file mode 100644 index 000000000..1f9218748 --- /dev/null +++ b/moabb/evaluations/protocols.py @@ -0,0 +1,83 @@ +from enum import Enum + + +class PredictMode(str, Enum): + BLOCKWISE = "blockwise" + TRIALWISE = "trialwise" + + +class CsMode(str, Enum): + HOS_SOURCE_ONLY_BLOCKWISE = "hos_source_only_blockwise" + HOS_SOURCE_ONLY_TRIALWISE = "hos_source_only_trialwise" + HOS_UNLABELED_20P = "hos_unlabeled_20p" + HOS_UNLABELED_50P = "hos_unlabeled_50p" + HOS_UNLABELED_100P = "hos_unlabeled_100p" + HOS_LABELED_20P = "hos_labeled_20p" + HOS_LABELED_50P = "hos_labeled_50p" + + +_CS_MODE_MAP = { + CsMode.HOS_SOURCE_ONLY_BLOCKWISE: dict( + calibration_size=0.0, + calibration_labeled=False, + predict_mode=PredictMode.BLOCKWISE.value, + ), + CsMode.HOS_SOURCE_ONLY_TRIALWISE: dict( + calibration_size=0.0, + calibration_labeled=False, + predict_mode=PredictMode.TRIALWISE.value, + ), + CsMode.HOS_UNLABELED_20P: dict( + calibration_size=0.2, + calibration_labeled=False, + predict_mode=PredictMode.BLOCKWISE.value, + ), + CsMode.HOS_UNLABELED_50P: dict( + calibration_size=0.5, + calibration_labeled=False, + predict_mode=PredictMode.BLOCKWISE.value, + ), + CsMode.HOS_UNLABELED_100P: dict( + calibration_size=1.0, + calibration_labeled=False, + predict_mode=PredictMode.BLOCKWISE.value, + ), + CsMode.HOS_LABELED_20P: dict( + calibration_size=0.2, + calibration_labeled=True, + predict_mode=PredictMode.BLOCKWISE.value, + ), + CsMode.HOS_LABELED_50P: dict( + calibration_size=0.5, + calibration_labeled=True, + predict_mode=PredictMode.BLOCKWISE.value, + ), +} + + +def validate_transfer_protocol(calibration_size, calibration_labeled): + if not 0.0 <= calibration_size <= 1.0: + raise ValueError( + f"calibration_size must be in [0, 1]. Got {calibration_size!r}." + ) + + if not isinstance(calibration_labeled, bool): + raise TypeError( + "calibration_labeled must be a bool. " + f"Got {type(calibration_labeled).__name__}." + ) + + if calibration_labeled and calibration_size > 0.5: + raise ValueError( + "calibration_labeled=True is only allowed with " + "calibration_size <= 0.5." + ) + + +def resolve_cs_mode(cs_mode): + params = dict(_CS_MODE_MAP[CsMode(cs_mode)]) + validate_transfer_protocol( + params["calibration_size"], + params["calibration_labeled"], + ) + return params \ No newline at end of file diff --git a/moabb/evaluations/splitters.py b/moabb/evaluations/splitters.py index 35931ed9a..a9b11e99b 100644 --- a/moabb/evaluations/splitters.py +++ b/moabb/evaluations/splitters.py @@ -482,42 +482,60 @@ def get_metadata(self): class CrossSubjectSplitter(BaseCrossValidator): - """Data splitter for cross subject evaluation. + """Data splitter for cross-subject evaluation. - This splitter enables cross-subject evaluation by performing a Leave-One-Session-Out (LOSO) - cross-validation on the dataset. + This splitter enables cross-subject evaluation by splitting data according + to subject identity. It assumes that the entire metainformation across all + subjects is already loaded. - It assumes that the entire metainformation across all subjects is already loaded. - - Unlike the `CrossSubjectEvaluation` class from `moabb.evaluation`, which manages - the complete evaluation process end-to-end, this splitter is solely responsible - for dividing the data into training and testing sets based on subjects. + Unlike the `CrossSubjectEvaluation` class from `moabb.evaluation`, which + manages the complete evaluation process end-to-end, this splitter is solely + responsible for dividing the data into training, optional calibration, and + testing sets based on subjects. .. image:: https://raw.githubusercontent.com/NeuroTechX/moabb/refs/heads/develop/docs/source/images/crosssubj.png - :alt: The schematic diagram of the CrossSubject split - :align: center + :alt: The schematic diagram of the CrossSubject split + :align: center The splitting strategy for the subjects can be changed by passing the - `cv_class` and `cv_kwargs` arguments. By default, it uses LeaveOneGroupOut, - which performs Leave-One-Subject-Out cross-validation. + `cv_class` and `cv_kwargs` arguments. By default, it uses + LeaveOneGroupOut, which performs leave-one-subject-out cross-validation. Parameters ---------- cv_class : type - Cross-validation strategy for splitting the subjects between train and test sets. - By default, use LeaveOneGroupOut, which keeps one subject as a test. - Defaults to ``LeaveOneGroupOut``. - calibration_size : float - Transfer-learning calibration, in ``[0, 1]``. ``0.0`` (default) is the - ordinary cross-subject split, yielding ``(train, test)``. When ``> 0``, - the first ``calibration_size`` fraction of each held-out subject's trials - is set aside for adaptation and the splitter yields - ``(train, calibration, test)`` instead; ``1.0`` is transductive - (calibration equals test). + Cross-validation strategy for splitting subjects between train and + test sets. By default, use LeaveOneGroupOut, which keeps one subject + as test. Defaults to ``LeaveOneGroupOut``. + + calibration_size : float, default=0.0 + Transfer-learning calibration fraction, in ``[0, 1]``. + + ``0.0`` is the ordinary cross-subject split and yields + ``(train, test)``. + + When ``> 0``, the first ``calibration_size`` fraction of each held-out + subject's trials is set aside for adaptation/calibration and the + splitter yields ``(train, calibration, test)``. + + ``1.0`` is transductive for unlabeled calibration: calibration and test + are the same target trials. + + calibration_labeled : bool, default=False + Whether the calibration slice may be treated as labeled target + calibration data. + + If ``False``, the calibration slice may only be used as unlabeled + target data. + + If ``True``, labels may be routed together with the calibration slice, + but ``calibration_size`` must be at most ``0.5``. This prevents labeled + calibration from using most or all of the held-out target subject. + random_state : int or None - Controls the randomness of the cross-validation. - Pass an int for reproducible output across multiple calls. - Defaults to ``None``. + Controls the randomness of the cross-validation. Pass an int for + reproducible output across multiple calls. Defaults to ``None``. + cv_kwargs : dict Additional arguments to pass to the inner cross-validation strategy. @@ -527,7 +545,8 @@ class CrossSubjectSplitter(BaseCrossValidator): The training set indices for that split. calibration : ndarray - The held-out calibration indices. Only yielded when ``calibration_size > 0``. + The held-out calibration indices. Only yielded when + ``calibration_size > 0``. test : ndarray The testing set indices for that split. @@ -537,80 +556,112 @@ def __init__( self, cv_class: type[BaseCrossValidator] = LeaveOneGroupOut, calibration_size: float = 0.0, + calibration_labeled: bool = False, random_state: int = None, **cv_kwargs, ): if not 0.0 <= calibration_size <= 1.0: raise ValueError( - f"calibration_size must be in [0, 1]. Got {calibration_size!r}." + f"calibration_size must be in [0, 1]. " + f"Got {calibration_size!r}." ) + + if not isinstance(calibration_labeled, bool): + raise TypeError( + "calibration_labeled must be a bool. " + f"Got {type(calibration_labeled).__name__}." + ) + + if calibration_labeled and calibration_size > 0.5: + raise ValueError( + "calibration_labeled=True is only allowed with " + "calibration_size <= 0.5." + ) + self.cv_class = cv_class self.calibration_size = calibration_size + self.calibration_labeled = calibration_labeled + self.random_state = random_state self.cv_kwargs = cv_kwargs self._cv_kwargs = dict(**cv_kwargs) params = inspect.signature(self.cv_class).parameters + if "random_state" in params: self._cv_kwargs["random_state"] = random_state - # Detect whether the cv_class uses the groups parameter + # Detect whether the cv_class uses the groups parameter. self._cv_uses_groups = issubclass(cv_class, GroupsConsumerMixin) + self._last_split_metadata = None def get_n_splits(self, metadata): """ Return the number of splits for the cross-validation. - The number of splits is the number of subjects times the number of splits - of the inner cross-validation strategy. - - We try to keep the same behaviour as the sklearn cross-validation classes. - Parameters ---------- - metadata: :class:`pandas.DataFrame` - The metadata containing the subject and session information. + metadata : pandas.DataFrame + Metadata containing subject and session information. Returns ------- - n_splits: int - The number of splits for the cross-validation + n_splits : int + Number of splits. """ - splitter = self.cv_class(**self._cv_kwargs) + get_n_splits_kwargs = {"X": metadata.index} + if self._cv_uses_groups: get_n_splits_kwargs["groups"] = metadata["subject"] - n_splits = splitter.get_n_splits(**get_n_splits_kwargs) - return n_splits + + return splitter.get_n_splits(**get_n_splits_kwargs) def split(self, y, metadata): - # here, I am getting the index across all the subject all_index = metadata.index.values - splitter = self.cv_class(**self._cv_kwargs) self._last_split_metadata = None # Only pass groups to cv_classes that actually use them # (detected via GroupsConsumerMixin). This avoids the # "The groups parameter is ignored" warning from e.g. TimeSeriesSplit. - split_kwargs = {"X": all_index, "y": y} + split_kwargs = { + "X": all_index, + "y": y, + } + if self._cv_uses_groups: split_kwargs["groups"] = metadata["subject"] - for train_session_idx, test_session_idx in splitter.split(**split_kwargs): - self._last_split_metadata = _splitter_metadata(splitter) - train_idx = all_index[train_session_idx] - group_idx = all_index[test_session_idx] + for train_subject_idx, test_subject_idx in splitter.split(**split_kwargs): + inner_metadata = _splitter_metadata(splitter) + + split_metadata = { + "calibration_size": self.calibration_size, + "calibration_labeled": self.calibration_labeled, + } + + if inner_metadata is not None: + split_metadata.update(inner_metadata) + + self._last_split_metadata = split_metadata + + train_idx = all_index[train_subject_idx] + target_idx = all_index[test_subject_idx] + if self.calibration_size > 0: - # Transfer learning: carve a calibration slice off the held-out - # subject -> (train, calibration, test). + # Transfer learning: carve a calibration slice off the + # held-out target subject -> (train, calibration, test). calib_idx, test_idx = _split_target_fraction( - group_idx, self.calibration_size + target_idx, + self.calibration_size, ) + yield train_idx, calib_idx, test_idx + else: - yield train_idx, group_idx + yield train_idx, target_idx def get_metadata(self): """Return metadata for the most recent split.""" @@ -618,29 +669,42 @@ def get_metadata(self): def _split_target_fraction( - target_idx: np.ndarray, calibration_size: float + target_idx: np.ndarray, + calibration_size: float, ) -> tuple[np.ndarray, np.ndarray]: - """Slice a held-out target subject's trials into (calibration, test). + """Slice a held-out target subject's trials into calibration and test. + + The first ``calibration_size`` fraction of ``target_idx`` in trial order is + the calibration/adaptation slice; the rest is evaluated. + + ``0.0`` yields no calibration. + + ``1.0`` is transductive: calibration and test are identical. This is valid + only for unlabeled target adaptation at the evaluation level. - The first ``calibration_size`` fraction of ``target_idx`` (in trial order) is - the calibration/adaptation slice; the rest is evaluated. ``0.0`` yields no - calibration; ``1.0`` is transductive (calibration and test are identical). - For any fraction strictly between 0 and 1, at least one trial is kept on each - side. + For any fraction strictly between 0 and 1, at least one trial is kept on + each side. """ target_idx = np.asarray(target_idx, dtype=int) + if len(target_idx) == 0: raise ValueError("Empty held-out target index.") + if not 0.0 <= calibration_size <= 1.0: - raise ValueError(f"calibration_size must be in [0, 1]. Got {calibration_size!r}.") + raise ValueError( + f"calibration_size must be in [0, 1]. " + f"Got {calibration_size!r}." + ) if calibration_size == 0.0: return np.array([], dtype=int), target_idx + if calibration_size == 1.0: return target_idx, target_idx n_calib = int(round(calibration_size * len(target_idx))) n_calib = min(max(n_calib, 1), len(target_idx) - 1) + return target_idx[:n_calib], target_idx[n_calib:] From 41d8a915ed4093bd3baab1c1a02cc3536a47b04a Mon Sep 17 00:00:00 2001 From: Anton Date: Mon, 22 Jun 2026 17:16:23 +0200 Subject: [PATCH 05/15] Documentation and verification improvements. --- moabb/evaluations/base.py | 1 + moabb/evaluations/evaluations.py | 16 +++++++++++++--- moabb/evaluations/protocols.py | 8 ++++++++ moabb/evaluations/splitters.py | 21 +++------------------ 4 files changed, 25 insertions(+), 21 deletions(-) diff --git a/moabb/evaluations/base.py b/moabb/evaluations/base.py index 9ae3e35d2..9d9988577 100644 --- a/moabb/evaluations/base.py +++ b/moabb/evaluations/base.py @@ -83,6 +83,7 @@ def _route_transfer_metadata( return {k: v for k, v in candidate.items() if k in kept} + class TrialwisePredictWrapper(ClassifierMixin, BaseEstimator): """Wrap a fitted estimator and force one-trial-at-a-time prediction.""" diff --git a/moabb/evaluations/evaluations.py b/moabb/evaluations/evaluations.py index 940dab8bd..43d9fe954 100644 --- a/moabb/evaluations/evaluations.py +++ b/moabb/evaluations/evaluations.py @@ -473,9 +473,19 @@ class CrossSubjectEvaluation(BaseEvaluation): ``[0, 1]``, default ``0.0``) enables transfer learning: when ``> 0`` each fold becomes ``(train, calibration, test)`` and the held-out calibration slice is routed (raw) to the pipeline steps that request it via - ``set_fit_request`` -- ``subjects`` plus either ``X_target_unlabeled`` - or ``X_target_labeled`` / ``y_target_labeled``, depending on - ``calibration_labeled``. + ``set_fit_request``. With ``calibration_labeled=False``, only + ``X_target_unlabeled`` may be routed. With ``calibration_labeled=True``, + ``X_target_labeled`` and ``y_target_labeled`` may be routed. + Labeled calibration is only allowed with ``calibration_size <= 0.5``. + cs_mode : CsMode or str, optional + Named cross-subject protocol preset. If provided, it sets + ``calibration_size``, ``calibration_labeled``, and ``predict_mode``. + Cannot be combined with manual ``calibration_size`` or + ``calibration_labeled`` in ``cv_kwargs``. + predict_mode : {"blockwise", "trialwise"}, default="blockwise" + Whether prediction is done on the full test block or one trial at a time. + Cannot be combined with ``cs_mode`` because ``cs_mode`` already defines it. + Notes ----- diff --git a/moabb/evaluations/protocols.py b/moabb/evaluations/protocols.py index 1f9218748..990fc9dfc 100644 --- a/moabb/evaluations/protocols.py +++ b/moabb/evaluations/protocols.py @@ -56,6 +56,14 @@ class CsMode(str, Enum): def validate_transfer_protocol(calibration_size, calibration_labeled): + if not isinstance(calibration_size, (int, float)): + raise TypeError( + "calibration_size must be a number. " + f"Got {type(calibration_size).__name__}." + ) + + calibration_size = float(calibration_size) + if not 0.0 <= calibration_size <= 1.0: raise ValueError( f"calibration_size must be in [0, 1]. Got {calibration_size!r}." diff --git a/moabb/evaluations/splitters.py b/moabb/evaluations/splitters.py index a9b11e99b..18c837ffd 100644 --- a/moabb/evaluations/splitters.py +++ b/moabb/evaluations/splitters.py @@ -13,6 +13,7 @@ ) from sklearn.model_selection._split import GroupsConsumerMixin from sklearn.utils import check_random_state +from moabb.evaluations.protocols import validate_transfer_protocol log = logging.getLogger(__name__) @@ -560,23 +561,7 @@ def __init__( random_state: int = None, **cv_kwargs, ): - if not 0.0 <= calibration_size <= 1.0: - raise ValueError( - f"calibration_size must be in [0, 1]. " - f"Got {calibration_size!r}." - ) - - if not isinstance(calibration_labeled, bool): - raise TypeError( - "calibration_labeled must be a bool. " - f"Got {type(calibration_labeled).__name__}." - ) - - if calibration_labeled and calibration_size > 0.5: - raise ValueError( - "calibration_labeled=True is only allowed with " - "calibration_size <= 0.5." - ) + validate_transfer_protocol(calibration_size, calibration_labeled) self.cv_class = cv_class self.calibration_size = calibration_size @@ -702,7 +687,7 @@ def _split_target_fraction( if calibration_size == 1.0: return target_idx, target_idx - n_calib = int(round(calibration_size * len(target_idx))) + n_calib = int(np.floor(calibration_size * len(target_idx))) n_calib = min(max(n_calib, 1), len(target_idx) - 1) return target_idx[:n_calib], target_idx[n_calib:] From 16d56c1fe76f025f8ea35b3c4a5dfee1dfa44ef0 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Tue, 23 Jun 2026 09:02:55 +0000 Subject: [PATCH 06/15] [pre-commit.ci] auto fixes from pre-commit.com hooks --- moabb/evaluations/base.py | 50 +++++++-------------------- moabb/evaluations/evaluations.py | 59 ++++++++------------------------ moabb/evaluations/protocols.py | 17 +++------ moabb/evaluations/splitters.py | 17 +++------ 4 files changed, 37 insertions(+), 106 deletions(-) diff --git a/moabb/evaluations/base.py b/moabb/evaluations/base.py index 9d9988577..a04e6303c 100644 --- a/moabb/evaluations/base.py +++ b/moabb/evaluations/base.py @@ -23,6 +23,7 @@ BaseDataset, CacheConfig, ) +from moabb.evaluations.protocols import PredictMode from moabb.evaluations.utils import ( Emissions, _carbonfootprint, @@ -40,19 +41,14 @@ ) from moabb.paradigms.base import BaseParadigm from moabb.utils import verbose -from moabb.evaluations.protocols import PredictMode + search_methods, optuna_available = check_search_available() log = logging.getLogger(__name__) -def _route_transfer_metadata( - estimator, - subjects, - calib=None, - calibration_labeled=False, -): +def _route_transfer_metadata(estimator, subjects, calib=None, calibration_labeled=False): """Keep only protocol-allowed transfer metadata requested at ``fit``. ``subjects`` is the per-trial source-subject array. @@ -94,45 +90,32 @@ def __init__(self, fitted_estimator): self.classes_ = self._get_classes(fitted_estimator) def fit(self, X, y=None): - raise RuntimeError( - "TrialwisePredictWrapper wraps an already fitted estimator." - ) + raise RuntimeError("TrialwisePredictWrapper wraps an already fitted estimator.") def predict(self, X): return np.asarray( - [ - self.fitted_estimator.predict(X[i : i + 1])[0] - for i in range(len(X)) - ] + [self.fitted_estimator.predict(X[i : i + 1])[0] for i in range(len(X))] ) def predict_proba(self, X): if not hasattr(self.fitted_estimator, "predict_proba"): - raise AttributeError( - "Wrapped estimator does not provide predict_proba." - ) + raise AttributeError("Wrapped estimator does not provide predict_proba.") return np.vstack( [ - np.asarray( - self.fitted_estimator.predict_proba(X[i : i + 1]) - )[0] + np.asarray(self.fitted_estimator.predict_proba(X[i : i + 1]))[0] for i in range(len(X)) ] ) def decision_function(self, X): if not hasattr(self.fitted_estimator, "decision_function"): - raise AttributeError( - "Wrapped estimator does not provide decision_function." - ) + raise AttributeError("Wrapped estimator does not provide decision_function.") rows = [] for i in range(len(X)): - out = np.asarray( - self.fitted_estimator.decision_function(X[i : i + 1]) - ) + out = np.asarray(self.fitted_estimator.decision_function(X[i : i + 1])) if out.ndim == 0: rows.append(out.item()) @@ -158,9 +141,7 @@ def _get_classes(estimator): if hasattr(final_estimator, "classes_"): return final_estimator.classes_ - raise AttributeError( - "Wrapped estimator does not expose classes_." - ) + raise AttributeError("Wrapped estimator does not expose classes_.") def _wrap_predict_mode(estimator, predict_mode): @@ -174,6 +155,7 @@ def _wrap_predict_mode(estimator, predict_mode): raise ValueError(f"Unknown predict_mode={predict_mode!r}.") + # Making the optuna soft dependency @@ -314,10 +296,7 @@ def _evaluate_fold( calib = None if calib_idx is not None and len(calib_idx): y_calib = y[calib_idx] if mne_labels else le.transform(y[calib_idx]) - calib = { - "X": X[calib_idx], - "y": y_calib, - } + calib = {"X": X[calib_idx], "y": y_calib} calibration_labeled = False if split_metadata is not None: @@ -325,10 +304,7 @@ def _evaluate_fold( subjects_train = metadata["subject"].to_numpy()[train_idx] fit_params = _route_transfer_metadata( - cvclf, - subjects_train, - calib=calib, - calibration_labeled=calibration_labeled, + cvclf, subjects_train, calib=calib, calibration_labeled=calibration_labeled ) # Fit model diff --git a/moabb/evaluations/evaluations.py b/moabb/evaluations/evaluations.py index 43d9fe954..1ca8b2a67 100644 --- a/moabb/evaluations/evaluations.py +++ b/moabb/evaluations/evaluations.py @@ -12,17 +12,17 @@ _route_transfer_metadata, _wrap_predict_mode, ) +from moabb.evaluations.protocols import ( + PredictMode, + resolve_cs_mode, + validate_transfer_protocol, +) from moabb.evaluations.splitters import ( CrossSessionSplitter, CrossSubjectSplitter, WithinSessionSplitter, WithinSubjectSplitter, ) -from moabb.evaluations.protocols import ( - PredictMode, - resolve_cs_mode, - validate_transfer_protocol, -) if TYPE_CHECKING: @@ -498,11 +498,7 @@ class CrossSubjectEvaluation(BaseEvaluation): _needs_all_subjects = True def __init__( - self, - *args, - cs_mode=None, - predict_mode=PredictMode.BLOCKWISE.value, - **kwargs, + self, *args, cs_mode=None, predict_mode=PredictMode.BLOCKWISE.value, **kwargs ): cv_kwargs = dict(kwargs.get("cv_kwargs") or {}) @@ -548,9 +544,7 @@ def _create_splitter(self): cv_class, cv_kwargs = self._resolve_cv(default_class, default_kwargs) return CrossSubjectSplitter( - cv_class=cv_class, - random_state=self.random_state, - **cv_kwargs, + cv_class=cv_class, random_state=self.random_state, **cv_kwargs ) # flake8: noqa: C901 @@ -573,10 +567,7 @@ def evaluate( for subject in dataset.subject_list: run_pipes.update( self.results.not_yet_computed( - pipelines, - dataset, - subject, - process_pipeline, + pipelines, dataset, subject, process_pipeline ) ) @@ -584,10 +575,7 @@ def evaluate( return X, y, metadata = self._load_data( - dataset, - run_pipes, - process_pipeline, - postprocess_pipeline, + dataset, run_pipes, process_pipeline, postprocess_pipeline ) le = LabelEncoder() @@ -603,11 +591,7 @@ def evaluate( if self.n_splits is not None and self.cv_class is None: n_subjects = self.n_splits - inner_cv = StratifiedKFold( - 3, - shuffle=True, - random_state=self.random_state, - ) + inner_cv = StratifiedKFold(3, shuffle=True, random_state=self.random_state) if _carbonfootprint: tracker = self.emissions.create_tracker() @@ -630,27 +614,18 @@ def evaluate( split_metadata = dict(split_metadata) run_pipes = self.results.not_yet_computed( - pipelines, - dataset, - subject, - process_pipeline, + pipelines, dataset, subject, process_pipeline ) for name, clf in run_pipes.items(): clf = self._grid_search( - param_grid=param_grid, - name=name, - grid_clf=clf, - inner_cv=inner_cv, + param_grid=param_grid, name=name, grid_clf=clf, inner_cv=inner_cv ) cvclf = clone(clf) calib_md = None if len(calib): - calib_md = { - "X": X[calib], - "y": y[calib], - } + calib_md = {"X": X[calib], "y": y[calib]} calibration_labeled = False if split_metadata is not None: @@ -674,13 +649,7 @@ def evaluate( ) self._maybe_save_model_cv( - cvclf, - dataset, - subject, - "", - name, - cv_ind, - eval_type="CrossSubject", + cvclf, dataset, subject, "", name, cv_ind, eval_type="CrossSubject" ) score_estimator = _wrap_predict_mode(cvclf, self.predict_mode) diff --git a/moabb/evaluations/protocols.py b/moabb/evaluations/protocols.py index 990fc9dfc..e98771d49 100644 --- a/moabb/evaluations/protocols.py +++ b/moabb/evaluations/protocols.py @@ -58,16 +58,13 @@ class CsMode(str, Enum): def validate_transfer_protocol(calibration_size, calibration_labeled): if not isinstance(calibration_size, (int, float)): raise TypeError( - "calibration_size must be a number. " - f"Got {type(calibration_size).__name__}." + f"calibration_size must be a number. Got {type(calibration_size).__name__}." ) calibration_size = float(calibration_size) if not 0.0 <= calibration_size <= 1.0: - raise ValueError( - f"calibration_size must be in [0, 1]. Got {calibration_size!r}." - ) + raise ValueError(f"calibration_size must be in [0, 1]. Got {calibration_size!r}.") if not isinstance(calibration_labeled, bool): raise TypeError( @@ -77,15 +74,11 @@ def validate_transfer_protocol(calibration_size, calibration_labeled): if calibration_labeled and calibration_size > 0.5: raise ValueError( - "calibration_labeled=True is only allowed with " - "calibration_size <= 0.5." + "calibration_labeled=True is only allowed with calibration_size <= 0.5." ) def resolve_cs_mode(cs_mode): params = dict(_CS_MODE_MAP[CsMode(cs_mode)]) - validate_transfer_protocol( - params["calibration_size"], - params["calibration_labeled"], - ) - return params \ No newline at end of file + validate_transfer_protocol(params["calibration_size"], params["calibration_labeled"]) + return params diff --git a/moabb/evaluations/splitters.py b/moabb/evaluations/splitters.py index 18c837ffd..c6c9d31b1 100644 --- a/moabb/evaluations/splitters.py +++ b/moabb/evaluations/splitters.py @@ -13,6 +13,7 @@ ) from sklearn.model_selection._split import GroupsConsumerMixin from sklearn.utils import check_random_state + from moabb.evaluations.protocols import validate_transfer_protocol @@ -611,10 +612,7 @@ def split(self, y, metadata): # Only pass groups to cv_classes that actually use them # (detected via GroupsConsumerMixin). This avoids the # "The groups parameter is ignored" warning from e.g. TimeSeriesSplit. - split_kwargs = { - "X": all_index, - "y": y, - } + split_kwargs = {"X": all_index, "y": y} if self._cv_uses_groups: split_kwargs["groups"] = metadata["subject"] @@ -639,8 +637,7 @@ def split(self, y, metadata): # Transfer learning: carve a calibration slice off the # held-out target subject -> (train, calibration, test). calib_idx, test_idx = _split_target_fraction( - target_idx, - self.calibration_size, + target_idx, self.calibration_size ) yield train_idx, calib_idx, test_idx @@ -654,8 +651,7 @@ def get_metadata(self): def _split_target_fraction( - target_idx: np.ndarray, - calibration_size: float, + target_idx: np.ndarray, calibration_size: float ) -> tuple[np.ndarray, np.ndarray]: """Slice a held-out target subject's trials into calibration and test. @@ -676,10 +672,7 @@ def _split_target_fraction( raise ValueError("Empty held-out target index.") if not 0.0 <= calibration_size <= 1.0: - raise ValueError( - f"calibration_size must be in [0, 1]. " - f"Got {calibration_size!r}." - ) + raise ValueError(f"calibration_size must be in [0, 1]. Got {calibration_size!r}.") if calibration_size == 0.0: return np.array([], dtype=int), target_idx From 61a63749a2ef29d347a29149256d4420d3e18c35 Mon Sep 17 00:00:00 2001 From: Bru Date: Tue, 23 Jun 2026 22:58:35 +0200 Subject: [PATCH 07/15] docs(examples): cross-subject transfer learning via transform_input Runnable example: a target-aware Riemannian Alignment step plugs into the standard CrossSubjectEvaluation using cv_kwargs={"calibration_size": ...} for the (train, calibration, test) split, metadata routing (set_fit_request) to receive `subjects` + the unlabeled target slice, and Pipeline(transform_input=["X_target_unlabeled"]) so the raw slice is transformed through Covariances() before the alignment step -- no dedicated evaluation engine, no separate transfer module. --- .../noplot_cross_subject_transfer.py | 123 ++++++++++++++++++ 1 file changed, 123 insertions(+) create mode 100644 examples/how_to_benchmark/noplot_cross_subject_transfer.py diff --git a/examples/how_to_benchmark/noplot_cross_subject_transfer.py b/examples/how_to_benchmark/noplot_cross_subject_transfer.py new file mode 100644 index 000000000..a6ada5f4d --- /dev/null +++ b/examples/how_to_benchmark/noplot_cross_subject_transfer.py @@ -0,0 +1,123 @@ +""" +========================================================================= +Cross-subject transfer learning through the standard CrossSubjectEvaluation +========================================================================= + +Target-aware transfer methods (e.g. Riemannian Alignment / RPA) need two things +the ordinary cross-subject split does not give them: + +1. which source subject each training trial belongs to, and +2. controlled access to an (unlabeled) slice of the held-out target subject. + +Both fit into the **standard** :class:`~moabb.evaluations.CrossSubjectEvaluation` +with three scikit-learn-native pieces -- no dedicated evaluation engine: + +* ``cv_kwargs={"calibration_size": ...}`` makes each fold yield + ``(train, calibration, test)`` + (see :class:`~moabb.evaluations.splitters.CrossSubjectSplitter`); +* **metadata routing** (``set_fit_request``) delivers ``subjects`` and the raw + calibration slice to the steps that request them; +* ``Pipeline(transform_input=...)`` transforms that raw slice through the earlier + steps (here ``Covariances``), so a mid-pipeline alignment step receives the + target in the *same representation* as ``X`` (covariances, not raw epochs). +""" + +# License: BSD (3-clause) + +import numpy as np +from pyriemann.estimation import Covariances +from pyriemann.tangentspace import TangentSpace +from pyriemann.utils.base import invsqrtm +from pyriemann.utils.mean import mean_riemann +from sklearn import config_context +from sklearn.base import BaseEstimator, TransformerMixin +from sklearn.linear_model import LogisticRegression +from sklearn.pipeline import make_pipeline + +from moabb.datasets.fake import FakeDataset +from moabb.evaluations import CrossSubjectEvaluation +from moabb.paradigms import LeftRightImagery + + +# --------------------------------------------------------------------------- +# pyriemann's ``Covariances`` is stateless and does not mark itself fitted, +# which ``Pipeline(transform_input=...)`` requires (it calls ``transform`` on +# the sub-pipeline, which runs ``check_is_fitted``). One line fixes it; this +# ideally lands upstream in pyriemann. +# --------------------------------------------------------------------------- +class Covariances_(Covariances): + def __sklearn_is_fitted__(self): + return True + + +# --------------------------------------------------------------------------- +# A minimal target-aware step: Riemannian Alignment (recenter each domain to the +# identity). It consumes ``subjects`` (per-source-subject reference) and the +# unlabeled target covariances (target reference). Both come from the evaluation +# via metadata routing. +# --------------------------------------------------------------------------- +class RiemannianAlignment(TransformerMixin, BaseEstimator): + def fit(self, X, y=None, subjects=None, X_target_unlabeled=None): + X = np.asarray(X) + self.fit_subjects_ = np.asarray(subjects) + # one whitening reference per source subject + self.source_refs_ = { + s: invsqrtm(mean_riemann(X[self.fit_subjects_ == s])) + for s in np.unique(self.fit_subjects_) + } + # X_target_unlabeled arrives as covariances thanks to transform_input + self.target_ref_ = ( + invsqrtm(mean_riemann(np.asarray(X_target_unlabeled))) + if X_target_unlabeled is not None and len(X_target_unlabeled) + else None + ) + return self + + def transform(self, X): + X = np.asarray(X) + if len(X) == len(self.fit_subjects_): # training data: per-source-subject + out = np.empty_like(X) + for s, ref in self.source_refs_.items(): + m = self.fit_subjects_ == s + out[m] = ref @ X[m] @ ref + return out + # held-out target at predict time: recenter by the target reference + ref = self.target_ref_ + if ref is None: + ref = next(iter(self.source_refs_.values())) + return ref @ X @ ref + + +def make_transfer_pipeline(): + with config_context(enable_metadata_routing=True): + align = RiemannianAlignment().set_fit_request( + subjects=True, X_target_unlabeled=True + ) + # transform_input -> the raw target slice is passed through Covariances_() + # before RiemannianAlignment.fit receives it (so it is covariances, not raw). + return make_pipeline( + Covariances_("oas"), + align, + TangentSpace("riemann"), + LogisticRegression(max_iter=500), + transform_input=["X_target_unlabeled"], + ) + + +if __name__ == "__main__": + dataset = FakeDataset( + ["left_hand", "right_hand"], n_subjects=4, n_sessions=2, seed=42 + ) + paradigm = LeftRightImagery() + + # calibration_size=0.2: the first 20% of each held-out subject is the + # unlabeled target slice used by RiemannianAlignment; the remaining 80% is + # scored. calibration_size=0.0 reproduces the ordinary cross-subject split. + evaluation = CrossSubjectEvaluation( + paradigm=paradigm, + datasets=[dataset], + cv_kwargs={"calibration_size": 0.2}, + overwrite=True, + ) + results = evaluation.process({"RA+TS+LR": make_transfer_pipeline()}) + print(results[["subject", "session", "pipeline", "score"]].to_string(index=False)) From 81ba8656dde7168dbd56c3019adac7d2f759215e Mon Sep 17 00:00:00 2001 From: Bru Date: Fri, 26 Jun 2026 12:26:32 +0200 Subject: [PATCH 08/15] docs(examples): add scikit-learn-native trialwise (FrozenEstimator + cross_val_predict/LeaveOneOut) --- .../noplot_cross_subject_transfer.py | 25 +++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/examples/how_to_benchmark/noplot_cross_subject_transfer.py b/examples/how_to_benchmark/noplot_cross_subject_transfer.py index a6ada5f4d..2c272912a 100644 --- a/examples/how_to_benchmark/noplot_cross_subject_transfer.py +++ b/examples/how_to_benchmark/noplot_cross_subject_transfer.py @@ -121,3 +121,28 @@ def make_transfer_pipeline(): ) results = evaluation.process({"RA+TS+LR": make_transfer_pipeline()}) print(results[["subject", "session", "pipeline", "score"]].to_string(index=False)) + + # ------------------------------------------------------------------ + # Trialwise / one-shot prediction with pure scikit-learn: freeze the + # source-trained model and predict each held-out trial in isolation via + # LeaveOneOut (test fold size == 1). No custom wrapper. + # ------------------------------------------------------------------ + from sklearn.frozen import FrozenEstimator + from sklearn.model_selection import LeaveOneOut, cross_val_predict + + X, y, meta = paradigm.get_data(dataset=dataset) + target = meta["subject"].to_numpy() == np.unique(meta["subject"])[-1] + source_model = make_pipeline( + Covariances_("oas"), TangentSpace("riemann"), LogisticRegression(max_iter=500) + ).fit(X[~target], y[~target]) + trialwise = cross_val_predict( + FrozenEstimator(source_model), + X[target], + y[target], + cv=LeaveOneOut(), + method="predict", + ) + print( + "trialwise == blockwise (inductive):", + np.array_equal(trialwise, source_model.predict(X[target])), + ) From 8477734ddbc07acfa926ca329af42d05d3d51c7f Mon Sep 17 00:00:00 2001 From: Anton Date: Fri, 26 Jun 2026 13:20:05 +0200 Subject: [PATCH 09/15] Simplify cross-subject protocol modes Remove the public predict_mode API and related prediction-mode enum/wrapper.Keep trialwise evaluation as an official CsMode preset instead of exposing it as a general user-configurable option. The trialwise behavior is now handled by a small private helper function at scoring time, not by a public wrapper class. --- moabb/evaluations/base.py | 128 ++++++++++++--------------------- moabb/evaluations/protocols.py | 52 ++++++++++---- 2 files changed, 85 insertions(+), 95 deletions(-) diff --git a/moabb/evaluations/base.py b/moabb/evaluations/base.py index a04e6303c..430fa4307 100644 --- a/moabb/evaluations/base.py +++ b/moabb/evaluations/base.py @@ -13,7 +13,7 @@ import pandas as pd from joblib import Parallel, delayed from sklearn import config_context -from sklearn.base import BaseEstimator, ClassifierMixin, clone +from sklearn.base import BaseEstimator, clone from sklearn.model_selection import StratifiedKFold from sklearn.preprocessing import LabelEncoder from sklearn.utils.metadata_routing import get_routing_for_object @@ -23,7 +23,6 @@ BaseDataset, CacheConfig, ) -from moabb.evaluations.protocols import PredictMode from moabb.evaluations.utils import ( Emissions, _carbonfootprint, @@ -48,6 +47,49 @@ log = logging.getLogger(__name__) +def _one_shot_estimator(estimator): + class _OneShotEstimator: + def __init__(self, estimator): + self.estimator = estimator + if hasattr(estimator, "classes_"): + self.classes_ = estimator.classes_ + elif hasattr(estimator, "steps") and hasattr(estimator.steps[-1][1], "classes_"): + self.classes_ = estimator.steps[-1][1].classes_ + + def predict(self, X): + return np.asarray( + [self.estimator.predict(X[i : i + 1])[0] for i in range(len(X))] + ) + + def __getattr__(self, name): + if name == "predict_proba" and hasattr(self.estimator, "predict_proba"): + def predict_proba(X): + return np.vstack( + [ + self.estimator.predict_proba(X[i : i + 1]) + for i in range(len(X)) + ] + ) + + return predict_proba + + if name == "decision_function" and hasattr( + self.estimator, "decision_function" + ): + def decision_function(X): + values = [ + self.estimator.decision_function(X[i : i + 1]) + for i in range(len(X)) + ] + return np.asarray([v[0] for v in values]) + + return decision_function + + raise AttributeError(name) + + return _OneShotEstimator(estimator) + + def _route_transfer_metadata(estimator, subjects, calib=None, calibration_labeled=False): """Keep only protocol-allowed transfer metadata requested at ``fit``. @@ -80,82 +122,6 @@ def _route_transfer_metadata(estimator, subjects, calib=None, calibration_labele return {k: v for k, v in candidate.items() if k in kept} -class TrialwisePredictWrapper(ClassifierMixin, BaseEstimator): - """Wrap a fitted estimator and force one-trial-at-a-time prediction.""" - - _estimator_type = "classifier" - - def __init__(self, fitted_estimator): - self.fitted_estimator = fitted_estimator - self.classes_ = self._get_classes(fitted_estimator) - - def fit(self, X, y=None): - raise RuntimeError("TrialwisePredictWrapper wraps an already fitted estimator.") - - def predict(self, X): - return np.asarray( - [self.fitted_estimator.predict(X[i : i + 1])[0] for i in range(len(X))] - ) - - def predict_proba(self, X): - if not hasattr(self.fitted_estimator, "predict_proba"): - raise AttributeError("Wrapped estimator does not provide predict_proba.") - - return np.vstack( - [ - np.asarray(self.fitted_estimator.predict_proba(X[i : i + 1]))[0] - for i in range(len(X)) - ] - ) - - def decision_function(self, X): - if not hasattr(self.fitted_estimator, "decision_function"): - raise AttributeError("Wrapped estimator does not provide decision_function.") - - rows = [] - - for i in range(len(X)): - out = np.asarray(self.fitted_estimator.decision_function(X[i : i + 1])) - - if out.ndim == 0: - rows.append(out.item()) - elif out.shape[0] == 1: - rows.append(out[0]) - else: - rows.append(out) - - out = np.asarray(rows) - - if out.ndim == 2 and out.shape[1] == 1: - return out.ravel() - - return out - - @staticmethod - def _get_classes(estimator): - if hasattr(estimator, "classes_"): - return estimator.classes_ - - if hasattr(estimator, "steps"): - final_estimator = estimator.steps[-1][1] - if hasattr(final_estimator, "classes_"): - return final_estimator.classes_ - - raise AttributeError("Wrapped estimator does not expose classes_.") - - -def _wrap_predict_mode(estimator, predict_mode): - predict_mode = PredictMode(predict_mode) - - if predict_mode == PredictMode.BLOCKWISE: - return estimator - - if predict_mode == PredictMode.TRIALWISE: - return TrialwisePredictWrapper(estimator) - - raise ValueError(f"Unknown predict_mode={predict_mode!r}.") - - # Making the optuna soft dependency @@ -253,7 +219,7 @@ def _evaluate_fold( score_per_session = config["score_per_session"] mne_labels = config["mne_labels"] codecarbon_config = config["codecarbon_config"] - predict_mode = config.get("predict_mode", PredictMode.BLOCKWISE.value) + one_shot_predict = config.get("one_shot_predict", False) # Label encode per fold (matching old per-session/per-subject scoping) if not mne_labels: @@ -340,7 +306,7 @@ def _evaluate_fold( ) _save_model_cv(model=cvclf, save_path=model_save_path, cv_index=str(cv_ind)) - score_estimator = _wrap_predict_mode(cvclf, predict_mode) + score_estimator = _one_shot_estimator(cvclf) if one_shot_predict else cvclf scorer = _create_scorer(score_estimator, scoring) # Build score groups: per-session or full test set @@ -804,7 +770,7 @@ def _build_eval_config(self, param_grid): self.emissions.codecarbon_config if _carbonfootprint else None ), "score_per_session": self._score_per_session, - "predict_mode": getattr(self, "predict_mode", PredictMode.BLOCKWISE.value), + "one_shot_predict": getattr(self, "one_shot_predict", False), "param_grid": None, # overridden per-task below if needed } diff --git a/moabb/evaluations/protocols.py b/moabb/evaluations/protocols.py index e98771d49..07d3bbc10 100644 --- a/moabb/evaluations/protocols.py +++ b/moabb/evaluations/protocols.py @@ -1,56 +1,75 @@ -from enum import Enum +"""Cross-subject transfer protocol presets. +This module defines named presets for held-out-subject (HOS) evaluation. -class PredictMode(str, Enum): - BLOCKWISE = "blockwise" - TRIALWISE = "trialwise" +Blockwise prediction means that the estimator receives the whole target test +block at scoring time, as in the standard MOABB CrossSubjectEvaluation path. +Trialwise prediction means that the estimator is scored one target trial at a +time. This prevents methods from using statistics of the full target test block +during prediction. +""" +from enum import Enum class CsMode(str, Enum): - HOS_SOURCE_ONLY_BLOCKWISE = "hos_source_only_blockwise" + # Default source-only cross-subject evaluation. + # Train only on source subjects; no target calibration data is used. + # The held-out target test data is predicted blockwise. + HOS_SOURCE_ONLY = "hos_source_only" + + # Strict source-only one-shot evaluation. + # Train only on source subjects; no target calibration data is used. + # The held-out target test data is predicted one trial at a time. HOS_SOURCE_ONLY_TRIALWISE = "hos_source_only_trialwise" + + # Use 20% of the held-out subject as unlabeled target calibration/adaptation data. + # Evaluate on the remaining target test data, predicted blockwise. HOS_UNLABELED_20P = "hos_unlabeled_20p" + + # Use 50% of the held-out subject as unlabeled target calibration/adaptation data. + # Evaluate on the remaining target test data, predicted blockwise. HOS_UNLABELED_50P = "hos_unlabeled_50p" + + # Use all held-out target data as unlabeled adaptation data. + # Evaluate transductively on the target test block, predicted blockwise. HOS_UNLABELED_100P = "hos_unlabeled_100p" + + # Use 20% of the held-out subject as labeled target calibration/adaptation data. + # Evaluate on the remaining target test data, predicted blockwise. HOS_LABELED_20P = "hos_labeled_20p" - HOS_LABELED_50P = "hos_labeled_50p" + # Use 50% of the held-out subject as labeled target calibration/adaptation data. + # Evaluate on the remaining target test data, predicted blockwise. + HOS_LABELED_50P = "hos_labeled_50p" _CS_MODE_MAP = { - CsMode.HOS_SOURCE_ONLY_BLOCKWISE: dict( + CsMode.HOS_SOURCE_ONLY: dict( calibration_size=0.0, calibration_labeled=False, - predict_mode=PredictMode.BLOCKWISE.value, ), CsMode.HOS_SOURCE_ONLY_TRIALWISE: dict( calibration_size=0.0, calibration_labeled=False, - predict_mode=PredictMode.TRIALWISE.value, ), CsMode.HOS_UNLABELED_20P: dict( calibration_size=0.2, calibration_labeled=False, - predict_mode=PredictMode.BLOCKWISE.value, ), CsMode.HOS_UNLABELED_50P: dict( calibration_size=0.5, calibration_labeled=False, - predict_mode=PredictMode.BLOCKWISE.value, ), CsMode.HOS_UNLABELED_100P: dict( calibration_size=1.0, calibration_labeled=False, - predict_mode=PredictMode.BLOCKWISE.value, ), CsMode.HOS_LABELED_20P: dict( calibration_size=0.2, calibration_labeled=True, - predict_mode=PredictMode.BLOCKWISE.value, ), CsMode.HOS_LABELED_50P: dict( calibration_size=0.5, calibration_labeled=True, - predict_mode=PredictMode.BLOCKWISE.value, ), } @@ -82,3 +101,8 @@ def resolve_cs_mode(cs_mode): params = dict(_CS_MODE_MAP[CsMode(cs_mode)]) validate_transfer_protocol(params["calibration_size"], params["calibration_labeled"]) return params + + +def is_one_shot_mode(cs_mode): + return CsMode(cs_mode) == CsMode.HOS_SOURCE_ONLY_TRIALWISE + \ No newline at end of file From c11a451bc7ff9987e1af6935a886387e27895398 Mon Sep 17 00:00:00 2001 From: Anton Date: Fri, 26 Jun 2026 13:22:40 +0200 Subject: [PATCH 10/15] File missing from previos commit --- moabb/evaluations/evaluations.py | 59 ++++++++++++++++++-------------- 1 file changed, 33 insertions(+), 26 deletions(-) diff --git a/moabb/evaluations/evaluations.py b/moabb/evaluations/evaluations.py index 1ca8b2a67..1d4202c65 100644 --- a/moabb/evaluations/evaluations.py +++ b/moabb/evaluations/evaluations.py @@ -9,11 +9,12 @@ from moabb.evaluations.base import ( BaseEvaluation, + _one_shot_estimator, _route_transfer_metadata, - _wrap_predict_mode, ) from moabb.evaluations.protocols import ( - PredictMode, + CsMode, + is_one_shot_mode, resolve_cs_mode, validate_transfer_protocol, ) @@ -477,15 +478,13 @@ class CrossSubjectEvaluation(BaseEvaluation): ``X_target_unlabeled`` may be routed. With ``calibration_labeled=True``, ``X_target_labeled`` and ``y_target_labeled`` may be routed. Labeled calibration is only allowed with ``calibration_size <= 0.5``. - cs_mode : CsMode or str, optional - Named cross-subject protocol preset. If provided, it sets - ``calibration_size``, ``calibration_labeled``, and ``predict_mode``. - Cannot be combined with manual ``calibration_size`` or - ``calibration_labeled`` in ``cv_kwargs``. - predict_mode : {"blockwise", "trialwise"}, default="blockwise" - Whether prediction is done on the full test block or one trial at a time. - Cannot be combined with ``cs_mode`` because ``cs_mode`` already defines it. - + cs_mode : CsMode or str, default=CsMode.HOS_SOURCE_ONLY + Named cross-subject protocol preset. By default, this is the standard + source-only cross-subject evaluation with no target calibration. The + ``HOS_SOURCE_ONLY_TRIALWISE`` mode additionally enforces one-trial-at-a-time + prediction during scoring. Cannot be combined with manual + ``calibration_size`` or ``calibration_labeled`` in ``cv_kwargs``, except + for the default ``HOS_SOURCE_ONLY`` mode. Notes ----- @@ -497,34 +496,40 @@ class CrossSubjectEvaluation(BaseEvaluation): _score_per_session = True _needs_all_subjects = True - def __init__( - self, *args, cs_mode=None, predict_mode=PredictMode.BLOCKWISE.value, **kwargs - ): + def __init__(self, *args, cs_mode=CsMode.HOS_SOURCE_ONLY, **kwargs): cv_kwargs = dict(kwargs.get("cv_kwargs") or {}) + self.one_shot_predict = False - if cs_mode is not None: - if "calibration_size" in cv_kwargs or "calibration_labeled" in cv_kwargs: - raise ValueError( - "Pass either cs_mode or calibration_size/calibration_labeled, " - "not both." - ) + if cs_mode is None: + cs_mode = CsMode.HOS_SOURCE_ONLY + + cs_mode = CsMode(cs_mode) + + # manual cv_kwargs still work when the + # default source-only blockwise mode is used. + has_manual_calibration = ( + "calibration_size" in cv_kwargs or "calibration_labeled" in cv_kwargs + ) - if PredictMode(predict_mode) != PredictMode.BLOCKWISE: - raise ValueError("Pass either cs_mode or predict_mode, not both.") + if has_manual_calibration and cs_mode != CsMode.HOS_SOURCE_ONLY: + raise ValueError( + "Pass either cs_mode or calibration_size/calibration_labeled, " + "not both." + ) + if not has_manual_calibration: params = resolve_cs_mode(cs_mode) cv_kwargs["calibration_size"] = params["calibration_size"] cv_kwargs["calibration_labeled"] = params["calibration_labeled"] - predict_mode = params["predict_mode"] + + self.one_shot_predict = is_one_shot_mode(cs_mode) validate_transfer_protocol( cv_kwargs.get("calibration_size", 0.0), cv_kwargs.get("calibration_labeled", False), ) - self.predict_mode = PredictMode(predict_mode).value kwargs["cv_kwargs"] = cv_kwargs - super().__init__(*args, **kwargs) def _create_splitter(self): @@ -652,7 +657,9 @@ def evaluate( cvclf, dataset, subject, "", name, cv_ind, eval_type="CrossSubject" ) - score_estimator = _wrap_predict_mode(cvclf, self.predict_mode) + score_estimator = ( + _one_shot_estimator(cvclf) if self.one_shot_predict else cvclf + ) scorer = _create_scorer(score_estimator, self.paradigm.scoring) for session in np.unique(sessions[test]): From 5a194b3dc037190a629bb5161124f49b4384c8bd Mon Sep 17 00:00:00 2001 From: Anton Date: Wed, 1 Jul 2026 16:39:09 +0200 Subject: [PATCH 11/15] Renamed the modes to more natural names. --- moabb/evaluations/base.py | 3 +- moabb/evaluations/evaluations.py | 48 ++++++++--------- moabb/evaluations/protocols.py | 89 ++++++++++++++++++++------------ 3 files changed, 83 insertions(+), 57 deletions(-) diff --git a/moabb/evaluations/base.py b/moabb/evaluations/base.py index 430fa4307..4979a9db7 100644 --- a/moabb/evaluations/base.py +++ b/moabb/evaluations/base.py @@ -48,6 +48,7 @@ def _one_shot_estimator(estimator): + """Wrap an estimator so predictions are made one sample at a time.""" class _OneShotEstimator: def __init__(self, estimator): self.estimator = estimator @@ -85,7 +86,7 @@ def decision_function(X): return decision_function - raise AttributeError(name) + return getattr(self.estimator, name) return _OneShotEstimator(estimator) diff --git a/moabb/evaluations/evaluations.py b/moabb/evaluations/evaluations.py index 1d4202c65..04abdefc2 100644 --- a/moabb/evaluations/evaluations.py +++ b/moabb/evaluations/evaluations.py @@ -13,9 +13,9 @@ _route_transfer_metadata, ) from moabb.evaluations.protocols import ( - CsMode, - is_one_shot_mode, - resolve_cs_mode, + CrossSubjectMode, + is_trialwise_mode, + resolve_cross_subject_mode, validate_transfer_protocol, ) from moabb.evaluations.splitters import ( @@ -478,13 +478,13 @@ class CrossSubjectEvaluation(BaseEvaluation): ``X_target_unlabeled`` may be routed. With ``calibration_labeled=True``, ``X_target_labeled`` and ``y_target_labeled`` may be routed. Labeled calibration is only allowed with ``calibration_size <= 0.5``. - cs_mode : CsMode or str, default=CsMode.HOS_SOURCE_ONLY + cs_mode : CrossSubjectMode or str, default=CrossSubjectMode.TRAIN Named cross-subject protocol preset. By default, this is the standard - source-only cross-subject evaluation with no target calibration. The - ``HOS_SOURCE_ONLY_TRIALWISE`` mode additionally enforces one-trial-at-a-time + train-only cross-subject evaluation with no target calibration. The + ``TRAIN_TRIALWISE`` mode additionally enforces one-trial-at-a-time prediction during scoring. Cannot be combined with manual ``calibration_size`` or ``calibration_labeled`` in ``cv_kwargs``, except - for the default ``HOS_SOURCE_ONLY`` mode. + for the default ``TRAIN`` mode. Notes ----- @@ -496,42 +496,42 @@ class CrossSubjectEvaluation(BaseEvaluation): _score_per_session = True _needs_all_subjects = True - def __init__(self, *args, cs_mode=CsMode.HOS_SOURCE_ONLY, **kwargs): + def __init__(self, *args, cs_mode=CrossSubjectMode.TRAIN, **kwargs): cv_kwargs = dict(kwargs.get("cv_kwargs") or {}) self.one_shot_predict = False - + if cs_mode is None: - cs_mode = CsMode.HOS_SOURCE_ONLY - - cs_mode = CsMode(cs_mode) - - # manual cv_kwargs still work when the - # default source-only blockwise mode is used. + cs_mode = CrossSubjectMode.TRAIN + + cs_mode = CrossSubjectMode(cs_mode) + + # Manual cv_kwargs still work when the default train-only blockwise + # mode is used. has_manual_calibration = ( "calibration_size" in cv_kwargs or "calibration_labeled" in cv_kwargs ) - - if has_manual_calibration and cs_mode != CsMode.HOS_SOURCE_ONLY: + + if has_manual_calibration and cs_mode != CrossSubjectMode.TRAIN: raise ValueError( "Pass either cs_mode or calibration_size/calibration_labeled, " "not both." ) - + if not has_manual_calibration: - params = resolve_cs_mode(cs_mode) + params = resolve_cross_subject_mode(cs_mode) cv_kwargs["calibration_size"] = params["calibration_size"] cv_kwargs["calibration_labeled"] = params["calibration_labeled"] - - self.one_shot_predict = is_one_shot_mode(cs_mode) - + + self.one_shot_predict = is_trialwise_mode(cs_mode) + validate_transfer_protocol( cv_kwargs.get("calibration_size", 0.0), cv_kwargs.get("calibration_labeled", False), ) - + kwargs["cv_kwargs"] = cv_kwargs super().__init__(*args, **kwargs) - + def _create_splitter(self): """Create the CrossSubjectSplitter for parallel evaluation. diff --git a/moabb/evaluations/protocols.py b/moabb/evaluations/protocols.py index 07d3bbc10..52f3cd4a9 100644 --- a/moabb/evaluations/protocols.py +++ b/moabb/evaluations/protocols.py @@ -1,6 +1,24 @@ """Cross-subject transfer protocol presets. -This module defines named presets for held-out-subject (HOS) evaluation. +This module defines named presets for cross-subject evaluation in which one +subject is held out as the target subject and the remaining subjects are used +as training/source subjects. + +The presets describe what information from the held-out target subject is +allowed to be used by an estimator, and how the remaining target data are +scored. They make the evaluation protocol explicit, so that different +cross-subject and transfer-learning methods can be compared under controlled +and reproducible conditions. + +A preset controls two related aspects of the evaluation: + +1. target calibration/adaptation access: + whether the estimator receives no target data, an unlabeled target slice, or + a labeled target slice during fitting; + +2. prediction access at scoring time: + whether the estimator receives the target test data as a full block or one + trial at a time. Blockwise prediction means that the estimator receives the whole target test block at scoring time, as in the standard MOABB CrossSubjectEvaluation path. @@ -9,65 +27,71 @@ time. This prevents methods from using statistics of the full target test block during prediction. """ + from enum import Enum -class CsMode(str, Enum): - # Default source-only cross-subject evaluation. - # Train only on source subjects; no target calibration data is used. + +class CrossSubjectMode(str, Enum): + # Train only on training/source subjects; no target calibration data is used. # The held-out target test data is predicted blockwise. - HOS_SOURCE_ONLY = "hos_source_only" + TRAIN = "train" - # Strict source-only one-shot evaluation. - # Train only on source subjects; no target calibration data is used. + # Train only on training/source subjects; no target calibration data is used. # The held-out target test data is predicted one trial at a time. - HOS_SOURCE_ONLY_TRIALWISE = "hos_source_only_trialwise" + TRAIN_TRIALWISE = "train_trialwise" - # Use 20% of the held-out subject as unlabeled target calibration/adaptation data. + # Train on source subjects and use 20% of the held-out subject as + # unlabeled target calibration/adaptation data. # Evaluate on the remaining target test data, predicted blockwise. - HOS_UNLABELED_20P = "hos_unlabeled_20p" + TRAIN_AND_TARGET_UNLABELED_20P = "train_and_target_unlabeled_20p" - # Use 50% of the held-out subject as unlabeled target calibration/adaptation data. + # Train on source subjects and use 50% of the held-out subject as + # unlabeled target calibration/adaptation data. # Evaluate on the remaining target test data, predicted blockwise. - HOS_UNLABELED_50P = "hos_unlabeled_50p" + TRAIN_AND_TARGET_UNLABELED_50P = "train_and_target_unlabeled_50p" - # Use all held-out target data as unlabeled adaptation data. + # Train on source subjects and use all held-out target data as + # unlabeled target adaptation data. # Evaluate transductively on the target test block, predicted blockwise. - HOS_UNLABELED_100P = "hos_unlabeled_100p" + TRAIN_AND_TARGET_UNLABELED_FULL = "train_and_target_unlabeled_full" - # Use 20% of the held-out subject as labeled target calibration/adaptation data. + # Train on source subjects and use 20% of the held-out subject as + # labeled target calibration/adaptation data. # Evaluate on the remaining target test data, predicted blockwise. - HOS_LABELED_20P = "hos_labeled_20p" + TRAIN_AND_TARGET_LABELED_20P = "train_and_target_labeled_20p" - # Use 50% of the held-out subject as labeled target calibration/adaptation data. + # Train on source subjects and use 50% of the held-out subject as + # labeled target calibration/adaptation data. # Evaluate on the remaining target test data, predicted blockwise. - HOS_LABELED_50P = "hos_labeled_50p" + TRAIN_AND_TARGET_LABELED_50P = "train_and_target_labeled_50p" + -_CS_MODE_MAP = { - CsMode.HOS_SOURCE_ONLY: dict( +_CROSS_SUBJECT_MODE_MAP = { + CrossSubjectMode.TRAIN: dict( calibration_size=0.0, calibration_labeled=False, ), - CsMode.HOS_SOURCE_ONLY_TRIALWISE: dict( + CrossSubjectMode.TRAIN_TRIALWISE: dict( calibration_size=0.0, calibration_labeled=False, ), - CsMode.HOS_UNLABELED_20P: dict( + CrossSubjectMode.TRAIN_AND_TARGET_UNLABELED_20P: dict( calibration_size=0.2, calibration_labeled=False, ), - CsMode.HOS_UNLABELED_50P: dict( + CrossSubjectMode.TRAIN_AND_TARGET_UNLABELED_50P: dict( calibration_size=0.5, calibration_labeled=False, ), - CsMode.HOS_UNLABELED_100P: dict( + CrossSubjectMode.TRAIN_AND_TARGET_UNLABELED_FULL: dict( calibration_size=1.0, calibration_labeled=False, ), - CsMode.HOS_LABELED_20P: dict( + CrossSubjectMode.TRAIN_AND_TARGET_LABELED_20P: dict( calibration_size=0.2, calibration_labeled=True, ), - CsMode.HOS_LABELED_50P: dict( + CrossSubjectMode.TRAIN_AND_TARGET_LABELED_50P: dict( calibration_size=0.5, calibration_labeled=True, ), @@ -75,7 +99,9 @@ class CsMode(str, Enum): def validate_transfer_protocol(calibration_size, calibration_labeled): - if not isinstance(calibration_size, (int, float)): + if isinstance(calibration_size, bool) or not isinstance( + calibration_size, (int, float) + ): raise TypeError( f"calibration_size must be a number. Got {type(calibration_size).__name__}." ) @@ -97,12 +123,11 @@ def validate_transfer_protocol(calibration_size, calibration_labeled): ) -def resolve_cs_mode(cs_mode): - params = dict(_CS_MODE_MAP[CsMode(cs_mode)]) +def resolve_cross_subject_mode(cross_subject_mode): + params = dict(_CROSS_SUBJECT_MODE_MAP[CrossSubjectMode(cross_subject_mode)]) validate_transfer_protocol(params["calibration_size"], params["calibration_labeled"]) return params -def is_one_shot_mode(cs_mode): - return CsMode(cs_mode) == CsMode.HOS_SOURCE_ONLY_TRIALWISE - \ No newline at end of file +def is_trialwise_mode(cross_subject_mode): + return CrossSubjectMode(cross_subject_mode) == CrossSubjectMode.TRAIN_TRIALWISE \ No newline at end of file From 9ee2b8bfbcf350ee8166e0d52f35ca13e524903f Mon Sep 17 00:00:00 2001 From: Anton Date: Wed, 1 Jul 2026 17:01:14 +0200 Subject: [PATCH 12/15] Add the possibility to request the CrossSubject mode. This allows an algorithm to adapt to the selected Cross Subject benchmark mode and also to refuse running in certain benchmark modes for which it is was designed for. --- moabb/evaluations/base.py | 19 +++++++++++++++++-- moabb/evaluations/evaluations.py | 2 ++ 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/moabb/evaluations/base.py b/moabb/evaluations/base.py index 4979a9db7..27544f44b 100644 --- a/moabb/evaluations/base.py +++ b/moabb/evaluations/base.py @@ -91,7 +91,13 @@ def decision_function(X): return _OneShotEstimator(estimator) -def _route_transfer_metadata(estimator, subjects, calib=None, calibration_labeled=False): +def _route_transfer_metadata( + estimator, + subjects, + calib=None, + calibration_labeled=False, + cs_mode=None, +): """Keep only protocol-allowed transfer metadata requested at ``fit``. ``subjects`` is the per-trial source-subject array. @@ -110,6 +116,9 @@ def _route_transfer_metadata(estimator, subjects, calib=None, calibration_labele """ candidate = {"subjects": subjects} + if cs_mode is not None: + candidate["cs_mode"] = cs_mode + if calib is not None: if calibration_labeled: candidate["X_target_labeled"] = calib["X"] @@ -221,6 +230,7 @@ def _evaluate_fold( mne_labels = config["mne_labels"] codecarbon_config = config["codecarbon_config"] one_shot_predict = config.get("one_shot_predict", False) + cs_mode = config.get("cs_mode", None) # Label encode per fold (matching old per-session/per-subject scoping) if not mne_labels: @@ -271,7 +281,11 @@ def _evaluate_fold( subjects_train = metadata["subject"].to_numpy()[train_idx] fit_params = _route_transfer_metadata( - cvclf, subjects_train, calib=calib, calibration_labeled=calibration_labeled + cvclf, + subjects_train, + calib=calib, + calibration_labeled=calibration_labeled, + cs_mode=cs_mode, ) # Fit model @@ -772,6 +786,7 @@ def _build_eval_config(self, param_grid): ), "score_per_session": self._score_per_session, "one_shot_predict": getattr(self, "one_shot_predict", False), + "cs_mode": getattr(self, "cs_mode", None), "param_grid": None, # overridden per-task below if needed } diff --git a/moabb/evaluations/evaluations.py b/moabb/evaluations/evaluations.py index 04abdefc2..04fb1f843 100644 --- a/moabb/evaluations/evaluations.py +++ b/moabb/evaluations/evaluations.py @@ -504,6 +504,7 @@ def __init__(self, *args, cs_mode=CrossSubjectMode.TRAIN, **kwargs): cs_mode = CrossSubjectMode.TRAIN cs_mode = CrossSubjectMode(cs_mode) + self.cs_mode = cs_mode # Manual cv_kwargs still work when the default train-only blockwise # mode is used. @@ -643,6 +644,7 @@ def evaluate( groups[train], calib=calib_md, calibration_labeled=calibration_labeled, + cs_mode=self.cs_mode, ) duration, emissions, task_name = self._fit_cv( From fcdf5c6818ed994dff0ebc1fd90a614c05f4c921 Mon Sep 17 00:00:00 2001 From: Anton Date: Wed, 1 Jul 2026 19:53:44 +0200 Subject: [PATCH 13/15] The example code has been split into two examples. The first is again the RPA, but now it also uses the cs_mode. The second is subject-prototype MDM classifier, which also allows to be used only in TRAIN_TRIALWISE mode. Some small comment updates to protocols.py. --- .../noplot_cross_subject_transfer.py | 148 ---------- .../noplot_cross_subject_transfer_rpa.py | 275 ++++++++++++++++++ .../noplot_cross_subject_transfer_sp_mdm.py | 253 ++++++++++++++++ moabb/evaluations/protocols.py | 5 +- 4 files changed, 532 insertions(+), 149 deletions(-) delete mode 100644 examples/how_to_benchmark/noplot_cross_subject_transfer.py create mode 100644 examples/how_to_benchmark/noplot_cross_subject_transfer_rpa.py create mode 100644 examples/how_to_benchmark/noplot_cross_subject_transfer_sp_mdm.py diff --git a/examples/how_to_benchmark/noplot_cross_subject_transfer.py b/examples/how_to_benchmark/noplot_cross_subject_transfer.py deleted file mode 100644 index 2c272912a..000000000 --- a/examples/how_to_benchmark/noplot_cross_subject_transfer.py +++ /dev/null @@ -1,148 +0,0 @@ -""" -========================================================================= -Cross-subject transfer learning through the standard CrossSubjectEvaluation -========================================================================= - -Target-aware transfer methods (e.g. Riemannian Alignment / RPA) need two things -the ordinary cross-subject split does not give them: - -1. which source subject each training trial belongs to, and -2. controlled access to an (unlabeled) slice of the held-out target subject. - -Both fit into the **standard** :class:`~moabb.evaluations.CrossSubjectEvaluation` -with three scikit-learn-native pieces -- no dedicated evaluation engine: - -* ``cv_kwargs={"calibration_size": ...}`` makes each fold yield - ``(train, calibration, test)`` - (see :class:`~moabb.evaluations.splitters.CrossSubjectSplitter`); -* **metadata routing** (``set_fit_request``) delivers ``subjects`` and the raw - calibration slice to the steps that request them; -* ``Pipeline(transform_input=...)`` transforms that raw slice through the earlier - steps (here ``Covariances``), so a mid-pipeline alignment step receives the - target in the *same representation* as ``X`` (covariances, not raw epochs). -""" - -# License: BSD (3-clause) - -import numpy as np -from pyriemann.estimation import Covariances -from pyriemann.tangentspace import TangentSpace -from pyriemann.utils.base import invsqrtm -from pyriemann.utils.mean import mean_riemann -from sklearn import config_context -from sklearn.base import BaseEstimator, TransformerMixin -from sklearn.linear_model import LogisticRegression -from sklearn.pipeline import make_pipeline - -from moabb.datasets.fake import FakeDataset -from moabb.evaluations import CrossSubjectEvaluation -from moabb.paradigms import LeftRightImagery - - -# --------------------------------------------------------------------------- -# pyriemann's ``Covariances`` is stateless and does not mark itself fitted, -# which ``Pipeline(transform_input=...)`` requires (it calls ``transform`` on -# the sub-pipeline, which runs ``check_is_fitted``). One line fixes it; this -# ideally lands upstream in pyriemann. -# --------------------------------------------------------------------------- -class Covariances_(Covariances): - def __sklearn_is_fitted__(self): - return True - - -# --------------------------------------------------------------------------- -# A minimal target-aware step: Riemannian Alignment (recenter each domain to the -# identity). It consumes ``subjects`` (per-source-subject reference) and the -# unlabeled target covariances (target reference). Both come from the evaluation -# via metadata routing. -# --------------------------------------------------------------------------- -class RiemannianAlignment(TransformerMixin, BaseEstimator): - def fit(self, X, y=None, subjects=None, X_target_unlabeled=None): - X = np.asarray(X) - self.fit_subjects_ = np.asarray(subjects) - # one whitening reference per source subject - self.source_refs_ = { - s: invsqrtm(mean_riemann(X[self.fit_subjects_ == s])) - for s in np.unique(self.fit_subjects_) - } - # X_target_unlabeled arrives as covariances thanks to transform_input - self.target_ref_ = ( - invsqrtm(mean_riemann(np.asarray(X_target_unlabeled))) - if X_target_unlabeled is not None and len(X_target_unlabeled) - else None - ) - return self - - def transform(self, X): - X = np.asarray(X) - if len(X) == len(self.fit_subjects_): # training data: per-source-subject - out = np.empty_like(X) - for s, ref in self.source_refs_.items(): - m = self.fit_subjects_ == s - out[m] = ref @ X[m] @ ref - return out - # held-out target at predict time: recenter by the target reference - ref = self.target_ref_ - if ref is None: - ref = next(iter(self.source_refs_.values())) - return ref @ X @ ref - - -def make_transfer_pipeline(): - with config_context(enable_metadata_routing=True): - align = RiemannianAlignment().set_fit_request( - subjects=True, X_target_unlabeled=True - ) - # transform_input -> the raw target slice is passed through Covariances_() - # before RiemannianAlignment.fit receives it (so it is covariances, not raw). - return make_pipeline( - Covariances_("oas"), - align, - TangentSpace("riemann"), - LogisticRegression(max_iter=500), - transform_input=["X_target_unlabeled"], - ) - - -if __name__ == "__main__": - dataset = FakeDataset( - ["left_hand", "right_hand"], n_subjects=4, n_sessions=2, seed=42 - ) - paradigm = LeftRightImagery() - - # calibration_size=0.2: the first 20% of each held-out subject is the - # unlabeled target slice used by RiemannianAlignment; the remaining 80% is - # scored. calibration_size=0.0 reproduces the ordinary cross-subject split. - evaluation = CrossSubjectEvaluation( - paradigm=paradigm, - datasets=[dataset], - cv_kwargs={"calibration_size": 0.2}, - overwrite=True, - ) - results = evaluation.process({"RA+TS+LR": make_transfer_pipeline()}) - print(results[["subject", "session", "pipeline", "score"]].to_string(index=False)) - - # ------------------------------------------------------------------ - # Trialwise / one-shot prediction with pure scikit-learn: freeze the - # source-trained model and predict each held-out trial in isolation via - # LeaveOneOut (test fold size == 1). No custom wrapper. - # ------------------------------------------------------------------ - from sklearn.frozen import FrozenEstimator - from sklearn.model_selection import LeaveOneOut, cross_val_predict - - X, y, meta = paradigm.get_data(dataset=dataset) - target = meta["subject"].to_numpy() == np.unique(meta["subject"])[-1] - source_model = make_pipeline( - Covariances_("oas"), TangentSpace("riemann"), LogisticRegression(max_iter=500) - ).fit(X[~target], y[~target]) - trialwise = cross_val_predict( - FrozenEstimator(source_model), - X[target], - y[target], - cv=LeaveOneOut(), - method="predict", - ) - print( - "trialwise == blockwise (inductive):", - np.array_equal(trialwise, source_model.predict(X[target])), - ) diff --git a/examples/how_to_benchmark/noplot_cross_subject_transfer_rpa.py b/examples/how_to_benchmark/noplot_cross_subject_transfer_rpa.py new file mode 100644 index 000000000..3f049a169 --- /dev/null +++ b/examples/how_to_benchmark/noplot_cross_subject_transfer_rpa.py @@ -0,0 +1,275 @@ +""" +========================================================================= +Cross-subject transfer learning example using several CrossSubject modes +========================================================================= + +Target-aware transfer methods such as Riemannian Alignment / RPA need two +things that are now available in CrossSubjectEvaluation: + +1. which source subject each training trial belongs to; +2. controlled access to an unlabeled slice of the held-out target subject. + +The available cross-subject modes are defined and documented in +:mod:`moabb.evaluations.protocols`. + +This example demonstrates how these modes extend the standard +:class:`~moabb.evaluations.CrossSubjectEvaluation` to support transfer-learning +methods. + +This example runs the standard source-only ``TRAIN`` mode as a baseline, then +runs RPA with the currently supported unlabeled target-adaptation modes. + +RPA itself is intentionally restricted here to the currently supported unlabeled +target-adaptation modes. The example shows how to select and benchmark those +modes against the baseline. It also demonstrates how an estimator can reject +protocol modes that are not appropriate or not currently supported by the +example, such as source-only, labeled target-calibration modes, or standalone +``TRAIN_TRIALWISE``. + +Modes combining unlabeled target calibration with trialwise scoring could be +meaningful for RPA, but they are not currently represented by the available +``CrossSubjectMode`` presets. + +""" + +import numpy as np +import pandas as pd +from pyriemann.estimation import Covariances +from pyriemann.tangentspace import TangentSpace +from pyriemann.utils.base import invsqrtm +from pyriemann.utils.mean import mean_riemann +from sklearn import config_context +from sklearn.base import BaseEstimator, TransformerMixin +from sklearn.linear_model import LogisticRegression +from sklearn.pipeline import make_pipeline + +from moabb.datasets.fake import FakeDataset +from moabb.evaluations import CrossSubjectEvaluation +from moabb.evaluations.protocols import CrossSubjectMode +from moabb.paradigms import LeftRightImagery + + +# --------------------------------------------------------------------------- +# pyriemann's ``Covariances`` is stateless and does not mark itself fitted, +# which ``Pipeline(transform_input=...)`` requires. It calls ``transform`` on +# the sub-pipeline, which runs ``check_is_fitted``. +# +# This small wrapper should no longer be needed with pyRiemann version > 0.11. +# --------------------------------------------------------------------------- +class Covariances_(Covariances): + def __sklearn_is_fitted__(self): + return True + + +RPA_COMPATIBLE_MODES = { + CrossSubjectMode.TRAIN_AND_TARGET_UNLABELED_20P, + CrossSubjectMode.TRAIN_AND_TARGET_UNLABELED_50P, + CrossSubjectMode.TRAIN_AND_TARGET_UNLABELED_FULL, +} + + +# --------------------------------------------------------------------------- +# A minimal target-aware step: Riemannian Alignment. +# +# It consumes: +# * ``subjects``: source-subject label for each training trial; +# * ``X_target_unlabeled``: unlabeled target calibration covariances; +# * ``cs_mode``: the selected cross-subject protocol. +# +# The estimator refuses modes that do not match the currently implemented RPA +# assumption: unlabeled target adaptation. +# --------------------------------------------------------------------------- +class RiemannianAlignment(TransformerMixin, BaseEstimator): + def fit(self, X, y=None, subjects=None, X_target_unlabeled=None, cs_mode=None): + if cs_mode is None: + raise ValueError( + "RiemannianAlignment requires `cs_mode` fit metadata. " + "Use set_fit_request(cs_mode=True)." + ) + + cs_mode = CrossSubjectMode(cs_mode) + self.cs_mode_ = cs_mode + + if cs_mode not in RPA_COMPATIBLE_MODES: + allowed = ", ".join( + mode.value + for mode in sorted(RPA_COMPATIBLE_MODES, key=lambda m: m.value) + ) + raise ValueError( + "RiemannianAlignment / RPA supports only the currently implemented " + "unlabeled target-adaptation modes in this example. " + f"Got {cs_mode.value!r}. " + "Modes combining unlabeled target calibration with trialwise scoring " + "could be meaningful, but are not currently represented by the " + "available CrossSubjectMode presets. " + f"Allowed modes are: {allowed}." + ) + + if subjects is None: + raise ValueError( + "RiemannianAlignment requires source-subject metadata. " + "Use set_fit_request(subjects=True)." + ) + + if X_target_unlabeled is None or len(X_target_unlabeled) == 0: + raise ValueError( + "RiemannianAlignment requires unlabeled target calibration data. " + "Use CrossSubjectMode.TRAIN_AND_TARGET_UNLABELED_20P, " + "TRAIN_AND_TARGET_UNLABELED_50P, or " + "TRAIN_AND_TARGET_UNLABELED_FULL." + ) + + X = np.asarray(X) + self.fit_subjects_ = np.asarray(subjects) + + # One whitening reference per source subject. + self.source_refs_ = { + subject: invsqrtm(mean_riemann(X[self.fit_subjects_ == subject])) + for subject in np.unique(self.fit_subjects_) + } + + # ``X_target_unlabeled`` arrives as covariances thanks to transform_input. + self.target_ref_ = invsqrtm(mean_riemann(np.asarray(X_target_unlabeled))) + + return self + + def transform(self, X): + X = np.asarray(X) + + # Training data: align each source subject with its own reference. + if len(X) == len(self.fit_subjects_): + out = np.empty_like(X) + for subject, ref in self.source_refs_.items(): + mask = self.fit_subjects_ == subject + out[mask] = ref @ X[mask] @ ref + return out + + # Held-out target data at prediction time: align with target reference. + return self.target_ref_ @ X @ self.target_ref_ + + +def run_mode(dataset, paradigm, mode, pipeline_name, pipeline): + print("\n" + "=" * 78) + print(f"Running mode: {mode.value}") + print(f"Pipeline : {pipeline_name}") + print("=" * 78) + + evaluation = CrossSubjectEvaluation( + paradigm=paradigm, + datasets=[dataset], + cs_mode=mode, + overwrite=True, + suffix=f"rpa_example_{mode.value}_{pipeline_name}", + ) + + results = evaluation.process({pipeline_name: pipeline}) + view = results[["subject", "session", "pipeline", "score"]] + print(view.to_string(index=False)) + return results + + +if __name__ == "__main__": + dataset = FakeDataset( + ["left_hand", "right_hand"], + n_subjects=4, + n_sessions=2, + seed=42, + ) + paradigm = LeftRightImagery() + + all_results = [] + + # ------------------------------------------------------------------ + # 1. Standard source-only baseline. + # + # This is the old ordinary CrossSubjectEvaluation behavior: + # train on source subjects, test on the held-out subject, no target + # calibration data. + # ------------------------------------------------------------------ + baseline_results = run_mode( + dataset=dataset, + paradigm=paradigm, + mode=CrossSubjectMode.TRAIN, # set benchmark mode + pipeline_name="SourceOnly+TS+LR", + pipeline=make_pipeline( + Covariances_("oas"), + TangentSpace("riemann"), + LogisticRegression(max_iter=500), + ), + ) + all_results.append(baseline_results) + + # ------------------------------------------------------------------ + # 2. RPA with compatible unlabeled target-adaptation modes. + # + # These modes differ in how much of the held-out target subject is made + # available as unlabeled calibration/adaptation data. + # ------------------------------------------------------------------ + for mode in sorted(RPA_COMPATIBLE_MODES, key=lambda m: m.value): + with config_context(enable_metadata_routing=True): + align = RiemannianAlignment().set_fit_request( + subjects=True, + X_target_unlabeled=True, + cs_mode=True, + ) + + results = run_mode( + dataset=dataset, + paradigm=paradigm, + mode=mode, + pipeline_name="RPA+TS+LR", + pipeline=make_pipeline( + Covariances_("oas"), + align, + TangentSpace("riemann"), + LogisticRegression(max_iter=500), + transform_input=["X_target_unlabeled"], + ), + ) + all_results.append(results) + + # ------------------------------------------------------------------ + # 3. One intentionally incompatible mode. + # + # RPA is not a labeled target-calibration method. This block shows the + # explicit error message produced by the estimator. + # ------------------------------------------------------------------ + try: + with config_context(enable_metadata_routing=True): + align = RiemannianAlignment().set_fit_request( + subjects=True, + X_target_unlabeled=True, + cs_mode=True, + ) + + run_mode( + dataset=dataset, + paradigm=paradigm, + mode=CrossSubjectMode.TRAIN_AND_TARGET_LABELED_20P, + pipeline_name="RPA+TS+LR", + pipeline=make_pipeline( + Covariances_("oas"), + align, + TangentSpace("riemann"), + LogisticRegression(max_iter=500), + transform_input=["X_target_unlabeled"], + ), + ) + except ValueError as err: + print("\nExpected error for incompatible mode:") + print(err) + + # ------------------------------------------------------------------ + # Optional compact comparison. + # ------------------------------------------------------------------ + summary = pd.concat(all_results, ignore_index=True) + + print("\n" + "=" * 78) + print("Compact comparison") + print("=" * 78) + print( + summary.groupby("pipeline")["score"] + .mean() + .sort_values(ascending=False) + .to_string() + ) \ No newline at end of file diff --git a/examples/how_to_benchmark/noplot_cross_subject_transfer_sp_mdm.py b/examples/how_to_benchmark/noplot_cross_subject_transfer_sp_mdm.py new file mode 100644 index 000000000..962564ef3 --- /dev/null +++ b/examples/how_to_benchmark/noplot_cross_subject_transfer_sp_mdm.py @@ -0,0 +1,253 @@ +""" +============================================================================= +Subject-prototype distance features using CrossSubject TRAIN_TRIALWISE mode +============================================================================= + +Some cross-subject methods need to know which source subject each training trial +belongs to. This example implements a simple subject-aware feature extractor: + +1. for each source subject, compute one Riemannian prototype per class; +2. represent each trial by its distances to all subject-specific prototypes; +3. train a linear SVM on these subject-prototype distance features. + +The available cross-subject modes are defined and documented in +:mod:`moabb.evaluations.protocols`. + +This example demonstrates how the standard +:class:`~moabb.evaluations.CrossSubjectEvaluation` can support subject-aware +methods through metadata routing. The feature extractor requests two fit +metadata fields: + +* ``subjects``: the source-subject label of each training trial; +* ``cs_mode``: the selected cross-subject protocol preset. + +The subject-prototype feature extractor is intentionally restricted here to +TRAIN_TRIALWISE. The example also runs a standard source-only MDM baseline with +TRAIN to compare against the subject-prototype method. Finally, it demonstrates +how the subject-prototype feature extractor can reject a mode that is not +appropriate for this example. + +``TRAIN_TRIALWISE`` means: train on source subjects only, use no target +calibration data, and score the held-out target subject one trial at a time. + +""" + +import numpy as np +import pandas as pd +from pyriemann.classification import MDM +from pyriemann.estimation import Covariances +from pyriemann.utils.distance import distance_riemann +from pyriemann.utils.mean import mean_riemann +from sklearn import config_context +from sklearn.base import BaseEstimator, TransformerMixin +from sklearn.pipeline import make_pipeline +from sklearn.preprocessing import StandardScaler +from sklearn.svm import LinearSVC + +from moabb.datasets.fake import FakeDataset +from moabb.evaluations import CrossSubjectEvaluation +from moabb.evaluations.protocols import CrossSubjectMode +from moabb.paradigms import LeftRightImagery + + +class SubjectPrototypeDistanceFeatures(TransformerMixin, BaseEstimator): + """Distances to subject-specific class prototypes. + + The transformer learns one Riemannian class prototype per source subject. + Each trial is then represented by its distance to every available + subject/class prototype. + + This is a subject-aware, source-only transformer. In this example it is + intentionally restricted to ``CrossSubjectMode.TRAIN_TRIALWISE``. + """ + + def fit(self, X, y=None, subjects=None, cs_mode=None): + if cs_mode is None: + raise ValueError( + "SubjectPrototypeDistanceFeatures requires `cs_mode` fit metadata. " + "Use set_fit_request(cs_mode=True)." + ) + + cs_mode = CrossSubjectMode(cs_mode) + self.cs_mode_ = cs_mode + + if cs_mode != CrossSubjectMode.TRAIN_TRIALWISE: + raise ValueError( + "SubjectPrototypeDistanceFeatures supports only " + f"{CrossSubjectMode.TRAIN_TRIALWISE.value!r} in this example. " + f"Got {cs_mode.value!r}. " + "This transformer is demonstrated as a subject-aware source-only " + "method scored one target trial at a time." + ) + + if subjects is None: + raise ValueError( + "SubjectPrototypeDistanceFeatures requires source-subject metadata. " + "Use set_fit_request(subjects=True)." + ) + + if y is None: + raise ValueError("SubjectPrototypeDistanceFeatures requires labels `y`.") + + X = np.asarray(X) + y = np.asarray(y) + subjects = np.asarray(subjects) + + self.classes_ = np.unique(y) + self.subjects_ = np.unique(subjects) + + # We compute one prototype per (source subject, class), not just one + # prototype per subject. Otherwise the features would describe only + # subject/domain similarity and would not directly encode + # class-discriminative distances. + self.prototype_keys_ = [] + self.prototypes_ = {} + + for subject in self.subjects_: + subject_mask = subjects == subject + + for klass in self.classes_: + mask = subject_mask & (y == klass) + + if np.any(mask): + key = (subject, klass) + self.prototype_keys_.append(key) + self.prototypes_[key] = mean_riemann(X[mask]) + + if not self.prototype_keys_: + raise ValueError("No subject/class prototypes could be estimated.") + + return self + + def transform(self, X): + X = np.asarray(X) + features = np.empty((len(X), len(self.prototype_keys_))) + + for i, cov in enumerate(X): + for j, key in enumerate(self.prototype_keys_): + features[i, j] = distance_riemann(cov, self.prototypes_[key]) + + return features + + +def run_mode(dataset, paradigm, mode, pipeline_name, pipeline): + print("\n" + "=" * 78) + print(f"Running mode: {mode.value}") + print(f"Pipeline : {pipeline_name}") + print("=" * 78) + + evaluation = CrossSubjectEvaluation( + paradigm=paradigm, + datasets=[dataset], + cs_mode=mode, + overwrite=True, + suffix=f"subject_prototype_distance_{mode.value}_{pipeline_name}", + ) + + results = evaluation.process({pipeline_name: pipeline}) + view = results[["subject", "session", "pipeline", "score"]] + print(view.to_string(index=False)) + return results + + +if __name__ == "__main__": + dataset = FakeDataset( + ["left_hand", "right_hand"], + n_subjects=4, + n_sessions=2, + seed=42, + ) + paradigm = LeftRightImagery() + + all_results = [] + + # ------------------------------------------------------------------ + # 1. Standard source-only MDM baseline. + # + # This is the ordinary CrossSubjectEvaluation behavior: + # train on source subjects, test on the held-out subject, no target + # calibration data, and score the target block normally. + # ------------------------------------------------------------------ + baseline_results = run_mode( + dataset=dataset, + paradigm=paradigm, + mode=CrossSubjectMode.TRAIN, + pipeline_name="SourceOnlyMDM", + pipeline=make_pipeline( + Covariances("oas"), + MDM(metric="riemann"), + ), + ) + all_results.append(baseline_results) + + # ------------------------------------------------------------------ + # 2. Subject-prototype distance features in TRAIN_TRIALWISE mode. + # + # The feature extractor receives the source-subject metadata during fit. + # The evaluation then scores the held-out target subject one trial at a + # time. No target calibration data is used. + # ------------------------------------------------------------------ + with config_context(enable_metadata_routing=True): + subject_prototype_features = SubjectPrototypeDistanceFeatures().set_fit_request( + subjects=True, + cs_mode=True, + ) + + prototype_svm_results = run_mode( + dataset=dataset, + paradigm=paradigm, + mode=CrossSubjectMode.TRAIN_TRIALWISE, + pipeline_name="SubjectPrototypeSVM", + pipeline=make_pipeline( + Covariances("oas"), + subject_prototype_features, + StandardScaler(), + LinearSVC(dual=False, max_iter=5000), + ), + ) + all_results.append(prototype_svm_results) + + # ------------------------------------------------------------------ + # 3. One intentionally incompatible mode. + # + # This transformer is restricted to TRAIN_TRIALWISE in this example. + # It is not a target-adaptation method, so it rejects modes that provide + # unlabeled target calibration data. + # ------------------------------------------------------------------ + try: + with config_context(enable_metadata_routing=True): + bad_mode_features = SubjectPrototypeDistanceFeatures().set_fit_request( + subjects=True, + cs_mode=True, + ) + + run_mode( + dataset=dataset, + paradigm=paradigm, + mode=CrossSubjectMode.TRAIN_AND_TARGET_UNLABELED_20P, + pipeline_name="SubjectPrototypeSVM", + pipeline=make_pipeline( + Covariances("oas"), + bad_mode_features, + StandardScaler(), + LinearSVC(dual=False, max_iter=5000), + ), + ) + except ValueError as err: + print("\nExpected error for incompatible mode:") + print(err) + + # ------------------------------------------------------------------ + # Compact comparison for the successful runs. + # ------------------------------------------------------------------ + summary = pd.concat(all_results, ignore_index=True) + + print("\n" + "=" * 78) + print("Compact comparison") + print("=" * 78) + print( + summary.groupby("pipeline")["score"] + .mean() + .sort_values(ascending=False) + .to_string() + ) \ No newline at end of file diff --git a/moabb/evaluations/protocols.py b/moabb/evaluations/protocols.py index 52f3cd4a9..0be957998 100644 --- a/moabb/evaluations/protocols.py +++ b/moabb/evaluations/protocols.py @@ -8,7 +8,9 @@ allowed to be used by an estimator, and how the remaining target data are scored. They make the evaluation protocol explicit, so that different cross-subject and transfer-learning methods can be compared under controlled -and reproducible conditions. +and reproducible conditions. + +These presets are designed specifically to be used in publications. A preset controls two related aspects of the evaluation: @@ -34,6 +36,7 @@ class CrossSubjectMode(str, Enum): # Train only on training/source subjects; no target calibration data is used. # The held-out target test data is predicted blockwise. + # This is the DEFAULT mode and it is not transfer learning aware. TRAIN = "train" # Train only on training/source subjects; no target calibration data is used. From e6fdc44bfde9f32058478f85837ac7a282bafd09 Mon Sep 17 00:00:00 2001 From: Anton Date: Wed, 1 Jul 2026 20:00:14 +0200 Subject: [PATCH 14/15] Improved comments. --- .../how_to_benchmark/noplot_cross_subject_transfer_rpa.py | 4 ---- moabb/evaluations/protocols.py | 6 ++++++ 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/examples/how_to_benchmark/noplot_cross_subject_transfer_rpa.py b/examples/how_to_benchmark/noplot_cross_subject_transfer_rpa.py index 3f049a169..84a38ac9a 100644 --- a/examples/how_to_benchmark/noplot_cross_subject_transfer_rpa.py +++ b/examples/how_to_benchmark/noplot_cross_subject_transfer_rpa.py @@ -26,10 +26,6 @@ example, such as source-only, labeled target-calibration modes, or standalone ``TRAIN_TRIALWISE``. -Modes combining unlabeled target calibration with trialwise scoring could be -meaningful for RPA, but they are not currently represented by the available -``CrossSubjectMode`` presets. - """ import numpy as np diff --git a/moabb/evaluations/protocols.py b/moabb/evaluations/protocols.py index 0be957998..68cad820d 100644 --- a/moabb/evaluations/protocols.py +++ b/moabb/evaluations/protocols.py @@ -28,6 +28,12 @@ Trialwise prediction means that the estimator is scored one target trial at a time. This prevents methods from using statistics of the full target test block during prediction. + +Alternatively, the protocol can be configured manually with +``calibration_size`` and ``calibration_labeled``. For publications and +benchmarks, however, a predefined ``CrossSubjectMode`` preset is strongly +recommended. + """ from enum import Enum From 192e89fe8f721610c2d48d37510340f0d320acae Mon Sep 17 00:00:00 2001 From: Anton Date: Wed, 1 Jul 2026 20:22:26 +0200 Subject: [PATCH 15/15] Small comment update --- moabb/evaluations/base.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/moabb/evaluations/base.py b/moabb/evaluations/base.py index 27544f44b..ac8824474 100644 --- a/moabb/evaluations/base.py +++ b/moabb/evaluations/base.py @@ -101,6 +101,8 @@ def _route_transfer_metadata( """Keep only protocol-allowed transfer metadata requested at ``fit``. ``subjects`` is the per-trial source-subject array. + + ``cs_mode`` is the selected cross-subject protocol preset. ``calib`` is an optional dict with raw calibration data: ``{"X": X[calib_idx], "y": y_calib}``.