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