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..84a38ac9a --- /dev/null +++ b/examples/how_to_benchmark/noplot_cross_subject_transfer_rpa.py @@ -0,0 +1,271 @@ +""" +========================================================================= +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``. + +""" + +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/base.py b/moabb/evaluations/base.py index e572f5f11..ac8824474 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,94 @@ log = logging.getLogger(__name__) + +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 + 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 + + return getattr(self.estimator, name) + + return _OneShotEstimator(estimator) + + +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. + + ``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}``. + + 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``. + + Steps opt in via ``set_fit_request(...)``. Plain pipelines request nothing, + so this returns ``{}`` and the fit is unchanged. + """ + 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"] + 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} + + # Making the optuna soft dependency @@ -100,6 +190,7 @@ def _evaluate_fold( session, cv_ind, split_metadata=None, + calib_idx=None, ): """Evaluate a single CV fold. Pure function, no shared mutable state. @@ -140,11 +231,16 @@ def _evaluate_fold( score_per_session = config["score_per_session"] 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: 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]) @@ -174,6 +270,26 @@ def _evaluate_fold( tracker = emissions_obj.create_tracker() tracker.start() + # 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": 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=calib, + calibration_labeled=calibration_labeled, + cs_mode=cs_mode, + ) + # Fit model task_name = None emissions = math.nan @@ -181,7 +297,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() @@ -206,7 +323,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 = _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 if score_per_session: @@ -222,7 +340,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 @@ -463,13 +581,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( @@ -565,7 +684,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 +692,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() @@ -667,6 +787,8 @@ def _build_eval_config(self, param_grid): self.emissions.codecarbon_config if _carbonfootprint else None ), "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 } @@ -674,13 +796,16 @@ def _build_eval_config(self, param_grid): 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 +816,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 +844,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..04fb1f843 100644 --- a/moabb/evaluations/evaluations.py +++ b/moabb/evaluations/evaluations.py @@ -7,7 +7,17 @@ from sklearn.preprocessing import LabelEncoder from tqdm import tqdm -from moabb.evaluations.base import BaseEvaluation +from moabb.evaluations.base import ( + BaseEvaluation, + _one_shot_estimator, + _route_transfer_metadata, +) +from moabb.evaluations.protocols import ( + CrossSubjectMode, + is_trialwise_mode, + resolve_cross_subject_mode, + validate_transfer_protocol, +) from moabb.evaluations.splitters import ( CrossSessionSplitter, CrossSubjectSplitter, @@ -455,6 +465,26 @@ 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``. + 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``. ``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``. 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 : CrossSubjectMode or str, default=CrossSubjectMode.TRAIN + Named cross-subject protocol preset. By default, this is the standard + 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 ``TRAIN`` mode. Notes ----- @@ -466,8 +496,51 @@ class CrossSubjectEvaluation(BaseEvaluation): _score_per_session = True _needs_all_subjects = True + 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 = 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. + has_manual_calibration = ( + "calibration_size" in cv_kwargs or "calibration_labeled" in cv_kwargs + ) + + 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_cross_subject_mode(cs_mode) + cv_kwargs["calibration_size"] = params["calibration_size"] + cv_kwargs["calibration_labeled"] = params["calibration_labeled"] + + 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.""" + """Create the CrossSubjectSplitter for parallel evaluation. + + ``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 default_kwargs = {} @@ -492,12 +565,10 @@ 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( @@ -505,64 +576,94 @@ def evaluate( pipelines, dataset, subject, process_pipeline ) ) + if len(run_pipes) == 0: return X, y, metadata = self._load_data( 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) if _carbonfootprint: - # Initialise CodeCarbon per cross-validation tracker = self.emissions.create_tracker() tracker.start() - # Progressbar at subject level - for cv_ind, (train, test) in enumerate( + 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 + + 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 ) - # 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 ) cvclf = clone(clf) + calib_md = None + if len(calib): + calib_md = {"X": X[calib], "y": y[calib]} + + 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, + cs_mode=self.cs_mode, + ) + 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" ) - # Create scorer once per pipeline - scorer = _create_scorer(cvclf, self.paradigm.scoring) + score_estimator = ( + _one_shot_estimator(cvclf) if self.one_shot_predict else cvclf + ) + scorer = _create_scorer(score_estimator, self.paradigm.scoring) - # Evaluate on each session for session in np.unique(sessions[test]): ix = sessions[test] == session @@ -575,9 +676,10 @@ def evaluate( nchan, duration, scorer, - cvclf, + score_estimator, X[test[ix]], y[test[ix]], + split_metadata=split_metadata, ) if _carbonfootprint: @@ -594,11 +696,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..68cad820d --- /dev/null +++ b/moabb/evaluations/protocols.py @@ -0,0 +1,142 @@ +"""Cross-subject transfer protocol presets. + +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. + +These presets are designed specifically to be used in publications. + +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. + +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 + + +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. + # The held-out target test data is predicted one trial at a time. + TRAIN_TRIALWISE = "train_trialwise" + + # 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. + TRAIN_AND_TARGET_UNLABELED_20P = "train_and_target_unlabeled_20p" + + # 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. + TRAIN_AND_TARGET_UNLABELED_50P = "train_and_target_unlabeled_50p" + + # 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. + TRAIN_AND_TARGET_UNLABELED_FULL = "train_and_target_unlabeled_full" + + # 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. + TRAIN_AND_TARGET_LABELED_20P = "train_and_target_labeled_20p" + + # 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. + TRAIN_AND_TARGET_LABELED_50P = "train_and_target_labeled_50p" + + +_CROSS_SUBJECT_MODE_MAP = { + CrossSubjectMode.TRAIN: dict( + calibration_size=0.0, + calibration_labeled=False, + ), + CrossSubjectMode.TRAIN_TRIALWISE: dict( + calibration_size=0.0, + calibration_labeled=False, + ), + CrossSubjectMode.TRAIN_AND_TARGET_UNLABELED_20P: dict( + calibration_size=0.2, + calibration_labeled=False, + ), + CrossSubjectMode.TRAIN_AND_TARGET_UNLABELED_50P: dict( + calibration_size=0.5, + calibration_labeled=False, + ), + CrossSubjectMode.TRAIN_AND_TARGET_UNLABELED_FULL: dict( + calibration_size=1.0, + calibration_labeled=False, + ), + CrossSubjectMode.TRAIN_AND_TARGET_LABELED_20P: dict( + calibration_size=0.2, + calibration_labeled=True, + ), + CrossSubjectMode.TRAIN_AND_TARGET_LABELED_50P: dict( + calibration_size=0.5, + calibration_labeled=True, + ), +} + + +def validate_transfer_protocol(calibration_size, calibration_labeled): + 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__}." + ) + + 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}.") + + 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_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_trialwise_mode(cross_subject_mode): + return CrossSubjectMode(cross_subject_mode) == CrossSubjectMode.TRAIN_TRIALWISE \ No newline at end of file diff --git a/moabb/evaluations/splitters.py b/moabb/evaluations/splitters.py index ab3a7a77f..c6c9d31b1 100644 --- a/moabb/evaluations/splitters.py +++ b/moabb/evaluations/splitters.py @@ -14,6 +14,8 @@ 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__) @@ -482,35 +484,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``. + 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. @@ -519,6 +546,10 @@ 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,52 +557,55 @@ class CrossSubjectSplitter(BaseCrossValidator): def __init__( self, cv_class: type[BaseCrossValidator] = LeaveOneGroupOut, + calibration_size: float = 0.0, + calibration_labeled: bool = False, random_state: int = None, **cv_kwargs, ): + validate_transfer_protocol(calibration_size, calibration_labeled) + 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 @@ -579,18 +613,79 @@ def split(self, y, metadata): # (detected via GroupsConsumerMixin). This avoids the # "The groups parameter is ignored" warning from e.g. TimeSeriesSplit. 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) - yield all_index[train_session_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 target subject -> (train, calibration, test). + calib_idx, test_idx = _split_target_fraction( + target_idx, self.calibration_size + ) + + yield train_idx, calib_idx, test_idx + + else: + yield train_idx, target_idx def get_metadata(self): """Return metadata for the most recent split.""" 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 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. + + 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(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:] + + 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..56d0fe394 100644 --- a/moabb/tests/test_evaluations.py +++ b/moabb/tests/test_evaluations.py @@ -1034,3 +1034,95 @@ 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], + cv_kwargs={"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], + 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) diff --git a/moabb/tests/test_splits.py b/moabb/tests/test_splits.py index 8b6464cd5..d78c8751f 100644 --- a/moabb/tests/test_splits.py +++ b/moabb/tests/test_splits.py @@ -673,3 +673,38 @@ 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_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 + 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_cross_subject_calibration_invalid_size(): + with pytest.raises(ValueError): + CrossSubjectSplitter(calibration_size=1.5)