bci_essentials.paradigm.mi_paradigm

  1import numpy as np
  2
  3from .paradigm import Paradigm
  4
  5
  6class MiParadigm(Paradigm):
  7    """
  8    MI paradigm.
  9    """
 10
 11    def __init__(
 12        self,
 13        filters=[5, 30],
 14        iterative_training=False,
 15        live_update=False,
 16        buffer_time=0.01,
 17    ):
 18        """
 19        Parameters
 20        ----------
 21        filters : list of floats, *optional*
 22            Filter bands.
 23            - Default is `[5, 30]`.
 24        iterative_training : bool, *optional*
 25            Flag to indicate if the classifier will be updated iteratively.
 26            - Default is `False`.
 27        live_update : bool, *optional*
 28            Flag to indicate if the classifier will be used to provide
 29            live updates on trial classification.
 30            - Default is `False`.
 31        buffer_time : float, *optional*
 32            Defines the time in seconds after an epoch for which we require EEG data to ensure that all EEG is present in that epoch.
 33            - Default is `0.01`.
 34        """
 35        super().__init__(filters)
 36
 37        self.live_update = live_update
 38        self.iterative_training = iterative_training
 39
 40        self.lowcut = filters[0]
 41        self.highcut = filters[1]
 42
 43        if self.live_update:
 44            self.classify_each_epoch = True
 45            self.classify_each_trial = False
 46        else:
 47            self.classify_each_trial = True
 48            self.classify_each_epoch = False
 49
 50        self.buffer_time = buffer_time
 51
 52        self.paradigm_name = "MI"
 53
 54    def get_eeg_start_and_end_times(self, markers, timestamps):
 55        """
 56        Get the start and end times of the EEG data based on the markers.
 57
 58        Parameters
 59        ----------
 60        markers : list of str
 61            List of markers.
 62        timestamps : list of float
 63            List of timestamps.
 64
 65        Returns
 66        -------
 67        float
 68            Start time.
 69        float
 70            End time.
 71        """
 72        start_time = timestamps[0] - self.buffer_time
 73
 74        end_time = timestamps[-1] + float(markers[-1].split(",")[-1]) + self.buffer_time
 75
 76        return start_time, end_time
 77
 78    def process_markers(self, markers, marker_timestamps, eeg, eeg_timestamps, fsample):
 79        """
 80        This takes in the markers and EEG data and processes them into epochs accordingt to the MI paradigm.
 81
 82        Parameters
 83        ----------
 84        markers : list of str
 85            List of markers.
 86        marker_timestamps : list of float
 87            List of timestamps.
 88        eeg : np.array
 89            EEG data. Shape is (n_channels, n_samples).
 90        eeg_timestamps : np.array
 91            EEG timestamps. Shape is (n_samples).
 92        fsample : float
 93            Sampling frequency.
 94
 95        Returns
 96        -------
 97        np.array
 98            Processed EEG data. Shape is (n_epochs, n_channels, n_samples).
 99        np.array
100            Labels. Shape is (n_epochs).
101        """
102
103        # Initialize y
104        y = np.zeros(len(markers), dtype=int)
105
106        for i, marker in enumerate(markers):
107            marker = marker.split(",")
108            label = int(marker[2])
109            epoch_length = float(marker[3])
110
111            n_channels, _ = eeg.shape
112
113            marker_timestamp = marker_timestamps[i]
114
115            # Subtract the marker timestamp from the EEG timestamps so that 0 becomes the marker onset
116            marker_eeg_timestamps = eeg_timestamps - marker_timestamp
117
118            # Create the epoch time vector
119            epoch_time = np.arange(0, epoch_length, 1 / fsample)
120
121            # Initialize the EEG data array
122            epoch_eeg = np.zeros((1, n_channels, len(epoch_time)))
123
124            # Interpolate the EEG data to the epoch time vector for each channel
125            for c in range(n_channels):
126                epoch_eeg[0, c, :] = np.interp(
127                    epoch_time, marker_eeg_timestamps, eeg[c, :]
128                )
129
130            epoch_eeg[0, :, :] = super()._preprocess(
131                epoch_eeg[0, :, :], fsample, self.lowcut, self.highcut
132            )
133
134            if i == 0:
135                X = epoch_eeg
136            else:
137                X = np.concatenate((X, epoch_eeg), axis=0)
138
139            y[i] = label
140
141        return X, y
142
143    # TODO: Implement this to check compatibility between paradigm and classifier
144    def check_compatibility(self):
145        pass