在 app.callback 期间使用 python plotly/dash 访问 self

Question

我正在使用 plotly/dash 和 python 创建实时更新图。不幸的是，在 python 中传递变量self 会引发错误。我在下面做了一个最小的例子来概述问题。

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


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
            ),
        ]
        )

    @_app.callback(
        Output('live-graph', 'figure'),
        [Input('graph-update', 'n_intervals') ]
    )
    def update_graph_scatter(self, 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()

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

图表每秒更新一次，其中包含通过update 函数提供的新数据。 update 函数更新存储在self 中的数据队列。在@app.callback 期间，每秒也会发生一次，它会在绘图之前咨询self 以获取数据的当前状态，但是这样做时出现以下错误

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

如果我在 update_graph_scatter 方法中将 self 作为 arg 删除，错误就会消失，但是该方法的内部无法访问 self ，这会导致不同的错误。任何关于我如何访问 self 或更新方法和@app.callback 标记方法都可以咨询的等效公共商店的帮助将不胜感激。

编辑：通过使用 rob 的解决方案和一些线程，我能够解决这个问题。请参阅下面关于如何在一个线程中实例化类的解决方案（必须完成，因为 app_server 被阻塞） 然后使用主线程更新该类正在绘制的数据。

import dash
from dash.dependencies import Output, Input
import dash_core_components as dcc
import dash_html_components as html
import plotly
import random
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):
            # let's update data here to show class callbacks are working
            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)

Answer 1

• 我已经回答了您的问题范围。如何将回调作为类方法
• 基本上将它们限定为类方法中的静态方法。
• 您假设run_server() 是非阻塞的，即在服务器启动后立即将控制权返回给控制进程。不是 - 因此我将调用 self.update() 放入回调中

import dash
from dash.dependencies import Output, Input
import dash_core_components as dcc
import dash_html_components as html
import plotly
import random
import plotly.graph_objs as go
from collections import deque
import time

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):
            # let's update data here to show class callbacks are working
            self.update()
            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):
        self._X.append(self._X[-1] + 1)
        self._Y.append(self._Y[-1] + self._Y[-1] * random.uniform(-0.1, 0.1))

    def start(self):
        print("starting")
        self._app.run_server(port=8050)
        # this will never be seen !!!!
        print("started")


live_plotter = PlotlyDashboard()


live_plotter.start()
# this code will never run as run_server() is a blocking not a sub-process
while True:
    live_plotter.update()
    time.sleep(1)