From e78e90569bfd60175d77d1c4c76646950a9a9d57 Mon Sep 17 00:00:00 2001 From: Tab Memmott Date: Mon, 9 Mar 2026 14:58:35 -0700 Subject: [PATCH] EEG Triggering via stim channels --- .DS_Store | Bin 6148 -> 0 bytes .flake8 | 4 +- .gitignore | 3 +- Makefile | 22 + README.md | 32 +- bci_essentials/.DS_Store | Bin 6148 -> 0 bytes bci_essentials/bci_controller.py | 11 +- bci_essentials/io/.DS_Store | Bin 6148 -> 0 bytes bci_essentials/io/__init__.py | 44 ++ bci_essentials/io/eeg_trigger_sources.py | 218 ++++++++ bci_essentials/io/fake_stim_source.py | 174 ++++++ bci_essentials/io/lsl_stim_marker_source.py | 152 ++++++ bci_essentials/paradigm/paradigm.py | 13 +- bci_essentials/triggers.py | 282 ++++++++++ examples/p300_trigger_stim_backend.py | 94 ++++ examples/trigger_stim_example.py | 156 ++++++ pyproject.toml | 1 + tests/test_eeg_trigger_sources.py | 565 ++++++++++++++++++++ tests/test_triggers.py | 340 ++++++++++++ 19 files changed, 2092 insertions(+), 19 deletions(-) delete mode 100644 .DS_Store create mode 100644 Makefile delete mode 100644 bci_essentials/.DS_Store delete mode 100644 bci_essentials/io/.DS_Store create mode 100644 bci_essentials/io/__init__.py create mode 100644 bci_essentials/io/eeg_trigger_sources.py create mode 100644 bci_essentials/io/fake_stim_source.py create mode 100644 bci_essentials/io/lsl_stim_marker_source.py create mode 100644 bci_essentials/triggers.py create mode 100644 examples/p300_trigger_stim_backend.py create mode 100644 examples/trigger_stim_example.py create mode 100644 tests/test_eeg_trigger_sources.py create mode 100644 tests/test_triggers.py diff --git a/.DS_Store b/.DS_Store deleted file mode 100644 index e0a5e68439f9993772c6bd675da5271acd6fef5a..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 6148 zcmeHKyH3ME5S)V)k&sYI%KHWWz>2~b@Bu&)7m*A>a*!z9@!QNk0z?;D3N&l&jc;!E z&Yr^S1yGi6+go4(U`{v0hYhXSx_M(4$ru*jqsPdr#|sWk^m)Q^ud&7kJ3KJ+JAa2K z3>dfLW?1j{0}tR+^mr%zj8`r(__%m}Z?RH93P=GdAO)mAtALeM+Waa|qZE(=Qs7Gg z`#w~i-@6m;V1WNi!)R1^$%+WwyFsE%{1WTbq}& wT3hLl^rJD>!#Q{ITBH+^IowoM7DlF6Pvj!~`F6S<`~EuEpJZ-l;GK;22S=Hb011!)36KB@{0ITN zw_)=|C?g4w00}$^*!Q8pO>1fk^-l+aj{wja%5KPgmVg#3Kx=9XMFpnO3XN9vF~sWL z4lQ}Ernb;%7tP^A^T}#c3{0b4v|s_#>cT(*BrqZ{jdf@D{|^3X{vWk)NdhGBX9RS% z-ETK|sk~d?UeD@>tXkgSP(P0F@)LlCUByed8}^GO(3;vpQGxMCz%ek8z*h-80O7R| AGXMYp diff --git a/bci_essentials/bci_controller.py b/bci_essentials/bci_controller.py index c5153ea4..13e4cc6f 100644 --- a/bci_essentials/bci_controller.py +++ b/bci_essentials/bci_controller.py @@ -20,13 +20,13 @@ import time import os import numpy as np -from enum import Enum from .paradigm.paradigm import Paradigm from .data_tank.data_tank import DataTank from .classification.generic_classifier import GenericClassifier from .io.sources import EegSource, MarkerSource from .io.messenger import Messenger +from .triggers import MarkerTypes from .utils.logger import Logger # Instantiate a logger for the module at the default level of logging.INFO @@ -34,15 +34,6 @@ logger = Logger(name=__name__) -class MarkerTypes(Enum): - TRIAL_STARTED = "Trial Started" - TRIAL_ENDS = "Trial Ends" - TRAINING_COMPLETE = "Training Complete" - TRAIN_CLASSIFIER = "Train Classifier" - DONE_RS_COLLECTION = "Done with all RS collection" - UPDATE_CLASSIFIER = "Update Classifier" - - # EEG data class BciController: """ diff --git a/bci_essentials/io/.DS_Store b/bci_essentials/io/.DS_Store deleted file mode 100644 index 5008ddfcf53c02e82d7eee2e57c38e5672ef89f6..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 6148 zcmeH~Jr2S!425mzP>H1@V-^m;4Wg<&0T*E43hX&L&p$$qDprKhvt+--jT7}7np#A3 zem<@ulZcFPQ@L2!n>{z**++&mCkOWA81W14cNZlEfg7;MkzE(HCqgga^y>{tEnwC%0;vJ&^%eQ zLs35+`xjp>T0 255: + raise ValueError( + f"trigger_map keys must be integers in 0-255, got {key!r}" + ) + + self._eeg = eeg_source + self._stim_channel_name = stim_channel_name + tmap = trigger_map if trigger_map is not None else dict(DEFAULT_TRIGGER_MAP) + self._enabled = bool(enabled) + self._detector = TriggerDetector(tmap, detect_mode, include_unmapped) + + self._pending: list[tuple[list[str], float]] = [] + self._resolved_stim_idx: int | None = None + + self._samples_scanned: int = 0 + self._triggers_detected: int = 0 + self._warned_no_triggers: bool = False + + @property + def enabled(self) -> bool: + return self._enabled + + @enabled.setter + def enabled(self, value: bool) -> None: + self._enabled = bool(value) + + @property + def stim_channel_name(self) -> str: + return self._stim_channel_name + + @stim_channel_name.setter + def stim_channel_name(self, name: str) -> None: + if not isinstance(name, str) or not name: + raise ValueError(f"stim_channel_name must be non-empty, got {name!r}") + self._stim_channel_name = name + self._resolved_stim_idx = None + + @property + def name(self) -> str: + return self._eeg.name + + @property + def fsample(self) -> float: + return self._eeg.fsample + + @property + def n_channels(self) -> int: + return self._eeg.n_channels + + @property + def channel_types(self) -> list[str]: + return self._eeg.channel_types + + @property + def channel_units(self) -> list[str]: + return self._eeg.channel_units + + @property + def channel_labels(self) -> list[str]: + return self._eeg.channel_labels + + def get_samples(self) -> tuple[list[list], list]: + try: + samples, timestamps = self._eeg.get_samples() + except Exception: + logger.error( + "EegStimTriggerMarkerSource: error reading EEG source", + exc_info=True, + ) + return [[]], [] + + if not self._enabled or not timestamps: + return samples, timestamps + + stim_idx = self._resolve_stim_index() + if stim_idx is None: + return samples, timestamps + + for sample, ts in zip(samples, timestamps): + if stim_idx >= len(sample): + continue + self._samples_scanned += 1 + try: + stim_val = int(sample[stim_idx]) + except (ValueError, TypeError): + continue + + marker_str = self._detector.detect(stim_val) + if marker_str is not None: + self._triggers_detected += 1 + self._pending.append(([marker_str], ts)) + + self._check_no_trigger_warning() + return samples, timestamps + + def time_correction(self) -> float: + return self._eeg.time_correction() + + def get_markers(self) -> tuple[list[list], list]: + if not self._enabled or not self._pending: + return [[]], [] + markers = [m for m, _ in self._pending] + timestamps = [t for _, t in self._pending] + self._pending.clear() + return markers, timestamps + + def _resolve_stim_index(self) -> int | None: + if self._resolved_stim_idx is not None: + return self._resolved_stim_idx + try: + labels = self._eeg.channel_labels + except Exception: + return None + if self._stim_channel_name in labels: + self._resolved_stim_idx = labels.index(self._stim_channel_name) + return self._resolved_stim_idx + logger.warning( + f"Stim channel '{self._stim_channel_name}' not found in {labels}" + ) + return None + + def _check_no_trigger_warning(self) -> None: + if self._warned_no_triggers or self._triggers_detected > 0: + return + try: + threshold = int(self._eeg.fsample * 10) + except Exception: + threshold = 2560 + if self._samples_scanned >= threshold: + logger.warning( + f"Scanned {self._samples_scanned} samples with no triggers. " + f"Check stim_channel_name, trigger_map, and hardware connections." + ) + self._warned_no_triggers = True diff --git a/bci_essentials/io/fake_stim_source.py b/bci_essentials/io/fake_stim_source.py new file mode 100644 index 00000000..1da51c55 --- /dev/null +++ b/bci_essentials/io/fake_stim_source.py @@ -0,0 +1,174 @@ +import time + +from .sources import EegSource +from ..utils.logger import Logger # Logger wrapper + +# Instantiate a logger for the module at the default level of logging.INFO +# Logs to bci_essentials.__module__) where __module__ is the name of the module +logger = Logger(name=__name__) + +__all__ = ["FakeStimEegSource"] + + +class FakeStimEegSource(EegSource): + """Synthetic EEG source with programmable stim-channel trigger injection. + + Generates zero-valued EEG samples at the configured sample rate. One channel + (the stim channel) carries trigger byte values injected at regular intervals + or according to a user-supplied schedule. Designed for testing without hardware. + """ + + def __init__( + self, + n_channels: int = 9, + fsample: float = 256.0, + stim_channel_index: int = -1, + trigger_bytes: list[int] | None = None, + trigger_interval: float = 1.0, + trigger_schedule: list[tuple[float, int]] | None = None, + duration: float = 10.0, + channel_labels: list[str] | None = None, + ): + """Create a FakeStimEegSource for testing. + + Parameters + ---------- + n_channels : int, *optional* + Total number of channels (EEG + stim). Default is 9. + fsample : float, *optional* + Sample rate in Hz. Default is 256.0. + stim_channel_index : int, *optional* + Index of the stim channel. Default is -1 (last channel). + trigger_bytes : list[int], *optional* + Byte values cycled through at trigger_interval spacing. + Mutually exclusive with trigger_schedule. + trigger_interval : float, *optional* + Seconds between successive triggers when using trigger_bytes. + Default is 1.0. + trigger_schedule : list[tuple[float, int]], *optional* + Explicit (time_offset_sec, byte_value) pairs. + Mutually exclusive with trigger_bytes. + duration : float, *optional* + Total duration of fake data in seconds. Default is 10.0. + channel_labels : list[str], *optional* + Custom channel labels. Defaults to ["Ch0", "Ch1", ..., "STIM"]. + """ + if trigger_bytes is not None and trigger_schedule is not None: + raise ValueError( + "Specify either trigger_bytes or trigger_schedule, not both." + ) + + self._n_channels = n_channels + self._fsample = fsample + self._duration = duration + + # Resolve stim channel index + self._stim_idx = ( + stim_channel_index + if stim_channel_index >= 0 + else n_channels + stim_channel_index + ) + + # Channel labels + if channel_labels is not None: + if len(channel_labels) != n_channels: + raise ValueError( + f"channel_labels length ({len(channel_labels)}) " + f"!= n_channels ({n_channels})" + ) + self._channel_labels = list(channel_labels) + else: + self._channel_labels = [ + f"Ch{i}" if i != self._stim_idx else "STIM" for i in range(n_channels) + ] + + # Build the trigger schedule (sorted by time) + if trigger_schedule is not None: + self._schedule = sorted(trigger_schedule, key=lambda x: x[0]) + elif trigger_bytes is not None: + self._schedule = [] + t = 0.0 + idx = 0 + while t < duration: + self._schedule.append((t, trigger_bytes[idx % len(trigger_bytes)])) + t += trigger_interval + idx += 1 + else: + self._schedule = [] + + self._start_time: float | None = None + self._samples_generated: int = 0 + self._total_samples = int(duration * fsample) + self._schedule_idx = 0 + + @property + def name(self) -> str: + return "FakeStimEegSource" + + @property + def fsample(self) -> float: + return self._fsample + + @property + def n_channels(self) -> int: + return self._n_channels + + @property + def channel_types(self) -> list[str]: + return [ + "stim" if i == self._stim_idx else "eeg" for i in range(self._n_channels) + ] + + @property + def channel_units(self) -> list[str]: + return [ + "n/a" if i == self._stim_idx else "microvolts" + for i in range(self._n_channels) + ] + + @property + def channel_labels(self) -> list[str]: + return self._channel_labels + + def get_samples(self) -> tuple[list[list], list]: + if self._samples_generated >= self._total_samples: + return [[]], [] + + now = time.monotonic() + if self._start_time is None: + self._start_time = now + + elapsed = now - self._start_time + target_sample = min(int(elapsed * self._fsample), self._total_samples) + n_new = target_sample - self._samples_generated + + if n_new <= 0: + return [[]], [] + + samples = [] + timestamps = [] + + for i in range(n_new): + sample_idx = self._samples_generated + i + t = self._start_time + sample_idx / self._fsample + + # Determine stim value for this sample + stim_val = 0 + sample_time_offset = sample_idx / self._fsample + while ( + self._schedule_idx < len(self._schedule) + and self._schedule[self._schedule_idx][0] <= sample_time_offset + ): + stim_val = self._schedule[self._schedule_idx][1] + self._schedule_idx += 1 + + row = [0.0] * self._n_channels + row[self._stim_idx] = float(stim_val) + samples.append(row) + timestamps.append(t) + + self._samples_generated = target_sample + return samples, timestamps + + def time_correction(self) -> float: + return 0.0 diff --git a/bci_essentials/io/lsl_stim_marker_source.py b/bci_essentials/io/lsl_stim_marker_source.py new file mode 100644 index 00000000..84d232be --- /dev/null +++ b/bci_essentials/io/lsl_stim_marker_source.py @@ -0,0 +1,152 @@ +from mne_lsl.lsl import StreamInlet, StreamInfo, resolve_streams + +from .sources import MarkerSource +from ..triggers import DEFAULT_TRIGGER_MAP, TriggerDetector +from ..utils.logger import Logger # Logger wrapper + +# Instantiate a logger for the module at the default level of logging.INFO +# Logs to bci_essentials.__module__) where __module__ is the name of the module +logger = Logger(name=__name__) + +__all__ = ["LslStimMarkerSource"] + + +class LslStimMarkerSource(MarkerSource): + def __init__( + self, + stim_channel_name: str = "STIM", + trigger_map: dict[int, str] | None = None, + detect_mode: str = "rise", + include_unmapped: bool = False, + stream: StreamInfo | None = None, + buffer_size: int = 5, + timeout: float = 600, + ): + """Create a MarkerSource that reads triggers from the EEG stim channel. + + Opens its own LSL inlet to the EEG stream and scans the stim channel + for trigger events, converting byte values to marker strings. + + Parameters + ---------- + stim_channel_name : str, *optional* + Label of the stim channel in the EEG stream. Default is "STIM". + trigger_map : dict[int, str], *optional* + Byte to marker string mapping. Defaults to DEFAULT_TRIGGER_MAP. + detect_mode : str, *optional* + "rise" (default) detects 0 to non-zero transitions. + "change" detects any value change. + include_unmapped : bool, *optional* + If True, unmapped byte values produce "trigger_{value}" markers. + stream : StreamInfo, *optional* + Provide stream to use for EEG, if not provided, stream will be discovered. + buffer_size : int, *optional* + Size of the buffer to use for the inlet in seconds. Default is 5. + timeout : float, *optional* + How many seconds to wait for EEG stream to be discovered. + If no stream is discovered, an Exception is raised. + By default init will wait 10 minutes. + """ + if detect_mode not in ("rise", "change"): + raise ValueError( + f"detect_mode must be 'rise' or 'change', got {detect_mode!r}" + ) + + self._stim_channel_name = stim_channel_name + tmap = trigger_map if trigger_map is not None else dict(DEFAULT_TRIGGER_MAP) + self._detector = TriggerDetector(tmap, detect_mode, include_unmapped) + + try: + if stream is None: + streams = resolve_streams(stype="EEG", timeout=timeout) + if not streams: + raise RuntimeError(f"No EEG stream found within {timeout}s timeout") + stream = streams[0] + self._inlet = StreamInlet( + stream, max_buffered=buffer_size, processing_flags=["dejitter"] + ) + self._inlet.open_stream(timeout=5) + self.__info = self._inlet.get_sinfo() + except Exception: + raise Exception( + "LslStimMarkerSource: could not open LSL inlet to EEG stream" + ) + + self._stim_idx = self._find_stim_channel() + + self._samples_scanned: int = 0 + self._triggers_detected: int = 0 + self._warned_no_triggers: bool = False + + @property + def name(self) -> str: + return f"LslStimMarkerSource({self._stim_channel_name})" + + def get_markers(self) -> tuple[list[list], list]: + samples, timestamps = self._inlet.pull_chunk(timeout=0.001) + + if samples is None or len(samples) == 0: + return [[]], [] + + markers = [] + marker_timestamps = [] + + for sample, ts in zip(samples, timestamps): + self._samples_scanned += 1 + + if self._stim_idx is None or self._stim_idx >= len(sample): + continue + + try: + stim_val = int(sample[self._stim_idx]) + except (ValueError, TypeError): + continue + + marker_str = self._detector.detect(stim_val) + if marker_str is not None: + self._triggers_detected += 1 + markers.append([marker_str]) + marker_timestamps.append(ts) + logger.debug(f"trigger {stim_val} -> '{marker_str}' at t={ts:.4f}") + + # Warn once after ~10s of data with no triggers + if ( + not self._warned_no_triggers + and self._triggers_detected == 0 + and self._samples_scanned > 0 + ): + try: + warn_threshold = int(self.__info.sfreq * 10) + except Exception: + warn_threshold = 2560 + if self._samples_scanned >= warn_threshold: + logger.warning( + f"Scanned {self._samples_scanned} samples from channel " + f"'{self._stim_channel_name}' with no triggers detected" + ) + self._warned_no_triggers = True + + if not markers: + return [[]], [] + + return markers, marker_timestamps + + def time_correction(self) -> float: + return self._inlet.time_correction() + + def _find_stim_channel(self) -> int | None: + try: + ch_names = self.__info.get_channel_names() + except Exception: + logger.warning("Could not read channel names from EEG stream") + return None + + if self._stim_channel_name in ch_names: + idx = ch_names.index(self._stim_channel_name) + logger.info(f"Stim channel '{self._stim_channel_name}' at index {idx}") + return idx + + logger.warning( + f"Stim channel '{self._stim_channel_name}' not found in {ch_names}" + ) + return None diff --git a/bci_essentials/paradigm/paradigm.py b/bci_essentials/paradigm/paradigm.py index 06f4a27d..d71e7791 100644 --- a/bci_essentials/paradigm/paradigm.py +++ b/bci_essentials/paradigm/paradigm.py @@ -1,6 +1,7 @@ import numpy as np from abc import ABC, abstractmethod +from ..triggers import MarkerTypes from ..utils.logger import Logger from ..signal_processing import bandpass, lowpass @@ -170,36 +171,36 @@ def package_resting_state_data( current_time = eeg_timestamps[current_timestamp_loc] # get eyes open start times - if current_rs_data_marker == "Start Eyes Open RS: 1": + if current_rs_data_marker == MarkerTypes.START_EYES_OPEN.value: eyes_open_start_time.append(current_rs_timestamp) eyes_open_start_loc.append(current_timestamp_loc - 1) logger.debug("received eyes open start") # get eyes open end times - if current_rs_data_marker == "End Eyes Open RS: 1": + if current_rs_data_marker == MarkerTypes.END_EYES_OPEN.value: eyes_open_end_time.append(current_rs_timestamp) eyes_open_end_loc.append(current_timestamp_loc) logger.debug("received eyes open end") # get eyes closed start times - if current_rs_data_marker == "Start Eyes Closed RS: 2": + if current_rs_data_marker == MarkerTypes.START_EYES_CLOSED.value: eyes_closed_start_time.append(current_rs_timestamp) eyes_closed_start_loc.append(current_timestamp_loc - 1) logger.debug("received eyes closed start") # get eyes closed end times - if current_rs_data_marker == "End Eyes Closed RS: 2": + if current_rs_data_marker == MarkerTypes.END_EYES_CLOSED.value: eyes_closed_end_time.append(current_rs_timestamp) eyes_closed_end_loc.append(current_timestamp_loc) logger.debug("received eyes closed end") # get rest start times - if current_rs_data_marker == "Start Rest for RS: 0": + if current_rs_data_marker == MarkerTypes.START_REST.value: rest_start_time.append(current_rs_timestamp) rest_start_loc.append(current_timestamp_loc - 1) logger.debug("received rest start") # get rest end times - if current_rs_data_marker == "End Rest for RS: 0": + if current_rs_data_marker == MarkerTypes.END_REST.value: rest_end_time.append(current_rs_timestamp) rest_end_loc.append(current_timestamp_loc) logger.debug("received rest end") diff --git a/bci_essentials/triggers.py b/bci_essentials/triggers.py new file mode 100644 index 00000000..281accf8 --- /dev/null +++ b/bci_essentials/triggers.py @@ -0,0 +1,282 @@ +from enum import Enum + +__all__ = [ + "MarkerTypes", + "TriggerByte", + "TriggerDetector", + "DEFAULT_TRIGGER_MAP", + "SIMPLE_TRIGGER_MAP", + "make_p300_trigger_map", + "make_mi_trigger_map", + "make_ssvep_trigger_map", + "make_simple_trigger_map", +] + + +class MarkerTypes(Enum): + """Status marker strings recognised by BciController. + + These correspond 1-to-1 with the status bytes in TriggerByte. + """ + + TRIAL_STARTED = "Trial Started" + TRIAL_ENDS = "Trial Ends" + TRAINING_COMPLETE = "Training Complete" + TRAIN_CLASSIFIER = "Train Classifier" + UPDATE_CLASSIFIER = "Update Classifier" + DONE_RS_COLLECTION = "Done with all RS collection" + + # Resting state markers + START_EYES_OPEN = "Start Eyes Open RS: 1" + END_EYES_OPEN = "End Eyes Open RS: 1" + START_EYES_CLOSED = "Start Eyes Closed RS: 2" + END_EYES_CLOSED = "End Eyes Closed RS: 2" + START_REST = "Start Rest for RS: 0" + END_REST = "End Rest for RS: 0" + + +class TriggerByte: + """Default serial-port byte values sent by Unity's SerialTriggerWriter. + + Status bytes occupy the high range (240-255) to stay clear of + stimulus byte values (1-N). + """ + + # Status events + TRIAL_STARTED: int = 240 + TRIAL_ENDS: int = 241 + TRAINING_COMPLETE: int = 242 + TRAIN_CLASSIFIER: int = 243 + UPDATE_CLASSIFIER: int = 244 + DONE_RS_COLLECTION: int = 245 + + # Simple target/non-target encoding + TARGET: int = 1 + NON_TARGET: int = 2 + + +# Default trigger map for the six status events (byte -> marker string) +DEFAULT_TRIGGER_MAP: dict[int, str] = { + TriggerByte.TRIAL_STARTED: MarkerTypes.TRIAL_STARTED.value, + TriggerByte.TRIAL_ENDS: MarkerTypes.TRIAL_ENDS.value, + TriggerByte.TRAINING_COMPLETE: MarkerTypes.TRAINING_COMPLETE.value, + TriggerByte.TRAIN_CLASSIFIER: MarkerTypes.TRAIN_CLASSIFIER.value, + TriggerByte.UPDATE_CLASSIFIER: MarkerTypes.UPDATE_CLASSIFIER.value, + TriggerByte.DONE_RS_COLLECTION: MarkerTypes.DONE_RS_COLLECTION.value, +} + +# Simplified target/non-target trigger map +SIMPLE_TRIGGER_MAP: dict[int, str] = { + TriggerByte.TARGET: "target", + TriggerByte.NON_TARGET: "non_target", +} + + +def make_simple_trigger_map( + target_byte: int = TriggerByte.TARGET, + non_target_byte: int = TriggerByte.NON_TARGET, + include_status: bool = True, + custom: dict[int, str] | None = None, +) -> dict[int, str]: + """Build a simplified target / non-target trigger map. + + Parameters + ---------- + target_byte : int + Byte value for the target stimulus (default 1). + non_target_byte : int + Byte value for non-target stimuli (default 2). + include_status : bool + If True (default), merge DEFAULT_TRIGGER_MAP into the returned dict. + custom : dict[int, str] or None + Extra entries to merge. Overrides colliding keys. + + Returns + ------- + dict[int, str] + """ + if not (0 <= target_byte <= 255): + raise ValueError(f"target_byte must be 0-255, got {target_byte}") + if not (0 <= non_target_byte <= 255): + raise ValueError(f"non_target_byte must be 0-255, got {non_target_byte}") + if target_byte == non_target_byte: + raise ValueError( + f"target_byte and non_target_byte must differ, both are {target_byte}" + ) + + trigger_map: dict[int, str] = {} + if include_status: + trigger_map.update(DEFAULT_TRIGGER_MAP) + trigger_map[target_byte] = "target" + trigger_map[non_target_byte] = "non_target" + if custom: + trigger_map.update(custom) + return trigger_map + + +def make_p300_trigger_map( + n_options: int, + stimulus_base_byte: int = 1, +) -> dict[int, str]: + """Build a trigger map for a single-flash P300 paradigm. + + Unity sends byte = StimulusIndex + 1 for each flash. The byte encodes + which object flashed, not which object is the training target. All markers + use -1 (classification mode) since the target changes per trial. + + Marker format: "p300,s,{n_options},-1,{flash_index}" + + Parameters + ---------- + n_options : int + Number of stimulus presenters. + stimulus_base_byte : int + First byte value for stimuli. Default 1. + + Returns + ------- + dict[int, str] + """ + if n_options <= 0: + raise ValueError(f"n_options must be positive, got {n_options}") + if n_options > 239: + raise ValueError(f"n_options must be at most 239, got {n_options}") + + trigger_map = dict(DEFAULT_TRIGGER_MAP) + for i in range(n_options): + byte_code = stimulus_base_byte + i + trigger_map[byte_code] = f"p300,s,{n_options},-1,{i + 1}" + return trigger_map + + +def make_mi_trigger_map( + n_classes: int, + epoch_length: float, +) -> dict[int, str]: + """Build a trigger map for a Motor Imagery paradigm. + + Unity sends byte = TrainingTargetIndex + 1 for each MI trial. + Each byte maps to its own class label (1-indexed). + + Marker format: "mi,{n_classes},{label},{epoch_length}" + + Parameters + ---------- + n_classes : int + Number of MI classes. + epoch_length : float + EEG epoch length in seconds. + + Returns + ------- + dict[int, str] + """ + if n_classes <= 0: + raise ValueError(f"n_classes must be positive, got {n_classes}") + if epoch_length <= 0: + raise ValueError(f"epoch_length must be positive, got {epoch_length}") + if n_classes > 239: + raise ValueError(f"n_classes must be at most 239, got {n_classes}") + + trigger_map = dict(DEFAULT_TRIGGER_MAP) + for i in range(n_classes): + byte_code = i + 1 + trigger_map[byte_code] = f"mi,{n_classes},{i + 1},{epoch_length:.2f}" + return trigger_map + + +def make_ssvep_trigger_map( + frequencies: list[float], + epoch_length: float, +) -> dict[int, str]: + """Build a trigger map for an SSVEP paradigm. + + Unity sends byte = TrainingTargetIndex + 1 for each SSVEP trial. + Each byte maps to its own class label (1-indexed). + + Marker format: "ssvep,{n_frequencies},{label},{epoch_length},{freq1},{freq2},..." + + Parameters + ---------- + frequencies : list[float] + Flashing frequencies for stimulus presenters. + epoch_length : float + EEG epoch length in seconds. + + Returns + ------- + dict[int, str] + """ + n_freqs = len(frequencies) + if n_freqs == 0: + raise ValueError("frequencies must not be empty") + if epoch_length <= 0: + raise ValueError(f"epoch_length must be positive, got {epoch_length}") + if n_freqs > 239: + raise ValueError(f"number of frequencies must be at most 239, got {n_freqs}") + + freq_str = ",".join(str(f) for f in frequencies) + trigger_map = dict(DEFAULT_TRIGGER_MAP) + for i in range(n_freqs): + byte_code = i + 1 + trigger_map[byte_code] = ( + f"ssvep,{n_freqs},{i + 1}," f"{epoch_length:.2f},{freq_str}" + ) + return trigger_map + + +class TriggerDetector: + """Stateful trigger detector for stim-channel values. + + Encapsulates the rise/change detection logic used by + LslStimMarkerSource and EegStimTriggerMarkerSource. + """ + + def __init__( + self, + trigger_map: dict[int, str], + detect_mode: str = "rise", + include_unmapped: bool = False, + ): + """Create a TriggerDetector. + + Parameters + ---------- + trigger_map : dict[int, str] + Byte to marker string mapping. + detect_mode : str, *optional* + "rise" or "change". Default is "rise". + include_unmapped : bool, *optional* + If True, unmapped bytes produce "trigger_{value}". + """ + if detect_mode not in ("rise", "change"): + raise ValueError( + f"detect_mode must be 'rise' or 'change', got {detect_mode!r}" + ) + self.trigger_map = trigger_map + self.detect_mode = detect_mode + self.include_unmapped = include_unmapped + self.last_value: int = 0 + self.warned_unmapped: set[int] = set() + + def detect(self, stim_val: int) -> str | None: + """Process a stim-channel sample value. Returns the marker string + if a trigger event is detected, otherwise None.""" + is_event = False + if self.detect_mode == "rise": + is_event = stim_val != 0 and self.last_value == 0 + else: # "change" + is_event = stim_val != self.last_value + + self.last_value = stim_val + + if not is_event or stim_val == 0: + return None + + if stim_val in self.trigger_map: + return self.trigger_map[stim_val] + elif self.include_unmapped: + return f"trigger_{stim_val}" + else: + self.warned_unmapped.add(stim_val) + return None diff --git a/examples/p300_trigger_stim_backend.py b/examples/p300_trigger_stim_backend.py new file mode 100644 index 00000000..3ce178ad --- /dev/null +++ b/examples/p300_trigger_stim_backend.py @@ -0,0 +1,94 @@ +"""P300 BCI backend using the serial-port stim-channel trigger workflow. + +Drop-in replacement for ``p300_unity_backend.py`` that reads BCI event markers +from the EEG stim channel instead of a separate LSL marker stream. + +How it works +------------ +1. Unity's ``SerialTriggerWriter`` sends a single byte over serial for every + BCI event (stimulus flash, trial start/end, train classifier, etc.). +2. The hardware trigger box forwards the byte to the EEG amplifier, which + embeds it in a dedicated stim channel alongside the EEG data. +3. ``LslEegSource`` receives the combined EEG+stim stream over LSL. +4. ``LslStimMarkerSource`` opens its own LSL inlet to the same EEG stream, + reads rising edges on the stim channel, and converts them to marker + strings using ``TRIGGER_MAP``. +5. ``BciController`` receives ``eeg_source`` and ``marker_source`` as + separate objects — identical to the standard LSL marker workflow. +6. Predictions flow back to Unity over LSL via ``LslMessenger``. + +Unity setup +----------- +On the BCIController GameObject: + +1. Replace ``MarkerWriter`` with ``SerialCapableMarkerWriter``. +2. Add ``SerialTriggerWriter`` as a sibling component. +3. Set ``SerialTriggerWriter.PortName`` to your COM port (e.g. ``COM3``). +4. Leave ``UseSimpleTargetEncoding`` **disabled** (default) so that each + stimulus flash is encoded as a unique byte (``stimulusIndex + 1``). + +Configuration +------------- +Edit the constants in the ``CONFIGURATION`` section below, then run:: + + python p300_trigger_stim_backend.py +""" + +from bci_essentials.io.lsl_sources import LslEegSource +from bci_essentials.io.lsl_messenger import LslMessenger +from bci_essentials.io.lsl_stim_marker_source import LslStimMarkerSource +from bci_essentials.triggers import make_p300_trigger_map +from bci_essentials.bci_controller import BciController +from bci_essentials.paradigm.p300_paradigm import P300Paradigm +from bci_essentials.data_tank.data_tank import DataTank +from bci_essentials.classification.erp_rg_classifier import ErpRgClassifier + +# ====================================================================== +# CONFIGURATION — adjust these values to match your setup +# ====================================================================== + +# Label of the stim channel as it appears in your EEG LSL stream. +STIM_CHANNEL_NAME = "TRG" + +# Number of P300 stimulus presenters (rows/columns or individual objects). +N_OPTIONS = 6 + +# ====================================================================== + +# Build the trigger map: status events (bytes 240-245) + one P300 entry +# per stimulus. Default byte for stimulus i = i + 1. +TRIGGER_MAP = make_p300_trigger_map( + n_options=N_OPTIONS, +) + +# Connect to the live EEG stream (blocks until an LSL outlet appears). +eeg_source = LslEegSource() + +# Marker source that reads triggers from the EEG stim channel. +# Opens its own LSL inlet — no dual-interface, no wrapping. +marker_source = LslStimMarkerSource( + stim_channel_name=STIM_CHANNEL_NAME, + trigger_map=TRIGGER_MAP, +) + +# Outbound predictions still use LSL — no change from the standard backend. +messenger = LslMessenger() + +# P300 paradigm + classifier (identical to p300_unity_backend.py) +paradigm = P300Paradigm() +data_tank = DataTank() +classifier = ErpRgClassifier() +classifier.set_p300_clf_settings( + n_splits=5, + lico_expansion_factor=1, + oversample_ratio=0, + undersample_ratio=0, +) + +# Same pattern as the standard backend — separate eeg_source and marker_source. +controller = BciController( + classifier, eeg_source, marker_source, paradigm, data_tank, messenger +) + +controller.setup(online=True) +controller.run() diff --git a/examples/trigger_stim_example.py b/examples/trigger_stim_example.py new file mode 100644 index 00000000..5c7aabdd --- /dev/null +++ b/examples/trigger_stim_example.py @@ -0,0 +1,156 @@ +"""Example: Serial-port trigger ingestion — real and fake modes. + +Demonstrates how to use the EEG stim-channel trigger pipeline with: + +1. **Fake mode** — no hardware needed. A ``FakeStimEegSource`` generates + synthetic EEG data with trigger bytes injected on a schedule. Uses + ``EegStimTriggerMarkerSource`` (wrapper) since there is no LSL stream. + +2. **Real mode** — uses ``LslStimMarkerSource`` with a live EEG stream + whose last channel carries trigger codes from a hardware trigger box. + +Usage:: + + # Run in fake mode (default — no hardware required) + python trigger_stim_example.py + + # Run in real mode (requires a live LSL EEG stream with stim channel) + python trigger_stim_example.py --real +""" + +import argparse +import time + +from bci_essentials.triggers import make_simple_trigger_map +from bci_essentials.io.eeg_trigger_sources import EegStimTriggerMarkerSource +from bci_essentials.io.fake_stim_source import FakeStimEegSource +from bci_essentials.utils.logger import Logger + +logger = Logger(name="trigger_stim_example") + + +# ====================================================================== +# TRIGGER MAP — customise here +# ====================================================================== +TRIGGER_MAP = make_simple_trigger_map( + target_byte=10, + non_target_byte=11, + include_status=True, +) +# ====================================================================== + + +def run_fake_mode() -> None: + """Run the trigger pipeline with a FakeStimEegSource (no hardware).""" + logger.info("=" * 60) + logger.info("FAKE MODE — no hardware required") + logger.info("=" * 60) + + schedule = [ + (0.0, 240), # Trial Started + (0.3, 10), # TARGET flash + (0.6, 11), # non-target flash + (0.9, 11), # non-target flash + (1.2, 10), # TARGET flash + (1.5, 241), # Trial Ends + (2.0, 243), # Train Classifier + (3.0, 240), # Trial Started + (3.3, 11), # non-target + (3.6, 11), # non-target + (3.9, 11), # non-target + (4.2, 241), # Trial Ends + ] + + fake_eeg = FakeStimEegSource( + trigger_schedule=schedule, + n_channels=9, # 8 EEG + 1 stim + stim_channel_index=-1, + fsample=256.0, + duration=5.0, + ) + + logger.info(f"Trigger map: {TRIGGER_MAP}") + + # Wrap in EegStimTriggerMarkerSource (for non-LSL sources) + stim_source = EegStimTriggerMarkerSource( + eeg_source=fake_eeg, + stim_channel_name="STIM", + trigger_map=TRIGGER_MAP, + detect_mode="rise", + include_unmapped=True, + ) + + logger.info("Reading samples and markers...") + total_markers = 0 + + for _ in range(100): + samples, timestamps = stim_source.get_samples() + if not timestamps or timestamps == []: + if fake_eeg._samples_generated >= fake_eeg._total_samples: + break + time.sleep(0.05) + continue + + markers, marker_ts = stim_source.get_markers() + for marker, ts in zip(markers, marker_ts): + total_markers += 1 + logger.info(f" MARKER @ t={ts:.4f}s : {marker[0]}") + + time.sleep(0.05) + + logger.info(f"Done — detected {total_markers} trigger markers.") + + +def run_real_mode() -> None: + """Run with a live LSL EEG stream using LslStimMarkerSource.""" + logger.info("=" * 60) + logger.info("REAL MODE — requires live LSL EEG stream with stim channel") + logger.info("=" * 60) + + try: + from bci_essentials.io.lsl_stim_marker_source import LslStimMarkerSource + except ImportError: + logger.error("mne-lsl not installed — cannot use real mode.") + return + + logger.info("Searching for LSL EEG stream (timeout 10 s)...") + try: + marker_source = LslStimMarkerSource( + stim_channel_name="STIM", + trigger_map=TRIGGER_MAP, + detect_mode="rise", + timeout=10, + ) + except Exception as e: + logger.error(f"Could not connect: {e}") + return + + logger.info("Listening for triggers for 30 seconds...") + deadline = time.monotonic() + 30 + total_markers = 0 + + while time.monotonic() < deadline: + markers, marker_ts = marker_source.get_markers() + for marker, ts in zip(markers, marker_ts): + total_markers += 1 + logger.info(f" MARKER @ t={ts:.4f}s : {marker[0]}") + time.sleep(0.01) + + logger.info(f"Done — detected {total_markers} trigger markers in 30 s.") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description="Demonstrate serial-port trigger ingestion." + ) + parser.add_argument( + "--real", + action="store_true", + help="Use a live LSL EEG stream instead of the fake source.", + ) + args = parser.parse_args() + + if args.real: + run_real_mode() + else: + run_fake_mode() diff --git a/pyproject.toml b/pyproject.toml index 367b147f..f85b4864 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,6 +40,7 @@ exclude = ''' | \.mypy_cache | \.tox | \.venv + | venv | _build | buck-out | build diff --git a/tests/test_eeg_trigger_sources.py b/tests/test_eeg_trigger_sources.py new file mode 100644 index 00000000..b85429dc --- /dev/null +++ b/tests/test_eeg_trigger_sources.py @@ -0,0 +1,565 @@ +import unittest + +from bci_essentials.io.eeg_trigger_sources import ( + DEFAULT_TRIGGER_MAP, + EegStimTriggerMarkerSource, + make_p300_trigger_map, + make_mi_trigger_map, + make_ssvep_trigger_map, + make_simple_trigger_map, + TARGET_BYTE, + NON_TARGET_BYTE, +) +from bci_essentials.io.sources import EegSource, MarkerSource + + +class FakeEegSource(EegSource): + def __init__( + self, + samples=None, + timestamps=None, + channel_labels=None, + fsample=256.0, + ): + self._channel_labels = channel_labels or ["Ch0", "Ch1", "Ch2", "STIM"] + self._n_channels = len(self._channel_labels) + self._batches = [] + if samples is not None: + self._batches.append((samples, timestamps or [])) + self._batch_idx = 0 + self._fsample = fsample + + def add_batch(self, samples, timestamps): + self._batches.append((samples, timestamps)) + + @property + def name(self) -> str: + return "FakeEegSource" + + @property + def fsample(self) -> float: + return self._fsample + + @property + def n_channels(self) -> int: + return self._n_channels + + @property + def channel_types(self) -> list[str]: + return ["eeg"] * self._n_channels + + @property + def channel_units(self) -> list[str]: + return ["microvolts"] * self._n_channels + + @property + def channel_labels(self) -> list[str]: + return list(self._channel_labels) + + def get_samples(self) -> tuple[list[list], list]: + if self._batch_idx < len(self._batches): + result = self._batches[self._batch_idx] + self._batch_idx += 1 + return result + return [[]], [] + + def time_correction(self) -> float: + return 0.0 + + +class TestDefaultTriggerMap(unittest.TestCase): + def test_has_six_status_entries(self): + self.assertEqual(len(DEFAULT_TRIGGER_MAP), 6) + + def test_known_byte_values(self): + self.assertEqual(DEFAULT_TRIGGER_MAP[240], "Trial Started") + self.assertEqual(DEFAULT_TRIGGER_MAP[241], "Trial Ends") + self.assertEqual(DEFAULT_TRIGGER_MAP[242], "Training Complete") + self.assertEqual(DEFAULT_TRIGGER_MAP[243], "Train Classifier") + self.assertEqual(DEFAULT_TRIGGER_MAP[244], "Update Classifier") + self.assertEqual(DEFAULT_TRIGGER_MAP[245], "Done with all RS collection") + + +class TestMakeSimpleTriggerMap(unittest.TestCase): + def test_target_and_non_target_present(self): + m = make_simple_trigger_map(include_status=False) + self.assertEqual(m[TARGET_BYTE], "target") + self.assertEqual(m[NON_TARGET_BYTE], "non_target") + + def test_includes_status_by_default(self): + m = make_simple_trigger_map() + self.assertEqual(m[TARGET_BYTE], "target") + self.assertEqual(m[240], "Trial Started") + + def test_custom_bytes(self): + m = make_simple_trigger_map( + target_byte=10, non_target_byte=20, include_status=False + ) + self.assertEqual(m[10], "target") + self.assertEqual(m[20], "non_target") + + def test_raises_on_duplicate_bytes(self): + with self.assertRaises(ValueError): + make_simple_trigger_map(target_byte=5, non_target_byte=5) + + def test_custom_entries_merged(self): + m = make_simple_trigger_map(custom={99: "special"}, include_status=False) + self.assertEqual(m[99], "special") + + def test_includes_all_status_events(self): + m = make_simple_trigger_map(include_status=True) + self.assertEqual(m[240], "Trial Started") + self.assertEqual(m[241], "Trial Ends") + self.assertEqual(m[242], "Training Complete") + self.assertEqual(m[243], "Train Classifier") + self.assertEqual(m[244], "Update Classifier") + self.assertEqual(m[245], "Done with all RS collection") + self.assertEqual(m[TARGET_BYTE], "target") + self.assertEqual(m[NON_TARGET_BYTE], "non_target") + self.assertEqual(len(m), 8) + + +class TestMakeP300TriggerMap(unittest.TestCase): + def test_entry_count(self): + m = make_p300_trigger_map(n_options=4) + self.assertEqual(len(m), 6 + 4) + + def test_stimulus_marker_format(self): + m = make_p300_trigger_map(n_options=3) + self.assertEqual(m[1], "p300,s,3,-1,1") + self.assertEqual(m[2], "p300,s,3,-1,2") + self.assertEqual(m[3], "p300,s,3,-1,3") + + def test_stimulus_bytes_are_one_indexed(self): + m = make_p300_trigger_map(n_options=2) + self.assertIn(1, m) + self.assertIn(2, m) + self.assertNotIn(0, m) + + def test_status_codes_preserved(self): + m = make_p300_trigger_map(n_options=1) + for k, v in DEFAULT_TRIGGER_MAP.items(): + self.assertEqual(m[k], v) + + def test_raises_on_zero_options(self): + with self.assertRaises(ValueError): + make_p300_trigger_map(n_options=0) + + def test_raises_on_too_many_options(self): + with self.assertRaises(ValueError): + make_p300_trigger_map(n_options=240) + + +class TestMakeMiTriggerMap(unittest.TestCase): + def test_entry_count(self): + m = make_mi_trigger_map(n_classes=2, epoch_length=2.0) + self.assertEqual(len(m), 6 + 2) + + def test_marker_contains_paradigm_prefix(self): + m = make_mi_trigger_map(n_classes=3, epoch_length=1.5) + self.assertIn("mi,", m[1]) + + def test_raises_on_invalid_classes(self): + with self.assertRaises(ValueError): + make_mi_trigger_map(n_classes=-1, epoch_length=1.0) + + +class TestMakeSsvepTriggerMap(unittest.TestCase): + def test_entry_count(self): + m = make_ssvep_trigger_map(frequencies=[8.0, 10.0, 12.0], epoch_length=4.0) + self.assertEqual(len(m), 6 + 3) + + def test_marker_contains_paradigm_prefix(self): + m = make_ssvep_trigger_map(frequencies=[8.0, 12.0], epoch_length=3.0) + self.assertIn("ssvep,", m[1]) + + def test_raises_on_empty_frequencies(self): + with self.assertRaises(ValueError): + make_ssvep_trigger_map(frequencies=[], epoch_length=1.0) + + +class TestConstructionValidation(unittest.TestCase): + def test_requires_stim_channel_name(self): + with self.assertRaises(TypeError): + EegStimTriggerMarkerSource(eeg_source=FakeEegSource()) + + def test_rejects_empty_channel_name(self): + with self.assertRaises(ValueError): + EegStimTriggerMarkerSource(eeg_source=FakeEegSource(), stim_channel_name="") + + def test_rejects_non_string_channel_name(self): + with self.assertRaises(ValueError): + EegStimTriggerMarkerSource(eeg_source=FakeEegSource(), stim_channel_name=3) + + def test_rejects_invalid_detect_mode(self): + with self.assertRaises(ValueError): + EegStimTriggerMarkerSource( + eeg_source=FakeEegSource(), + stim_channel_name="STIM", + detect_mode="invalid", + ) + + def test_rejects_non_eeg_source(self): + with self.assertRaises(TypeError): + EegStimTriggerMarkerSource( + eeg_source="not_a_source", stim_channel_name="STIM" + ) + + def test_rejects_out_of_range_trigger_map_key(self): + with self.assertRaises(ValueError): + EegStimTriggerMarkerSource( + eeg_source=FakeEegSource(), + stim_channel_name="STIM", + trigger_map={256: "too big"}, + ) + + def test_is_both_eeg_and_marker_source(self): + src = EegStimTriggerMarkerSource( + eeg_source=FakeEegSource(), stim_channel_name="STIM" + ) + self.assertIsInstance(src, EegSource) + self.assertIsInstance(src, MarkerSource) + + def test_enabled_true_by_default(self): + src = EegStimTriggerMarkerSource( + eeg_source=FakeEegSource(), stim_channel_name="STIM" + ) + self.assertTrue(src.enabled) + + def test_enabled_false_constructor(self): + src = EegStimTriggerMarkerSource( + eeg_source=FakeEegSource(), stim_channel_name="STIM", enabled=False + ) + self.assertFalse(src.enabled) + + +class TestEegSourcePassthrough(unittest.TestCase): + def setUp(self) -> None: + self.eeg = FakeEegSource(channel_labels=["Fp1", "Fp2", "STIM"], fsample=512.0) + self.src = EegStimTriggerMarkerSource( + eeg_source=self.eeg, stim_channel_name="STIM" + ) + + def test_name(self): + self.assertEqual(self.src.name, "FakeEegSource") + + def test_fsample(self): + self.assertEqual(self.src.fsample, 512.0) + + def test_n_channels(self): + self.assertEqual(self.src.n_channels, 3) + + def test_channel_labels(self): + self.assertEqual(self.src.channel_labels, ["Fp1", "Fp2", "STIM"]) + + def test_time_correction(self): + self.assertEqual(self.src.time_correction(), 0.0) + + +class TestRiseModeDetection(unittest.TestCase): + def test_single_trigger_on_zero_to_nonzero(self): + eeg = FakeEegSource( + samples=[[0, 0, 0, 0], [0, 0, 0, 1], [0, 0, 0, 0]], + timestamps=[1.0, 2.0, 3.0], + ) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, + stim_channel_name="STIM", + trigger_map={1: "Trial Started"}, + ) + src.get_samples() + markers, ts = src.get_markers() + self.assertEqual(markers, [["Trial Started"]]) + self.assertAlmostEqual(ts[0], 2.0) + + def test_no_repeat_while_sustained_high(self): + eeg = FakeEegSource( + samples=[[0, 0, 0, 1], [0, 0, 0, 1], [0, 0, 0, 1]], + timestamps=[1.0, 2.0, 3.0], + ) + src = EegStimTriggerMarkerSource(eeg_source=eeg, stim_channel_name="STIM") + src.get_samples() + _, ts = src.get_markers() + self.assertEqual(ts, []) + + def test_multiple_rises_detected(self): + eeg = FakeEegSource( + samples=[ + [0, 0, 0, 0], + [0, 0, 0, 1], + [0, 0, 0, 0], + [0, 0, 0, 2], + [0, 0, 0, 0], + ], + timestamps=[1.0, 2.0, 3.0, 4.0, 5.0], + ) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, + stim_channel_name="STIM", + trigger_map={1: "Trial Started", 2: "Trial Ends"}, + ) + src.get_samples() + markers, _ = src.get_markers() + self.assertEqual(markers, [["Trial Started"], ["Trial Ends"]]) + + def test_timestamp_matches_trigger_sample(self): + eeg = FakeEegSource( + samples=[[0, 0, 0, 0], [0, 0, 0, 0], [0, 0, 0, 240]], + timestamps=[10.0, 10.004, 10.008], + ) + src = EegStimTriggerMarkerSource(eeg_source=eeg, stim_channel_name="STIM") + src.get_samples() + _, ts = src.get_markers() + self.assertAlmostEqual(ts[0], 10.008) + + +class TestChangeModeDetection(unittest.TestCase): + def test_detects_any_nonzero_change(self): + eeg = FakeEegSource( + samples=[[0, 0, 0, 0], [0, 0, 0, 1], [0, 0, 0, 2]], + timestamps=[1.0, 2.0, 3.0], + ) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, + stim_channel_name="STIM", + trigger_map={1: "Trial Started", 2: "Trial Ends"}, + detect_mode="change", + ) + src.get_samples() + markers, _ = src.get_markers() + self.assertEqual(markers, [["Trial Started"], ["Trial Ends"]]) + + def test_zero_change_not_emitted(self): + eeg = FakeEegSource( + samples=[[0, 0, 0, 0], [0, 0, 0, 0]], + timestamps=[1.0, 2.0], + ) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, + stim_channel_name="STIM", + detect_mode="change", + ) + src.get_samples() + _, ts = src.get_markers() + self.assertEqual(ts, []) + + +class TestUnmappedTriggers(unittest.TestCase): + def test_unmapped_ignored_by_default(self): + eeg = FakeEegSource( + samples=[[0, 0, 0, 0], [0, 0, 0, 99]], + timestamps=[1.0, 2.0], + ) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, + stim_channel_name="STIM", + trigger_map={1: "Trial Started"}, + ) + src.get_samples() + markers, ts = src.get_markers() + self.assertEqual(ts, []) + + def test_unmapped_included_when_flag_set(self): + eeg = FakeEegSource( + samples=[[0, 0, 0, 0], [0, 0, 0, 99]], + timestamps=[1.0, 2.0], + ) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, + stim_channel_name="STIM", + trigger_map={1: "Trial Started"}, + include_unmapped=True, + ) + src.get_samples() + markers, _ = src.get_markers() + self.assertEqual(markers, [["trigger_99"]]) + + +class TestChannelNameLookup(unittest.TestCase): + def test_unknown_name_returns_no_markers(self): + eeg = FakeEegSource( + samples=[[0, 0, 0, 0], [1, 0, 0, 1]], + timestamps=[1.0, 2.0], + ) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, + stim_channel_name="DOES_NOT_EXIST", + trigger_map={1: "Trial Started"}, + ) + src.get_samples() + markers, timestamps = src.get_markers() + self.assertEqual(markers, [[]]) + self.assertEqual(timestamps, []) + + def test_correct_name_detects_triggers(self): + eeg = FakeEegSource( + samples=[[0, 0, 0, 0], [0, 0, 0, 1]], + timestamps=[1.0, 2.0], + ) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, + stim_channel_name="STIM", + trigger_map={1: "Trial Started"}, + ) + src.get_samples() + markers, _ = src.get_markers() + self.assertEqual(markers, [["Trial Started"]]) + + def test_name_setter_clears_cache(self): + eeg = FakeEegSource( + channel_labels=["Ch0", "OLD", "NEW"], + samples=[[0, 0, 0], [0, 1, 0]], + timestamps=[1.0, 2.0], + ) + eeg.add_batch( + samples=[[0, 0, 0], [0, 0, 1]], + timestamps=[3.0, 4.0], + ) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, + stim_channel_name="OLD", + trigger_map={1: "Trial Started"}, + ) + src.get_samples() + m1, _ = src.get_markers() + self.assertEqual(m1, [["Trial Started"]]) + + src.stim_channel_name = "NEW" + src.get_samples() + m2, _ = src.get_markers() + self.assertEqual(m2, [["Trial Started"]]) + + +class TestEnabledToggle(unittest.TestCase): + def test_disabled_skips_trigger_detection(self): + eeg = FakeEegSource( + samples=[[0, 0, 0, 0], [0, 0, 0, 1]], + timestamps=[1.0, 2.0], + ) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, stim_channel_name="STIM", enabled=False + ) + src.get_samples() + _, ts = src.get_markers() + self.assertEqual(ts, []) + + def test_eeg_data_passes_through_when_disabled(self): + original = [[1.0, 2.0, 3.0, 4.0]] + eeg = FakeEegSource(samples=original, timestamps=[1.0]) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, stim_channel_name="STIM", enabled=False + ) + samples, ts = src.get_samples() + self.assertEqual(samples, original) + + def test_reenable_resumes_detection(self): + eeg = FakeEegSource( + channel_labels=["Ch0", "Ch1", "Ch2", "STIM"], + samples=[[0, 0, 0, 0], [0, 0, 0, 1]], + timestamps=[1.0, 2.0], + ) + eeg.add_batch( + samples=[[0, 0, 0, 0], [0, 0, 0, 2]], + timestamps=[3.0, 4.0], + ) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, + stim_channel_name="STIM", + trigger_map={1: "Trial Started", 2: "Trial Ends"}, + enabled=False, + ) + src.get_samples() + _, ts1 = src.get_markers() + self.assertEqual(ts1, []) + + src.enabled = True + src.get_samples() + markers, ts2 = src.get_markers() + self.assertEqual(len(ts2), 1) + self.assertEqual(markers[0], ["Trial Ends"]) + + +class TestEdgeCases(unittest.TestCase): + def test_empty_samples_no_error(self): + eeg = FakeEegSource(samples=[], timestamps=[]) + src = EegStimTriggerMarkerSource(eeg_source=eeg, stim_channel_name="STIM") + samples, ts = src.get_samples() + self.assertEqual(samples, []) + markers, mts = src.get_markers() + self.assertEqual(mts, []) + + def test_get_markers_drains_queue(self): + eeg = FakeEegSource( + samples=[[0, 0, 0, 0], [0, 0, 0, 240]], + timestamps=[1.0, 2.0], + ) + src = EegStimTriggerMarkerSource(eeg_source=eeg, stim_channel_name="STIM") + src.get_samples() + _, ts1 = src.get_markers() + self.assertEqual(len(ts1), 1) + _, ts2 = src.get_markers() + self.assertEqual(ts2, []) + + def test_get_markers_empty_before_any_samples(self): + src = EegStimTriggerMarkerSource( + eeg_source=FakeEegSource(), stim_channel_name="STIM" + ) + markers, ts = src.get_markers() + self.assertEqual(markers, [[]]) + self.assertEqual(ts, []) + + def test_p300_map_integration(self): + tmap = make_p300_trigger_map(n_options=4) + eeg = FakeEegSource( + samples=[[0, 0, 0, 0], [0, 0, 0, 1]], + timestamps=[1.0, 2.0], + ) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, + stim_channel_name="STIM", + trigger_map=tmap, + ) + src.get_samples() + markers, _ = src.get_markers() + self.assertEqual(markers, [["p300,s,4,-1,1"]]) + + def test_simple_map_with_status_integration(self): + # status events (240+) + target/non-target on low bytes + tmap = make_simple_trigger_map(include_status=True) + eeg = FakeEegSource( + samples=[ + [0, 0, 0, 0], + [0, 0, 0, 240], + [0, 0, 0, 0], + [0, 0, 0, 1], + [0, 0, 0, 0], + [0, 0, 0, 2], + [0, 0, 0, 0], + [0, 0, 0, 241], + ], + timestamps=[1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0], + ) + src = EegStimTriggerMarkerSource( + eeg_source=eeg, + stim_channel_name="STIM", + trigger_map=tmap, + ) + src.get_samples() + markers, ts = src.get_markers() + self.assertEqual( + markers, + [["Trial Started"], ["target"], ["non_target"], ["Trial Ends"]], + ) + self.assertEqual(ts, [2.0, 4.0, 6.0, 8.0]) + + def test_underlying_source_error_handled(self): + eeg = FakeEegSource() + eeg.get_samples = lambda: (_ for _ in ()).throw(RuntimeError("boom")) + src = EegStimTriggerMarkerSource(eeg_source=eeg, stim_channel_name="STIM") + samples, ts = src.get_samples() + self.assertEqual(samples, [[]]) + self.assertEqual(ts, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_triggers.py b/tests/test_triggers.py new file mode 100644 index 00000000..c68bd624 --- /dev/null +++ b/tests/test_triggers.py @@ -0,0 +1,340 @@ +import time +import unittest + +from bci_essentials.triggers import ( + MarkerTypes, + TriggerByte, + TriggerDetector, + DEFAULT_TRIGGER_MAP, + SIMPLE_TRIGGER_MAP, + make_p300_trigger_map, + make_mi_trigger_map, + make_ssvep_trigger_map, + make_simple_trigger_map, +) +from bci_essentials.io.fake_stim_source import FakeStimEegSource +from bci_essentials.io.eeg_trigger_sources import EegStimTriggerMarkerSource + + +class TestMarkerTypes(unittest.TestCase): + def test_trial_started_value(self): + self.assertEqual(MarkerTypes.TRIAL_STARTED.value, "Trial Started") + + def test_trial_ends_value(self): + self.assertEqual(MarkerTypes.TRIAL_ENDS.value, "Trial Ends") + + def test_train_classifier_value(self): + self.assertEqual(MarkerTypes.TRAIN_CLASSIFIER.value, "Train Classifier") + + +class TestTriggerByte(unittest.TestCase): + def test_status_byte_range(self): + self.assertEqual(TriggerByte.TRIAL_STARTED, 240) + self.assertEqual(TriggerByte.TRIAL_ENDS, 241) + self.assertEqual(TriggerByte.TRAINING_COMPLETE, 242) + self.assertEqual(TriggerByte.TRAIN_CLASSIFIER, 243) + self.assertEqual(TriggerByte.UPDATE_CLASSIFIER, 244) + self.assertEqual(TriggerByte.DONE_RS_COLLECTION, 245) + + def test_target_bytes(self): + self.assertEqual(TriggerByte.TARGET, 1) + self.assertEqual(TriggerByte.NON_TARGET, 2) + + +class TestDefaultTriggerMap(unittest.TestCase): + def test_maps_status_bytes_to_marker_strings(self): + self.assertEqual(DEFAULT_TRIGGER_MAP[240], "Trial Started") + self.assertEqual(DEFAULT_TRIGGER_MAP[241], "Trial Ends") + self.assertEqual(DEFAULT_TRIGGER_MAP[243], "Train Classifier") + + def test_has_six_entries(self): + self.assertEqual(len(DEFAULT_TRIGGER_MAP), 6) + + +class TestSimpleTriggerMap(unittest.TestCase): + def test_maps_target_and_non_target(self): + self.assertEqual(SIMPLE_TRIGGER_MAP[1], "target") + self.assertEqual(SIMPLE_TRIGGER_MAP[2], "non_target") + + +class TestMakeP300TriggerMap(unittest.TestCase): + def test_includes_status_markers(self): + tmap = make_p300_trigger_map(6) + self.assertIn(240, tmap) + self.assertEqual(tmap[240], "Trial Started") + + def test_stimulus_entries(self): + tmap = make_p300_trigger_map(6) + self.assertEqual(tmap[1], "p300,s,6,-1,1") + self.assertEqual(tmap[6], "p300,s,6,-1,6") + + def test_entry_count(self): + tmap = make_p300_trigger_map(6) + # 6 status + 6 stimulus + self.assertEqual(len(tmap), 12) + + def test_raises_on_zero_options(self): + with self.assertRaises(ValueError): + make_p300_trigger_map(0) + + def test_raises_on_too_many_options(self): + with self.assertRaises(ValueError): + make_p300_trigger_map(240) + + +class TestMakeMiTriggerMap(unittest.TestCase): + def test_stimulus_entries(self): + tmap = make_mi_trigger_map(3, 2.0) + self.assertEqual(tmap[1], "mi,3,1,2.00") + self.assertEqual(tmap[2], "mi,3,2,2.00") + self.assertEqual(tmap[3], "mi,3,3,2.00") + + def test_includes_status_markers(self): + tmap = make_mi_trigger_map(3, 2.0) + self.assertIn(240, tmap) + + def test_raises_on_bad_inputs(self): + with self.assertRaises(ValueError): + make_mi_trigger_map(0, 2.0) + with self.assertRaises(ValueError): + make_mi_trigger_map(3, 0) + + +class TestMakeSsvepTriggerMap(unittest.TestCase): + def test_stimulus_entries(self): + tmap = make_ssvep_trigger_map([8.0, 10.0, 12.0], 4.0) + self.assertEqual(tmap[1], "ssvep,3,1,4.00,8.0,10.0,12.0") + + def test_includes_status_markers(self): + tmap = make_ssvep_trigger_map([8.0], 1.0) + self.assertIn(240, tmap) + + def test_raises_on_empty_frequencies(self): + with self.assertRaises(ValueError): + make_ssvep_trigger_map([], 1.0) + + +class TestMakeSimpleTriggerMap(unittest.TestCase): + def test_default_bytes(self): + tmap = make_simple_trigger_map() + self.assertEqual(tmap[1], "target") + self.assertEqual(tmap[2], "non_target") + + def test_custom_bytes(self): + tmap = make_simple_trigger_map(target_byte=10, non_target_byte=11) + self.assertEqual(tmap[10], "target") + self.assertEqual(tmap[11], "non_target") + + def test_raises_on_same_bytes(self): + with self.assertRaises(ValueError): + make_simple_trigger_map(target_byte=5, non_target_byte=5) + + +class TestTriggerDetectorRiseMode(unittest.TestCase): + def setUp(self) -> None: + self.tmap = {240: "Trial Started", 1: "stim_0", 2: "stim_1"} + self.detector = TriggerDetector(self.tmap, detect_mode="rise") + + def test_baseline_zero_returns_none(self): + self.assertIsNone(self.detector.detect(0)) + + def test_rise_from_zero_triggers(self): + self.detector.detect(0) + self.assertEqual(self.detector.detect(240), "Trial Started") + + def test_held_value_does_not_retrigger(self): + self.detector.detect(0) + self.detector.detect(240) + self.assertIsNone(self.detector.detect(240)) + + def test_return_to_zero_and_new_trigger(self): + self.detector.detect(0) + self.detector.detect(240) + self.detector.detect(0) + self.assertEqual(self.detector.detect(1), "stim_0") + + def test_unmapped_value_returns_none(self): + self.detector.detect(0) + self.assertIsNone(self.detector.detect(99)) + + def test_raises_on_invalid_mode(self): + with self.assertRaises(ValueError): + TriggerDetector(self.tmap, detect_mode="invalid") + + +class TestTriggerDetectorChangeMode(unittest.TestCase): + def setUp(self) -> None: + self.tmap = {240: "Trial Started", 1: "stim_0"} + self.detector = TriggerDetector(self.tmap, detect_mode="change") + + def test_change_triggers_on_any_transition(self): + self.detector.detect(0) + self.assertEqual(self.detector.detect(240), "Trial Started") + # 240 -> 1 triggers on change + self.assertEqual(self.detector.detect(1), "stim_0") + + def test_change_to_zero_returns_none(self): + self.detector.detect(0) + self.detector.detect(240) + self.assertIsNone(self.detector.detect(0)) + + +class TestTriggerDetectorUnmapped(unittest.TestCase): + def test_include_unmapped_produces_fallback_string(self): + detector = TriggerDetector({}, detect_mode="rise", include_unmapped=True) + detector.detect(0) + self.assertEqual(detector.detect(99), "trigger_99") + + def test_exclude_unmapped_tracks_warned_set(self): + detector = TriggerDetector({}, detect_mode="rise", include_unmapped=False) + detector.detect(0) + detector.detect(99) + self.assertIn(99, detector.warned_unmapped) + + +class TestFakeStimEegSource(unittest.TestCase): + def test_properties(self): + source = FakeStimEegSource(n_channels=9, fsample=256.0) + self.assertEqual(source.name, "FakeStimEegSource") + self.assertEqual(source.fsample, 256.0) + self.assertEqual(source.n_channels, 9) + self.assertEqual(len(source.channel_types), 9) + self.assertEqual(len(source.channel_units), 9) + self.assertEqual(len(source.channel_labels), 9) + + def test_stim_channel_label(self): + source = FakeStimEegSource(n_channels=9) + self.assertEqual(source.channel_labels[-1], "STIM") + self.assertEqual(source.channel_types[-1], "stim") + + def test_custom_channel_labels(self): + labels = ["C1", "C2", "C3", "C4", "C5", "C6", "C7", "C8", "TRG"] + source = FakeStimEegSource(n_channels=9, channel_labels=labels) + self.assertEqual(source.channel_labels, labels) + + def test_raises_on_wrong_label_count(self): + with self.assertRaises(ValueError): + FakeStimEegSource(n_channels=9, channel_labels=["a", "b"]) + + def test_raises_on_both_trigger_modes(self): + with self.assertRaises(ValueError): + FakeStimEegSource(trigger_bytes=[1], trigger_schedule=[(0.0, 1)]) + + def test_time_correction_is_zero(self): + source = FakeStimEegSource() + self.assertEqual(source.time_correction(), 0.0) + + def test_get_samples_returns_data(self): + source = FakeStimEegSource(n_channels=4, fsample=256.0, duration=1.0) + source.get_samples() # starts the internal clock + time.sleep(0.1) + samples, timestamps = source.get_samples() + self.assertGreater(len(timestamps), 0) + self.assertEqual(len(samples[0]), 4) + + def test_get_samples_empty_after_duration(self): + source = FakeStimEegSource(n_channels=4, fsample=100.0, duration=0.01) + time.sleep(0.05) + # drain all samples + source.get_samples() + samples, timestamps = source.get_samples() + self.assertEqual(samples, [[]]) + self.assertEqual(len(timestamps), 0) + + +class TestEegStimTriggerMarkerSource(unittest.TestCase): + def setUp(self) -> None: + self.schedule = [ + (0.0, 240), # Trial Started + (0.3, 1), # flash 0 + (0.6, 2), # flash 1 + (1.5, 241), # Trial Ends + ] + self.fake_eeg = FakeStimEegSource( + trigger_schedule=self.schedule, + n_channels=9, + fsample=256.0, + duration=2.0, + ) + self.trigger_map = make_p300_trigger_map(n_options=6) + self.source = EegStimTriggerMarkerSource( + eeg_source=self.fake_eeg, + stim_channel_name="STIM", + trigger_map=self.trigger_map, + detect_mode="rise", + ) + + def test_eeg_properties_delegate(self): + self.assertEqual(self.source.name, "FakeStimEegSource") + self.assertEqual(self.source.fsample, 256.0) + self.assertEqual(self.source.n_channels, 9) + self.assertEqual(len(self.source.channel_labels), 9) + self.assertEqual(len(self.source.channel_types), 9) + self.assertEqual(len(self.source.channel_units), 9) + + def test_time_correction(self): + self.assertEqual(self.source.time_correction(), 0.0) + + def test_get_markers_empty_before_samples(self): + markers, timestamps = self.source.get_markers() + self.assertEqual(markers, [[]]) + self.assertEqual(len(timestamps), 0) + + def test_end_to_end_trigger_detection(self): + detected = [] + for _ in range(200): + samples, ts = self.source.get_samples() + if not ts: + if self.fake_eeg._samples_generated >= self.fake_eeg._total_samples: + break + time.sleep(0.01) + continue + markers, mts = self.source.get_markers() + for m, t in zip(markers, mts): + detected.append(m[0]) + time.sleep(0.01) + + self.assertEqual(len(detected), 4) + self.assertEqual(detected[0], "Trial Started") + self.assertEqual(detected[1], "p300,s,6,-1,1") + self.assertEqual(detected[2], "p300,s,6,-1,2") + self.assertEqual(detected[3], "Trial Ends") + + def test_raises_on_bad_eeg_source(self): + with self.assertRaises(TypeError): + EegStimTriggerMarkerSource( + eeg_source="not_an_eeg_source", + stim_channel_name="STIM", + ) + + def test_raises_on_empty_stim_channel_name(self): + with self.assertRaises(ValueError): + EegStimTriggerMarkerSource( + eeg_source=self.fake_eeg, + stim_channel_name="", + ) + + def test_raises_on_bad_detect_mode(self): + with self.assertRaises(ValueError): + EegStimTriggerMarkerSource( + eeg_source=self.fake_eeg, + stim_channel_name="STIM", + detect_mode="invalid", + ) + + def test_enabled_property(self): + self.assertTrue(self.source.enabled) + self.source.enabled = False + self.assertFalse(self.source.enabled) + + def test_disabled_returns_no_markers(self): + self.source.enabled = False + time.sleep(0.1) + self.source.get_samples() + markers, timestamps = self.source.get_markers() + self.assertEqual(markers, [[]]) + self.assertEqual(len(timestamps), 0) + + +if __name__ == "__main__": + unittest.main()