时间戳

首页 > 解决方案 > 来自 python 脚本的实时绘图破折号图

python - 来自 python 脚本的实时绘图破折号图

问题描述

我正在尝试使用 plotly 创建实时虚线图。我正在关注本教程链接，不同之处在于我想从脚本的另一部分更新数据。换句话说，我想让我的脚本运行并收集数据，然后它应该实时显示在图表上。

这是我到目前为止所拥有的

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 pandas as pd


class Plot:

    def __init__(self):
        self.points=[]
        self.app = dash.Dash(__name__)
        self.app.layout = html.Div(
            [
                dcc.Graph(id='live-graph', animate=True),
                dcc.Interval(
                    id='graph-update',
                    interval=500
                ),
            ]
        )
        self. app.callback(Output('live-graph', 'figure'),
        [Input('graph-update', 'n_intervals')])(self.update_graph_scatter)
        
    def start_server(self):
        self.app.run_server(debug=True,host='127.0.0.1', port=8011)

    def update_graph_scatter(self):
        
        pos = pd.DataFrame(data=self.points, columns=['x', 'y'])

        data = plotly.graph_objs.Scatter(
                x=pos['x'],
                y=pos['y'],
                name='Scatter',
                mode= 'lines+markers'
                )

        return {'data': [data],'layout' : go.Layout(xaxis=dict(range=[min(pos['x']),max(pos['x'])]),
                                                    yaxis=dict(range=[min(pos['y']),max(pos['y'])]),)}

    def add_point(self,point):
        self.points.append(point)

我想要的是做类似的事情：

plot=Plot()
plot.start_server()

for i in data_from_my_sensor_callback:
    plot.add_point(i)

但是start_server()脚本没有执行之后。我也尝试在单独的线程中运行它，但结果是一样的。

标签： pythonpython-3.xplotly-dash

解决方案

据我所知，目前没有官方的 Dash 组件支持从服务器启动的实时更新，这似乎是您所追求的（该Interval组件从客户端启动更新）。一种可能的解决方案是使用 websocket，这是WebSocket来自dash-extensions. 这是一个小例子，

import json
import dash_html_components as html
import random
from gevent import sleep
from dash import Dash
from dash.dependencies import Input, Output
from dash_extensions import WebSocket
from dash_extensions.websockets import SocketPool, run_server


# Generator to simulate continuous readings from sensor.
def query_sensor():
    while True:
        sleep(random.uniform(0, 20))  # delay between sensor readings
        yield random.uniform(0, 1)  # the sensor value


# This block runs asynchronously.
def ws_handler(ws):
    for i in query_sensor():
        ws.send(json.dumps(i))  # send data


# Create example app.
app = Dash(prevent_initial_callbacks=True)
socket_pool = SocketPool(app, handler=ws_handler)
app.layout = html.Div([html.Div(id="data"), WebSocket(id="ws")])


@app.callback(Output("data", "children"), [Input("ws", "message")])
def update_data(msg):
    return json.dumps(msg["data"])


if __name__ == '__main__':
    run_server(app, port=5000)  # 5000 if the default port

用dash-extensions==0.0.41. 免责声明：我是dash-extensions.

推荐阅读