diff --git a/MuseLSL2/__main__.py b/MuseLSL2/__main__.py index f460c35..bf559bf 100644 --- a/MuseLSL2/__main__.py +++ b/MuseLSL2/__main__.py @@ -1,8 +1,6 @@ import argparse import sys -from .cli import CLI - def main(): parser = argparse.ArgumentParser( @@ -18,6 +16,8 @@ def main(): # exclude the rest of the args too, or validation will fail args = parser.parse_args(sys.argv[1:2]) + from .cli import CLI + if not hasattr(CLI, args.command): print("Incorrect usage. See help below.") parser.print_help() diff --git a/MuseLSL2/backends.py b/MuseLSL2/backends.py index 282d486..f816f1a 100644 --- a/MuseLSL2/backends.py +++ b/MuseLSL2/backends.py @@ -74,9 +74,29 @@ def char_write_handle( ) ) + def char_write_uuid(self, uuid, value, wait_for_response=True, timeout=30): + _wait(self._client.write_gatt_char(uuid, bytearray(value), wait_for_response)) + def subscribe(self, uuid, callback=None, indication=False, wait_for_response=True): def wrap(gatt_characteristic, data): value_handle = gatt_characteristic.handle + 1 callback(value_handle, data) _wait(self._client.start_notify(uuid, wrap)) + + def has_characteristic(self, uuid: str) -> bool: + services = self._client.services + if services is None: + # Ensure services are resolved + services = _wait(self._client.get_services()) + for service in services: + for char in service.characteristics: + if str(char.uuid).lower() == uuid.lower(): + return True + return False + + def get_services(self): + services = self._client.services + if services is None: + services = _wait(self._client.get_services()) + return services diff --git a/MuseLSL2/cli.py b/MuseLSL2/cli.py index e154ad4..b47565b 100644 --- a/MuseLSL2/cli.py +++ b/MuseLSL2/cli.py @@ -14,7 +14,9 @@ def find(self): find_devices(max_duration=10, verbose=True) def stream(self): - parser = argparse.ArgumentParser(description="Start an LSL stream from Muse headset.") + parser = argparse.ArgumentParser( + description="Start an LSL stream from Muse headset." + ) parser.add_argument( "-a", "--address", @@ -51,8 +53,8 @@ def stream(self): type=str, help="Select preset which dictates data channels to be streamed. Default is p50, but can also be 'none'", ) - args = parser.parse_args(sys.argv[2:]) + from .stream import stream stream(args.address, args.ppg, args.acc, args.gyro, args.preset) @@ -61,3 +63,25 @@ def view(self): from .view import view view() + + def inspect(self): + parser = argparse.ArgumentParser( + description="Inspect BLE GATT services/characteristics for a device." + ) + parser.add_argument("--address", "-a", required=True, help="Device MAC address") + args = parser.parse_args(sys.argv[2:]) + from .backends import BleakBackend + + adapter = BleakBackend() + dev = adapter.connect(args.address) + try: + services = dev.get_services() + for svc in services: + print(f"Service {svc.uuid} ({getattr(svc, 'description', '')})") + for ch in svc.characteristics: + props = ",".join(sorted(getattr(ch, "properties", []))) + print( + f" Char {ch.uuid} handle={ch.handle} props=[{props}] desc={getattr(ch, 'description', '')}" + ) + finally: + dev.disconnect() diff --git a/MuseLSL2/find.py b/MuseLSL2/find.py index 01199d5..528bbd6 100644 --- a/MuseLSL2/find.py +++ b/MuseLSL2/find.py @@ -5,7 +5,8 @@ def find_devices(max_duration=10, verbose=True): adapter = BleakBackend() adapter.start() - print(f"Searching for Muses (max. {max_duration} seconds)...") + if verbose: + print(f"Searching for Muses (max. {max_duration} seconds)...") devices = adapter.scan(timeout=max_duration) # Muse scan timeout adapter.stop() muses = [d for d in devices if d["name"] and "Muse" in d["name"]] diff --git a/MuseLSL2/muse.py b/MuseLSL2/muse.py index c34e6e5..b8d4b63 100644 --- a/MuseLSL2/muse.py +++ b/MuseLSL2/muse.py @@ -1,6 +1,7 @@ import bitstring import mne_lsl.lsl import numpy as np +from typing import cast from .backends import BleakBackend @@ -27,7 +28,18 @@ ATTR_PPG1 = "273e000f-4c4d-454d-96be-f03bac821358" # ambient 0x37-0x39 ATTR_PPG2 = "273e0010-4c4d-454d-96be-f03bac821358" # infrared 0x3a-0x3c ATTR_PPG3 = "273e0011-4c4d-454d-96be-f03bac821358" # red 0x3d-0x3f -ATTR_THERMISTOR = "273e0012-4c4d-454d-96be-f03bac821358" # muse S only, not implemented yet 0x40-0x42 +ATTR_THERMISTOR = ( + "273e0012-4c4d-454d-96be-f03bac821358" # muse S only, not implemented yet 0x40-0x42 +) + +# Muse S Athena notes +# --------------------------------------------------------------------------------------------- +# Newer devices (Muse S Athena) appear to use different characteristics layout: +# - One characteristic for all EEG channels +# - One characteristic for other sensors +# Based on BrainFlow PR #779 discussion, the following UUIDs are observed: +ATTR_ATHENA_EEG_ALL = "273e0013-4c4d-454d-96be-f03bac821358" +ATTR_ATHENA_SENSORS_ALL = "273e0014-4c4d-454d-96be-f03bac821358" class Muse: @@ -74,17 +86,18 @@ def __init__( self.preset = preset self.disable_light = disable_light + self.eeg_channel_count = 5 + self._athena_local_counter = -1 def connect(self): """Connect to the device""" - print(f"Connecting to {self.address}...") self.adapter = BleakBackend() self.adapter.start() self.device = self.adapter.connect(self.address) # Send a preset to the device to enable some functionalities - if self.preset not in ["none", "None"]: + if self.preset and self.preset not in ["none", "None"]: self.select_preset(self.preset) # subscribes to EEG stream @@ -116,7 +129,9 @@ def connect(self): def _write_cmd(self, cmd): """Wrapper to write a command to the Muse device. cmd -- list of bytes""" - self.device.char_write_handle(0x000E, cmd, False) + # Write to the control/stream toggle characteristic by UUID to be robust + # across devices where handles differ (e.g., Muse Athena). + self.device.char_write_uuid(ATTR_STREAM_TOGGLE, cmd, False) def _write_cmd_str(self, cmd): """Wrapper to encode and write a command string to the Muse device. @@ -176,6 +191,18 @@ def start(self): def resume(self): """Resume streaming, sending 'd' command""" self._write_cmd_str("d") + # Athena devices may require an alternate start command + try: + if hasattr(self, "device") and ( + getattr(self.device, "has_characteristic", None) + and ( + self.device.has_characteristic(ATTR_ATHENA_EEG_ALL) + or self.device.has_characteristic(ATTR_ATHENA_SENSORS_ALL) + ) + ): + self._write_cmd_str("dc001") + except Exception: + pass def stop(self): """Stop streaming.""" @@ -217,11 +244,60 @@ def disconnect(self): def _subscribe_eeg(self): """subscribe to eeg stream.""" - self.device.subscribe(ATTR_TP9, callback=self._handle_eeg) - self.device.subscribe(ATTR_AF7, callback=self._handle_eeg) - self.device.subscribe(ATTR_AF8, callback=self._handle_eeg) - self.device.subscribe(ATTR_TP10, callback=self._handle_eeg) - self.device.subscribe(ATTR_RIGHTAUX, callback=self._handle_eeg) + # Prefer Athena combined EEG characteristic if present + if hasattr( + self.device, "has_characteristic" + ) and self.device.has_characteristic(ATTR_ATHENA_EEG_ALL): + print("Detected Muse S Athena combined EEG characteristic.") + # Default to 5 channels; actual shape will be adjusted on first packet + self.eeg_channel_count = 5 + self._init_sample() + self.device.subscribe( + ATTR_ATHENA_EEG_ALL, callback=self._handle_eeg_combined + ) + return + + # Fallback to legacy per-channel characteristics + missing = [] + mapping = [ + (ATTR_TP9, 0), + (ATTR_AF7, 1), + (ATTR_AF8, 2), + (ATTR_TP10, 3), + (ATTR_RIGHTAUX, 4), + ] + found_any = False + found_map = [] + for uuid, idx in mapping: + if hasattr( + self.device, "has_characteristic" + ) and not self.device.has_characteristic(uuid): + missing.append(uuid) + else: + found_any = True + found_map.append((uuid, idx)) + if found_any and len(found_map) == 5: + # Size buffers to 5 channels for legacy devices + self.eeg_channel_count = 5 + self._init_sample() + for uuid, idx in found_map: + self.device.subscribe( + uuid, callback=lambda h, d, i=idx: self._handle_eeg_idx(i, h, d) + ) + elif found_any and len(found_map) != 5: + print( + f"Found only {len(found_map)} EEG legacy characteristics; expected 5. Skipping subscription." + ) + if not found_any: + print("No legacy EEG characteristics found, and Athena EEG not detected.") + print( + "Run 'MuseLSL2 inspect --address ' and share output for support." + ) + elif missing: + print("Some EEG characteristics not found on this device:") + for u in missing: + print(f" - {u}") + print("Continuing with available channels.") def _unpack_eeg_channel(self, packet): """Decode data packet of one EEG channel. @@ -230,20 +306,69 @@ def _unpack_eeg_channel(self, packet): samples with a 12 bit resolution. """ aa = bitstring.Bits(bytes=packet) - pattern = "uint:16,uint:12,uint:12,uint:12,uint:12,uint:12,uint:12, \ - uint:12,uint:12,uint:12,uint:12,uint:12,uint:12" - + pattern = ( + "uint:16,uint:12,uint:12,uint:12,uint:12,uint:12,uint:12, " + "uint:12,uint:12,uint:12,uint:12,uint:12,uint:12" + ) res = aa.unpack(pattern) - packetIndex = res[0] + packet_index = cast(int, res[0]) data = res[1:] # 12 bits on a 2 mVpp range data = 0.48828125 * (np.array(data) - 2048) - return packetIndex, data + return packet_index, data def _init_sample(self): """initialize array to store the samples""" - self.timestamps = np.full(5, np.nan) - self.data = np.zeros((5, 12)) + self.timestamps = np.full(self.eeg_channel_count, np.nan) + self.data = np.zeros((self.eeg_channel_count, 12)) + + def _unpack_eeg_combined(self, packet): + """Attempt to decode Athena combined EEG packet. + + This is experimental. We try two strategies: + 1) Legacy-like: 16-bit timestamp + 5*12 int16 samples interleaved. + 2) If sizes do not match, fall back to legacy 12-bit unpack per-channel is not applicable. + + Returns (packet_index, data) where data is shape (5, 12) float array. + Raises ValueError if format is unknown. + """ + b = memoryview(packet) + tm = None + if len(b) >= 2: + tm = int.from_bytes(b[0:2], byteorder="little", signed=False) + payload = b[2:] + else: + payload = b + + # Heuristic 1: prefer tail blocks of expected sizes + for size, ch in ((120, 5), (96, 4)): + if len(payload) >= size: + arr = np.frombuffer(payload[-size:], dtype="= size: + limit = len(payload) - size + 1 + for start in range(0, limit, 2): # align to int16 + segment = payload[start : start + size] + arr = np.frombuffer(segment, dtype=" 0.2: + try: + data = arr.reshape(12, ch).T + return tm, data + except Exception: + continue + + raise ValueError( + "Unknown Athena EEG packet format (len=%d, payload=%d)" + % (len(packet), len(payload)) + ) def _init_ppg_sample(self): """Initialise array to store PPG samples @@ -285,7 +410,7 @@ def _update_timestamp_correction(self, t_source, t_receiver): self.reg_params[1] = R self._P = P - def _handle_eeg(self, handle, data): + def _handle_eeg_idx(self, index, handle, data): """Callback for receiving a sample. samples are received in this order : 44, 41, 38, 32, 35 @@ -296,7 +421,6 @@ def _handle_eeg(self, handle, data): self.first_sample = False timestamp = mne_lsl.lsl.local_clock() - index = int((handle - 32) / 3) tm, d = self._unpack_eeg_channel(data) if self.last_tm == 0: @@ -304,8 +428,8 @@ def _handle_eeg(self, handle, data): self.data[index] = d self.timestamps[index] = timestamp - # last data received - if handle == 35: + # When we've received all channel packets, push a frame + if not np.isnan(self.timestamps).any(): if tm != self.last_tm + 1: if (tm - self.last_tm) != -65535: # counter reset print("missing sample %d : %d" % (tm, self.last_tm)) @@ -318,17 +442,15 @@ def _handle_eeg(self, handle, data): idxs = np.arange(0, 12) + self.sample_index self.sample_index += 12 - # update timestamp correction - # We received the first packet as soon as the last timestamp got - # sampled + # update timestamp correction based on earliest packet timestamp self._update_timestamp_correction(idxs[-1], np.nanmin(self.timestamps)) - # timestamps are extrapolated backwards based on sampling rate - # and current time + # timestamps are extrapolated based on sampling rate and start time timestamps = self.reg_params[1] * idxs + self.reg_params[0] # push data - self.callback_eeg(self.data, timestamps) + if self.callback_eeg: + self.callback_eeg(self.data, timestamps) # save last timestamp for disconnection timer self.last_timestamp = timestamps[-1] @@ -336,6 +458,47 @@ def _handle_eeg(self, handle, data): # reset sample self._init_sample() + def _handle_eeg_combined(self, handle, packet): + """Handle combined EEG notification for Muse S Athena. + + Experimental: attempts to decode to 5x12 samples and reuse legacy pipeline. + """ + if self.first_sample: + self._init_timestamp_correction() + self.first_sample = False + + timestamp = mne_lsl.lsl.local_clock() + try: + tm, d = self._unpack_eeg_combined(packet) + except Exception as e: + # Log once per minute to avoid spamming + print(f"Athena EEG packet decode failed: {e}") + self.last_timestamp = timestamp + return + + # Adjust buffer shape if needed + if d.shape[0] != self.data.shape[0]: + self.eeg_channel_count = d.shape[0] + print(f"Athena EEG channels detected: {self.eeg_channel_count}") + self._init_sample() + # Fill current data buffer + self.data[:, :] = d + self.timestamps[:] = timestamp + + # Use continuous sample index without counter jumps for stability + idxs = np.arange(0, 12) + self.sample_index + self.sample_index += 12 + + # Anchor dejittering to the current receive timestamp + self._update_timestamp_correction(idxs[-1], timestamp) + timestamps = self.reg_params[1] * idxs + self.reg_params[0] + + if self.callback_eeg: + self.callback_eeg(self.data, timestamps) + + # Keep watchdog alive based on receive time + self.last_timestamp = timestamp + def _init_control(self): """Variable to store the current incoming message.""" self._current_msg = "" @@ -363,78 +526,80 @@ def _handle_control(self, handle, packet): each line is a message, the 4 messages are a json object. """ - if handle != 14: + # No handle guard; subscription is bound by UUID + data_bytes = bytes(packet) + if not data_bytes: return - - # Decode data - bit_decoder = bitstring.Bits(bytes=packet) - pattern = "uint:8,uint:8,uint:8,uint:8,uint:8,uint:8,uint:8,uint:8,uint:8,uint:8, \ - uint:8,uint:8,uint:8,uint:8,uint:8,uint:8,uint:8,uint:8,uint:8,uint:8" - - chars = bit_decoder.unpack(pattern) - - # Length of the string - n_incoming = chars[0] - - # Parse as chars, only useful bytes - incoming_message = "".join(map(chr, chars[1:]))[:n_incoming] + n_incoming = data_bytes[0] + incoming_message = data_bytes[1 : 1 + n_incoming].decode( + "ascii", errors="ignore" + ) # Add to current message self._current_msg += incoming_message if incoming_message[-1] == "}": # Message ended completely - self.callback_control(self._current_msg) + if self.callback_control: + self.callback_control(self._current_msg) self._init_control() def _subscribe_telemetry(self): + if hasattr( + self.device, "has_characteristic" + ) and not self.device.has_characteristic(ATTR_TELEMETRY): + print(f"TELEMETRY characteristic not found: {ATTR_TELEMETRY}") + print("Run 'MuseLSL2 inspect --address ' to discover proper UUID.") + return self.device.subscribe(ATTR_TELEMETRY, callback=self._handle_telemetry) def _handle_telemetry(self, handle, packet): """Handle the telemetry (battery, temperature and stuff) incoming data""" - - if handle != 26: # handle 0x1a - return timestamp = mne_lsl.lsl.local_clock() bit_decoder = bitstring.Bits(bytes=packet) pattern = "uint:16,uint:16,uint:16,uint:16,uint:16" # The rest is 0 padding data = bit_decoder.unpack(pattern) + battery = float(cast(int, data[1])) / 512.0 + fuel_gauge = float(cast(int, data[2])) * 2.2 + adc_volt = cast(int, data[3]) + temperature = cast(int, data[4]) - battery = data[1] / 512 - fuel_gauge = data[2] * 2.2 - adc_volt = data[3] - temperature = data[4] - - self.callback_telemetry(timestamp, battery, fuel_gauge, adc_volt, temperature) + if self.callback_telemetry: + self.callback_telemetry( + timestamp, battery, fuel_gauge, adc_volt, temperature + ) - def _unpack_imu_channel(self, packet, scale=1): + def _unpack_imu_channel(self, packet, scale: float = 1.0): """Decode data packet of the accelerometer and gyro (imu) channels. Each packet is encoded with a 16bit timestamp followed by 9 samples with a 16 bit resolution. """ bit_decoder = bitstring.Bits(bytes=packet) - pattern = "uint:16,int:16,int:16,int:16,int:16, \ - int:16,int:16,int:16,int:16,int:16" - + pattern = ( + "uint:16,int:16,int:16,int:16,int:16, " "int:16,int:16,int:16,int:16,int:16" + ) data = bit_decoder.unpack(pattern) - - packet_index = data[0] - - samples = np.array(data[1:]).reshape((3, 3), order="F") * scale - + packet_index = cast(int, data[0]) + samples = np.array(data[1:], dtype=float).reshape((3, 3), order="F") * float( + scale + ) return packet_index, samples def _subscribe_acc(self): + if hasattr( + self.device, "has_characteristic" + ) and not self.device.has_characteristic(ATTR_ACCELEROMETER): + print(f"ACC characteristic not found: {ATTR_ACCELEROMETER}") + print("Run 'MuseLSL2 inspect --address ' to discover proper UUID.") + return self.device.subscribe(ATTR_ACCELEROMETER, callback=self._handle_acc) def _handle_acc(self, handle, packet): """Handle incoming accelerometer data. sampling rate: ~17 x second (3 samples in each message, roughly 50Hz)""" - if handle != 23: # handle 0x17 - return timestamps = [mne_lsl.lsl.local_clock()] * 3 # save last timestamp for disconnection timer @@ -443,18 +608,22 @@ def _handle_acc(self, handle, packet): # MUSE_ACCELEROMETER_SCALE_FACTOR (no idea where this comes from) packet_index, samples = self._unpack_imu_channel(packet, scale=0.0000610352) - self.callback_acc(samples, timestamps) + if self.callback_acc: + self.callback_acc(samples, timestamps) def _subscribe_gyro(self): + if hasattr( + self.device, "has_characteristic" + ) and not self.device.has_characteristic(ATTR_GYRO): + print(f"GYRO characteristic not found: {ATTR_GYRO}") + print("Run 'MuseLSL2 inspect --address ' to discover proper UUID.") + return self.device.subscribe(ATTR_GYRO, callback=self._handle_gyro) def _handle_gyro(self, handle, packet): """Handle incoming gyroscope data. sampling rate: ~17 x second (3 samples in each message, roughly 50Hz)""" - if handle != 20: # handle 0x14 - return - timestamps = [mne_lsl.lsl.local_clock()] * 3 # save last timestamp for disconnection timer @@ -463,22 +632,37 @@ def _handle_gyro(self, handle, packet): # MUSE_GYRO_SCALE_FACTOR (no idea where this number comes from) packet_index, samples = self._unpack_imu_channel(packet, scale=0.0074768) - self.callback_gyro(samples, timestamps) + if self.callback_gyro: + self.callback_gyro(samples, timestamps) def _subscribe_ppg(self): """subscribe to ppg stream.""" - self.device.subscribe(ATTR_PPG1, callback=self._handle_ppg) - self.device.subscribe(ATTR_PPG2, callback=self._handle_ppg) - self.device.subscribe(ATTR_PPG3, callback=self._handle_ppg) - - def _handle_ppg(self, handle, data): + mapping = [(ATTR_PPG1, 0), (ATTR_PPG2, 1), (ATTR_PPG3, 2)] + missing = [] + for uuid, idx in mapping: + if hasattr( + self.device, "has_characteristic" + ) and not self.device.has_characteristic(uuid): + missing.append(uuid) + else: + self.device.subscribe( + uuid, callback=lambda h, d, i=idx: self._handle_ppg_idx(i, h, d) + ) + if missing: + print("PPG characteristics not found on this device:") + for u in missing: + print(f" - {u}") + print( + "Run 'MuseLSL2 inspect --address ' to list available characteristics." + ) + + def _handle_ppg_idx(self, index, handle, data): """Callback for receiving a sample. samples are received in this order : 56, 59, 62 wait until we get x and call the data callback """ timestamp = mne_lsl.lsl.local_clock() - index = int((handle - 56) / 3) tm, d = self._unpack_ppg_channel(data) if self.last_tm_ppg == 0: @@ -486,8 +670,8 @@ def _handle_ppg(self, handle, data): self.data_ppg[index] = d self.timestamps_ppg[index] = timestamp - # last data received - if handle == 62: + # When we've received all 3 channel packets, push a frame + if not np.isnan(self.timestamps_ppg).any(): if tm != self.last_tm_ppg + 1: print("missing sample %d : %d" % (tm, self.last_tm_ppg)) self.last_tm_ppg = tm @@ -497,7 +681,9 @@ def _handle_ppg(self, handle, data): self.sample_index_ppg += 6 # timestamps are extrapolated backwards based on sampling rate and current time - timestamps = self.reg_ppg_sample_rate[1] * idxs + self.reg_ppg_sample_rate[0] + timestamps = ( + self.reg_ppg_sample_rate[1] * idxs + self.reg_ppg_sample_rate[0] + ) # save last timestamp for disconnection timer self.last_timestamp = timestamps[-1] @@ -514,14 +700,12 @@ def _unpack_ppg_channel(self, packet): Each packet is encoded with a 16bit timestamp followed by 3 samples with an x bit resolution. """ - aa = bitstring.Bits(bytes=packet) pattern = "uint:16,uint:24,uint:24,uint:24,uint:24,uint:24,uint:24" res = aa.unpack(pattern) - packetIndex = res[0] - data = res[1:] - - return packetIndex, data + packet_index = cast(int, res[0]) + data = np.array(res[1:], dtype=float) + return packet_index, data def _disable_light(self): self._write_cmd_str("L0") diff --git a/MuseLSL2/stream.py b/MuseLSL2/stream.py index 651a3a4..ffb0bd7 100644 --- a/MuseLSL2/stream.py +++ b/MuseLSL2/stream.py @@ -1,13 +1,61 @@ from functools import partial - -import mne_lsl.lsl +import numpy as np from . import backends -from .muse import Muse -# Begins LSL stream(s) from a Muse with a given address with data sources determined by arguments -def stream(address, ppg=True, acc=True, gyro=True, preset=None): +def _configure_lsl_api_cfg(): + """Configure liblsl via a temporary config file when not provided. + + Disables IPv6 multicast (removes yellow warnings) and lowers log level to -1 + to silence info/warn messages, without requiring a repo-level config file. + + See https://github.com/hbldh/bleak/discussions/1423 + """ + import os, tempfile, atexit + + if "LSLAPICFG" in os.environ: + return + cfg_fd, cfg_path = tempfile.mkstemp(prefix="lsl_api_", suffix=".cfg") + try: + with os.fdopen(cfg_fd, "w") as f: + f.write( + """ +[ports] +IPv6 = disable + +[log] +level = -1 +""".lstrip() + ) + except Exception: + # If writing fails, close and remove the file and continue without config + try: + os.close(cfg_fd) + except Exception: + pass + try: + os.remove(cfg_path) + except Exception: + pass + return + os.environ["LSLAPICFG"] = cfg_path + + def _cleanup_cfg(): + try: + os.remove(cfg_path) + except Exception: + pass + + atexit.register(_cleanup_cfg) + + +def stream(address, ppg=True, acc=True, gyro=True, preset=None, quiet=True): + if quiet: + _configure_lsl_api_cfg() # Silence LSL warnings + import mne_lsl.lsl + from .muse import Muse + # Find device if not address: from .find import find_devices @@ -16,22 +64,34 @@ def stream(address, ppg=True, acc=True, gyro=True, preset=None): address = device["address"] # EEG ==================================================== + # Determine EEG channel count based on preset hints (p21 => 4 EEG only) + eeg_channels = 5 + if preset is not None: + ps = str(preset).lower() + if ps.startswith("p"): + ps = ps[1:] + if ps in ("21",): + eeg_channels = 4 eeg_info = mne_lsl.lsl.StreamInfo( "Muse", stype="EEG", - n_channels=5, + n_channels=eeg_channels, sfreq=256, dtype="float32", source_id=f"Muse_{address}", ) eeg_info.desc.append_child_value("manufacturer", "Muse") - eeg_info.set_channel_names(["TP9", "AF7", "AF8", "TP10", "AUX"]) - eeg_info.set_channel_types(["eeg"] * 5) + if eeg_channels == 4: + eeg_info.set_channel_names(["TP9", "AF7", "AF8", "TP10"]) # No AUX + else: + eeg_info.set_channel_names(["TP9", "AF7", "AF8", "TP10", "AUX"]) + eeg_info.set_channel_types(["eeg"] * eeg_channels) eeg_info.set_channel_units("microvolts") eeg_outlet = mne_lsl.lsl.StreamOutlet(eeg_info, chunk_size=6) # PPG ==================================================== + ppg_outlet = None if ppg is True: ppg_info = mne_lsl.lsl.StreamInfo( "Muse", @@ -47,9 +107,11 @@ def stream(address, ppg=True, acc=True, gyro=True, preset=None): ppg_info.set_channel_types(["ppg"] * 3) ppg_info.set_channel_units("mmHg") - ppg_outlet = mne_lsl.lsl.StreamOutlet(ppg_info, chunk_size=1) + if ppg_info is not None: + ppg_outlet = mne_lsl.lsl.StreamOutlet(ppg_info, chunk_size=1) # ACC ==================================================== + acc_outlet = None if acc: acc_info = mne_lsl.lsl.StreamInfo( "Muse", @@ -64,9 +126,11 @@ def stream(address, ppg=True, acc=True, gyro=True, preset=None): acc_info.set_channel_types(["accelerometer"] * 3) acc_info.set_channel_units("g") - acc_outlet = mne_lsl.lsl.StreamOutlet(acc_info, chunk_size=1) + if acc_info is not None: + acc_outlet = mne_lsl.lsl.StreamOutlet(acc_info, chunk_size=1) # GYRO ==================================================== + gyro_outlet = None if gyro: gyro_info = mne_lsl.lsl.StreamInfo( "Muse", @@ -81,12 +145,24 @@ def stream(address, ppg=True, acc=True, gyro=True, preset=None): gyro_info.set_channel_types(["gyroscope"] * 3) gyro_info.set_channel_units("dps") - gyro_outlet = mne_lsl.lsl.StreamOutlet(gyro_info, chunk_size=1) + if gyro_info is not None: + gyro_outlet = mne_lsl.lsl.StreamOutlet(gyro_info, chunk_size=1) def push(data, timestamps, outlet): - outlet.push_chunk(data.T, timestamps[-1]) + arr = np.asarray(data.T, dtype=np.float32) + outlet.push_chunk(arr, timestamps[-1]) + + def push_eeg(data, timestamps): + # Ensure channel dimension matches declared outlet + ch = data.shape[0] + if ch != eeg_channels: + if ch < eeg_channels: + pad = np.zeros((eeg_channels - ch, data.shape[1]), dtype=data.dtype) + data = np.vstack([data, pad]) + else: + data = data[:eeg_channels, :] + push(data, timestamps, outlet=eeg_outlet) - push_eeg = partial(push, outlet=eeg_outlet) push_ppg = partial(push, outlet=ppg_outlet) if ppg else None push_acc = partial(push, outlet=acc_outlet) if acc else None push_gyro = partial(push, outlet=gyro_outlet) if gyro else None @@ -110,14 +186,19 @@ def push(data, timestamps, outlet): acc_txt = ", ACC" if acc else "" gyro_txt = ", GYRO" if gyro else "" - print(f"Streaming... EEG{ppg_txt}{acc_txt}{gyro_txt}... (CTRL + C to interrupt)") + print( + f"Streaming... EEG{ppg_txt}{acc_txt}{gyro_txt}... (CTRL + C to interrupt)" + ) # Disconnect if no data is received for 60 seconds while mne_lsl.lsl.local_clock() - muse.last_timestamp < 60: try: backends.sleep(1) except KeyboardInterrupt: - muse.stop() + try: + muse.stop() + except Exception: + pass print("Stream interrupted. Stopping...") break diff --git a/MuseLSL2/view.py b/MuseLSL2/view.py index dbd6930..1e62b7a 100644 --- a/MuseLSL2/view.py +++ b/MuseLSL2/view.py @@ -93,7 +93,9 @@ def view(): class Canvas(app.Canvas): def __init__(self, eeg, ppg=None): - app.Canvas.__init__(self, title="MuseLSL2 - Use your wheel to zoom!", keys="interactive") + app.Canvas.__init__( + self, title="MuseLSL2 - Use your wheel to zoom!", keys="interactive" + ) # Get info from stream eeg_info = _view_info(eeg) @@ -108,6 +110,10 @@ def __init__(self, eeg, ppg=None): (103 / 255, 58 / 255, 183 / 255), # Dark Purple (0 / 255, 0 / 255, 0 / 255), # Black ] + # Trim to actual EEG channel count (4 or 5) + eeg_n = eeg_info["n_channels"] + if eeg_n < len(colors): + colors = colors[:eeg_n] # Colors for impedence self.colors_quality = plt.get_cmap("RdYlGn")(np.linspace(0, 1, 11))[::-1] @@ -141,7 +147,9 @@ def __init__(self, eeg, ppg=None): self.program = gloo.Program(VERT_SHADER, FRAG_SHADER) self.program["a_position"] = self.data.T.astype(np.float32).reshape(-1, 1) self.program["a_index"] = index - self.program["a_color"] = np.repeat(colors[::-1], eeg_info["n_samples"], axis=0).astype(np.float32) + self.program["a_color"] = np.repeat( + colors[::-1], eeg_info["n_samples"], axis=0 + ).astype(np.float32) self.program["u_scale"] = (1.0, 1.0) self.program["u_size"] = (n_rows, n_cols) self.program["u_n"] = eeg_info["n_samples"] @@ -158,9 +166,10 @@ def __init__(self, eeg, ppg=None): # Store self.eeg = eeg_info["inlet"] - self.ppg = False if ppg is None else ppg_info["inlet"] + self.ppg = ppg_info["inlet"] if ppg_info is not None else False self.n_samples = eeg_info["n_samples"] self.sfreq = eeg_info["sfreq"] + self.eeg_channels = eeg_info["n_channels"] # View self._timer = app.Timer("auto", connect=self.on_timer, start=True) @@ -200,7 +209,9 @@ def on_timer(self, event): # PPG ------------------------------------------------ if self.ppg: - samples = self.update_data(outlet=self.ppg, samples=samples, time=time, n_channels=3) + samples = self.update_data( + outlet=self.ppg, samples=samples, time=time, n_channels=3 + ) self.data = np.vstack([self.data, samples]) # Concat self.data = self.data[-self.n_samples :] # Keep only last window length @@ -208,14 +219,17 @@ def on_timer(self, event): # Rescaling plot_data = self.data.copy() - # Normalize EEG (last 5 channels) -------------------- - plot_data[:, -5:] = (plot_data[:, -5:] - plot_data[:, -5:].mean(axis=0)) / 500 + # Normalize EEG (last 4 or 5 channels depending on stream) -------------------- + eeg_ch = self.eeg_channels + plot_data[:, -eeg_ch:] = ( + plot_data[:, -eeg_ch:] - plot_data[:, -eeg_ch:].mean(axis=0) + ) / 500 # Compute Impedence - sd = np.std(plot_data[-int(self.sfreq) :, -5:], axis=0)[::-1] * 500 + sd = np.std(plot_data[-int(self.sfreq) :, -eeg_ch:], axis=0)[::-1] * 500 # Discretize the impedence into 11 levels for coloring - co = np.int32(np.tanh((sd - 30) / 15) * 5 + 5) - # Loop through the 5 last channels indices (EEG channels) - for i in range(5): + co = np.clip((np.tanh((sd - 30) / 15) * 5 + 5).astype(int), 0, 10) + # Loop through the EEG channels + for i in range(eeg_ch): self.display_quality[i].text = f"{sd[i]:.2f}" self.display_quality[i].color = self.colors_quality[co[i]] self.display_quality[i].font_size = 12 + co[i] @@ -225,9 +239,11 @@ def on_timer(self, event): # Normalize PPG (3 channels) -------------------- if self.ppg: - plot_data[:, 0:3] = (plot_data[:, 0:3] - plot_data[:, 0:3].mean(axis=0)) / np.nanstd( - plot_data[:, 0:3], axis=0 - ) + std = np.nanstd(plot_data[:, 0:3], axis=0) + if np.all(std > 0): + plot_data[:, 0:3] = ( + plot_data[:, 0:3] - plot_data[:, 0:3].mean(axis=0) + ) / std self.program["a_position"].set_data(plot_data.T.ravel().astype(np.float32)) self.update() diff --git a/README.md b/README.md index 457d1f8..9faf81e 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,14 @@ This is a light reimplementation of [muse-lsl](https://github.com/alexandrebarac By default, MuseLSL2 streams *all* channels (including gyroscope, accelerometer, and the signal form the Auxiliary port "AUX", which can be used to add [an additional electrode](https://github.com/andrewjsauer/Muse-EEG-Extra-Electrode-Tutorial)). Note that without an additional electrode, the AUX channel will just pick up noise and should be discarded. +## Muse S Athena (experimental) + +- Initial support is added for the new "Muse S Athena" devices that expose a combined EEG characteristic. +- If detected, the package subscribes to `273e0013-4c4d-454d-96be-f03bac821358` and attempts to decode 5x12 EEG frames similarly to legacy devices. This is based on public notes from BrainFlow PR #779 and may change. +- Use `MuseLSL2 inspect --address ` to list GATT services and confirm whether the device exposes the combined EEG characteristic. +- Start with preset `p21` or `p1045` to enable EEG only; broader sensor support is still under investigation. +If streaming fails, please share the output of the `inspect` command and any console logs. + ## Usage Install with: @@ -30,6 +38,12 @@ Once you have the mac address of your device, run for instance (but replace the ``` MuseLSL2 stream --address 00:55:DA:B5:E8:CF + +For Athena devices, you can also try forcing an EEG-only preset first: + +``` +MuseLSL2 stream --address 00:55:DA:B5:E8:CF --preset p21 +``` ``` In a new console, while streaming, run: