Programming Reference:SignalSharing Python Demo

Demo example of the flexibility of visualizations with Python

Location

src/core/SignalProcessing/SignalSharingDemo/PythonClientApp

Synopsis

The SignalSharing Python Demo demonstrates how to make a complex real-time visualization in Python using data parallelly collected in BCI2000. It makes use of the SignalSharing feature in BCI2000.

Function

The SignalSharing Python Demo creates a basic visualization in Python using the data collected in BCI2000. Since this is done outside of the BCI2000 processing loop, rendering visualizations can take as long as needed. Using Python also allows for the use of the numerous packages that are available to create complex figures.

This Demo uses the SignalSharing feature of BCI2000. The Python demo is a client that visualizes the data coming from BCI2000. It is a simple Python script that provides a TCP connection with the port that is being used to synchronize the data transfer, and updates the visualization as data is being streamed.

Source vs Client

The BCI2000 data is sent according to the format specified in BCI2000 Messages Wiki page, as Descriptor=4: Visualization and Brain Signal Data Format, then Descriptor Supplement=1: Signal Data, then data type 2.

With this demo, BCI2000 and the Python script must be run on the same computer. The script expects the data stream to be sending the name of the shared memory, which will only happen if they are both on the same computer. It is possible to grab BCI2000 data from another computer, and is implemented in the C++ SignalSharing Demo and the SignalSharing Client Lib Demo . If on separate computers, the actual data points are being shared instead of the memory name. Implementing this would require a simple extension of this demo.

How to run

1. Build BCI2000 as you would, make sure to check BUILD_DEMOS
2. Make sure the SignalSharingDemo works first
3. Navigate to src\core\SignalProcessing\SignalSharingDemo\PythonClientApp, where there is a batch file and a Python file. Copy the batch file to your BCI2000 batch folder
4. Run the Python file. For example from the command line, navigate to the folder and run python SignalSharingPythonDemo.py
5. Run the batch file, and press "Start Run" (it already sets the configuration, it won't work it you press it multiple times)
   • If you change the Parameter ShareTransmissionFilter, make sure to also change it in the Python script
6. In the visualization window, pressing the Set Running 0 button will set the Running state to 0, stopping BCI2000. This is an example how to modify state values from the Python side.
7. In the visualization window, pressing the Insert NaNs button will insert a data block of NaNs into the BCI2000 processing chain. In the SpatialFilter signal visualization, you will be able to see the NaN block in form of an empty area in the display. This is an example of how to modify signal values from the Python side.

Python Script

Here is the same code that is on the SVN (r8388)

#! /usr/bin/env python3

import socket
from select import select
from multiprocessing import shared_memory
import matplotlib.pyplot as plt
from matplotlib.widgets import Button
import numpy as np
import io
import struct
import traceback
import platform
from enum import Enum

class Object(object):
    pass

def waitForRead(sock):
    """polling wait for data on the socket so we may react to a keyboard interrupt"""
    pollingIntervalSeconds = 0.1
    ready, _, _ = select([sock], [], [], pollingIntervalSeconds)
    while not ready:
        try:
            ready, _, _ = select([sock], [], [], pollingIntervalSeconds)
        except KeyboardInterrupt:
            print('Keyboard interrupt, exiting')
            quit()
    

def readLine(stream, terminator = b'\n'):
    """read a line from a stream up to terminator character"""
    chars = []
    c = stream.read(1)
    while c != terminator and c != b'':
        chars.append(c)
        c = stream.read(1)
    return str(b''.join(chars), 'utf-8')

class BciDescSupp(Enum):
    """BCI2000 descriptor and supplement for relevant messages"""
    Parameter = b'\x02\x00'
    State = b'\x03\x00'
    SignalData = b'\x04\x01'
    SignalProperties = b'\x04\x03'
    SysCommand = b'\x06\x00'

def readBciLengthField(stream, fieldSize):
    """read a length field of specified size from a stream"""
    # read fieldSize bytes that make up a little-endian number
    b = stream.read(fieldSize)
    if b == b'':
        raise EOFError()
    if len(b) != fieldSize:
        raise RuntimeError('Could not read size field')
    n = int.from_bytes(b, 'little')
    # if all bytes are 0xff, ignore them and read the field value as a string
    if n == (1 << (fieldSize * 8)) - 1:
        n = int(readLine(stream, b'\x00'))
    return n

def writeBciLengthField(stream, fieldSize, value):
    """write a length field of specified size to a stream"""
    n = (1 << (fieldSize * 8)) - 1
    if value < n:
        b = value.to_bytes(fieldSize, 'little')
        stream.write(b)
    else:
        b = n.to_bytes(fieldSize, 'little')
        stream.write(b)
        b = value.to_string()
        stream.write(b)
        stream.write(b'\x00')

def readBciIndexCount(stream):
    """read a channel or element index, ignoring the actual indices"""
    s = readLine(stream, b' ')
    if s == '{':
        n = 0
        s = readLine(stream, b' ')
        while s != '}':
            n += 1
            s = readLine(stream, b' ')
    else:
        n = int(s)
    return n

def readBciPhysicalUnit(stream):
    """read the members of a physical unit from a stream"""
    pu = Object()
    pu.offset = float(readLine(stream, b' '))
    pu.gain = float(readLine(stream, b' '))
    pu.unit = readLine(stream, b' ')
    pu.rawMin = float(readLine(stream, b' '))
    pu.rawMax = float(readLine(stream, b' '))
    return pu

def readBciSourceIdentifier(stream):
    """read a BCI2000 source identifier from a stream"""
    b = stream.read(1)
    if b != b'\xff':
        return str(b[0])
    return readLine(stream, b'\x00')

def readBciRawMessage(stream):
    """read a full raw BCI2000 message from a stream"""
    descsupp = stream.read(2) # get descriptor and descriptor supplement
    if descsupp == b'':
        raise EOFError()
    if len(descsupp) != 2:
        raise RuntimeError('Could not read descriptor fields')
    messageLength = readBciLengthField(stream, 2)
    chunks = []
    bytesRead = 0
    while bytesRead < messageLength:
        chunk = stream.read(min(messageLength - bytesRead, 2048))
        if chunk == b'':
            raise EOFError()
        chunks.append(chunk)
        bytesRead = bytesRead + len(chunk)
    return descsupp, b''.join(chunks)

def parseBciSignalProperties(stream):
    """parse a raw signal properties message into an object"""
    sp = Object()
    sp.kind = 'SignalProperties'
    sp.sourceID = readBciSourceIdentifier(stream)
    sp.name = readLine(stream, b' ')
    sp.channels = readBciIndexCount(stream)
    sp.elements = readBciIndexCount(stream)
    sp.type = readLine(stream, b' ')
    sp.channelUnit = readBciPhysicalUnit(stream)
    sp.elementUnit = readBciPhysicalUnit(stream)
    return sp

def parseBciSignalData(stream):
    """parse a raw signal data message into an object"""
    signal = Object()
    signal.kind = 'Signal'
    signal.sourceID = readBciSourceIdentifier(stream)
    signal.type = ord(stream.read(1))
    signal.channels = readBciLengthField(stream, 2)
    signal.elements = readBciLengthField(stream, 2)
    signal.shm = readLine(stream, b'\x00')

    if signal.channels != 0 and signal.elements != 0:
        if signal.type & 64 == 0:
            raise RuntimeError('Signal data not located in shared memory')
        signal.type = signal.type & ~64
        if signal.type == 0:
            signal.type = 'int16'
        elif signal.type == 1:
            signal.type = 'float24'
        elif signal.type == 2:
            signal.type = 'float32'
        elif signal.type == 3:
            signal.type = 'int32'
        else:
            raise RuntimeError('Invalid signal type')
        if platform.system() == 'Windows':
              signal.shm = signal.shm.split("/")[1]

    return signal

def parseBciParameter(stream):
    """parse a raw parameter message into an object"""
    param = Object()
    param.kind = 'Parameter'

    return param;

def parseBciSysCommand(stream):
    """parse a raw syscommand message into an object"""
    syscmd = Object()
    syscmd.kind = 'SysCommand'
    syscmd.command = readLine(stream, b'\x00')
    return syscmd;

def receiveBciMessage(stream):
    """read and parse a single BCI2000 message from a stream"""
    descsupp, data = readBciRawMessage(stream)
    stream2 = io.BytesIO(data)
    if descsupp == BciDescSupp.SignalProperties.value:
        return parseBciSignalProperties(stream2)
    elif descsupp == BciDescSupp.SignalData.value:
        return parseBciSignalData(stream2)
    elif descsupp == BciDescSupp.Parameter.value:
        return parseBciParameter(stream2)
    elif descsupp == BciDescSupp.SysCommand.value:
        return parseBciSysCommand(stream2)
    else:
        raise RuntimeError('Unexpected BCI2000 message type')

def writeBciMessage(stream, descSupp, payload):
    """write a signal BCI2000 message to a stream"""
    stream.write(descSupp)
    length = len(payload)
    writeBciLengthField(stream, 2, length)
    stream.write(payload)
    stream.flush()

def writeBciStateMessage(stream, stateLine):
    """write a single BCI2000 state message to a stream"""
    writeBciMessage(stream, BciDescSupp.State.value, bytes(stateLine, 'utf-8') + b'\r\n')

def writeBciSysCommandMessage(stream, syscmd):
    """write a single BCI2000 sys command message to a stream"""
    writeBciMessage(stream, BciDescSupp.SysCommand.value, bytes(syscmd, 'utf-8') + b'\0')

def writeBciSignalMessage(stream, data):
    """write a numpy array's contents as a signal message to a stream"""
    stream2 = io.BytesIO()
    stream2.write(b'\xff' + bytes('Signal', 'utf-8') + b'\x00' + b'\x02')
    writeBciLengthField(stream2, 2, data.shape[0])
    writeBciLengthField(stream2, 2, data.shape[1])
    for ch in range(0, data.shape[0]):
        for el in range(0, data.shape[1]):
            stream2.write(struct.pack('<f', data[ch, el]))
    writeBciMessage(stream, BciDescSupp.SignalData.value, stream2.getvalue())
    stream2.close()

#user input
HOST, PORT = "localhost", 1879

#initialize variables
CHANNELS = -1
ELEMENTS = -1
setProps = False
chNames = []
memoryName = ""
conn = []

def setRunning0(event):
    """Send a state value to BCI2000"""
    stream = conn.makefile('wb')
    writeBciStateMessage(stream, 'Running 1 0')
    writeBciSysCommandMessage(stream, 'EndOfData')
    stream.close()

def insertNaNs(event):
    """Send a signal of NaNs to BCI2000"""
    data = np.ndarray((CHANNELS,ELEMENTS))
    for ch in range(0,CHANNELS):
        for el in range(0, ELEMENTS):
            data[ch, el] = np.nan
    stream = conn.makefile('wb')
    writeBciSignalMessage(stream, data)
    writeBciSysCommandMessage(stream, 'EndOfData')
    stream.close()

figure, ax = plt.subplots(figsize=(10, 8))
ax.set_xlim(-2,2)
ax.set_ylim(-2,2)
figure.set_facecolor((0,0,0,1))
ax.set_axis_off()
ax.set_frame_on(0)
figure.canvas.draw()

axnans = figure.add_axes([0.05, 0.05, 0.4, 0.075])
axstop = figure.add_axes([0.55, 0.05, 0.4, 0.075])
bnans = Button(axnans, 'Insert NaNs')
bnans.on_clicked(insertNaNs)
bstop = Button(axstop, 'Set Running 0')
bstop.on_clicked(setRunning0)

#listen for connection on specified port
try:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind((HOST, PORT))
        s.listen(1)
        while True:
            print("Waiting for BCI2000 on %s at port %i" %(HOST, PORT))
            waitForRead(s)
            conn, addr = s.accept()
            print('Connected by', addr)
            try:
                stream = conn.makefile('rb')
                while True: #go until we receive an EOFError exception
                    waitForRead(conn)
                    msg = receiveBciMessage(stream)
                        
                    if msg.kind == 'SignalProperties' and msg.sourceID == 'Signal':
                        CHANNELS = msg.channels
                        chNames = range(1,CHANNELS+1)
                        ELEMENTS = msg.elements
                        #initialize variables once we have channels and elements
                        phi = np.zeros((CHANNELS, ELEMENTS))
                        bla = np.zeros((ELEMENTS,1))
                        lineArr = list(range(CHANNELS))
                        for i in range(0,CHANNELS):
                            lineArr[i], = ax.plot(bla, bla)

                        print("Properties: Channels: %i, Elements: %i" %(CHANNELS, ELEMENTS))

                    elif msg.kind == 'SignalProperties' and msg.sourceID == 'States':
                        pass

                    elif msg.kind == 'Signal' and msg.sourceID == 'Signal':

                        if msg.channels != CHANNELS:
                            raise RuntimeError('Mismatch in number of channels')
                        if msg.elements != ELEMENTS:
                            raise RuntimeError('Mismatch in number of elements')

                        if memoryName != msg.shm:
                            # update shared memory object
                            memoryName = msg.shm
                            mem = shared_memory.SharedMemory(memoryName)
                            print(f"Connected to shared memory: {memoryName}")
                            print("Visualizing data...")
                        
                        #update visualization with new data
                        data = np.ndarray((CHANNELS,ELEMENTS),dtype=np.double, buffer=mem.buf)
                        for ch in range(0,CHANNELS):
                            for el in range(0, ELEMENTS):
                                phi[ch, el] = el*2*np.pi/(ELEMENTS-1)

                            xdata = np.multiply(1+0.003*data[ch,:],np.cos(phi[ch,:]))
                            ydata = np.multiply(1+0.003*data[ch,:],np.sin(phi[ch,:]))
                            #update plots
                            lineArr[ch].set_xdata(xdata)
                            lineArr[ch].set_ydata(ydata)

                        #update figure
                        figure.canvas.draw()
                        plt.pause(0.01) #render update

                    elif msg.kind == 'Parameter':
                        continue

                    elif msg.kind == 'Signal' and msg.sourceID == 'States':
                        pass

                    elif msg.kind == 'SysCommand' and msg.command == 'EndOfData':
                        continue;

                    elif msg.kind == 'SysCommand' and msg.command == 'EndOfTransmission':
                        continue;

                    else:
                        raise RuntimeError('Unexpected BCI2000 message')

            except EOFError:
                print('disconnected')
                continue;

            except Exception:
                traceback.print_exc()

except KeyboardInterrupt:
    print('aborted by user')

Conclusion

This demo shows how to grab data from BCI2000 and plot it in Python! The main advantage of this demo shows how to access the data in Python. Once this is done, Python's extensive library can be used to create complex, real-time visualizations and calculations! Also, this demo shows how to modify BCI2000 signals and states from Python.

See also

Programming Reference:SignalSharingClientLibDemo, Programming Reference:SignalSharingDemoClient C++ App, Technical Reference:BCI2000 Messages, User Tutorial:BCI2000Remote