Skip to content
Branch: master
Find file Copy path
Fetching contributors…
Cannot retrieve contributors at this time
executable file 81 lines (65 sloc) 2.25 KB
#!/usr/bin/env python2
import time
import threading
import csv
from multiprocessing import Process
import sys
from open_bci import *
#Notes for using
#initiate CSVCollector, takes filename, port, and baud as inputs
#start recording with start() method
#tag recording with tag() method
#what needs to be implemented
# A if statement for multiprocessing/threading.
class CSVCollector(object):
def __init__(self, fname = 'collect.csv',
port=None, baud=115200):
self.board = OpenBCIBoard(port, baud)
self.fname = fname
self.counter = 0
self.file = None
self.fieldnames = []
self.channel_names = []
for i in range(8):
self.channel_names.append("channel_" + str(i))
self.fieldnames.extend(['time', 'tag'])
self.epoch = 0
self.bg_thread = None
def stop(self):
# resolve files and stuff
self.board.should_stream = False
self.csv_writer = None = []
self.bg_thread = None
self.file = None
self.epoch = 0
def disconnect(self):
def receive_sample(self, sample):
t = time.time()
sample = sample.channels
# print '\r', sample,
d = dict(zip(self.channel_names, sample))
d['time'] = t
d['tag'] = self.epoch
if self.csv_writer and self.file:
def start(self):
if self.bg_thread:
self.file = open(self.fname, 'w')
self.csv_writer = csv.DictWriter(self.file, self.fieldnames)
#create a new thread in which the OpenBCIBoard object will stream data
self.bg_thread = threading.Thread(target=self.board.start,
args=(self.receive_sample, ))
# self.bg_thread = Process(target=self.board.start,
# args=(self.receive_sample, ))
def tag(self, epoch):
self.epoch = epoch
You can’t perform that action at this time.