Python Plotly/Dash的app.callback回调中访问self实例的解决方案

问题场景 #

使用Python+Plotly/Dash开发实时更新图表时，在被@app.callback修饰的类方法中访问self参数会触发参数缺失报错，错误信息如下：

output_value = func(*func_args, **func_kwargs)  # %% callback invoked %%
TypeError: update_graph_scatter() missing 1 required positional argument: 'n'

若直接移除方法的self参数，报错消失但方法内部无法访问实例存储的数据队列，会触发新的属性未定义错误。

根因分析 #

• 直接在类定义阶段用@_app.callback修饰实例方法时，Dash的回调注册逻辑不会绑定实例的self参数，回调触发时只会传入Input配置对应的参数，导致参数数量不匹配
• app.run_server()是阻塞方法，直接调用后会阻塞后续的代码执行，主线程的数据更新逻辑无法运行

解决方案 #

实现逻辑 #

1. 把回调注册逻辑移到类的callbacks实例方法中，在实例初始化__init__阶段完成注册，利用闭包特性捕获当前实例的self，回调函数即可正常访问实例属性
2. 启动Dash服务时用独立线程运行，避免阻塞主线程的数据更新逻辑

完整可运行代码 #

import dash
from dash.dependencies import Output, Input
import dash_core_components as dcc
import dash_html_components as html
import plotly
import plotly.graph_objs as go
from collections import deque
import time
import numpy as np
from threading import Thread

class PlotlyDashboard:
    _X = deque(maxlen = 20)
    _X.append(1)
    
    _Y = deque(maxlen = 20)
    _Y.append(1)
    
    _app = dash.Dash(__name__)

    def __init__(self):
        self._app.layout = html.Div(
        [
            dcc.Graph(id = 'live-graph', animate = True),
            dcc.Interval(
                id = 'graph-update',
                interval = 1000,
                n_intervals = 0
            ),
        ]
        )
        if self._app is not None and hasattr(self, "callbacks"):
            self.callbacks(self._app)

    def callbacks(self, _app):
        @_app.callback(
            Output("live-graph", "figure"), [Input("graph-update", "n_intervals")]
        )
        def update_graph_scatter(n):
            data = plotly.graph_objs.Scatter(
                x=list(self._X), y=list(self._Y), name="Scatter", mode="lines+markers"
            )

            return {
                "data": [data],
                "layout": go.Layout(
                    xaxis=dict(range=[min(self._X), max(self._X)]),
                    yaxis=dict(range=[min(self._Y), max(self._Y)]),
                ),
            }
            
    def update(self, y):
        self._X.append(self._X[-1] + 1)
        self._Y.append(y)
    
    def start(self):
        self._app.run_server(port=8050)

live_plotter = PlotlyDashboard()

def starter():
    live_plotter.start()

plot_thread = Thread(target=starter)
plot_thread.start()

while True:
        live_plotter.update(np.random.normal(0,1))
        time.sleep(1)

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