import numpy as np
from bokeh.plotting import figure
from bokeh.models import Panel, Button, ColumnDataSource, Slider, Div
from bokeh.events import ButtonClick
from bokeh.layouts import layout, row, column, Spacer
from tornado import gen
import time
[docs]class AudioStreamTab:
"""Class to create a Bokeh tab that displays a life wingbeat stream of the latest audio chunk and a ringbuffer \
whose length can be specified with the `timeframe` parameter in the :ref:`config.yaml` file. The ringbuffer can be \
saved on button click.
:param config: configuration including :ref:`config.yaml` dictionary and current Bokeh document
:type config: object
"""
def __init__(self, config):
"""config - configuration object
"""
self.config = config
self.chunk_size = config.chunk_size
self.sampling_rate = config.sampling_rate
self.timeframe = config.timeframe
self.number_of_saved_chunks = config.number_of_saved_chunks
# derived parameters
self.max_freq = config.sampling_rate/2
self.max_freq_khz = self.max_freq*0.001
self.timeslice = self.chunk_size/self.sampling_rate * 1000
self.multiples = int(np.ceil(self.timeframe/self.timeslice))
self.template = """<div class='content' style="background-color: {background};;">
<div class='info'> {status_text} </div> </div>
"""
# call tab method to create tab
self._tab = self.make_tab()
self.update_data = False
self.save_all = False
@property
def tab(self):
"""Calls method :meth:`make_tab`.
"""
return self._tab
[docs] def make_tab(self):
""" Creates and arranges the elements of the corresponding Bokeh tab.
:return: Bokeh panel for ringbuffer audio stream visualization
:rtype: Bokeh object
"""
self.btn_wingbeats = Button(label='Start Stream', width=310)
self.btn_wingbeats.on_event(ButtonClick, self.do_update)
self.btn_save = Button(label=f'Save Ringbuffer ({int(self.multiples*self.timeslice)} ms)', width=310)
self.btn_save.disabled = True
self.btn_save.on_event(ButtonClick, self.save_ringbuffer)
self.text_status = Div(text=self.template.format(status_text='Wingbeat stream inactive',
background='#DC524C'),
render_as_text=False)
PLOTARGS = dict(tools="", toolbar_location=None, outline_line_color='#595959')
self.signal_source = ColumnDataSource(data=dict(t=[], y=[]))
self.signal_plot = figure(plot_width=600, plot_height=200, title="Chunk signal",
x_range=[0, self.timeslice], y_range=[-1, 1],
x_axis_label='discrete time [ms]',
y_axis_label='gain*normalized input [ ]', **PLOTARGS)
self.signal_plot.background_fill_color = "#eaeaea"
self.signal_plot.line(x="t", y="y", line_color="#024768", source=self.signal_source)
self.spectrum_source = ColumnDataSource(data=dict(f=[], y=[]))
self.spectrum_plot = figure(plot_width=600, plot_height=200, title="PDS of chunk signal",
y_range=[-150, -90], x_range=[0, self.max_freq_khz],
x_axis_label='discrete frequency [kHz]', y_axis_label='[dB]', **PLOTARGS)
self.spectrum_plot.background_fill_color = "#eaeaea"
self.spectrum_plot.line(x="f", y="y", line_color="#024768", source=self.spectrum_source)
self.ringbuffer_source = ColumnDataSource(data=dict(t=[], y=[]))
self.ringbuffer_plot = figure(plot_width=1200, plot_height=200, title="Ringbuffer",
x_range=[0, self.multiples*self.timeslice], y_range=[-1, 1],
x_axis_label='discrete time [ms]',
y_axis_label='gain*normalized input [ ]', **PLOTARGS)
self.ringbuffer_plot.background_fill_color = "#eaeaea"
self.ringbuffer_plot.line(x="t", y="y", line_color="#024768", source=self.ringbuffer_source)
self.freq = Slider(start=1, end=self.max_freq, value=self.max_freq, step=1, title="Frequency", default_size=200)
self.gain = Slider(start=1, end=300, value=200, step=10, title="Gain", default_size=200)
# define tab layout
_layout = layout(
[self.btn_wingbeats, self.text_status],
[column(row(Spacer(width=47), self.gain), self.signal_plot),
column(row(Spacer(width=47), self.freq), self.spectrum_plot)],
self.btn_save,
self.ringbuffer_plot,
)
return Panel(child=_layout, title='WB - Ringbuffer')
[docs] @gen.coroutine
def signal_update(self, data):
""" Updates the chunk and the ringbuffer figures. Targeted by next tick callback from \
:meth:`sensors.Utilities.pyaudiohandler.PyAudioHandler.update_audio_data`.
:param data: resampled chunk signal, PDS of chunk signal, resampled ringbuffer, raw triggered signal and \
filename provided by :meth:`PyAudioHandler.update_audio_data` \
:type data: 1D float array, 1D float array, 1D float array, 1D int16 array and string
"""
if self.update_data:
if data['values'] is None:
return
signal, PDS_signal, buffer, _, _ = data['values']
# reduce buffer data to reduce computational cost when displaying
buffer = buffer[::2]
# the if-else below are small optimization: avoid computing and sending
# all the x-values, if the length has not changed
if len(signal) == len(self.signal_source.data['y']):
self.signal_source.data['y'] = signal*self.gain.value
else:
t = np.linspace(0, self.timeslice, len(signal))
self.signal_source.data = dict(t=t, y=signal*self.gain.value)
if len(buffer) == len(self.ringbuffer_source.data['y']):
self.ringbuffer_source.data['y'] = buffer*self.gain.value
else:
t = np.linspace(0, self.multiples*self.timeslice, len(buffer))
self.ringbuffer_source.data = dict(t=t, y=buffer*self.gain.value)
if len(PDS_signal) == len(self.spectrum_source.data['y']):
self.spectrum_source.data['y'] = PDS_signal
else:
f = np.linspace(0, self.max_freq_khz, len(PDS_signal))
self.spectrum_source.data = dict(f=f, y=PDS_signal)
self.spectrum_plot.x_range.end = self.freq.value*0.001
[docs] def do_update(self):
""" Gets called on button click. Starts respectively ends the preview of the chunk signal, \
the ringbuffer and changes the text status corresponding to the button label.
"""
if self.btn_wingbeats.label == 'Start Stream':
self.btn_wingbeats.label = 'Stop Stream'
self.text_status.text = self.template.format(status_text='Wingbeat stream active',
background='#3EA639')
self.btn_save.disabled = False
self.update_data = True
else:
self.btn_wingbeats.label = 'Start Stream'
self.text_status.text = self.template.format(status_text='Wingbeat stream inactive',
background='#DC524C')
self.btn_save.disabled = True
self.update_data = False
[docs] def save_ringbuffer(self):
"""Gets called on button click. Saves the ringbuffer by setting a save flag that gets passed to \
:meth:`sensors.Utilities.pyaudiohandler.PyAudioHandler.update_audio_data` calling \
:meth:`sensors.Utilities.pyaudiohandler.PyAudioHandler.save_all` if true.
"""
self.save_all = True