Bokeh Server: общие данные сюжета в нескольких сеансах ⇐ Python

[Anonymous]

Я относительно новичок в Боке, но очень знаком с Python.

Я хочу использовать сервер Bokeh для создания графического просмотра в прямом эфире для моих данных. Я хочу получить данные в одном потоке, а затем передать их всем открытым документам /сеансам.

Что было бы хорошим решением для этого? Есть ли даже хорошее решение? Кроме того, после загрузки страницы существуют только начальные данные, видимые до появления новых данных. < /P>

Код: Выделить всё

# 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
< /code>
# 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)
< /code>
Error message:
2025-06-23 08:43:14,863 Exception in callback functools.partial(, )
Traceback (most recent call last):
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\tornado\ioloop.py", line 740, in _run_callback
ret = callback()
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\tornado\ioloop.py", line 764, in _discard_future_result
future.result()
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\server\session.py", line 94, in _needs_document_lock_wrapper
result = func(self, *args, **kwargs)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\server\session.py", line 226, in with_document_locked
return func(*args, **kwargs)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\document\callbacks.py", line 495, in wrapper
return invoke_with_curdoc(doc, invoke)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\document\callbacks.py", line 453, in invoke_with_curdoc
return f()
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\document\callbacks.py", line 494, in invoke
return f(*args, **kwargs)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\document\callbacks.py", line 184, in remove_then_invoke
return callback()
File "C:\XtProgramFiles\python_libs_v3\dev\bokeh_tests\app_hooks.py", line 42, in 
session.document.add_next_tick_callback(lambda: callback(session.document, new_data))
File "C:\XtProgramFiles\python_libs_v3\dev\bokeh_tests\app_hooks.py", line 22, in callback
source.data = new_data
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\core\has_props.py", line 336, in __setattr__
return super().__setattr__(name, value)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\core\property\descriptors.py", line 761, in __set__
self._set(obj, old, value, hint=hint, setter=setter)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\core\property\descriptors.py", line 614, in _set
self._trigger(obj, old, value, hint=hint, setter=setter)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\core\property\descriptors.py", line 692, in _trigger
obj.trigger(self.name, old, value, hint, setter)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\model\model.py", line 583, in trigger
super().trigger(descriptor.name, old, new, hint=hint, setter=setter)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\util\callback_manager.py", line 177, in trigger
self.document.callbacks.notify_change(cast(Model, self), attr, old, new, hint, setter, invoke)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\document\callbacks.py", line 251, in notify_change
self.trigger_on_change(event)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\document\callbacks.py", line 423, in trigger_on_change
invoke_with_curdoc(doc, invoke_callbacks)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\document\callbacks.py", line 453, in invoke_with_curdoc
return f()
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\document\callbacks.py", line 422, in invoke_callbacks
cb(event)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\document\callbacks.py", line 278, in 
self._change_callbacks[receiver] = lambda event: event.dispatch(receiver)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\document\events.py", line 426, in dispatch
super().dispatch(receiver)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\document\events.py", line 218, in dispatch
cast(DocumentPatchedMixin, receiver)._document_patched(self)
File "C:\XtProgramFiles\Miniconda3\envs\dev3\lib\site-packages\bokeh\server\session.py", line 244, in _document_patched
raise RuntimeError("_pending_writes should be non-None when we have a document lock, and we should have the lock when the document changes")
RuntimeError: _pending_writes should be non-None when we have a document lock, and we should have the lock when the document changes
< /code>
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)

и внутри этот обратный вызов, отправляющий HTTP-запрос, который дает мне текущие данные. Но я хочу избежать опроса, и я хочу свести к минимуму задержку, когда доступны новые данные. Так что мне приходилось очень часто опросить. < /Li>
< /ul>
Есть ли другие идеи? Есть ли лучшие практики для подобных проблем? Я даже выбираю хорошую библиотеку для этого?

Подробнее здесь: https://stackoverflow.com/questions/796 ... l-sessions