bci_essentials.io.lsl_sources

  1from mne_lsl.lsl import StreamInlet, StreamInfo, resolve_streams
  2from .sources import MarkerSource, EegSource
  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__all__ = ["LslMarkerSource", "LslEegSource"]
 10
 11
 12class LslMarkerSource(MarkerSource):
 13    def __init__(
 14        self, stream: StreamInfo = None, buffer_size: int = 5, timeout: float = 600
 15    ):
 16        """Create a MarkerSource object that obtains markers from an LSL outlet
 17
 18        Parameters
 19        ----------
 20        stream : StreamInfo, *optional*
 21            Provide stream to use for Markers, if not provided, stream will be discovered.
 22        buffer_size : int, *optional*
 23            Size of the buffer is `buffer_size * 100`. Default is 5 (i.e., 500 samples).
 24        timeout : float, *optional*
 25            How many seconds to wait for marker outlet stream to be discovered.
 26            If no stream is discovered, an Exception is raised.
 27            By default init will wait 10 minutes.
 28        """
 29        try:
 30            if stream is None:
 31                stream = discover_first_stream("LSL_Marker_Strings", timeout=timeout)
 32            self._inlet = StreamInlet(
 33                stream, max_buffered=buffer_size, processing_flags=["dejitter"]
 34            )
 35            self._inlet.open_stream(timeout=5)
 36            self.__info = self._inlet.get_sinfo()
 37        except Exception:
 38            raise Exception("LslMarkerSource: could not create inlet")
 39
 40    @property
 41    def name(self) -> str:
 42        return self.__info.name
 43
 44    def get_markers(self) -> tuple[list[list], list]:
 45        return pull_from_lsl_inlet(self._inlet)
 46
 47    def time_correction(self) -> float:
 48        return self._inlet.time_correction()
 49
 50
 51class LslEegSource(EegSource):
 52    def __init__(
 53        self, stream: StreamInfo = None, buffer_size: int = 5, timeout: float = 600
 54    ):
 55        """Create a MarkerSource object that obtains EEG from an LSL outlet
 56
 57        Parameters
 58        ----------
 59        stream : StreamInfo, *optional*
 60            Provide stream to use for EEG, if not provided, stream will be discovered.
 61        buffer_size : int, *optional*
 62            Size of the buffer to use for the inlet in seconds. Default is 5.
 63        timeout : float, *optional*
 64            How many seconds to wait for marker outlet stream to be discovered.
 65            If no stream is discovered, an Exception is raised.
 66            By default init will wait 10 minutes.
 67        """
 68        try:
 69            if stream is None:
 70                stream = discover_first_stream("EEG", timeout=timeout)
 71            self._inlet = StreamInlet(
 72                stream, max_buffered=buffer_size, processing_flags=["dejitter"]
 73            )
 74            self._inlet.open_stream(timeout=5)
 75            self.__info = self._inlet.get_sinfo()
 76        except Exception:
 77            raise Exception("LslEegSource: could not create inlet")
 78
 79    @property
 80    def name(self) -> str:
 81        return self.__info.name
 82
 83    @property
 84    def fsample(self) -> float:
 85        return self.__info.sfreq
 86
 87    @property
 88    def n_channels(self) -> int:
 89        return self.__info.n_channels
 90
 91    @property
 92    def channel_types(self) -> list[str]:
 93        return self.__info.get_channel_types()
 94
 95    @property
 96    def channel_units(self) -> list[str]:
 97        """Get channel units. Default to "microvolts"."""
 98        try:
 99            units = self.__info.get_channel_units()
100            # If no units found or empty strings, use default
101            if not units or all(unit == "" for unit in units):
102                logger.warning("No channel units found, defaulting to microvolts")
103                units = ["microvolts"] * self.n_channels
104            return units
105        except Exception:
106            logger.warning("Could not get channel units, defaulting to microvolts")
107            return ["microvolts"] * self.n_channels
108
109    @property
110    def channel_labels(self) -> list[str]:
111        """Get channel labels.  Default to Ch1, Ch2, etc."""
112
113        try:
114            ch_names = self.__info.get_channel_names()
115            # If no labels found or empty strings, use default
116            if not ch_names or all(label == "" for label in ch_names):
117                logger.warning("No channel labels found, defaulting to Ch1, Ch2, etc.")
118                ch_names = [f"Ch{i+1}" for i in range(self.n_channels)]
119            return ch_names
120        except Exception:
121            logger.warning("Could not get channel labels, defaulting to Ch1, Ch2, etc.")
122            return [f"Ch{i+1}" for i in range(self.n_channels)]
123
124        # if hasattr(self.__info, 'ch_names') and self.__info.ch_names:
125        #     return list(self.__info.ch_names)
126        # return [f"Ch{i+1}" for i in range(self.n_channels)]
127
128    def get_samples(self) -> tuple[list[list], list]:
129        return pull_from_lsl_inlet(self._inlet)
130
131    def time_correction(self) -> float:
132        return self._inlet.time_correction()
133
134    def get_channel_properties(self, property: str) -> list[str]:
135        """Get channel properties from mne_lsl stream info.
136
137        Parameters
138        ----------
139        property : str
140            Property to get ('name', 'unit', 'type', etc)
141
142        Returns
143        -------
144        list[str]
145            List of property values for each channel
146        """
147        if property == "name":
148            return self.name
149        elif property == "unit":
150            return self.channel_units
151        elif property == "type":
152            return self.channel_types
153        elif property == "label":
154            return self.channel_labels
155        else:
156            logger.warning(f"Property '{property}' not supported in mne_lsl")
157            return [""] * self.n_channels
158
159
160def discover_first_stream(type: str, timeout: float = 600) -> StreamInfo:
161    """This helper returns the first stream of the specified type.
162
163    If no stream is found, an exception is raised."""
164    streams = resolve_streams(stype=type, timeout=timeout)
165    return streams[0]
166
167
168def pull_from_lsl_inlet(inlet: StreamInlet) -> tuple[list[list], list]:
169    """StreamInlet.pull_chunk() may return None for samples.
170
171    This helper prevents `None` from propagating by converting it into [[]].
172
173    If None is detected, the timestamps list is also forced to [].
174    """
175
176    # read from inlet
177    samples, timestamps = inlet.pull_chunk(timeout=0.001)
178
179    # convert None or empty samples into empty lists
180    if samples is None or len(samples) == 0:
181        samples = [[]]
182        timestamps = []
183
184    # return tuple[list[list], list]
185    return [samples, timestamps]