Within Session splitter

docs/source/evaluations.rst

@@ -19,6 +19,13 @@ Evaluations
     CrossSubjectEvaluation
 
 
+.. autosummary::
+    :toctree: generated/
+    :template: class.rst
+
+    WithinSessionSplitter
+
+
 ------------
 Base & Utils
 ------------

docs/source/whats_new.rst

@@ -17,6 +17,7 @@ Develop branch
 
 Enhancements
 ~~~~~~~~~~~~
+- Adding :class:`moabb.evaluations.splitters.WithinSessionSplitter` (:gh:`664` by `Bruna Lopes_`)
 
 - Update version of pyRiemann to 0.7 (:gh:`671` by `Gregoire Cattan`_)
 - Add columns definitions in the datasets doc (:gh:`672` by `Pierre Guetschel`_)

moabb/evaluations/__init__.py

@@ -9,4 +9,5 @@
     CrossSubjectEvaluation,
     WithinSessionEvaluation,
 )
+from .splitters import WithinSessionSplitter
 from .utils import create_save_path, save_model_cv, save_model_list

moabb/evaluations/metasplitters.py

@@ -0,0 +1,114 @@
+from sklearn.model_selection import BaseCrossValidator
+
+from moabb.evaluations.utils import sort_group
+
+
+class PseudoOnlineSplit(BaseCrossValidator):
+    """Pseudo-online split for evaluation test data.
+
+    It takes into account the time sequence for obtaining the test data, and uses first run,
+    or first #calib_size trials as calibration data, and the rest as evaluation data.
+    Calibration data is important in the context where data alignment or filtering is used on
+    training data.
+
+    OBS: Be careful! Since this inference split is based on time disposition of obtained data,
+    if your data is not organized by time, but by other parameter, such as class, you may want to
+    be extra careful when using this split.
+
+    Parameters
+    ----------
+    calib_size: int
+    Size of calibration set, used if there is just one run.
+
+    Examples
+    --------
+    >>> import numpy as np
+    >>> import pandas as pd
+    >>> from moabb.evaluations.splitters import WithinSessionSplitter
+    >>> from moabb.evaluations.metasplitters import PseudoOnlineSplit
+    >>> X = np.array([[1, 2], [3, 4], [5, 6], [7, 8], [8, 9], [5, 4], [2, 5], [1, 7]])
+    >>> y = np.array([1, 2, 1, 2, 1, 2, 1, 2])
+    >>> subjects = np.array([1, 1, 1, 1, 1, 1, 1, 1])
+    >>> sessions = np.array([0, 0, 0, 0, 1, 1, 1, 1])
+    >>> runs = np.array(['0', '0', '1', '1', '0', '0', '1', '1'])
+    >>> metadata = pd.DataFrame(data={'subject': subjects, 'session': sessions, 'run':runs})
+    >>> posplit = PseudoOnlineSplit
+    >>> csubj = WithinSessionSplitter(cv=posplit, calib_size=1, custom_cv=True)
+    >>> posplit.get_n_splits(metadata)
+    2
+    >>> for i, (train_index, test_index) in enumerate(csubj.split(y, metadata)):
+    >>>     print(f"Fold {i}:")
+    >>>     print(f"  Calibration: index={train_index}, group={subjects[train_index]}, sessions={sessions[train_index]}, runs={runs[train_index]}")
+    >>>     print(f"  Test:  index={test_index}, group={subjects[test_index]}, sessions={sessions[test_index]}, runs={runs[test_index]}")
+
+    Fold 0:
+      Calibration: index=[4], group=[1], sessions=[1], runs=['0']
+      Test:  index=[5], group=[1], sessions=[1], runs=['0']
+    Fold 1:
+      Calibration: index=[6], group=[1], sessions=[1], runs=['1']
+      Test:  index=[7], group=[1], sessions=[1], runs=['1']
+    Fold 2:
+      Calibration: index=[0], group=[1], sessions=[0], runs=['0']
+      Test:  index=[1], group=[1], sessions=[0], runs=['0']
+    Fold 3:
+      Calibration: index=[2], group=[1], sessions=[0], runs=['1']
+      Test:  index=[3], group=[1], sessions=[0], runs=['1']
+
+    """
+
+    def __init__(self, calib_size: int = None):
+        self.calib_size = calib_size
+
+    def get_n_splits(self, metadata):
+        return len(metadata.groupby(["subject", "session"]))
+
+    def split(self, indices, y, metadata=None):
+
+        if metadata is not None:
+            for _, group in metadata.groupby(["subject", "session"]):
+                runs = group.run.unique()
+                if len(runs) > 1:
+                    # To guarantee that the runs are on the right order
+                    runs = sort_group(runs)
+                    for run in runs:
+                        calib_mask = group["run"] == run
+                        calib_ix = group[calib_mask].index
+
+                        if self.calib_size is None:
+                            test_ix = group[~calib_mask].index
+                            yield calib_ix, test_ix
+                            break  # Take the fist run as calibration
+                        else:
+                            mask_run = group["run"] == run
+                            if self.calib_size > len(mask_run):
+                                raise ValueError(
+                                    "Calibration size must be less than number of runs."
+                                )
+                            yield calib_ix[: self.calib_size], calib_ix[self.calib_size :]
+
+                # Else, get the first #calib_size trials
+                else:
+                    if self.calib_size is None:
+                        raise ValueError(
+                            "Data contains just one run. Need to provide calibration size."
+                        )
+                    # Take first #calib_size samples as calibration
+                    calib_size = self.calib_size
+                    if calib_size < len(group):
+                        raise ValueError(
+                            "Data contains just one run. Need to provide calibration size."
+                        )
+
+                    # Get indexes of respective groups
+                    calib_ix = group[:calib_size].index
+                    test_ix = group[calib_size:].index
+
+                    yield calib_ix, test_ix
+        # If not
+        else:
+            if self.calib_size is None:
+                raise ValueError(
+                    "No metadata information. Need to provide calibration size."
+                )
+            calib_size = self.calib_size
+            yield list(range(calib_size)), list(range(calib_size, len(indices)))

moabb/evaluations/splitters.py

@@ -0,0 +1,134 @@
+
+from sklearn.model_selection import BaseCrossValidator, StratifiedKFold
+from sklearn.utils import check_random_state
+
+from moabb.evaluations.metasplitters import PseudoOnlineSplit
+
+
+class WithinSessionSplitter(BaseCrossValidator):
+    """Data splitter for within session evaluation.
+
+    Within-session evaluation uses k-fold cross_validation to determine train
+    and test sets for each subject in each session. This splitter
+    assumes that all data from all subjects is already known and loaded.
+
+    .. image:: images/withinsess.png
+        :alt: The schematic diagram of the WithinSession split
+        :align: center
+
+
+    Parameters
+    ----------
+    n_folds : int
+        Number of folds. Must be at least 2. If
+    random_state: int, RandomState instance or None, default=None
+        Controls the randomness of splits. Only used when `shuffle` is True.
+        Pass an int for reproducible output across multiple function calls.
+    shuffle : bool, default=True
+        Whether to shuffle each class's samples before splitting into batches.
+        Note that the samples within each split will not be shuffled.
+    custom_cv: bool, default=False
+        Indicates if you are using PseudoOnlineSplit as cv strategy
+    calib_size: int, default=None
+        Size of calibration set if custom_cv==True
+    cv: cros-validation object, default=StratifiedKFold
+        Inner cross-validation strategy for splitting the sessions. Be careful, if
+        PseudoOnlineSplit is used, it will return calibration and test indexes.
+
+
+    Examples
+    -----------
+
+    >>> import pandas as pd
+    >>> import numpy as np
+    >>> from moabb.evaluations.splitters import WithinSessionSplitter
+    >>> X = np.array([[1, 2], [3, 4], [5, 6], [1,4], [7, 4], [5, 8], [0,3], [2,4]])
+    >>> y = np.array([1, 2, 1, 2, 1, 2, 1, 2])
+    >>> subjects = np.array([1, 1, 1, 1, 1, 1, 1, 1])
+    >>> sessions = np.array(['T', 'T', 'T', 'T', 'E', 'E', 'E', 'E'])
+    >>> metadata = pd.DataFrame(data={'subject': subjects, 'session': sessions})
+    >>> csess = WithinSessionSplitter(n_folds=2)
+    >>> csess.get_n_splits(metadata)
+    4
+    >>> for i, (train_index, test_index) in enumerate(csess.split(y, metadata)):
+    ...    print(f"Fold {i}:")
+    ...    print(f"  Train: index={train_index}, group={subjects[train_index]}, session={sessions[train_index]}")
+    ...    print(f"  Test:  index={test_index}, group={subjects[test_index]}, sessions={sessions[test_index]}")
+    Fold 0:
+      Train: index=[4 7], group=[1 1], session=['E' 'E']
+      Test:  index=[5 6], group=[1 1], sessions=['E' 'E']
+    Fold 1:
+      Train: index=[5 6], group=[1 1], session=['E' 'E']
+      Test:  index=[4 7], group=[1 1], sessions=['E' 'E']
+    Fold 2:
+      Train: index=[2 3], group=[1 1], session=['T' 'T']
+      Test:  index=[0 1], group=[1 1], sessions=['T' 'T']
+    Fold 3:
+      Train: index=[0 1], group=[1 1], session=['T' 'T']
+      Test:  index=[2 3], group=[1 1], sessions=['T' 'T']
+
+    """
+
+    def __init__(
+        self,
+        cv=StratifiedKFold,
+        custom_cv=False,
+        n_folds: int = 5,
+        random_state: int = 42,
+        shuffle: bool = True,
+        calib_size: int = None,
+    ):
+        self.n_folds = n_folds
+        self.shuffle = shuffle
+        self.random_state = check_random_state(random_state) if shuffle else None
+        self.cv = cv
+        self.calib_size = calib_size
+        self.custom_cv = custom_cv
+
+    def get_n_splits(self, metadata):
+        num_sessions_subjects = metadata.groupby(["subject", "session"]).ngroups
+        return (
+            self.cv.get_n_splits(metadata)
+            if self.custom_cv
+            else self.n_folds * num_sessions_subjects
+        )
+
+    def split(self, y, metadata, **kwargs):
+        all_index = metadata.index.values
+        subjects = metadata["subject"].unique()
+
+        # Shuffle subjects if required
+        if self.shuffle:
+            self.random_state.shuffle(subjects)
+
+        for i, subject in enumerate(subjects):
+            subject_mask = metadata.subject == subject
+            subject_indices = all_index[subject_mask]
+            subject_metadata = metadata[subject_mask]
+            sessions = subject_metadata.session.unique()
+
+            # Shuffle sessions if required
+            if self.shuffle:
+                self.random_state.shuffle(sessions)
+
+            for j, session in enumerate(sessions):
+                session_mask = subject_metadata.session == session
+                indices = subject_indices[session_mask]
+                group_y = y[indices]
+
+                # Handle custom splitter
+                if isinstance(self.cv(), PseudoOnlineSplit):
+                    splitter = self.cv(calib_size=self.calib_size)
+                    for calib_ix, test_ix in splitter.split(
+                        indices, group_y, subject_metadata[session_mask]
+                    ):
+                        yield indices[calib_ix], indices[test_ix]
+                else:
+                    # Handle standard CV like StratifiedKFold
+                    splitter = self.cv(
+                        n_splits=self.n_folds,
+                        shuffle=self.shuffle,
+                        random_state=self.random_state.randint(0, 2**10),
+                    )
+                    for train_ix, test_ix in splitter.split(indices, group_y):
+                        yield indices[train_ix], indices[test_ix]

moabb/evaluations/utils.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+import re
 from pathlib import Path
 from pickle import HIGHEST_PROTOCOL, dump
 from typing import Sequence
 
+import numpy as np
 from numpy import argmax
 from sklearn.pipeline import Pipeline
 
@@ -222,6 +224,17 @@ def create_save_path(
         print("No hdf5_path provided, models will not be saved.")
 
 
+def sort_group(groups):
+    runs_sort = []
+    pattern = r"([0-9]+)(|[a-zA-Z]+[a-zA-Z0-9]*)"
+    for i, group in enumerate(groups):
+        index, description = re.fullmatch(pattern, group).groups()
+        index = int(index)
+        runs_sort.append(index)
+    sorted_ix = np.argsort(runs_sort)
+    return groups[sorted_ix]
+
+
 def _convert_sklearn_params_to_optuna(param_grid: dict) -> dict:
     """
     Function to convert the parameter in Optuna format. This function will

moabb/tests/splits.py

@@ -0,0 +1,58 @@
+import numpy as np
+import pytest
+from sklearn.model_selection import StratifiedKFold
+from sklearn.utils import check_random_state
+
+from moabb.datasets.fake import FakeDataset
+from moabb.evaluations.splitters import WithinSessionSplitter
+from moabb.paradigms.motor_imagery import FakeImageryParadigm
+
+
+dataset = FakeDataset(["left_hand", "right_hand"], n_subjects=3, seed=12)
+paradigm = FakeImageryParadigm()
+
+
+# Split done for the Within Session evaluation
+def eval_split_within_session(shuffle, random_state):
+    random_state = check_random_state(random_state) if shuffle else None
+    for subject in dataset.subject_list:
+        X, y, metadata = paradigm.get_data(dataset=dataset, subjects=[subject])
+        sessions = metadata.session
+        for session in np.unique(sessions):
+            ix = sessions == session
+            cv = StratifiedKFold(n_splits=5, shuffle=shuffle, random_state=random_state)
+            X_, metadata_, y_ = X[ix], y[ix], metadata[ix]
+            for train, test in cv.split(y_, metadata_):
+                yield X_[train], X_[test]
+
+
+@pytest.mark.parametrize("shuffle", [True, False])
+@pytest.mark.parametrize("random_state", [0, 42])
+def test_within_session(shuffle, random_state):
+    X, y, metadata = paradigm.get_data(dataset=dataset)
+
+    split = WithinSessionSplitter(n_folds=5, shuffle=shuffle, random_state=random_state)
+
+    for (X_train_t, X_test_t), (train, test) in zip(
+        eval_split_within_session(shuffle=shuffle, random_state=random_state),
+        split.split(y, metadata),
+    ):
+        X_train, X_test = X[train], X[test]
+
+        # Check if the output is the same as the input
+        assert np.array_equal(X_train, X_train_t)
+        assert np.array_equal(X_test, X_test_t)
+
+
+def test_is_shuffling():
+    X, y, metadata = paradigm.get_data(dataset=dataset)
+
+    split = WithinSessionSplitter(n_folds=5, shuffle=False)
+    split_shuffle = WithinSessionSplitter(n_folds=5, shuffle=True, random_state=3)
+
+    for (train, test), (train_shuffle, test_shuffle) in zip(
+        split.split(y, metadata), split_shuffle.split(y, metadata)
+    ):
+        # Check if the output is the same as the input
+        assert np.array_equal(train, train_shuffle) == False
+        assert np.array_equal(test, test_shuffle) == False