Source code for bci_framework.default_extensions.Tutorial_Fourier_stream_produser.main
"""
=================
Analysis Produser
=================
"""
from bci_framework.extensions.data_analysis import DataAnalysis, loop_consumer
from bci_framework.extensions import properties as prop
from scipy.fftpack import fft, fftfreq
########################################################################
[docs]class AnalysisProduser(DataAnalysis):
"""Analysis Produser"""
# ----------------------------------------------------------------------
def __init__(self, *args, **kwargs):
""""""
super().__init__(*args, **kwargs)
self.create_buffer(seconds=30, resampling=1000)
self.stream()
# ----------------------------------------------------------------------
@loop_consumer('eeg')
def stream(self):
""""""
EEG = fft(self.buffer_eeg, axis=0)
W = fftfreq(EEG.shape[1], d=1 / prop.SAMPLE_RATE)
data = {'amplitude': EEG,
'frequency': W,
}
self.generic_produser('spectrum', data)
if __name__ == '__main__':
AnalysisProduser(enable_produser=True)