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 : 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 : 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
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.