API Reference

This section provides detailed documentation for all classes and functions in the MW75 EEG Streamer package.

Main Module

MW75 EEG Streamer - Main Entry Point

Clean main function and CLI interface for the MW75 EEG streamer.

class mw75_streamer.main.MW75Streamer(csv_file=None, extra_file=None, websocket_url=None, lsl_stream_name=None, panel_server=None, verbose=False, eeg_callback=None, raw_data_callback=None, other_event_callback=None)[source]

Bases: object

Main MW75 EEG streamer application

device: Any

set_verbose(verbose)[source]

Enable or disable verbose logging including checksum error messages

Parameters:: verbose (bool) – True to enable verbose logging, False to suppress it
Return type:: None

async start_streaming()[source]

Start the MW75 streaming process

Return type:: bool
Returns:: True if streaming completed successfully, False otherwise

mw75_streamer.main.parse_arguments()[source]

Parse command line arguments

Return type:: Namespace

mw75_streamer.main.show_output_configuration(args, logger)[source]

Display output configuration to user

Return type:: None

async mw75_streamer.main.main()[source]

Main entry point for the MW75 EEG streamer

Return type:: None

mw75_streamer.main.cli_main()[source]

Synchronous entry point for CLI console scripts

Return type:: None

Device Management

MW75 Device Controller

BLE Manager

BLE Manager for MW75 EEG Streamer

Handles Bluetooth Low Energy device discovery, connection, and activation sequence for the MW75 Neuro headphones.

class mw75_streamer.device.ble_manager.BLEManager[source]

Bases: object

Manages BLE connection and MW75 activation sequence

async discover_and_activate()[source]

Discover MW75 device and execute activation sequence

Return type:: Optional[str]
Returns:: Device name if successful, None if failed

async disconnect_after_activation()[source]

Disconnect BLE connection after activation is complete.

This is required on macOS Taho (26+) where keeping the BLE connection open blocks RFCOMM delegate callbacks from being delivered.

Return type:: None

async cleanup()[source]

Send disable commands and disconnect from BLE

Return type:: None

RFCOMM Manager

Data Processing

Packet Processor

Packet Processor for MW75 EEG Data

Handles parsing, validation, and processing of MW75 EEG data packets.

class mw75_streamer.data.packet_processor.EEGPacket(timestamp, event_id, counter, ref, drl, channels, feature_status, checksum_valid)[source]

Bases: object

Represents a parsed EEG packet from MW75 device

timestamp: float

event_id: int

counter: int

ref: float

drl: float

channels: List[float]

feature_status: int

checksum_valid: bool

class mw75_streamer.data.packet_processor.ChecksumStats(valid_packets=0, invalid_packets=0, total_packets=0)[source]

Bases: object

Tracks packet validation statistics

valid_packets: int = 0

invalid_packets: int = 0

total_packets: int = 0

property error_rate: float: Calculate error rate as percentage

class mw75_streamer.data.packet_processor.PacketProcessor(verbose=False)[source]

Bases: object

Handles MW75 packet parsing and validation

validate_checksum(packet)[source]

Validate MW75 packet checksum

MW75 checksum calculation: - Sum of first 61 bytes (index 0-60) - Masked to 16 bits (& 0xFFFF) - Stored in bytes 61-62 as little endian

Parameters:: packet (bytes) – 63-byte MW75 packet
Return type:: Tuple[bool, int, int]
Returns:: Tuple of (is_valid, calculated_checksum, received_checksum)

parse_eeg_packet(packet)[source]

Parse a 63-byte EEG packet into structured data

Parameters:: packet (bytes) – 63-byte packet from MW75 device
Return type:: Optional[EEGPacket]
Returns:: EEGPacket if successfully parsed and valid, None otherwise

process_data_buffer(data, eeg_callback, other_event_callback=None)[source]

Process data buffer and extract packets using continuous buffering

This method accumulates data across RFCOMM chunks to handle packet framing correctly, since RFCOMM delivers arbitrary-sized chunks (e.g., 64 bytes) while EEG packets are exactly 63 bytes.

Parameters:

data (bytes) – Raw data buffer from device
eeg_callback (Callable[[EEGPacket], None]) – Function to call for EEG packets
other_event_callback (Optional[Callable[[bytes], None]]) – Function to call for non-EEG packets

Return type:

None

get_final_stats()[source]

Get final packet processing statistics

Return type:: ChecksumStats
Returns:: ChecksumStats with current statistics

Streamers

Data Streamers for MW75 EEG Data

Handles output of EEG data to various destinations: CSV files, WebSocket, and stdout.

class mw75_streamer.data.streamers.CSVWriter(eeg_path=None, extra_path=None)[source]

Bases: object

Handles CSV file output for EEG data and other events

write_eeg_packet(packet)[source]

Write EEG packet to CSV file or stdout

Parameters:: packet (EEGPacket) – EEG packet to write
Return type:: bool
Returns:: True if write successful, False otherwise

write_other_event(packet)[source]

Write non-EEG event to extra CSV file

Parameters:: packet (bytes) – Raw packet bytes
Return type:: bool
Returns:: True if write successful, False otherwise

close()[source]

Close CSV files safely

Return type:: None

class mw75_streamer.data.streamers.WebSocketStreamer(url=None)[source]

Bases: object

Handles real-time WebSocket streaming of EEG data

send_eeg_data(packet)[source]

Send EEG packet as JSON to WebSocket

Parameters:: packet (EEGPacket) – EEG packet to send
Return type:: bool
Returns:: True if send successful, False otherwise

close()[source]

Close WebSocket connection

Return type:: None

class mw75_streamer.data.streamers.StdoutStreamer(print_header=True)[source]

Bases: object

Handles EEG data output to stdout (console)

write_eeg_packet(packet)[source]

Write EEG packet to stdout

Parameters:: packet (EEGPacket) – EEG packet to write
Return type:: bool
Returns:: True if write successful, False otherwise

class mw75_streamer.data.streamers.LSLStreamer(stream_name='MW75_EEG', stream_type='EEG')[source]

Bases: object

Handles real-time LSL (Lab Streaming Layer) streaming of EEG data

send_eeg_data(packet)[source]

Send EEG packet to LSL stream

Parameters:: packet (EEGPacket) – EEG packet to send
Return type:: bool
Returns:: True if send successful, False otherwise

close()[source]

Close LSL stream outlet

Return type:: None

get_stream_info()[source]

Get information about the current LSL stream

Return type:: dict
Returns:: Dictionary with stream information

Configuration

MW75 EEG Streamer Configuration

Contains all constants, UUIDs, and configuration settings for the MW75 EEG streamer.

WebSocket Server (Remote Control)

WebSocket Server for Remote MW75 Control

Provides a WebSocket server that allows third-party applications to: - Connect/disconnect to MW75 device remotely - Receive real-time EEG data - Get status updates and logs - Configure auto-reconnect behavior

class mw75_streamer.server.ws_server.DeviceState(value)[source]

Bases: Enum

MW75 device connection states

IDLE = 'idle'

CONNECTING = 'connecting'

CONNECTED = 'connected'

DISCONNECTING = 'disconnecting'

DISCONNECTED = 'disconnected'

ERROR = 'error'

class mw75_streamer.server.ws_server.WebSocketLogHandler(server)[source]

Bases: Handler

Custom logging handler that sends logs to WebSocket client

emit(record)[source]

Send log record to WebSocket client

Return type:: None

class mw75_streamer.server.ws_server.MW75WebSocketServer(host='localhost', port=8080)[source]

Bases: object

WebSocket server for remote MW75 device control

async start()[source]

Start the WebSocket server

Return type:: None

Testing Utilities

WebSocket Test Server

WebSocket Test Servers for MW75 EEG Streamer

Provides WebSocket test servers for validating EEG streaming functionality.

class mw75_streamer.testing.websocket_server.SimpleWebSocketServer(host='localhost', port=8080, verbose=False)[source]

Bases: object

Minimal WebSocket server for quick testing

async handle_connection(websocket)[source]

Handle WebSocket connections

Return type:: None

async start()[source]

Start the WebSocket server

Return type:: None

class mw75_streamer.testing.websocket_server.WebSocketTestServer(host='localhost', port=8080, verbose=False)[source]

Bases: object

Full-featured WebSocket test server with advanced statistics

async handle_client(websocket)[source]

Handle incoming WebSocket connections

Return type:: None

async start()[source]

Start the advanced WebSocket server

Return type:: None

mw75_streamer.testing.websocket_server.main()[source]

CLI entry point for WebSocket test servers

Return type:: None

Test Guide

MW75 Testing Guide and Utilities

Provides testing guidance and utility functions for MW75 EEG streaming validation.

class mw75_streamer.testing.test_guide.TestGuide[source]

Bases: object

Interactive testing guide for MW75 EEG streaming

show_quick_start()[source]

Display quick start testing instructions

Return type:: None

open_browser_client(port=8080)[source]

Open the browser test client

Parameters:: port (int) – WebSocket server port for the browser client URL
Return type:: bool
Returns:: True if browser was opened successfully, False otherwise

validate_setup()[source]

Validate the testing setup

Return type:: dict
Returns:: Dictionary with validation results

show_test_checklist()[source]

Display testing checklist for validation

Return type:: None

mw75_streamer.testing.test_guide.show_quick_start()[source]

Convenience function to show quick start guide

Return type:: None

mw75_streamer.testing.test_guide.open_browser_test()[source]

Convenience function to open browser test client

Return type:: bool

mw75_streamer.testing.test_guide.validate_test_setup()[source]

Convenience function to validate testing setup

Return type:: dict

Panel Server

Panel WebSocket relay for MW75 streamer

Broadcasts EEG data, runtime statistics, and log messages to connected browser clients (e.g., panel.html). Sends data only when clients are connected and releases resources on disconnect.

class mw75_streamer.panel.panel_server.PanelServer(host='localhost', port=8090)[source]

Bases: object

Minimal WebSocket relay for browser panel.

async start()[source]

Start the WebSocket server in the current event loop.

Return type:: None

async stop()[source]

Stop the WebSocket server and disconnect clients.

Return type:: None

start_background()[source]

Start the server on a dedicated event loop thread.

Return type:: None

stop_background()[source]

Stop the background server thread cleanly.

Return type:: None

publish_eeg(packet)[source]

Schedule sending EEG packet to clients.

Return type:: None

publish_stats(stats)[source]

Schedule sending aggregated stats to clients.

Return type:: None

publish_log(record)[source]

Schedule sending a log record to clients.

Return type:: None

class mw75_streamer.panel.panel_server.WebSocketLogHandler(panel)[source]

Bases: Handler

Logging handler that forwards log records to the PanelServer.

emit(record)[source]

Do whatever it takes to actually log the specified logging record.

This version is intended to be implemented by subclasses and so raises a NotImplementedError.

Return type:: None

Utilities

Logging

Logging utilities for MW75 EEG Streamer

Provides consistent logging setup and configuration across the application.

mw75_streamer.utils.logging.setup_logging(verbose=False, logger_name=None)[source]

Setup logging configuration for the MW75 streamer

Parameters:

verbose (bool) – Enable debug level logging if True, otherwise info level
logger_name (Optional[str]) – Name for the logger, defaults to the package name

Return type:

Logger

Returns:

Configured logger instance

mw75_streamer.utils.logging.get_logger(name)[source]

Get a logger for a specific module

Parameters:: name (str) – Module name (usually __name__)
Return type:: Logger
Returns:: Logger instance with consistent configuration

Data Structures

The core data structures are documented within their respective modules above. The main data structure is the EEGPacket class in the packet processor module.