Save Muse 2/S PPG and IMU as sidecar CSVs

doc/getting_started/streaming.md

@@ -39,6 +39,19 @@ then you should see a GUI which looks something like the image below.
 Once you press the **Start Streaming** button, muse will be streaming data in the background and can the above code can
 be run to begin the notebooks interfacing with the bluemuse backend.
 
+### Interaxon Muse (BrainFlow)
+**Device Names:** *'muse2_bfn'*, *'muse2_bfb'*, *'museS_bfn'*, and *'museS_bfb'*
+**Backend:** Brainflow
+**Needed Parameters:** Native bluetooth (`*_bfn`) needs no extra parameters. BLED dongle devices (`*_bfb`) use a BLED112 adapter.
+
+These BrainFlow Muse 2 / Muse S names are the preferred way to record from Muse. EEG-ExPy still writes the usual EEG CSV, and also writes sidecar files for streams that run at different sampling rates:
+
+- `recording_....csv` — EEG + stim markers (~256 Hz)
+- `recording_...._ppg.csv` — photoplethysmography (PPG, ~64 Hz)
+- `recording_...._accel.csv` — accelerometer and gyroscope (~52 Hz)
+
+PPG is enabled automatically with Muse command `p51` (Muse 2) or `p61` (Muse S), which keep the four standard EEG channels. Pass `config=` to `EEG()` if you need a different Muse preset. Muse 2016 and the MuseLSL / BlueMuse backends do not save PPG.
+
 ### OpenBCI Ganglion
 ![fig](../img/ganglion.png)
 

eegnb/devices/eeg.py

@@ -7,14 +7,15 @@
 
 import sys
 import logging
+from pathlib import Path
 from time import sleep, time
 from datetime import datetime
 from multiprocessing import Process
 
 import numpy as np
 import pandas as pd
 
-from brainflow.board_shim import BoardShim, BoardIds, BrainFlowInputParams
+from brainflow.board_shim import BoardShim, BoardIds, BrainFlowInputParams, BrainFlowPresets
 from muselsl import stream, list_muses, record, constants as mlsl_cnsts
 from pylsl import StreamInfo, StreamOutlet, StreamInlet, resolve_byprop
 
@@ -68,6 +69,22 @@
     "muse2016_bfb",
 ]
 
+# Muse 2/S BrainFlow devices that can stream PPG. Commands keep 4 EEG channels
+# (p51 / p61) so the EEG CSV shape stays compatible. p50 would add a 5th EEG channel.
+MUSE_BRAINFLOW_PPG_COMMANDS = {
+    "muse2_bfn": "p51",
+    "muse2_bfb": "p51",
+    "museS_bfn": "p61",
+    "museS_bfb": "p61",
+}
+
+
+def muse_sidecar_path(save_fn, suffix: str) -> str:
+    """Return a sibling CSV path, e.g. recording.csv -> recording_ppg.csv."""
+    path = Path(save_fn)
+    return str(path.with_name(f"{path.stem}_{suffix}{path.suffix}"))
+
+
 xid_devices = [
    "nirsport2"
 ]
@@ -347,6 +364,8 @@ def _init_brainflow(self):
                     self.board.config_board(setting + 'X')
             else:
                 self.board.config_board(self.config)
+        else:
+            self._enable_muse_ppg_stream()
 
     def _start_brainflow(self):
         # only start stream if non exists
@@ -367,8 +386,15 @@ def _start_brainflow(self):
     def _stop_brainflow(self):
         """This functions kills the brainflow backend and saves the data to a CSV file."""
 
-        # Collect session data and kill session
-        data = self.board.get_board_data()  # will clear board buffer
+        # Collect session data. Muse boards keep EEG, IMU, and PPG in separate
+        # presets (different sampling rates), so pull all three before teardown.
+        data = self.board.get_board_data(preset=BrainFlowPresets.DEFAULT_PRESET)
+        aux_data = None
+        anc_data = None
+        if self.device_name in MUSE_BRAINFLOW_PPG_COMMANDS:
+            aux_data = self._get_board_data_soft(BrainFlowPresets.AUXILIARY_PRESET)
+            anc_data = self._get_board_data_soft(BrainFlowPresets.ANCILLARY_PRESET)
+
         self.board.stop_stream()
         self.board.release_session()
 
@@ -390,6 +416,96 @@ def _stop_brainflow(self):
         data_df = pd.DataFrame(total_data, columns=["timestamps"] + ch_names + ["stim"])
         data_df.to_csv(self.save_fn, index=False)
 
+        if self.device_name in MUSE_BRAINFLOW_PPG_COMMANDS:
+            self._write_muse_sidecars(aux_data, anc_data)
+
+    def _enable_muse_ppg_stream(self):
+        """Enable Muse PPG without adding a 5th EEG channel, unless the user set config."""
+        command = MUSE_BRAINFLOW_PPG_COMMANDS.get(self.device_name)
+        if not command:
+            return
+        try:
+            self.board.config_board(command)
+        except Exception:
+            logger.warning(
+                "Could not enable Muse PPG with command %s on %s",
+                command,
+                self.device_name,
+                exc_info=True,
+            )
+
+    def _get_board_data_soft(self, preset):
+        """Return board data for a preset, or None if the buffer is unavailable."""
+        try:
+            data = self.board.get_board_data(preset=preset)
+        except Exception:
+            logger.warning(
+                "Could not read BrainFlow preset %s from %s",
+                preset,
+                self.device_name,
+                exc_info=True,
+            )
+            return None
+        if data is None or getattr(data, "size", 0) == 0:
+            return None
+        return data
+
+    def _write_muse_sidecars(self, aux_data, anc_data):
+        """Write accel/gyro and PPG CSVs next to the EEG recording."""
+        if not self.save_fn:
+            return
+        self._write_brainflow_preset_csv(
+            aux_data,
+            BrainFlowPresets.AUXILIARY_PRESET,
+            muse_sidecar_path(self.save_fn, "accel"),
+            (("accel", BoardShim.get_accel_channels), ("gyro", BoardShim.get_gyro_channels)),
+        )
+        self._write_brainflow_preset_csv(
+            anc_data,
+            BrainFlowPresets.ANCILLARY_PRESET,
+            muse_sidecar_path(self.save_fn, "ppg"),
+            (("ppg", BoardShim.get_ppg_channels),),
+        )
+
+    def _write_brainflow_preset_csv(self, data, preset, out_path, channel_getters):
+        """Save one BrainFlow preset buffer. Failures must not block the EEG CSV."""
+        if data is None or getattr(data, "size", 0) == 0:
+            return
+        try:
+            data_t = np.asarray(data).T
+            timestamp_idx = BoardShim.get_timestamp_channel(self.brainflow_id, preset)
+            timestamps = data_t[:, timestamp_idx]
+
+            columns = ["timestamps"]
+            arrays = [timestamps[..., None]]
+
+            for prefix, getter in channel_getters:
+                try:
+                    idxs = list(getter(self.brainflow_id, preset))
+                except Exception:
+                    continue
+                if not idxs:
+                    continue
+                chunk = data_t[:, idxs]
+                columns.extend(f"{prefix}_{i}" for i in range(chunk.shape[1]))
+                arrays.append(chunk)
+
+            if len(arrays) == 1:
+                return
+
+            total_data = np.concatenate(arrays, axis=1)
+            try:
+                sfreq = BoardShim.get_sampling_rate(self.brainflow_id, preset)
+                trim = int(5 * sfreq)
+                if 0 < trim < len(total_data):
+                    total_data = total_data[trim:]
+            except Exception:
+                pass
+
+            pd.DataFrame(total_data, columns=columns).to_csv(out_path, index=False)
+        except Exception:
+            logger.warning("Failed to write Muse sidecar %s", out_path, exc_info=True)
+
     def _brainflow_extract(self, data):
         """
         Formats the data returned from brainflow to get

tests/test_acquisition.py

@@ -1,8 +1,13 @@
-import os
 import time
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import numpy as np
 import pandas as pd
-import pytest
-from eegnb.devices.eeg import EEG
+from brainflow.board_shim import BoardIds, BoardShim, BrainFlowPresets
+
+from eegnb.devices.eeg import EEG, MUSE_BRAINFLOW_PPG_COMMANDS, muse_sidecar_path
+
 
 def test_synthetic_acquisition(tmp_path):
     """
@@ -52,5 +57,135 @@ def test_synthetic_acquisition(tmp_path):
     # Check if markers were recorded (may vary slightly based on timing)
     # but we should at least see non-zero values in the stim column
     assert (data['stim'] != 0).any()
+
+    # Non-Muse boards should not grow sidecar files
+    assert not Path(muse_sidecar_path(save_fn, "ppg")).exists()
+    assert not Path(muse_sidecar_path(save_fn, "accel")).exists()
 
     print(f"Acquired {len(data)} samples with columns: {list(data.columns)}")
+
+
+def test_muse_sidecar_path():
+    path = Path("recording_2026-01-01-12.00.00.csv")
+    assert Path(muse_sidecar_path(path, "ppg")).name == "recording_2026-01-01-12.00.00_ppg.csv"
+    assert Path(muse_sidecar_path(path, "accel")).name == "recording_2026-01-01-12.00.00_accel.csv"
+
+
+def test_muse_ppg_commands_keep_four_eeg_channels():
+    assert MUSE_BRAINFLOW_PPG_COMMANDS["muse2_bfn"] == "p51"
+    assert MUSE_BRAINFLOW_PPG_COMMANDS["muse2_bfb"] == "p51"
+    assert MUSE_BRAINFLOW_PPG_COMMANDS["museS_bfn"] == "p61"
+    assert MUSE_BRAINFLOW_PPG_COMMANDS["museS_bfb"] == "p61"
+    assert "muse2016_bfn" not in MUSE_BRAINFLOW_PPG_COMMANDS
+
+
+def _fake_muse_eeg(device_name="muse2_bfn"):
+    eeg = EEG.__new__(EEG)
+    eeg.device_name = device_name
+    eeg.brainflow_id = BoardIds.MUSE_2_BOARD.value
+    eeg.sfreq = BoardShim.get_sampling_rate(eeg.brainflow_id)
+    eeg.ch_names = ["TP9", "AF7", "AF8", "TP10"]
+    eeg.markers = []
+    eeg.stream_started = True
+    return eeg
+
+
+def _preset_array(board_id, preset, n_samples, fill=1.0):
+    n_rows = BoardShim.get_num_rows(board_id, preset)
+    data = np.full((n_rows, n_samples), fill, dtype=float)
+    ts_idx = BoardShim.get_timestamp_channel(board_id, preset)
+    data[ts_idx] = np.arange(n_samples, dtype=float)
+    return data
+
+
+def test_enable_muse_ppg_stream_uses_p51_for_muse2():
+    eeg = _fake_muse_eeg("muse2_bfn")
+    eeg.board = MagicMock()
+    eeg._enable_muse_ppg_stream()
+    eeg.board.config_board.assert_called_once_with("p51")
+
+
+def test_enable_muse_ppg_stream_uses_p61_for_muses():
+    eeg = _fake_muse_eeg("museS_bfn")
+    eeg.brainflow_id = BoardIds.MUSE_S_BOARD.value
+    eeg.board = MagicMock()
+    eeg._enable_muse_ppg_stream()
+    eeg.board.config_board.assert_called_once_with("p61")
+
+
+def test_stop_brainflow_writes_muse_sidecars(tmp_path):
+    board_id = BoardIds.MUSE_2_BOARD.value
+    sfreq = BoardShim.get_sampling_rate(board_id)
+    aux_sfreq = BoardShim.get_sampling_rate(board_id, BrainFlowPresets.AUXILIARY_PRESET)
+    anc_sfreq = BoardShim.get_sampling_rate(board_id, BrainFlowPresets.ANCILLARY_PRESET)
+
+    eeg_data = _preset_array(board_id, BrainFlowPresets.DEFAULT_PRESET, 5 * sfreq + 10, 1.0)
+    aux_data = _preset_array(board_id, BrainFlowPresets.AUXILIARY_PRESET, 5 * aux_sfreq + 5, 2.0)
+    anc_data = _preset_array(board_id, BrainFlowPresets.ANCILLARY_PRESET, 5 * anc_sfreq + 5, 3.0)
+
+    def get_board_data(num_samples=None, preset=BrainFlowPresets.DEFAULT_PRESET):
+        if preset == BrainFlowPresets.AUXILIARY_PRESET:
+            return aux_data
+        if preset == BrainFlowPresets.ANCILLARY_PRESET:
+            return anc_data
+        return eeg_data
+
+    board = MagicMock()
+    board.get_board_data.side_effect = get_board_data
+
+    eeg = _fake_muse_eeg()
+    eeg.board = board
+    save_fn = tmp_path / "recording_2026-01-01.csv"
+    eeg.save_fn = str(save_fn)
+
+    eeg._stop_brainflow()
+
+    assert save_fn.exists()
+    data = pd.read_csv(save_fn)
+    assert "timestamps" in data.columns
+    assert "stim" in data.columns
+    assert "TP9" in data.columns
+    assert not any("ppg" in col.lower() for col in data.columns)
+
+    ppg_path = Path(muse_sidecar_path(save_fn, "ppg"))
+    accel_path = Path(muse_sidecar_path(save_fn, "accel"))
+    assert ppg_path.exists()
+    assert accel_path.exists()
+
+    ppg = pd.read_csv(ppg_path)
+    accel = pd.read_csv(accel_path)
+    assert "timestamps" in ppg.columns
+    assert any(col.startswith("ppg_") for col in ppg.columns)
+    assert "timestamps" in accel.columns
+    assert any(col.startswith("accel_") for col in accel.columns)
+
+    board.stop_stream.assert_called_once()
+    board.release_session.assert_called_once()
+
+
+def test_stop_brainflow_saves_eeg_if_sidecars_fail(tmp_path):
+    board_id = BoardIds.MUSE_2_BOARD.value
+    sfreq = BoardShim.get_sampling_rate(board_id)
+    eeg_data = _preset_array(board_id, BrainFlowPresets.DEFAULT_PRESET, 5 * sfreq + 10, 1.0)
+
+    def get_board_data(num_samples=None, preset=BrainFlowPresets.DEFAULT_PRESET):
+        if preset != BrainFlowPresets.DEFAULT_PRESET:
+            raise RuntimeError("preset unavailable")
+        return eeg_data
+
+    board = MagicMock()
+    board.get_board_data.side_effect = get_board_data
+
+    eeg = _fake_muse_eeg()
+    eeg.board = board
+    save_fn = tmp_path / "recording_2026-01-01.csv"
+    eeg.save_fn = str(save_fn)
+
+    eeg._stop_brainflow()
+
+    assert save_fn.exists()
+    data = pd.read_csv(save_fn)
+    assert "timestamps" in data.columns
+    assert "stim" in data.columns
+    assert not Path(muse_sidecar_path(save_fn, "ppg")).exists()
+    assert not Path(muse_sidecar_path(save_fn, "accel")).exists()