如何在Plotly Dash中实现当len(X1)=100时停止实时图表？

问题背景 #

现有Plotly Dash实时图表脚本，需实现以下需求：

• 打印"Hello"
• 浏览器打开实时图表
• 当len(X1)==100时停止实时更新
• 打印"World"

原脚本尝试过max_intervals参数，但该参数按时间计数而非实际数据步数，不符合需求；也了解过服务器关闭代码但不知如何落地。

方案一：动态禁用dcc.Interval（保留服务器） #

通过回调修改dcc.Interval的disabled属性，满足条件时停止定时器，可选择保留服务器或后续触发关闭。

修改核心代码

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, # 毫秒级更新间隔
                n_intervals=0
            ),
        ]
    )

    @app.callback(
        [Output('live-graph', 'figure'),
         Output('graph-update', 'disabled')],  # 新增输出：控制定时器是否禁用
        [Input('graph-update', 'n_intervals')],
        log=True
    )
    def update_graph_scatter(n):
        print(f"{n=}")
        fig = plot_graph()
        # 判断是否达到停止条件
        if len(X1) == 100:
            return fig, True  # 禁用定时器，停止更新
        return fig, False

    return app

方案二：满足条件时关闭服务器（终止程序） #

结合Werkzeug服务器的关闭函数，在数据达到指定长度时触发服务器关闭，程序继续执行后续的"World"打印。

完整修改代码

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
import webbrowser
import os
import requests  # 新增：发送关闭请求

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 shutdown_server():
    func = request.environ.get('werkzeug.server.shutdown')
    if func is None:
        raise RuntimeError('Not running with the Werkzeug Server')
    func()

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

    # 添加服务器关闭路由
    @app.server.route('/shutdown', methods=['POST'])
    def shutdown():
        shutdown_server()
        return 'Server shutting down...'

    app.layout = html.Div(
        [
            dcc.Graph(id='live-graph', animate=True, style={"height": "97vh", 'width': "97vw"}),
            dcc.Interval(
                id='graph-update',
                interval=1000, # 毫秒级更新间隔
                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()
        # 判断是否达到停止条件
        if len(X1) == 100:
            requests.post(f'{url}shutdown')  # 发送关闭请求
        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=False)  # 必须关闭debug模式，否则子进程无法正常关闭
    print("World")

关键注意事项 #

1. 方案二中必须关闭debug模式：debug模式下Werkzeug会启动子进程，werkzeug.server.shutdown仅能关闭主进程，子进程会残留。
2. 数据计数调整：X1初始长度为1，每次plot_graph会追加一条数据，因此len(X1)==100对应99次更新。若需刚好100次更新，可将初始X1设为空列表，或调整判断条件为len(X1) >= 100。
3. 混合方案：若需先停止图表更新，再延迟关闭服务器，可在方案一的回调中，设置disabled=True后添加定时器触发关闭请求。

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