diff --git a/docs/source/whats_new.rst b/docs/source/whats_new.rst index 3883e47ec..bdb3141a2 100644 --- a/docs/source/whats_new.rst +++ b/docs/source/whats_new.rst @@ -63,6 +63,7 @@ Requirements Bugs ~~~~ +- Keep :class:`moabb.evaluations.CrossSubjectEvaluation` result provenance subject-specific when a grouped cross-validation fold holds out multiple subjects at once. The estimator is still fitted once per fold, while scores, cache identities, and saved-model paths are emitted per held-out subject and session instead of assigning the whole fold to its first subject (by `lindicaphxag-tech`_). - Fix :func:`moabb.datasets.Dataset.convert_to_bids` crashing on datasets whose MOABB run-label suffix is literally ``"calibration"`` or ``"crosstalk"`` (Wang2025, Leeuwis2021, Brandl2020, Romani_BF2025_ERP). After the ``acq-`` diff --git a/moabb/evaluations/base.py b/moabb/evaluations/base.py index 9552bbd0d..7d8c88b11 100644 --- a/moabb/evaluations/base.py +++ b/moabb/evaluations/base.py @@ -139,6 +139,7 @@ def _evaluate_fold( cv_ind, split_metadata=None, calib_idx=None, + score_subjects=None, ): """Evaluate a single CV fold. Pure function, no shared mutable state. @@ -166,6 +167,10 @@ def _evaluate_fold( Cross-validation fold index. split_metadata : dict | None Extra metadata from the splitter. + score_subjects : sequence | None + Held-out subjects whose results are still missing for this pipeline. + When provided, the fold is fitted once and scored separately for these + subjects. Returns ------- @@ -177,6 +182,7 @@ def _evaluate_fold( param_grid = config["param_grid"] additional_columns = config["additional_columns"] score_per_session = config["score_per_session"] + score_per_subject = config["score_per_subject"] mne_labels = config["mne_labels"] codecarbon_config = config["codecarbon_config"] trialwise = config.get("trialwise", False) @@ -257,37 +263,66 @@ def _evaluate_fold( if tracker is not None: tracker.stop() - # Optionally save model + # Optionally save model. A multi-subject cross-subject fold still trains + # exactly one model, but each held-out subject owns its result/cache path. hdf5_path = config["hdf5_path"] eval_type = config["eval_type"] + result_subjects = ( + list(score_subjects) + if score_per_subject and score_subjects is not None + else [subject] + ) if hdf5_path is not None and config["save_model"]: - model_save_path = _create_save_path( - hdf5_path=hdf5_path, - code=dataset.code, - subject=subject, - session="" if score_per_session else session, - name=pipeline_name, - grid=is_search, - eval_type=eval_type, - paradigm=config["paradigm"], - suffix=config["suffix"], - ) - _save_model_cv(model=cvclf, save_path=model_save_path, cv_index=str(cv_ind)) + for save_subject in result_subjects: + model_save_path = _create_save_path( + hdf5_path=hdf5_path, + code=dataset.code, + subject=save_subject, + session="" if score_per_session else session, + name=pipeline_name, + grid=is_search, + eval_type=eval_type, + paradigm=config["paradigm"], + suffix=config["suffix"], + ) + _save_model_cv(model=cvclf, save_path=model_save_path, cv_index=str(cv_ind)) scorer = None if trialwise else _create_scorer(cvclf, scoring) - # Build score groups: per-session or full test set - if score_per_session: - test_sessions = metadata.iloc[test_idx]["session"].values - score_groups = [ - (test_idx[test_sessions == s], y_test[test_sessions == s], s) - for s in np.unique(test_sessions) - ] + # Build score groups. Cross-subject folds can contain multiple held-out + # subjects when using GroupKFold or another custom grouped splitter. + test_meta = metadata.iloc[test_idx] + test_sessions = test_meta["session"].to_numpy() + test_subjects = test_meta["subject"].to_numpy() + score_groups = [] + + if score_per_subject: + selected_subjects = ( + np.asarray(score_subjects) + if score_subjects is not None + else np.unique(test_subjects) + ) + for group_subject in selected_subjects: + subject_mask = test_subjects == group_subject + if score_per_session: + for group_session in np.unique(test_sessions[subject_mask]): + mask = subject_mask & (test_sessions == group_session) + score_groups.append( + (test_idx[mask], y_test[mask], group_subject, group_session) + ) + else: + score_groups.append( + (test_idx[subject_mask], y_test[subject_mask], group_subject, session) + ) + elif score_per_session: + for group_session in np.unique(test_sessions): + mask = test_sessions == group_session + score_groups.append((test_idx[mask], y_test[mask], subject, group_session)) else: - score_groups = [(test_idx, y_test, session)] + score_groups = [(test_idx, y_test, subject, session)] results = [] - for group_idx, group_y, group_session in score_groups: + for group_idx, group_y, group_subject, group_session in score_groups: is_error = False try: if trialwise: @@ -303,7 +338,7 @@ def _evaluate_fold( res = { "time": duration, "dataset": dataset, - "subject": subject, + "subject": group_subject, "session": group_session, "n_samples": len(train_idx), "n_samples_test": len(group_y), @@ -416,6 +451,7 @@ class BaseEvaluation(ABC): search = False _eval_type = None _score_per_session = False + _score_per_subject = False _needs_all_subjects = False _aggregate_folds = False @@ -759,6 +795,7 @@ def _build_eval_config(self, param_grid): self.emissions.codecarbon_config if _carbonfootprint else None ), "score_per_session": self._score_per_session, + "score_per_subject": self._score_per_subject, "trialwise": getattr(self, "trialwise", False), "param_grid": None, # overridden per-task below if needed } @@ -814,14 +851,32 @@ def _build_task_list( for cv_ind, train_idx, calib_idx, test_idx, split_meta in fold_preview: test_meta = metadata.iloc[test_idx] self._validate_test_fold_metadata(test_meta) - subject = test_meta["subject"].iloc[0] - - if subject not in work_plan: - continue - run_pipes = work_plan[subject] session = test_meta["session"].iloc[0] - for name, clf in run_pipes.items(): + if self._score_per_subject: + test_subjects = np.unique(test_meta["subject"].to_numpy()) + pipeline_subjects = { + name: [ + subject + for subject in test_subjects + if name in work_plan.get(subject, {}) + ] + for name in pipelines + } + task_specs = [ + (name, work_plan[subjects[0]][name], subjects[0], subjects) + for name, subjects in pipeline_subjects.items() + if subjects + ] + else: + subject = test_meta["subject"].iloc[0] + if subject not in work_plan: + continue + task_specs = [ + (name, clf, subject, None) for name, clf in work_plan[subject].items() + ] + + for name, clf, subject, score_subjects in task_specs: task_config = dict(config) if param_grid is not None and name in param_grid: task_param_grid = {name: deepcopy(param_grid[name])} @@ -841,6 +896,7 @@ def _build_task_list( "cv_ind": cv_ind, "split_metadata": split_meta, "calib_idx": calib_idx, + "score_subjects": score_subjects, } ) return tasks diff --git a/moabb/evaluations/evaluations.py b/moabb/evaluations/evaluations.py index c8f845563..13819866a 100644 --- a/moabb/evaluations/evaluations.py +++ b/moabb/evaluations/evaluations.py @@ -523,6 +523,7 @@ class CrossSubjectEvaluation(BaseEvaluation): _eval_type = "CrossSubject" _score_per_session = True + _score_per_subject = True _needs_all_subjects = True def __init__( diff --git a/moabb/tests/test_evaluations.py b/moabb/tests/test_evaluations.py index 3938b56c0..02a1d8d5a 100644 --- a/moabb/tests/test_evaluations.py +++ b/moabb/tests/test_evaluations.py @@ -1785,6 +1785,151 @@ def test_cross_subject_process_structure(self, tmp_path): # 2 subjects × 2 sessions = 4 results assert len(results) == 4 + def test_cross_subject_multisubject_fold_preserves_provenance(self, tmp_path): + """A fold holding out multiple subjects is scored per subject/session.""" + ds = FakeDataset(["left_hand", "right_hand"], n_subjects=4, n_sessions=2, seed=23) + paradigm = FakeImageryParadigm() + evaluation = ev.CrossSubjectEvaluation( + paradigm=paradigm, + datasets=[ds], + n_splits=2, + overwrite=True, + hdf5_path=str(tmp_path / "cross_subject_multisubject"), + save_model=True, + ) + + results = evaluation.process(pipelines) + + assert len(results) == 8 + assert set(results["subject"]) == {str(subject) for subject in ds.subject_list} + assert (results.groupby(["subject", "session"], observed=True).size() == 1).all() + + _, _, metadata = paradigm.get_data(ds) + metadata = metadata.copy() + metadata["subject"] = metadata["subject"].astype(str) + expected_sizes = metadata.groupby(["subject", "session"], observed=True).size() + actual_sizes = results.set_index(["subject", "session"])["samples_test"] + actual_sizes = actual_sizes.reindex(expected_sizes.index) + np.testing.assert_array_equal(actual_sizes.to_numpy(), expected_sizes.to_numpy()) + + for subject in ds.subject_list: + model_dir = ( + tmp_path + / "cross_subject_multisubject" + / "Models_CrossSubject" + / type(paradigm).__name__ + / ds.code + / str(subject) + / "C" + ) + assert list(model_dir.glob("fitted_model_*.pkl")) + + def test_cross_subject_multisubject_fold_respects_partial_work_plan(self, tmp_path): + """A cached first subject must not suppress work for another held-out subject.""" + ds = FakeDataset(["left_hand", "right_hand"], n_subjects=4, n_sessions=2, seed=42) + evaluation = ev.CrossSubjectEvaluation( + paradigm=FakeImageryParadigm(), + datasets=[ds], + hdf5_path=str(tmp_path / "cross_subject_partial_cache"), + n_splits=2, + ) + metadata = pd.DataFrame( + {"subject": np.repeat([1, 2, 3, 4], 2), "session": ["0", "1"] * 4} + ) + y = np.array([0, 1] * 4) + splitter = evaluation._create_splitter() + folds = list(splitter.split(y, metadata)) + held_out = list(pd.unique(metadata.iloc[folds[0][1]]["subject"])) + assert len(held_out) == 2 + + target_subject = held_out[1] + pipeline = Dummy(strategy="most_frequent") + work_plan = {target_subject: {"dummy": pipeline}} + tasks = evaluation._build_task_list( + ds, + None, + y, + metadata, + evaluation._create_splitter(), + work_plan, + {"dummy": pipeline}, + None, + ) + + assert len(tasks) == 1 + assert tasks[0]["subject"] == target_subject + assert tasks[0]["score_subjects"] == [target_subject] + assert target_subject in set(metadata.iloc[tasks[0]["test_idx"]]["subject"]) + + def test_cross_subject_partial_cache_fits_remaining_fold_once(self, tmp_path): + """A partially cached grouped fold is fitted once for the missing subject.""" + + class CountingDummy(Dummy): + fit_calls = 0 + + def fit(self, X, y, sample_weight=None): + type(self).fit_calls += 1 + return super().fit(X, y, sample_weight=sample_weight) + + ds = FakeDataset(["left_hand", "right_hand"], n_subjects=4, n_sessions=2, seed=42) + paradigm = FakeImageryParadigm() + evaluation = ev.CrossSubjectEvaluation( + paradigm=paradigm, + datasets=[ds], + hdf5_path=str(tmp_path / "cross_subject_cached_fold"), + n_splits=2, + ) + pipeline = CountingDummy(strategy="most_frequent") + pipeline_dict = {"counting": pipeline} + + X, y, metadata = paradigm.get_data(ds) + folds = list(evaluation._create_splitter().split(y, metadata)) + held_out_by_fold = [ + list(pd.unique(metadata.iloc[test_idx]["subject"])) for _, test_idx in folds + ] + assert all(len(subjects) == 2 for subjects in held_out_by_fold) + + cached_target = held_out_by_fold[0][0] + missing_target = held_out_by_fold[0][1] + cached_subjects = [cached_target, *held_out_by_fold[1]] + + process_pipeline = paradigm.make_process_pipelines(ds)[0] + for cached_subject in cached_subjects: + cached_result = { + "score": 0.0, + "time": 0.0, + "dataset": ds, + "subject": cached_subject, + "session": "cached", + "n_samples": 1, + "n_channels": X.shape[1], + "carbon_emission": 0.0, + } + evaluation.results.add( + {"counting": cached_result}, + pipelines=pipeline_dict, + process_pipeline=process_pipeline, + ) + + before = evaluation.results.to_dataframe( + pipelines=pipeline_dict, process_pipeline=process_pipeline + ) + before_counts = before.groupby("subject", observed=True).size().to_dict() + + CountingDummy.fit_calls = 0 + after = evaluation.process(pipeline_dict) + + assert CountingDummy.fit_calls == 1 + after_counts = after.groupby("subject", observed=True).size().to_dict() + increased_subjects = { + subject + for subject, count in after_counts.items() + if count > before_counts.get(subject, 0) + } + assert increased_subjects == {str(missing_target)} + assert after_counts[str(cached_target)] == before_counts[str(cached_target)] + assert after_counts[str(missing_target)] == ds.n_sessions + def test_learning_curve_parallel(self, tmp_path): """LearningCurve evaluation via parallel process().""" evaluation = ev.WithinSessionEvaluation(