如何无需Interval手动触发Dash Extendable Graph的extendData更新？

我有一个从互联网获取数据的WebSocket数据流，希望通过@app.callback(Output('graph', 'extendData'))的方式，用数据流中的数据更新图表，添加新的数据点。

已尝试方案 #

我参考了extendable-graph的示例代码，通过dcc.Interval组件每秒向图表添加随机点，代码如下：

import dash_extendable_graph as deg
import dash
from dash.dependencies import Input, Output, State
import dash_html_components as html
import dash_core_components as dcc
import random

app = dash.Dash(__name__)

app.scripts.config.serve_locally = True
app.css.config.serve_locally = True

app.layout = html.Div([
    deg.ExtendableGraph(
        id='extendablegraph_example',
        figure=dict(
            data=[{'x': [0],
                   'y': [0],
                   'mode':'lines+markers'
                   }],
        )
    ),
    dcc.Interval(
        id='interval_extendablegraph_update',
        interval=1000,
        n_intervals=0,
        max_intervals=-1),
    html.Div(id='output')
])


@app.callback(Output('extendablegraph_example', 'extendData'),
              [Input('interval_extendablegraph_update', 'n_intervals')],
              [State('extendablegraph_example', 'figure')])
def update_extendData(n_intervals, existing):
    x_new = existing['data'][0]['x'][-1] + 1
    y_new = random.random()
    return [dict(x=[x_new], y=[y_new])], [0], 100


if __name__ == '__main__':
    app.run_server(debug=True)

之后我尝试结合自己的WebSocket代码解析数据流：

def websocket_callback(msg):
    new_point = int(msg)
    print(f"Just received: {new_point}")

ws = websocket.WebSocketApp(url, on_message=websocket_callback)
wst = threading.Thread(target=ws.run_forever)
wst.daemon = True
wst.start()

while True:
    time.sleep(1)

我尝试直接在WebSocket回调上添加Dash装饰器，或者手动调用图表更新回调，但都失败了：

• 把WebSocket消息作为参数传入回调函数时，因为函数同时被Interval和WebSocket触发，逻辑混乱无法运行
• 手动调用update_graph函数时，抛出dash.exceptions.CallbackException: Inputs do not match callback definition错误，因为回调没有定义合法的输入源

解决方案 #

Dash的回调必须由组件的状态变化触发，不能直接调用。可以通过全局队列缓存WebSocket数据 + 中间组件触发回调的方式实现需求，具体代码如下：

import dash_extendable_graph as deg
import dash
from dash.dependencies import Input, Output, State
import dash_html_components as html
import dash_core_components as dcc
import websocket
import threading
from collections import deque

app = dash.Dash(__name__)

# 全局队列缓存WebSocket收到的数据，避免线程冲突
data_queue = deque(maxlen=100)

app.layout = html.Div([
    deg.ExtendableGraph(
        id='extendablegraph_example',
        figure=dict(
            data=[{'x': [0], 'y': [0], 'mode':'lines+markers'}],
        )
    ),
    # 定期检查队列是否有新数据
    dcc.Interval(
        id='interval_check_data',
        interval=100,  # 100ms轮询一次，可根据需求调整
        n_intervals=0,
        max_intervals=-1
    ),
    # 存储最新数据，作为图表更新的触发源
    dcc.Store(id='latest_websocket_data', data={'x': 0, 'y': 0})
])

def websocket_callback(msg):
    # 解析WebSocket消息，这里假设转换为整数
    new_point = int(msg)
    data_queue.append(new_point)

# 启动WebSocket后台线程
ws = websocket.WebSocketApp("你的WebSocket地址", on_message=websocket_callback)
ws_thread = threading.Thread(target=ws.run_forever)
ws_thread.daemon = True
ws_thread.start()

@app.callback(
    Output('latest_websocket_data', 'data'),
    Input('interval_check_data', 'n_intervals'),
    State('latest_websocket_data', 'data'),
    prevent_initial_call=False
)
def sync_queue_to_store(n, current_data):
    if data_queue:
        # 获取队列中的最新数据
        y_new = data_queue.popleft()
        x_new = current_data['x'] + 1
        return {'x': x_new, 'y': y_new}
    # 无新数据时不更新，避免触发不必要的图表刷新
    return dash.no_update

@app.callback(
    Output('extendablegraph_example', 'extendData'),
    Input('latest_websocket_data', 'data'),
    prevent_initial_call=True
)
def update_graph_with_websocket_data(latest_data):
    # 扩展图表数据，添加新点
    return [dict(x=[latest_data['x']], y=[latest_data['y']])], [0], 1000

if __name__ == '__main__':
    app.run_server(debug=True)

方案原理 #

1. 线程安全的数据缓存：用deque作为全局队列，缓存WebSocket后台线程收到的数据，避免直接在多线程中操作Dash组件
2. 轮询触发同步：通过dcc.Interval定期检查队列，将新数据同步到dcc.Store组件中
3. 组件变化触发图表更新：当dcc.Store的数据更新时，自动触发图表的extendData回调，添加新的数据点

这种方式完美分离了WebSocket数据解析和图表更新逻辑，符合Dash的回调机制要求。

内容的提问来源于stack exchange，提问作者Joe