停止 Plotly live Graph 的条件

Question

我想知道是否可以停止使用 Plotly 创建的实时图表。 这是我的剧本

import dash
from dash.dependencies import Output, Input
from threading import Timer
from dash import dcc, html
import plotly
import random
import plotly.graph_objs as go
from plotly.subplots import make_subplots
from flask import request
from collections import deque
import webbrowser
import os
import time

X1, Y1 = [1], [1.0]
X2, Y2 = [1], [1.0]
X3, Y3 = [1], [1.0]
X4, Y4 = [1], [1.0]

port = 8050
host = '127.0.0.1'
url = f'http://{host}:{str(port)}/'

def build_dash_app():
    app = dash.Dash(__name__)

    app.layout = html.Div(
        [
            dcc.Graph(id='live-graph', animate=True, style={"height": "97vh", 'width': "97vw"}),
            dcc.Interval(
                id='graph-update',
                interval=1000, # milliseconds
                n_intervals=0
            ),
        ]
    )

    @app.callback(
        Output('live-graph', 'figure'),
        [Input('graph-update', 'n_intervals')],
        log=True
    )
    def update_graph_scatter(n):
        print(f"{n=}")
        fig = plot_graph()
        return fig

    return app


def plot_graph():
    fig = make_subplots(rows=2, cols=2, subplot_titles=['1', '2', '3', '4'])

    X1.append(X1[-1] + 1)
    Y1.append(Y1[-1] + Y1[-1] * random.uniform(-0.1, 0.1))

    fig.add_trace(go.Scatter(
        x=list(X1),
        y=list(Y1),
        name='Scatter1',
        mode='lines+markers'
    ), col=1, row=1)
    fig.update_xaxes(col=1, row=1, range=[min(X1), max(X1)])
    fig.update_yaxes(col=1, row=1, range=[min(Y1), max(Y1)])

    X2.append(X2[-1] + 1)
    Y2.append(Y2[-1] + Y2[-1] * random.uniform(-0.1, 0.1))

    fig.add_trace(go.Scatter(
        x=list(X2),
        y=list(Y2),
        name='Scatter2',
        mode='lines+markers'
    ), col=2, row=1)

    fig.update_xaxes(col=2, row=1, range=[min(X2), max(X2)])
    fig.update_yaxes(col=2, row=1, range=[min(Y2), max(Y2)])

    X3.append(X3[-1] + 1)
    Y3.append(Y3[-1] + Y3[-1] * random.uniform(-0.1, 0.1))

    fig.add_trace(go.Scatter(
        x=list(X3),
        y=list(Y3),
        name='Scatter3',
        mode='lines+markers'
    ), col=1, row=2)
    fig.update_xaxes(col=1, row=2, range=[min(X3), max(X3)])
    fig.update_yaxes(col=1, row=2, range=[min(Y3), max(Y3)])

    X4.append(X4[-1] + 1)
    Y4.append(Y4[-1] + Y4[-1] * random.uniform(-0.1, 0.1))

    fig.add_trace(go.Scatter(
        x=list(X4),
        y=list(Y4),
        name='Scatter4',
        mode='lines+markers'
    ), col=2, row=2)

    fig.update_xaxes(col=2, row=2, range=[min(X4), max(X4)])
    fig.update_yaxes(col=2, row=2, range=[min(Y4), max(Y4)])
    fig.write_html("fig1.html")
    return fig


def open_browser():
    if not os.environ.get("WERKZEUG_RUN_MAIN"):
        webbrowser.open_new(url)


if __name__ == '__main__':
    print("Hello")
    Timer(1, open_browser).start()
    app = build_dash_app()
    app.run_server(port=8050, debug=True)
    print("World")

我想要脚本做的是：

1. print ("Hello")

2. 在浏览器上打开实时图表
3. 当

   len(X1)==100

   时，停止实时图表

4. print("World")

我不知道第3步怎么做。

我看到你可以通过添加关闭服务器

def shutdown_server():
    func = request.environ.get('werkzeug.server.shutdown')
    if func is None:
        raise RuntimeError('Not running with the Werkzeug Server')
    func()

和

    @app.server.route('/shutdown', methods=['POST'])
    def shutdown():
        shutdown_server()
        return 'Server shutting down...'

但是我不明白怎么用。 而且我看到在

app.layout

中，你可以添加一个

max_intervals

    app.layout = html.Div(
        [
            dcc.Graph(id='live-graph', animate=True, style={"height": "97vh", 'width': "97vw"}),
            dcc.Interval(
                id='graph-update',
                interval=1000, # milliseconds
                n_intervals=0,
                max_intervals=100
            ),
        ]
    )

然而，

n

跟随时间而不是执行的步骤数。

Answer 1

您可以针对区间组件的

disabled

道具

disabled（布尔值；可选）：如果为 True，计数器将不再更新。

https://dash.plotly.com/dash-core-components/interval

所以你可以像这样在布局中添加一个按钮：

html.Button(id="button-stop-update", children="Stop graph update", n_clicks=0),

然后你可以添加一个看起来像这样的回调：

@app.callback(
    Output("graph-update", "disabled"),
    Input("button-stop-update", "n_clicks"),
    prevent_initial_call=True
)
def update_graph_scatter(n_clicks):
    return False

如果您按下按钮，图表将停止更新。

如果你想实现一个开始/停止按钮，你可以在你的回调中添加一个

State

并返回当前

disabled

属性值的否定。