bci_essentials.io.xdf_sources

  1import pyxdf
  2
  3from .sources import EegSource, MarkerSource
  4from ..utils.logger import Logger  # Logger wrapper
  5
  6# Instantiate a logger for the module at the default level of logging.INFO
  7# Logs to bci_essentials.__module__) where __module__ is the name of the module
  8logger = Logger(name=__name__)
  9
 10__all__ = ["XdfMarkerSource", "XdfEegSource"]
 11
 12
 13class XdfMarkerSource(MarkerSource):
 14    def __init__(self, filename: str):
 15        """Create a MarkerSource object that obtains markers from an XDF file.
 16
 17        Parameters
 18        ----------
 19        filename : str
 20            The full name of file, including path.  If file isn't found, an Exception is raised.
 21        """
 22        samples, timestamps, info = load_xdf_stream(filename, "LSL_Marker_Strings")
 23        self.__samples = samples
 24        self.__timestamps = timestamps
 25        self.__info = info
 26
 27    @property
 28    def name(self) -> str:
 29        return self.__info["name"][0]
 30
 31    def get_markers(self) -> tuple[list[list], list]:
 32        """Read markers and related timestamps from the XDF file.
 33
 34        Returns
 35        -------
 36        markers_from_xdf : tuple[markers, timestamps]
 37            A tuple of (markers, timestamps). The contents of the file on the first call to get_markers().
 38                - Returns empty lists [[],[]] thereafter.
 39        """
 40        # return all data on first get
 41        samples = self.__samples
 42        timestamps = self.__timestamps
 43
 44        # reset to empty for next get
 45        self.__samples = [[]]
 46        self.__timestamps = []
 47
 48        return [samples, timestamps]
 49
 50    def time_correction(self) -> float:
 51        return 0.0
 52
 53
 54class XdfEegSource(EegSource):
 55    """Create a MarkerSource object that obtains EEG from an XDF file
 56
 57    Parameters
 58    ----------
 59    filename : str
 60        The full name of file, including path. If file isn't found, an Exception is raised.
 61    """
 62
 63    def __init__(self, filename: str):
 64        samples, timestamps, info = load_xdf_stream(filename, "EEG")
 65        self.__samples = samples
 66        self.__timestamps = timestamps
 67        self.__info = info
 68
 69    @property
 70    def name(self) -> str:
 71        return self.__info["name"][0]
 72
 73    @property
 74    def fsample(self) -> float:
 75        return float(self.__info["nominal_srate"][0])
 76
 77    @property
 78    def n_channels(self) -> int:
 79        return int(self.__info["channel_count"][0])
 80
 81    @property
 82    def channel_types(self) -> list[str]:
 83        types = []
 84        try:
 85            description = self.__info["desc"][0]["channels"][0]["channel"]
 86            types = [channel["type"][0] for channel in description]
 87            types = [str.lower(type) for type in types]
 88        except Exception:
 89            pass
 90
 91        return types
 92
 93    @property
 94    def channel_units(self) -> list[str]:
 95        units = []
 96        try:
 97            description = self.__info["desc"][0]["channels"][0]["channel"]
 98            units = [channel["unit"][0] for channel in description]
 99        except Exception:
100            pass
101        return units
102
103    @property
104    def channel_labels(self) -> list[str]:
105        labels = []
106        try:
107            description = self.__info["desc"][0]["channels"][0]["channel"]
108            labels = [channel["label"][0] for channel in description]
109        except Exception:
110            pass
111        return labels
112
113    def get_samples(self) -> tuple[list[list], list]:
114        """Read markers and related timestamps from the XDF file.  Returns the contents of the file
115        on the first call to get_markers(), returns empty lists thereafter.
116        """
117        # return all data on first get
118        samples = self.__samples
119        timestamps = self.__timestamps
120
121        # reset to empty for next get
122        self.__samples = [[]]
123        self.__timestamps = []
124
125        return [samples, timestamps]
126
127    def time_correction(self) -> float:
128        return 0.0
129
130
131def load_xdf_stream(filepath: str, streamtype: str) -> tuple[list, list, list]:
132    """A helper function to load the contents of the XDF file and return stream data, timestamps and info"""
133    #  Don't need the header returned by load_xdf()
134    streams, _ = pyxdf.load_xdf(filepath)
135
136    samples = []
137    timestamps = []
138    info = []
139
140    for i in range(len(streams)):
141        stream = streams[i]
142        info = stream["info"]
143        type = info["type"][0]
144        if type == streamtype:
145            try:
146                # Check if the stream is empty
147                if len(stream["time_series"]) == 0:
148                    logger.warning(
149                        "%s stream is empty, there may be multiple streams including ghost streams so maybe this is okay",
150                        streamtype,
151                    )
152                    continue
153
154                samples = stream["time_series"]
155                timestamps = stream["time_stamps"]
156
157            except Exception:
158                logger.error("%s data not available", streamtype)
159
160            break
161
162    return (samples, timestamps, info)