bci_essentials.paradigm.ssvep_paradigm

  1import numpy as np
  2
  3from .paradigm import Paradigm
  4
  5
  6class SsvepParadigm(Paradigm):
  7    """
  8    SSVEP paradigm.
  9
 10    TODO Determine how to automatically set the target frequencies in the classifier from here.
 11    """
 12
 13    def __init__(
 14        self,
 15        filters=[5, 30],
 16        iterative_training=False,
 17        live_update=False,
 18        buffer_time=0.01,
 19    ):
 20        """
 21        Parameters
 22        ----------
 23        filters : list of floats, *optional*
 24            Filter bands.
 25            - Default is `[5, 30]`.
 26        iterative_training : bool, *optional*
 27            Flag to indicate if the classifier will be updated iteratively.
 28            - Default is `False`.
 29        live_update : bool, *optional*
 30            Flag to indicate if the classifier will be used to provide
 31            live updates on trial classification.
 32            - Default is `False`.
 33        buffer_time : float, *optional*
 34            Defines the time in seconds after an epoch for which we require EEG data to ensure that all EEG is present in that epoch.
 35            - Default is `0.01`.
 36        """
 37        super().__init__(filters)
 38
 39        self.live_update = live_update
 40        self.iterative_training = iterative_training
 41
 42        if self.live_update:
 43            self.classify_each_epoch = True
 44            self.classify_each_trial = False
 45        else:
 46            self.classify_each_trial = True
 47            self.classify_each_epoch = False
 48
 49        self.buffer_time = buffer_time
 50
 51        self.paradigm_name = "SSVEP"
 52
 53    def get_eeg_start_and_end_times(self, markers, timestamps):
 54        """
 55        Get the start and end times of the EEG data based on the markers.
 56
 57        Parameters
 58        ----------
 59        markers : list of str
 60            List of markers.
 61        timestamps : list of float
 62            List of timestamps.
 63
 64        Returns
 65        -------
 66        float
 67            Start time.
 68        float
 69            End time.
 70        """
 71        start_time = timestamps[0] - self.buffer_time
 72
 73        end_time = timestamps[-1] + float(markers[-1].split(",")[3]) + self.buffer_time
 74
 75        return start_time, end_time
 76
 77    def process_markers(self, markers, marker_timestamps, eeg, eeg_timestamps, fsample):
 78        """
 79        This takes in the markers and EEG data and processes them into epochs according to the SSVEP paradigm.
 80
 81        Parameters
 82        ----------
 83        markers : list of str
 84            List of markers.
 85        marker_timestamps : list of float
 86            List of timestamps.
 87        eeg : np.array
 88            EEG data. Shape is (n_channels, n_samples).
 89        eeg_timestamps : np.array
 90            EEG timestamps. Shape is (n_samples).
 91        fsample : float
 92            Sampling frequency.
 93
 94        Returns
 95        -------
 96        np.array
 97            Processed EEG data. Shape is (n_epochs, n_channels, n_samples).
 98        np.array
 99            Labels. Shape is (n_epochs).
100        """
101
102        # Initialize y
103        y = np.zeros(len(markers), dtype=int)
104
105        for i, marker in enumerate(markers):
106            marker = marker.split(",")
107            label = int(marker[2])
108            epoch_length = float(marker[3])
109            self.target_freqs = [float(x) for x in marker[4:]]
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        return X, y
141
142    # TODO: Implement this
143    def check_compatibility(self):
144        pass