I am working on the Fast Fourier Transform (FFT) and I want to plot the frequency spectrum and waterfall diagram.
In a minimal working example, I am generating random data. I can produce a working plot using pure Python (that code is commented out), but then I must push the entire image every time. So I tried it using JavaScript, but now I am stuck.
If the image supported a one-dimensional array, I would use ColumnDataSource.stream
instead.
Minimal reproducible example:
from bokeh.layouts import column
from bokeh.models import ColumnDataSource, CustomJS, Slider, LogColorMapper, Range1d
import bokeh.plotting as bp
import numpy as np
def bkapp(doc):
# figure
fft_fig = bp.figure(
x_range=(0, 256), y_range=(-50, 150), sizing_mode="stretch_width"
)
# fft
fft_source = ColumnDataSource(dict(x=np.arange(0.5, 256), y=np.zeros(256)))
fft_fig.vbar(top="y", source=fft_source)
# waterfall
waterfall_image = np.zeros((256, 256), dtype=np.float32)
waterfall_source = ColumnDataSource(dict(image=[waterfall_image]), syncable=False)
color_mapper = LogColorMapper(
palette="Viridis256", low=1, high=150, high_color="red"
)
fft_fig.extra_y_ranges = {"waterfall_range": Range1d(start=0, end=1)}
r = fft_fig.image(
image="image",
color_mapper=color_mapper,
dh=1.0,
dw=256,
x=0,
y=-0.75,
source=waterfall_source,
y_range_name="waterfall_range",
)
color_bar = r.construct_color_bar(padding=1)
fft_fig.add_layout(color_bar, "right")
# image generation
callback = CustomJS(
args=dict(waterfall_source=waterfall_source),
code="""
const y = cb_obj.data.y;
const data= waterfall_source.data.image[0]
//This part of JavaScript could be written more efficiently, but I will leave it as-is for clarity.
for (let i = data.length-1; i >= 0 ; i--) {
if (i>=data.shape[0]){
data[i]=data[i-data.shape[0]];
}
else
{
data[i] = y[i];
}
}
// comment this line or just dissable breakpoits
debugger;
""",
)
fft_source.js_on_change("data", callback)
layout = column(fft_fig, sizing_mode="stretch_width")
doc.add_root(layout)
# generate random data
def updateData():
data=np.random.randint(0, 201, size=256)
fft_source.data["y"] = data
# this code below works but i want same from CustomJS to reduce data flow
# waterfall_image = waterfall_source.data["image"][0]
# waterfall_image = np.roll(waterfall_image, -1, 0)
# waterfall_image[-1] = data
# waterfall_source.data["image"] = [waterfall_image]
doc.add_periodic_callback(updateData, 200)
# server
from bokeh.server.server import Server
server = Server({"/": bkapp}, num_procs=1)
server.start()
if __name__ == "__main__":
print("Opening Bokeh application on http://localhost:5006/")
# server.io_loop.add_callback(server.show, "/")
server.io_loop.start()