Route every event marker through push_marker() / subscribe_marker()

conftest.py

@@ -14,6 +14,7 @@ def _is_available(module_name: str) -> bool:
     collect_ignore += [
         "eegnb/experiments",
         "eegnb/devices/vr.py",
+        "tests/test_push_marker.py",  # imports BaseExperiment, which pulls in psychopy via devices.vr
     ]
 elif not _is_available("psychxr"):
     collect_ignore += ["eegnb/devices/vr.py"]

eegnb/devices/eeg.py

@@ -185,6 +185,8 @@ def _stop_muse(self):
         pass
 
     def _muse_push_sample(self, marker, timestamp):
+        if not isinstance(marker, (list, tuple)):
+            marker = [marker]
         self.muse_StreamOutlet.push_sample(marker, timestamp)
 
     def _muse_get_recent(self, n_samples: int = 256, restart_inlet: bool = False):

eegnb/experiments/Experiment.py

@@ -9,13 +9,14 @@
 """
 
 from abc import abstractmethod, ABC
-from typing import Callable
+from typing import Callable, Optional
 from eegnb.devices.eeg import EEG
 from eegnb.devices.vr import VR
 from psychopy import prefs, visual, event, core
 
 import gc
 from time import time
+import logging
 import random
 import json
 
@@ -24,11 +25,18 @@
 
 from eegnb import generate_save_fn
 
+logger = logging.getLogger(__name__)
+
+
+# A marker subscriber: called as callback(marker, timestamp, trial_idx) on every
+# push_marker(); trial_idx may be None. Its return value is ignored.
+MarkerSubscriber = Callable[[int, float, Optional[int]], None]
+
 
 class BaseExperiment(ABC):
 
     def __init__(self, exp_name, duration, eeg, save_fn, n_trials: int, iti: float, soa: float, jitter: float,
-                 use_vr=False, use_fullscr = True, screen_num=0, stereoscopic = False, devices = list):
+                 use_vr=False, use_fullscr = True, screen_num=0, stereoscopic = False, devices=None):
         """ Initializer for the Base Experiment Class
 
         Args:
@@ -51,7 +59,7 @@ def __init__(self, exp_name, duration, eeg, save_fn, n_trials: int, iti: float,
         Press spacebar to continue. \n""".format(self.exp_name)
         self.duration = duration
         self.eeg: EEG = eeg
-        self.devices = devices
+        self.devices = devices if devices is not None else []
         self.save_fn = save_fn
         self.n_trials = n_trials
         self.iti = iti
@@ -77,6 +85,16 @@ def __init__(self, exp_name, duration, eeg, save_fn, n_trials: int, iti: float,
         # Initializing the marker names
         self.markernames = [1, 2]
 
+        # (callback, raise_on_error) pairs invoked on every push_marker().
+        self.marker_subscribers: list[tuple[MarkerSubscriber, bool]] = []
+
+        # Per-subscriber failure count: broken subscribers log once, not per trial.
+        self._subscriber_failures: dict = {}
+
+        # Setting event marker subscribers
+        self.subscribe_marker(self._write_marker_to_eeg)
+        self.subscribe_marker(self._write_marker_to_devices)
+
         # Setting up the trial and parameter list
         self.parameter = np.random.binomial(1, 0.5, self.n_trials)
         self.trials = DataFrame(dict(parameter=self.parameter, timestamp=np.zeros(self.n_trials)))
@@ -361,6 +379,10 @@ def run(self, instructions=True):
         # Clearing the screen for the next trial
         event.clearEvents()
 
+        # Surface any optional-subscriber failures whose repeats were suppressed.
+        for callback, count in self._subscriber_failures.items():
+            logger.warning("marker subscriber %s failed on %d marker(s)", callback, count)
+
         # Closing the EEG stream
         if self.eeg:
             self.eeg.stop()
@@ -371,12 +393,64 @@ def run(self, instructions=True):
         # Closing the window
         self.window.close()
 
-    def send_triggers(self, marker):
-        """Send timing triggers to recording device[s]"""
+    def subscribe_marker(self, callback: MarkerSubscriber, raise_on_error: bool = True):
+        """Register a marker subscriber: callable(marker, timestamp, trial_idx).
+
+        Invoked on every push_marker(). raise_on_error selects how a raised
+        exception is handled:
+
+        raise_on_error=True (the default) — essential recorders (the EEG board,
+            trigger/serial boxes) where a lost marker in the primary data is
+            unrecoverable. An exception propagates and aborts the run.
+        raise_on_error=False — optional telemetry (flip-time, eyetracker,
+            photodiode) that must never block core data. Exceptions are logged
+            and swallowed, so a bad observer can't take down a live recording.
+        """
+        self.marker_subscribers.append((callback, raise_on_error))
+
+    def _write_marker_to_eeg(self, marker, timestamp, trial_idx):
+        """Built-in essential subscriber: record the marker on the EEG board."""
+        if self.eeg:
+            self.eeg.push_sample(marker=marker, timestamp=timestamp)
+
+    def _write_marker_to_devices(self, marker, timestamp, trial_idx):
+        """Built-in essential subscriber: record the marker on every aux device."""
         for dev in self.devices:
-            timestamp = time()
             dev.push_sample(marker=marker, timestamp=timestamp)
 
+    def push_marker(self, marker, trial_idx=None):
+        """Send a stimulus marker to every subscriber under one shared timestamp.
+
+        A subscriber is any callable(marker, timestamp, trial_idx) registered via
+        subscribe_marker(). The built-in essential subscribers write the marker
+        onto the EEG board and any auxiliary devices (self.eeg, self.devices);
+        optional subscribers (raise_on_error=False) observe onset timing without
+        writing a marker to any stream — flip-time telemetry, eyetracker fixation,
+        photodiode metadata — and their exceptions are logged, not raised.
+
+        All subscribers share one timestamp so the marker onset is consistent
+        across streams. trial_idx is context for observers; callers without
+        observers may omit it. Returns the shared timestamp so a caller can record
+        the same onset in its own bookkeeping (e.g. a per-trial timestamp column)
+        without minting a second, slightly-skewed time().
+        """
+        timestamp = time()
+        for callback, raise_on_error in self.marker_subscribers:
+            try:
+                callback(marker, timestamp, trial_idx)
+            except Exception:
+                if raise_on_error:
+                    raise
+                # Log the traceback only on first failure (see subscribe_marker).
+                count = self._subscriber_failures.get(callback, 0) + 1
+                self._subscriber_failures[callback] = count
+                if count == 1:
+                    logger.exception(
+                        "marker subscriber failed; suppressing further tracebacks for it: %s",
+                        callback,
+                    )
+        return timestamp
+
     @property
     def name(self) -> str:
         """ This experiment's name """

eegnb/experiments/rest/eoec.py

@@ -1,6 +1,5 @@
 
 from typing import Optional
-from time import time
 
 from psychopy import prefs
 
@@ -49,8 +48,8 @@ def __init__(
         self.use_verbal_cues = use_verbal_cues
         self.open_audio = open_audio
         self.close_audio = close_audio
-        self.serial = None
-        self.outlet = None
+        self.serial: Optional["serial.Serial"] = None
+        self.outlet: Optional[StreamOutlet] = None
         self.open_sound = None
         self.close_sound = None
         super().__init__(
@@ -87,11 +86,20 @@ def setup(self, instructions: bool = True):
         # LSL outlet for markers
         info = StreamInfo("Markers", "Markers", 1, 0, "int32", "eyeclosure-baseline")
         self.outlet = StreamOutlet(info)
+        outlet = self.outlet  # local binding: mypy can't narrow self.outlet inside the closure
+        self.subscribe_marker(
+            lambda marker, timestamp, _idx: outlet.push_sample([marker], timestamp)
+        )
 
         # serial connection for hardware triggers
         if self.serial_port and serial is not None:
             try:
                 self.serial = serial.Serial(self.serial_port, 115200, timeout=1)
+                ser = self.serial  # local binding: mypy can't narrow self.serial inside the closure
+                self.subscribe_marker(
+                    lambda marker, timestamp, _idx: ser.write(bytes([marker])),
+                    raise_on_error=False,
+                )
             except Exception:  # pragma: no cover
                 self.serial = None
 
@@ -110,26 +118,8 @@ def present_stimulus(self, idx: int):
 
         label = self.trials["parameter"].iloc[idx]  # 0 open, 1 closed
         if self.trials["timestamp"].iloc[idx] == 0:
-            timestamp = time()
+            timestamp = self.push_marker(self.markernames[label], idx)
             self.trials.at[idx, "timestamp"] = timestamp
-            self.outlet.push_sample([self.markernames[label]], timestamp)
-
-            if self.eeg:
-              if self.eeg.backend == "muselsl":
-                  marker = [self.markernames[label]]
-              else:
-                  marker = self.markernames[label]
-              self.eeg.push_sample(marker=marker, timestamp=timestamp)
-
-            if self.devices:
-              marker = self.markernames[label]
-              self.send_triggers(marker)
-
-            if self.serial:
-                try:
-                    self.serial.write(bytes([self.markernames[label]]))
-                except Exception:  # pragma: no cover
-                    pass
 
             if label == 0:
                 self.open_sound.play()

eegnb/experiments/visual_n170/n170.py

@@ -1,7 +1,6 @@
 """  eeg-notebooks/eegnb/experiments/visual_n170/n170.py """
 
 import os
-from time import time
 from glob import glob
 from random import choice
 from psychopy import visual, core, event
@@ -44,23 +43,7 @@ def present_stimulus(self, idx: int):
         # Draw the image
         image.draw()
 
-
-        # Pushing the sample to the EEG
-        if self.eeg:
-            timestamp = time()
-
-            if self.eeg.backend == "muselsl":
-                marker = [self.markernames[label]]
-            else:
-                marker = self.markernames[label]
-
-            self.eeg.push_sample(marker=marker, timestamp=timestamp)
-
-
-        if self.devices:
-            marker = self.markernames[label]
-            self.send_triggers(marker)
-
+        self.push_marker(self.markernames[label], idx)
 
         self.window.flip()
 

eegnb/experiments/visual_p300/p300.py

@@ -2,7 +2,6 @@
 """  eeg-notebooks/eegnb/experiments/visual_p300/p300.py """
 
 import os
-from time import time
 from glob import glob
 from random import choice
 from optparse import OptionParser
@@ -41,13 +40,6 @@ def present_stimulus(self, idx: int):
         image = choice(self.targets if label == 1 else self.nontargets)
         image.draw()
 
-        # Push sample
-        if self.eeg:
-            timestamp = time()
-            if self.eeg.backend == "muselsl":
-                marker = [self.markernames[label]]
-            else:
-                marker = self.markernames[label]
-            self.eeg.push_sample(marker=marker, timestamp=timestamp)
+        self.push_marker(self.markernames[label], idx)
 
         self.window.flip()

eegnb/experiments/visual_ssvep/ssvep.py

@@ -1,7 +1,6 @@
 
 from eegnb.experiments import Experiment
 import os
-from time import time
 from glob import glob
 from random import choice
 
@@ -95,17 +94,11 @@ def present_stimulus(self, idx: int):
 
         # Select stimulus frequency
         ind = self.trials["parameter"].iloc[idx]
+        marker = self.markernames[ind]
+        marker_pushed = False
 
-        # Push sample
-        if self.eeg:
-            timestamp = time()
-            if self.eeg.backend == "muselsl":
-                marker = [self.markernames[ind]]
-            else:
-                marker = self.markernames[ind]
-            self.eeg.push_sample(marker=marker, timestamp=timestamp)
-
-        # Present flickering stim
+        # Present flickering stim; push the marker on the first flip so the
+        # captured onset is the true stimulus onset (not one frame early).
         for _ in range(int(self.stim_patterns[ind]["n_cycles"])):
 
             for _ in range(int(self.stim_patterns[ind]["cycle"][0])):
@@ -116,6 +109,9 @@ def present_stimulus(self, idx: int):
                 self.grating.draw()
                 self.fixation.draw()
                 self.window.flip()
+                if not marker_pushed:
+                    self.push_marker(marker, idx)
+                    marker_pushed = True
 
             for _ in range(self.stim_patterns[ind]["cycle"][1]):
                 if self.use_vr:

tests/test_acquisition.py

@@ -54,3 +54,25 @@ def test_synthetic_acquisition(tmp_path):
     assert (data['stim'] != 0).any()
 
     print(f"Acquired {len(data)} samples with columns: {list(data.columns)}")
+
+
+def test_muse_push_sample_wraps_a_scalar_but_leaves_a_list_alone():
+    """EEG._muse_push_sample sends a single-channel [marker] to the LSL outlet:
+    a scalar is wrapped, an already-wrapped list passes through (P300 path)."""
+
+    class FakeOutlet:
+        def __init__(self):
+            self.last = None
+
+        def push_sample(self, marker, timestamp):
+            self.last = marker
+
+    outlet = FakeOutlet()
+    eeg = object.__new__(EEG)                 # skip __init__ (it would connect hardware)
+    eeg.muse_StreamOutlet = outlet
+
+    eeg._muse_push_sample(2, 123.0)
+    assert outlet.last == [2]                 # scalar -> single-channel list
+
+    eeg._muse_push_sample([2], 123.0)
+    assert outlet.last == [2]                 # already a list -> unchanged