我两个都玩过：

https://www.dash-extensions.com/components/event_source和

https://github.com/bcliang/dash-extendable-graph/tree/master

当客户端推送新数据时更新跟踪（而不是 dcc.interval）。

server.py

import asyncio
import json
import random
import uvicorn
from sse_starlette import EventSourceResponse
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware

middleware = Middleware(CORSMiddleware, allow_origins=["*"], allow_headers=["*"])
server = Starlette(middleware=[middleware])

async def random_data():
    while True:
        await asyncio.sleep(0.5)
        yield json.dumps([random.random() for _ in range(100)]) 
    

@server.route("/random_data")
async def sse(request):
    generator = random_data()
    return EventSourceResponse(generator)

if __name__ == "__main__":
    uvicorn.run(server, port=5000)

client.py

import dash
from dash import dcc
import dash_bootstrap_components as dbc
from dash_extensions.enrich import html, Output, Input, State
from flask import Flask
import random
from dash_extensions import EventSource

# Start app
server = Flask(__name__)
app = dash.Dash(
    __name__,
    external_stylesheets = [dbc.themes.SPACELAB, dbc.icons.FONT_AWESOME],
    title = 'MT',
    prevent_initial_callbacks = "initial_duplicate"
)


app.layout =  html.Div([
    
    dcc.Graph(
        id='extendablegraph_example',
        figure=dict(
            data=[{'x': [0, 1, 2, 3, 4],
                       'y': [0, .5, 1, .5, 0],
                       'mode':'lines+markers'

                   }],
        )
    ),
    EventSource(id="sse", url="http://127.0.0.1:5000/random_data"),
    html.Div(id='output')
])
      


@app.callback(Output('extendablegraph_example', 'extendData'),
              [Input('sse', 'message')],
              [State('extendablegraph_example', 'figure')])
def update_extend_traces_traceselect(message, existing):
    print(message)
    return (dict(x=[
        [existing['data'][0]['x'][-1] + 1],
    ],
        y=[
        [random.random()],
    ]),
        [0], # trace 0
        20 #20 points total in the trace
    )

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

这仅适用于 1 点。但是，相反，我想在每次“推送”时（此处每 500 毫秒）更新 10 个点（来自“消息”）。我可以打印这些值update_extendData，因此可以访问它们，但我不知道进一步处理。

刚刚找到了一个不优雅的解决方案来推动 2 点（见下文）。

app.layout =  html.Div([
    
    dcc.Graph(
        id='extendablegraph_example',
        figure=dict(
            data=[{'x': [i for i in range(1, 11)],
                       'y': [0 for _ in range(10)],
                       'mode':'lines+markers'

                   }],
        )
    ),
    EventSource(id="sse", url="http://127.0.0.1:5000/random_data"),
    html.Div(id='output')
])
      


@app.callback(Output('extendablegraph_example', 'extendData'),
              [Input('sse', 'message')],
              [State('extendablegraph_example', 'figure')])
def update_extend_traces_traceselect(message, existing):
    print(existing['data'][0]['x'])
    return (dict(x=[
        [existing['data'][0]['x'][-1] + 1, existing['data'][0]['x'][-1] + 2 * 1],
        
    ],
        y=[
        [random.random(),random.random()],
    ]),
        [0], # trace 0
        10 #20 points total in the trace
    )

解决方案

暂无回答