py for BI
Project description
BIpy package
Python package to help build experiments at the Brain Institute. It’s main purpose is to help using the real-time BCI.
Module contents
BCI:
bci.classifier_process.run_classifier
bci.classifier_process.ClassifierProcess
bci.inlets.WindowInlet
bci.inlets.ClassifierInlet
bci.models.DummyClassifier
bci.models.get_trained_CSP_LDA
Data Processing:
data_processing.default_instructed_trigmap
data_processing.default_free_trigmap
data_processing.fc_c_cp_1through6
data_processing.organize_xdf
data_processing.get_sliding_window_partition
data_processing.lowpass_filter
data_processing.LowpassWrapper
Session:
session.Session
Other:
electrode_info.csv
electrode_info.json
Example usage of BCI
import BYpy.bci as bci
import numpy as np
from Psychopy import event, core, visual
from pylsl import resolve_stream, StreamInlet
# do Psychopy stuff
win = visual.Window(monitor='testMonitor', fullscr=True)
text_stim = visual.TextStim(win, color='grey')
# train classifier on previously recorded data
data = np.load('some_data.npy')
labels = np.load('some_labels.npy')
clf = bci.models.get_trained_CSP_LDA(data, labels)
# start a classifier process that runs in the background,
# in_source_id='myuid323457' corresponds to the stream of data from dry eeg that will be classified via CSP + LDA
# window_size=500 means the classifier will use the past 500 samples from the dry eeg as input
clfproc = bci.classifier_process.ClassifierProcess(clf, in_source_id='myuid323457', out_source_id='classifier_output', window_size=500)
# listen to the output of the classifier
inlet = bci.inlets.ClassifierInlet(source_id='classifier_output')
for _ in range(100):
# get a new sample from the classifer every second for 100 seconds
core.wait(1)
sample, timestamp = inlet.pull_sample()
# display the classifier's prediction on screen
text_stim.text = str(sample)
text_stim.draw()
win.flip()
# stop the classifier process, if not it could keep running in the background indefinetly
clfproc.kill()
clfproc.close()
# exit Psychopy
win.close()
core.quit()
Example usage of Session
from BIpy.sesssion import Session
import random
import numpy as np
# [OMITTED]: Psychopy stuff such as making window, stims...
def instructions(logger):
# display some instructions
text_stim.draw()
win.flip()
event.waitKeys()
# hide this trial so it doesn't show up in the saved csv file
logger.hide_trial()
def discrimination_RT_trial(logger):
# wait for keypress to start trial
event.waitKeys()
# present fixation cross (random amount of time)
cross.draw()
win.flip()
core.wait(random.uniform(1,2))
# present l/r arrow at random
is_left = random.choice([0,1])
if is_left:
left_arrow.draw()
else:
right_arrow.draw()
# get reaction time
win.flip()
key_pressed, RT = event.waitKeys(keyList=['0', '1'], timeStamped=core.Clock())[0]
# log key pressed, RT, and which arrow was displayed to a csv file
info = {'RT': RT, 'key_pressed': key_pressed, 'is_left': is_left}
logger.log(info)
# get current block and trial number
current_block_num, current_trial_num = logger.get_current()
# get all past trials this block
past_trials = logger.log_history[current_block_num][:current_trial_num]
# get average past correct reaction time for the current block
mean_rt = np.mean([ trial['rt'] for trial past_trials if str(trial['is_left']) == trial['key_pressed'] ])
# display feedback to the participant
if str(is_left) != key_pressed:
text_stim.text = 'Wrong side!'
elif RT > mean_rt:
text_stim.text = 'Slower than average'
else:
text_stim.text = 'Faster than average'
text_stim.draw()
win.flip()
core.wait(1)
# set up trials and blocks
# first block will just be instructions
instruction_block = [instructions]
# second block will be 4 discrimination RT trials
trial_block = [discrimination_RT_trial, discrimination_RT_trial, discrimination_RT_trial, discrimination_RT_trial]
blocks = [instruction_block, trial_block]
# some info about the experiment we might want saved
refresh_rate = win.monitorFramePeriod
sess_id = input('Enter session id:')
info = {'sess_id': sess_id, 'refresh_rate': refresh_rate}
session = Session(info, blocks)
session.run()
# exit Psychopy
win.close()
core.quit()
Submodules
BIpy.data_processing module
class BIpy.data_processing.LowpassWrapper(lowcut=70, sf=500, order=6, axis=2)
Bases: sklearn.base.TransformerMixin
Wrapper class for using lowpass_filter in an sklearn pipeline
fit_transform(data)
Low pass filters data
transform(data) = fit_transform
Also low pass filters data
_init_(lowcut=70, sf=500, order=6, axis=2)
lowcut
Upper limit in hz, above which frequencies are filtered out.
Default 70
sf
Sampling frequency, defualt 500
order
Passed to scipy.signal.butter, default 6
axis
Axis along which the filter is performed, with axis=2 it can filter the entire data cube at once
Default 2, and default should always be used with input data shape (trials, channels, time)
BIpy.data_processing.get_sliding_window_partition(data, labels, window_size)
Splits data into several windows
data
EEG data with shape (trials, channels, time)
labels
1d array of integer labels corresponding to left/right
window_size
Size in samples (not time) of the windows the data will be split into
If window_size corresponds to the number of recorded samples per trial, the function returns the input data and labels unchaged
windowed_data
Array of shape (windows, channels, time)
windowed_labels
1d array of labels corresponding to each window
BIpy.data_processing.lowpass_filter(data, lowcut=70, sf=500, order=6, axis=2)
Low pass filter
data
Shape (trials, channels, time)
lowcut
Upper limit in hz, above which frequencies are filtered out.
Default 70
sf
Sampling frequency, defualt 500
order
Passed to scipy.signal.butter, default 6
axis
Axis along which the filter is performed, with axis=2 it can filter the entire data cube at once
Default 2, and default should always be used with input data shape (trials, channels, time)
y
Filtered data, of same shape as input
BIpy.data_processing.organize_xdf(xdf_filename, trial_duration, gelled_indeces=[5, 6, 7, 10, 11, 21, 22, 24, 27, 28, 38, 39, 40, 42, 53, 55, 56, 57], stim_channel=67, instructed_trigger_map={'instructed_left': 12, 'instructed_right': 13})
Function to organize motor imagery xdf data into labeled epochs. Does not support free trials
Free tip: avoid using xdf data wherever possible
xdf_filename
The file location of the xdf data to load and organize
trial_duration
The duration in seconds of each motor imagery trial
gelled_indeces
The indices of relevant electrodes, the data from all other electrodes will be discarded.
By default fc_c_cp_1through6, the indeces corresponding to electrodes fc, c, and cp 1 through 6, found to be most useful for BCI
Index <=> electrode mappings can be found in BIpy.electrode_info.csv
stim_channel
The channel used for triggers/events, by default 67
instructed_trigger_map
Trigger/event values for instructed left/right motor imagery trials, with keys instructed_left, instructed_right
and int or list of int values
organized_data
Numpy array of shape (trials, channels, time) containing extracted epochs
labels
Numpy array of shape (trials,) where labels[trial_num] corresponds to organized_data[trial_num]
The labels are integers corresponing to instructed_trigger_map[instructed_left] and instructed_trigger_map[instructed_right]
BIpy.session module
class BIpy.session.Session(info, blocks, use_json=False)
Bases: object
A class that handles the execution and data collection of a Psychopy experiment
info
Information about the session, ex: {‘session_id’: 1234}
blocks
List of blocks, each block a list of trials, each trial a function that takes exactly one input: ‘logger’
each trial function should execute the intended trial
use_json
Flag to indicate whether or not to save data to json on calling save()
to_hide
List of trials to be ignored when saving to csv
log_history
Used to store trial data on call of log()
_iq
Index queue, list of tuples corresponding to the indeces of each trial function in blocks
run
Iterates through and runs all trial functions of blocks in order
log(to_log: dict, save_to_file=True)
Stores to_log in log_history[current_block_num][current_trial_num]
hide_trial
Marks current trial to be ignored when saving to csv
save
Attempts to find an apropriate filename, and calls save_to_csv(filename)
if use_json is True, also calls save_to_json(filename)
save_to_csv(filename)
Saves the Session’s info, and log_history to filename in csv format
get_current
Returns the current block and trial nuber: tuple (block, trial)
save_to_json(filename)
Saves as much information about the current Session object as possible to filename in json format
_init_(info, blocks, use_json=False)
info
Information about the session, ex: {‘session_id’: 1234}
blocks
List of blocks, each block a list of trials, each trial a function that takes exactly one input: ‘logger’
each trial function should execute the intended trial
use_json
Flag to indicate whether or not to save data to json on calling save()
get_current()
Returns the current block and trial nuber
hide_trial()
Marks current trial to be ignored when saving to csv
log(to_log, save_to_file=True)
Stores to_log in log_history[current_block_num][current_trial_num]
to_log: dict
Python dict with information to save from the current trial, ex: {‘RT’: .129486}
save_to_file: bool, defaul=True
Flag to save data to file
this can take time so set to False if you want to log some data but the current trial needs strict timing,
then call Session.save() a the end of the experiment
run()
Iterates through and runs all trial functions of blocks in order
save()
Attempts to find an apropriate filename, and calls save_to_csv(filename)
If self.use_json is True, also calls save_to_json(filename)
save_to_csv(filename)
Saves the Session’s info, and log_history to filename in csv format
filename : str
save_to_json(filename)
Saves as much information about the current Session object as possible to filename in json format
filename : str
BIpy.session.is_jsonable(x)
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.