From b97ba5571209faedecb4939d85a33ffd03311356 Mon Sep 17 00:00:00 2001 From: Khuzaymah Bin Haris Date: Wed, 19 Aug 2026 01:55:17 +0100 Subject: [PATCH] Save Muse 2/S PPG and IMU as sidecar CSVs. BrainFlow keeps those streams at different sampling rates, so writing them into the EEG table would break existing analysis. Enable p51/p61 by default while leaving the EEG CSV columns unchanged. Fixes #233 --- doc/getting_started/streaming.md | 13 +++ eegnb/devices/eeg.py | 122 +++++++++++++++++++++++++- tests/test_acquisition.py | 141 ++++++++++++++++++++++++++++++- 3 files changed, 270 insertions(+), 6 deletions(-) diff --git a/doc/getting_started/streaming.md b/doc/getting_started/streaming.md index f28ebde87..22c32fc80 100644 --- a/doc/getting_started/streaming.md +++ b/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) diff --git a/eegnb/devices/eeg.py b/eegnb/devices/eeg.py index 6c017ed21..1d1073776 100644 --- a/eegnb/devices/eeg.py +++ b/eegnb/devices/eeg.py @@ -7,6 +7,7 @@ import sys import logging +from pathlib import Path from time import sleep, time from datetime import datetime from multiprocessing import Process @@ -14,7 +15,7 @@ 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 diff --git a/tests/test_acquisition.py b/tests/test_acquisition.py index 5626f6ac0..1531d4c80 100644 --- a/tests/test_acquisition.py +++ b/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()