Python: Streaming Two Devices

1. Introduction

This document describes how to manage the connection and streaming of two Muse devices simultaneously. Streamed data and device ID of your Muse devices will be printed in real-time using the Bluetooth Low Energy (BLE) communication. All the examples work with the API and Python packages, available at the following links:

- Muse API: https://gitlab.com/221e-Repositories/muse/python/muse_api
- Bleak package (version 0.21.1): https://bleak.readthedocs.io/en/latest/
- Struct module: https://docs.python.org/3/library/struct.html
- Numpy package (version 1.24.1): https://pypi.org/project/numpy/

2. Collect Data from your Muse devices

First, check that the BLE devices are powered on and support connections by finding them using BleakScanner.discover function (see description in the tutorial Connection And Control). Remember to change the BLE device names according to your needs.

# Customizable variables
my_device_name_1 = 'muse_v3_01'
my_device_name_2 = 'muse_v3_02'
stream_duration_seconds = 30   #set streaming time duration in seconds
stream_start_timeout = 3   #set streaming start timeout counter in seconds

If both devices are discovered, then they will be saved in temporary variables before connection.

# DEVICE ENUMERATION
devices = await BleakScanner.discover()
myDev_1 = None
myDev_2 = None
for d in devices:
    print(d)
    if d.name == my_device_name_1:
        myDev_1 = d
    if d.name == my_device_name_2:
        myDev_2 = d

In order to connect to two devices, the use of a lock (asyncio.Lock()) and a stack (contextlib.AsyncExitStack()) is necessary, to avoid unwanted errors. See also the bleak example for two devices (https://github.com/hbldh/bleak/blob/develop/examples/two_devices.py).

client_1 = BleakClient(myDev_1)
await stack.enter_async_context(client_1)

Get the device IDs and the sensors’ fullscales.

# Get Muse n.1 device ID
await client_1.write_gatt_char(CMD_UUID,Muse_Utils.Cmd_GetDeviceID(),True)
response = await client_1.read_gatt_char(CMD_UUID)
device_1_ID = Muse_Utils.Dec_DeviceID(CommandResponse(response))

await client_1.write_gatt_char(CMD_UUID,Muse_Utils.Cmd_GetSensorsFullScale(),True)
response = await client_1.read_gatt_char(CMD_UUID)
gyrConfig1, axlConfig1, magConfig1, hdrConfig1 = Muse_Utils.Dec_SensorFullScales(CommandResponse(response))

Subscribe to data characteristic using the corresponding UUID.

await client_1.start_notify(DATA_UUID, partial(data_notification_handler, client_1))
await client_2.start_notify(DATA_UUID, partial(data_notification_handler, client_2))

Start a streaming acquisition of IMU data, as described in the Muse communication protocol. In this example, the sampling frequency is set to 25 Hz. The acquisition length is set to 30 seconds, but can be easily customized, and then automatically stopped.

# Set up the command
stream_mode = MH.DataMode.DATA_MODE_IMU
cmd_stream = Muse_Utils.Cmd_StartStream(stream_mode, MH.DataFrequency.DATA_FREQ_25Hz.value, True)

# Start Streaming using the above configuration (direct streaming, IMU mode and Sampling Frequency = 25 Hz)
await client_1.write_gatt_char(CMD_UUID, cmd_stream, True)
await client_2.write_gatt_char(CMD_UUID, cmd_stream, True)

# Wait for 'stream_duration_seconds' seconds
await asyncio.sleep(stream_duration_seconds)

The data acquisition and packet decoding are managed through a dedicated function called data_notification_handler. See the Insights section for further information.

The source code of this example is available here: Python_03_Example03_StreamingTwoDevices.py

3. Insights

For what concern message decoding, we provide the data notification handler function which shows how to decode the streaming packets from both devices. The decoded data is printed to the console showing also the sender device ID.

async def data_notification_handler(client: BleakClient, sender: int, data: bytearray):
    """Decode data"""

    header_offset = 8   # ignore packet header

    if (client.address == client_1.address):
        # decode packet data
        tempData = Muse_Utils.DecodePacket(data[header_offset:], 0, stream_mode.value, gyrConfig1.Sensitivity, 
                                     axlConfig1.Sensitivity, magConfig1.Sensitivity, hdrConfig1.Sensitivity)
        print("{0} {1} {2} {3} {4} {5} {6}".format(device_1_ID,tempData.axl[0],tempData.axl[1],tempData.axl[2],
                                     tempData.gyr[0],tempData.gyr[1],tempData.gyr[2]))
    else:
        # decode packet data
        tempData = Muse_Utils.DecodePacket(data[header_offset:], 0, stream_mode.value, gyrConfig2.Sensitivity, 
                                     axlConfig2.Sensitivity, magConfig2.Sensitivity, hdrConfig2.Sensitivity)
        print("{0} {1} {2} {3} {4} {5} {6}".format(device_2_ID,tempData.axl[0],tempData.axl[1],tempData.axl[2],
                                     tempData.gyr[0],tempData.gyr[1],tempData.gyr[2]))
        
    return

The result should be something like those represented below.

Reference

A. Muse_v3_communication_protocol
B. Muse Python: Connection and Control
C. Muse Python: Raw Data Aquisition

Tutorials