bci_essentials.io.sources

  1from abc import ABC, abstractmethod
  2
  3from ..utils.logger import Logger  # Logger wrapper
  4
  5# Instantiate a logger for the module at the default level of logging.INFO
  6# Logs to bci_essentials.__module__) where __module__ is the name of the module
  7logger = Logger(name=__name__)
  8
  9
 10class MarkerSource(ABC):
 11    """MarkerSource objects send time synchronized markers/commands to bci_controller."""
 12
 13    @property
 14    @abstractmethod
 15    def name(self) -> str:
 16        """Name of the marker source"""
 17        pass
 18
 19    @abstractmethod
 20    def get_markers(self) -> tuple[list[list], list]:
 21        """Get marker/command data and timestamps since last call
 22
 23        Returns
 24        -------
 25        marker_data : tuple(marker, timestemps)
 26            A tuple of (markers, timestamps):
 27            - markers : list[list]
 28                - A list of samples, where each sample corresponds to a timestamp.
 29                - Each sample is a list with a single string element that represents a command or a marker.
 30                - The string is formatted as follows:
 31                    - command = an arbitrary string, ex: "Trial Started"
 32                    - marker = "paradigm, num options, label number, trial length"
 33            - timestamps : list[float]
 34                - A list timestamps (float in seconds) corresponding to the markers
 35        """
 36        pass
 37
 38    @abstractmethod
 39    def time_correction(self) -> float:
 40        """Get the current time correction for timestamps.
 41
 42        Returns
 43        -------
 44        time_correction : float
 45            The current time correction estimate (seconds).
 46            - This is the number that needs to be added to a time tamp that was remotely generated via local_clock() to map it into the local clock domain of the machine.
 47
 48        """
 49        pass
 50
 51
 52class EegSource(ABC):
 53    """EegSource objects produce samples of EEG for use in bci_controller.
 54
 55    It can be used to represent an BCI headset providing EEG data, or it could be a source
 56    of Markers to control bci_controller behaviour, etc.
 57    """
 58
 59    @property
 60    @abstractmethod
 61    def name(self) -> str:
 62        """Name of the EEG source"""
 63        pass
 64
 65    @property
 66    @abstractmethod
 67    def fsample(self) -> float:
 68        """Sample rate of EEG source"""
 69        pass
 70
 71    @property
 72    @abstractmethod
 73    def n_channels(self) -> int:
 74        """Number of EEG channels per sample"""
 75        pass
 76
 77    @property
 78    @abstractmethod
 79    def channel_types(self) -> list[str]:
 80        """The type of each channel, ex: eeg, or stim"""
 81        pass
 82
 83    @property
 84    @abstractmethod
 85    def channel_units(self) -> list[str]:
 86        """The unit of each channel, ex: microvolts"""
 87        pass
 88
 89    @property
 90    @abstractmethod
 91    def channel_labels(self) -> list[str]:
 92        """The label for each channel, ex: FC3, C5"""
 93        pass
 94
 95    @abstractmethod
 96    def get_samples(self) -> tuple[list[list], list]:
 97        """Get EEG samples and timestamps since last call
 98
 99        Returns
100        -------
101        samples_data: tuple(samples, timestamps)
102            - A tuple of (samples, timestamps) where:
103                - samples : list[float]
104                    - A list of samples, where each sample corresponds to a timestamp. Each sample is a list of floats representing the value for each channel of EEG.
105                - timestamps : list[float]
106                    - A list timestamps (float in seconds) corresponding to the samples
107
108        """
109        pass
110
111    @abstractmethod
112    def time_correction(self) -> float:
113        """Get the current time correction for timestamps.
114
115        Returns
116        -------
117        time_correction : float
118            The current time correction estimate (seconds).
119            - This is the number that needs to be added to a time tamp that was remotely generated via local_clock() to map it into the local clock domain of the machine.
120
121        """
122        pass