Streaming floating point data in Python2

Question

I am in the process of creating a brain-computer interface, and one task is to stream electroencephalograph (EEG) data into a Python script for real-time analysis. Included in this question are two Python scripts: (1) to generate and stream fake EEG data (32 random floating points for testing purposes), and (2) to receive the EEG data and convert it to a numpy array of the shape (n_channels, n_samples). The second script extracts a sliding window of the last X seconds of data.

Generate and stream synthetic EEG data

# file is named "stream_eeg_realtime.py"
import random, threading, time

nominal_srate = 500.  # sampling rate in Hz. Must be float.
sleep_time = 1/nominal_srate

# Define all of the channel names.
channels = ['Fp1','Fp2','AF3','AF4','F3','F4','F7','F8','FC5','FC6','T7','T8',
'FC1','FC2','C3','C4','CP5','CP6','P7','P8','CP1','CP2','P3','P4','01','02',
'PO3','PO4','Oz','Pz','Cz','Fz']

one_sample = [round(random.uniform(-4.,4.), 5) for _ in channels]
data = []

datalock = threading.RLock()
def synthesize_data():
    while True:
        with datalock:
            data.append(one_sample)
        time.sleep(sleep_time)

t1 = threading.Thread(target=synthesize_data, name="EEG")
t1.daemon = True
t1.start()
print("Now streaming synthetic data...")

Receive and convert data to Numpy array (Jupyter Notebook)

In [1]:
import numpy as np
import stream_eeg_realtime as ser

DATA_DURATION = 20
sfreq = float(ser.nominal_srate)
last_rows_index = int(DATA_DURATION*sfreq)
thread_lock = ser.datalock
def get_data(raw_data, index, thread_lock, full_data=False):
    # Data comes in as array-like with shape (n_samples, n_channels).
    with thread_lock:
        if full_data==False:
            d = [row[:] for row in raw_data[-index:]]
        # If we want all of the data, instead of the X-second window.
        elif full_data==True:
            d = [row[:] for row in raw_data]
    data = np.array(d, dtype="float64").T
    return data

In [2]:
raw_array = get_data(ser.data, last_rows_index, thread_lock, full_data=False)
print raw_array.shape

This code has to be as fast as possible. I optimized the code as much as I could as an advanced-beginner, but can I make it faster? Is it OK to use threading in this case, or would multiprocessing be better? (I have no idea how to use multiprocessing)

Here are some things I tried before arriving at the code I posted. At first, I tried to append "columns" to the nested lists with list comprehension. That was fast, but it turns out it is much faster to append "rows" (no list comprehension needed) and then transpose. I also tried to deepcopy the list in the second script, and then I used nested list comprehension to make a copy. Using the slicing syntax in the list comprehension was the fastest method to copy the list.

Answer 1

datalock seems to be unneeded, since data.append is thread-safe and iterating while appending is safe.

multiprocessing might be faster, but it might be slower. It depends on where you're bottlenecked. In particular, data transfer is slower with multiple processes.

The slow aspect here is primarily repeated conversions to numpy.ndarray objects. This can be avoided by only doing the translation once, which is easiest in the producer thread:

data.append(np.array(one_sample, dtype="float64"))

get_data is then simpler as

def get_data(raw_data, number_values=None):
    if number_values == 0:
        return []
    return np.array(raw_data[-number_values or 0:]).T

If the block size you want is bounded by a fixed limit, it's probably better to write into a circular numpy buffer of the right shape. This way, you can just do a single concatenation, rather than a concatenation of a large number of arrays.