MindwaveMobile2

Data dataclass

Data(raw_data: list[int] = lambda: [0 for _ in range(512)](), attention: int = 0, meditation: int = 0, delta: int = 0, theta: int = 0, lowAlpha: int = 0, highAlpha: int = 0, lowBeta: int = 0, highBeta: int = 0, lowGamma: int = 0, highGamma: int = 0)

A data container for storing the parsed data from the stream.

Attributes:

Name	Type	Description
raw_data	list[int]	A list of 512 raw EEG readings.
attention	int	The attention level (0-100).
meditation	int	The meditation level (0-100).
delta	int	Power level in the delta frequency band.
theta	int	Power level in the theta frequency band.
lowAlpha	int	Power level in the low alpha frequency band.
highAlpha	int	Power level in the high alpha frequency band.
lowBeta	int	Power level in the low beta frequency band.
highBeta	int	Power level in the high beta frequency band.
lowGamma	int	Power level in the low gamma frequency band.
highGamma	int	Power level in the high gamma frequency band.

---

MindWaveMobile2

MindWaveMobile2(event_loop: AbstractEventLoop = None, tg_connector: ThinkGearConnector = None)

The main interface for the NeuroSky MindWaveMobile2 EEG headset.

This class provides an event-based interface for connecting to and receiving data from a MindWaveMobile2 headset using the ThinkGear connector service.

Parameters:

Name	Type	Description	Default
event_loop	AbstractEventLoop	The event loop to use for asynchronous operations.	None
tg_connector	ThinkGearConnector	The ThinkGearConnector instance to use.	None

Source code in mindwave/headset.py

def __init__(
    self,
    event_loop: asyncio.AbstractEventLoop = None,
    tg_connector: ThinkGearConnector = None,
) -> None:
    """Initialize a MindWaveMobile2 instance.

    Args:
        event_loop (asyncio.AbstractEventLoop, optional): The event loop to use for asynchronous operations.
        If not provided, it will check for a running event loop in the current thread.
        If no running event loop is found, a new event loop will be created in a new thread.
        tg_connector (ThinkGearConnector, optional): The ThinkGearConnector instance to use.
        If not provided, a new instance with default settings will be created.
    """
    super().__init__()
    self.is_running: bool = False
    self._event_loop: asyncio.AbstractEventLoop = event_loop

    self._logger: logging.Logger = Logger.get_logger(self.__class__.__name__)
    self._tg_connector = ThinkGearConnector() if tg_connector is None else tg_connector
    self._event_manager = EventManager()
    self._signal_quality: int = 0
    self._connection_status: ConnectionStatus = ConnectionStatus.DISCONNECTED
    self._stream_parser = StreamParser()
    self._parser_connector_supscription: Subscription = None
    self._lock = asyncio.Lock()
    self._read_loop_task: asyncio.Task = None

    self.on_connector_data(self._update_status)

signal_quality property writable

signal_quality: float

Float signal quality value normalized to the range of 0-100%.

The raw signal quality value is in the range of 0-200 (0 indicating a good signal and 200 indicating an off-head state).

If the raw signal quality value is needed, it can be accessed from the connector data through the on_connector_data events.

connection_status property writable

connection_status: ConnectionStatus

Current ConnectionStatus of the headset.

start

start(n_tries=3, timeout=15, blocking=True) -> bool

Start the connection to the MindWaveMobile2 device.

A connection to be considered successful, the Connection to the TGConnector is not an indication, The TGConnector must find and connect to the MindWaveMobile2 device and start streaming data. Which is indicated by the ConnectionStatus.CONNECTED.

Parameters:

Name	Type	Description	Default
n_tries	int	The number of connection attempts.	3
timeout	int	The timeout for each connection attempt in seconds.	15
blocking	bool	Whether to block until the connection is established.	True

Returns:

Name	Type	Description
bool	bool	True if the connection was successful, False otherwise.
	bool	If blocking is False, it returns True if the connection process was started successfully.

Source code in mindwave/headset.py

def start(self, n_tries=3, timeout=15, blocking=True) -> bool:
    """Start the connection to the MindWaveMobile2 device.

    A connection to be considered successful, the Connection to the TGConnector is not an indication,
    The TGConnector must find and connect to the MindWaveMobile2 device and start streaming data.
    Which is indicated by the ConnectionStatus.CONNECTED.

    Args:
        n_tries (int, optional): The number of connection attempts.
        timeout (int, optional): The timeout for each connection attempt in seconds.
        blocking (bool, optional): Whether to block until the connection is established.

    Returns:
        bool: True if the connection was successful, False otherwise.
        If blocking is False, it returns True if the connection process was started successfully.
    """
    self._setup_event_loop()
    future = asyncio.run_coroutine_threadsafe(self._start_async(n_tries=n_tries, timeout=timeout), self._event_loop)

    if blocking:
        return future.result()
    else:
        return True

stop

stop(blocking=True) -> bool

Stop the connection to the MindWaveMobile2 device.

Parameters:

Name	Type	Description	Default
blocking	bool	Whether to block until the disconnection is complete.	True

Returns:

Name	Type	Description
bool	bool	True if the disconnection was successful, False otherwise.
	bool	If blocking is False, it returns True if the disconnection process was started successfully.

Source code in mindwave/headset.py

def stop(self, blocking=True) -> bool:
    """Stop the connection to the MindWaveMobile2 device.

    Args:
        blocking (bool, optional): Whether to block until the disconnection is complete.

    Returns:
        bool: True if the disconnection was successful, False otherwise.
        If blocking is False, it returns True if the disconnection process was started successfully.
    """
    self._setup_event_loop()

    future = asyncio.run_coroutine_threadsafe(self._stop_async(), self._event_loop)
    if blocking:
        return future.result()
    else:
        return True

on_data

on_data(callback: Callable[[HeadsetDataEvent], Any]) -> Subscription

Rigister a callback for the processed headset data.

Parameters:

Name	Type	Description	Default
callback	function	The callback function to be called when data is received.	required

Returns:

Name	Type	Description
Subscription	Subscription	A Subscription object that can be used to detach the callback.

Source code in mindwave/headset.py

def on_data(self, callback: Callable[[HeadsetDataEvent], Any]) -> Subscription:
    """Rigister a callback for the processed headset data.

    Args:
        callback (function): The callback function to be called when data is received.

    Returns:
        Subscription: A Subscription object that can be used to detach the callback.
    """
    return self._stream_parser.on_data(callback)

on_blink

on_blink(callback: Callable[[BlinkEvent], Any]) -> Subscription

Register a callback for the blink detection event.

Parameters:

Name	Type	Description	Default
callback	function	The callback function to be called when a blink is detected.	required

Returns:

Name	Type	Description
Subscription	Subscription	the subscription object that can be used to detach the callback

Source code in mindwave/headset.py

def on_blink(self, callback: Callable[[BlinkEvent], Any]) -> Subscription:
    """Register a callback for the blink detection event.

    Args:
        callback (function): The callback function to be called when a blink is detected.

    Returns:
        Subscription: the subscription object that can be used to detach the callback
    """
    return self._event_manager.add_listener(EventType.Blink, callback)

on_status_change

on_status_change(callback: Callable[[HeadsetStatusEvent], Any]) -> Subscription

Register a callback for the headset connection status change event.

Parameters:

Name	Type	Description	Default
callback	function	The callback function to be called when the connection status changes.	required

Returns:

Name	Type	Description
Subscription	Subscription	the subscription object that can be used to detach the callback

Source code in mindwave/headset.py

def on_status_change(self, callback: Callable[[HeadsetStatusEvent], Any]) -> Subscription:
    """Register a callback for the headset connection status change event.

    Args:
        callback (function): The callback function to be called when the connection status changes.

    Returns:
        Subscription: the subscription object that can be used to detach the callback
    """
    return self._event_manager.add_listener(EventType.HeadsetStatus, callback)

on_signal_quality_change

on_signal_quality_change(callback: Callable[[SignalQualityEvent], Any]) -> Subscription

Register a callback for the signal quality change event.

Parameters:

Name	Type	Description	Default
callback	function	The callback function to be called when the signal quality changes.	required

Returns:

Name	Type	Description
Subscription	Subscription	the subscription object that can be used to detach the callback

Source code in mindwave/headset.py

def on_signal_quality_change(self, callback: Callable[[SignalQualityEvent], Any]) -> Subscription:
    """Register a callback for the signal quality change event.

    Args:
        callback (function): The callback function to be called when the signal quality changes.

    Returns:
        Subscription: the subscription object that can be used to detach the callback
    """
    return self._event_manager.add_listener(EventType.SignalQuality, callback)

on_connector_data

on_connector_data(callback: Callable[[ConnectorDataEvent], Any]) -> Subscription

Register a callback for the ThinkGear Connector data event.

Parameters:

Name	Type	Description	Default
callback	function	The callback function to be called when data is received from the ThinkGear Connector.	required

Returns:

Name	Type	Description
Subscription	Subscription	the subscription object that can be used to detach the callback

Source code in mindwave/headset.py

def on_connector_data(self, callback: Callable[[ConnectorDataEvent], Any]) -> Subscription:
    """Register a callback for the ThinkGear Connector data event.

    Args:
        callback (function): The callback function to be called when data is received from the ThinkGear Connector.

    Returns:
        Subscription: the subscription object that can be used to detach the callback
    """
    return self._event_manager.add_listener(EventType.ConnectorData, callback)

on_timeout

on_timeout(callback: Callable[[TimeoutEvent], Any]) -> Subscription

Register a callback for the timeout event.

Parameters:

Name	Type	Description	Default
callback	function	The callback function to be called when a timeout occurs.	required

Returns:

Name	Type	Description
Subscription	Subscription	the subscription object that can be used to detach the callback

Source code in mindwave/headset.py

def on_timeout(self, callback: Callable[[TimeoutEvent], Any]) -> Subscription:
    """Register a callback for the timeout event.

    Args:
        callback (function): The callback function to be called when a timeout occurs.

    Returns:
        Subscription: the subscription object that can be used to detach the callback
    """
    return self._event_manager.add_listener(EventType.Timeout, callback)