[Python] Bokeh server: plot shared data in several sessions

I am relatively new to bokeh, but very familiar with python.
I want to use a Bokeh server to create a graphical live viewer for my data. I want to retrieve the data in a single thread and then pass them to all open documents / sessions.
What would be a good solution to do so? Is there even a good solution?

What I tried:


My current approach was to start the thread as soon as the server is started, and inside that iterate over all sessions to update the plots, as soon as new data are available. It is working for one session, but as soon as I open another session, I start getting errors. Also, after loading the page there are only the initial data visible until new data arrive.

# app_hooks.py
import time
import threading

import bokeh.server.contexts
from bokeh.plotting import Document
from bokeh.models import ColumnDataSource

keep_running = True
data = { # some initial data
'x': [1, 2, 3, 4, 5],
'y': [6, 7, 2, 4, 7]
}

def callback(document: Document, new_data: dict):
model = document.get_model_by_name("line")
if model is None:
print("Model not found in document")
return
assert isinstance(model, bokeh.models.renderers.glyph_renderer.GlyphRenderer)
source: ColumnDataSource = model.data_source
# source.stream(new_data, rollover=100)
source.data = new_data

def retrieve_data(i: int) -> dict[str, list[float]]:
# idea: use a PUB/SUB pattern to retrieve data
global data
time.sleep(1) # it might take some time until new data is available. Can be seconds to hours.
data['x'].append(i)
data['y'].append(i * 0.5 % 5)
return data

def on_server_loaded(server_context: bokeh.server.contexts.BokehServerContext):
# If present, this function executes when the server starts.
def add_data_worker():
global keep_running
i=0
while keep_running:
print(f"Iteration {i + 1}")
new_data = retrieve_data(i)
for session in server_context.sessions:
print(session.destroyed, session.expiration_requested)
session.document.add_next_tick_callback(lambda: callback(session.document, new_data))
i += 1

thread = threading.Thread(target=add_data_worker, daemon=True)
thread.start()

def on_server_unloaded(server_context: bokeh.server.contexts.BokehServerContext):
# If present, this function executes when the server shuts down.
global keep_running
keep_running = False


# main.py
from bokeh.plotting import figure, curdoc
from bokeh.models import ColumnDataSource
from .app_hooks import data

source = ColumnDataSource(data=data)

p = figure(title="Simple Line Example", x_axis_label='x', y_axis_label='y')
p.line(x="x", y="y", legend_label="My Value", line_width=2, source=source, name="line")

curdoc().add_root(p)

Other ideas I had:

• If I use a publish/subscribe pattern anyway I could create one subscriber and update-thread per session. But for me it feels better to have them centralized.
• Let each session poll the data from another server. I would do so by using curdoc().add_periodic_callback(callback, 1000) and inside that callback sending a HTTP-request which gives me the current data. But I want to avoid polling and I want to minimize the delay when new data are available. So I would have to poll very often.

Are there other ideas? Are there best practices for similar problems? Do I even choose a good library for that?

Continue reading...