Data visualization¶
Bare minimum¶
[ ]:
# visualizations backend classes
from bci_framework.extensions.visualizations import EEGStream
from bci_framework.extensions.visualizations.utils import loop_consumer
# Main class must inherit `EEGStream`
class Stream(EEGStream): # This is `matplotlib.Figure` based class
def __init__(self):
# Initialize `StimuliAPI` class
super().__init__()
# -------------------------------------------------------------
self.axis = self.add_subplot(111)
self.axis.set_title('Title')
self.axis.set_xlabel('Time')
self.axis.set_ylabel('Amplitude')
self.axis.grid(True)
self.stream()
# -------------------------------------------------------------
@loop_consumer('eeg')
def stream(self, *args, **kwargs):
self.feed()
if __name__ == '__main__':
# Create and run the server
Stream()
Data stream access¶
The decorator @loop_consumer
is explained in Data analysis.
[ ]:
@loop_consumer('eeg')
def stream(self, data):
eeg, aux = data
# First eeg channel
ch0 = eeg[0]
self.line.set_ydata(ch0)
self.axis.set_ylim(ch0.min(), ch0.max())
# Time axis
time = range(len(ch0))
self.line.set_xdata(time)
self.axis.set_xlim(time[0], time[-1])
self.feed()
So chaotic, lets create a buffer to visualize more than 100 samples.
Buffer¶
We can use self.create_buffer
to implement an automatic buffer with a fixed time view:
[ ]:
self.create_buffer(30) # create a buffer of 30 seconds
The data can be accessed with self.buffer_eeg
, self.buffer_aux
and self.buffer_timestamp
:
[ ]:
@loop_consumer('eeg')
def stream(self):
eeg = self.buffer_eeg
# First eeg channel
ch0 = eeg[0]
self.line.set_ydata(ch0)
self.axis.set_ylim(-1, 1)
# Time axis
time = np.linspace(0, 30, len(ch0))
self.line.set_xdata(time)
self.axis.set_xlim(time[0], time[-1])
self.feed()
Resampling¶
This visualization has a big issue, there is a lot of data visualize, in this particular case it’s sampling at 1000 kHz
and 30
seconds, which means 30000
points to visualize on the window with (likely) an inferior number of pixels, moreover, this amount of data will affect the speed of the visualization, matplotlib
use all data to render the plot.
We can resample the data, to speed up the visualization.
[ ]:
self.create_buffer(30, resampling=1000) # create a buffer of 30 seconds and a resampling to 1000 samples
[ ]:
@loop_consumer('eeg')
def stream(self):
eeg = self.buffer_eeg_resampled
# First eeg channel
ch0 = eeg[0]
self.line.set_ydata(ch0)
self.axis.set_ylim(-1, 1)
# Time axis
time = np.linspace(0, 30, len(ch0))
self.line.set_xdata(time)
self.axis.set_xlim(time[0], time[-1])
self.feed()
[ ]:
self.create_buffer(30, resampling=300)
Plot raw data¶
There is a method to create matplotlib.lines
automatically for the current configurations.
[ ]:
self.axis, self.time, self.lines = self.create_lines(time=-30, window=1000)
It will return the axis
object the time
array and a list of lines
according to the number of channels.
We can also configure the visualization with channel names as ticks.
[ ]:
self.axis.set_ylim(0, len(prop.CHANNELS) + 1)
self.axis.set_yticks(range(1, len(prop.CHANNELS) + 1))
self.axis.set_yticklabels(prop.CHANNELS.values())
To update the visualization we only need to iterate over self.lines
and self.buffer_eeg_resampled
.
[ ]:
@loop_consumer('eeg')
def stream(self):
eeg = self.buffer_eeg_resampled
for i, line in enumerate(self.lines):
line.set_data(self.time, eeg[i] + 1 + i)
self.feed()
The complete example looks like:
[ ]:
from bci_framework.extensions.visualizations import EEGStream
from bci_framework.extensions.data_analysis import loop_consumer
from bci_framework.extensions import properties as prop
import numpy as np
class Stream(EEGStream):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
DATAWIDTH = 1000
self.axis, self.time, self.lines = self.create_lines(time=-30, window=DATAWIDTH)
self.axis.set_title('Raw EEG')
self.axis.set_xlabel('Time')
self.axis.set_ylabel('Channels')
self.axis.grid(True)
self.axis.set_ylim(0, len(prop.CHANNELS) + 1)
self.axis.set_yticks(range(1, len(prop.CHANNELS) + 1))
self.axis.set_yticklabels(prop.CHANNELS.values())
self.create_buffer(30, resampling=DATAWIDTH)
self.stream()
@loop_consumer('eeg')
def stream(self):
eeg = self.buffer_eeg_resampled
for i, line in enumerate(self.lines):
line.set_data(self.time, eeg[i] + 1 + i)
self.feed()
if __name__ == '__main__':
Stream()
Plot a reverse buffer¶
This option will optimize the image rendering, although the buffer will be transformed too.
This feature can be activated with 2 simple changes, set the time
positive:
[ ]:
self.create_lines(time=30, window=1000)
And call self.reverse_buffer
with the generated axis
:
[ ]:
self.reverse_buffer(self.axis)
Here the full code:
[ ]:
from bci_framework.extensions.visualizations import EEGStream
from bci_framework.extensions.data_analysis import loop_consumer
from bci_framework.extensions import properties as prop
import numpy as np
class Stream(EEGStream):
def __init__(self, *args, **kwargs):
""""""
super().__init__(*args, **kwargs)
DATAWIDTH = 1000
self.axis, self.time, self.lines = self.create_lines(time=30, window=DATAWIDTH)
self.axis.set_title('Raw EEG')
self.axis.set_xlabel('Time')
self.axis.set_ylabel('Channels')
self.axis.grid(True)
self.axis.set_ylim(0, len(prop.CHANNELS) + 1)
self.axis.set_yticks(range(1, len(prop.CHANNELS) + 1))
self.axis.set_yticklabels(prop.CHANNELS.values())
self.create_buffer(30, resampling=DATAWIDTH, fill=np.nan)
self.reverse_buffer(self.axis)
self.stream()
@loop_consumer('eeg')
def stream(self):
eeg = self.buffer_eeg_resampled
for i, line in enumerate(self.lines):
line.set_data(self.time, eeg[i] + 1 + i)
self.feed()
if __name__ == '__main__':
Stream()
Topoplot¶
Following the same instructions is possible to draw topoplots by setting the explicit axes.
[ ]:
from bci_framework.extensions.visualizations import EEGStream, loop_consumer
import mne
class Stream(EEGStream):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.axis = self.add_subplot(1, 1, 1)
self.tight_layout()
self.info = self.get_mne_info()
self.stream()
@loop_consumer('eeg')
def stream(self, data):
eeg, _ = data.value['data']
self.axis.clear()
mne.viz.plot_topomap(eeg.mean(axis=1) - eeg.mean(), self.info, axes=self.axis, show=False, outlines='skirt', cmap='cool')
self.feed()
if __name__ == '__main__':
Stream()
Skip frames¶
Sometimes when the data processing is slow than acquisition, is necessary just to drop some data packages, this will prevent the buffer saturation and improve the visualization.
[ ]:
@fake_loop_consumer('eeg')
def stream(self, data, frame):
if (frame % 5) == 0: # use only 1 of 5 packages
eeg, _ = data.value['data']
self.axis.clear()
mne.viz.plot_topomap(eeg.mean(axis=1) - eeg.mean(), self.info, axes=self.axis, show=False, outlines='skirt', cmap='cool')
self.feed()
Framework integration¶
In this interface is possible to load multiple visualizations at the same time (just keep an eye on your machine resources).