Блог

Я пытаюсь обновить пунктирную диаграмму, открытую в моем веб — браузере, данными, поступающими с сервера websocket, который я создал. У меня нет проблем с частью websocket. Мой сервер отправляет клиенту данные, которые я добавляю в список, и преобразую этот список в объект фрейма данных pandas. Затем этот объект добавляется в мультипроцесс.Очередь.

В моем обратном вызове Dash от dcc.Интервал, я получаю данные из предыдущей очереди (эта часть, похоже, работает, потому что я могу правильно печатать данные) и обновляю свой график.

Но график в моем браузере не обновляется. Я не вижу никаких ошибок, возникающих в моем коде python или в консоли браузера. (Я пробовал с Firefox, Chrome и Brave) Моя консоль python печатает dash-update-component HTTP/1.1" 200 - мои данные при каждом интервале обратного вызова, и название вкладки Updating... хуже, иногда я вижу, что правильно отображается одно или 10 данных, а затем ничего без изменения кода, просто перезапустив свой код…

Здесь код (на стороне клиента) :

 import multiprocess as mp
import threading
import dash
from dash.dependencies import Output, Input
from dash import dcc
from dash import html
import plotly.graph_objs as go
from socket import *
import time
import json
import pandas as pd
from datetime import datetime


FLAG_QUIT = False  
FLAG_ASK_DATA = False

MAX_SIZE = 10

queue = mp.Queue()
datas = []


app = dash.Dash(__name__)
app.layout = html.Div(
    [
        html.H4('Live Chart'),
        dcc.Graph(id='live-update-graph', animate =True), 
        dcc.Interval(
            id='interval-component',
            interval=200,
            n_intervals=0
        )
    ]
)

@app.callback(Output('live-update-graph', 'figure'),
              [Input('interval-component', 'n_intervals')])
def update_graph(n):
    df = queue.get()
    print(df)
    fig = go.Scatter(x=df.index, y = df['Data'])
    return fig

    

# function for receiving message from client
def send_to_server(clsock):
    global FLAG_QUIT
    global FLAG_ASK_DATA
    while True:
        try:
            if FLAG_QUIT == True:
                break
        
            if FLAG_ASK_DATA:
                time.sleep(0.5)
                clsock.sendall("ok".encode())
        except Exception as ex:
            print(str(ex))
        

# function for receiving message from server
def recv_from_server(clsock):
    global FLAG_QUIT
    global FLAG_ASK_DATA
    while True:
        try:
            data = clsock.recv(1024).decode()
            if data == 'q':
                print('Closing connection')
                FLAG_QUIT = True
                break
            process_data(data)
            FLAG_ASK_DATA = True
        except Exception as ex:
            print(str(ex))
        

def process_data(data):
    global datas
    global queue
    process_websocket_data(data)
    datas_dataframe = convert_data_to_dataframe(datas)
    queue.put(datas_dataframe)

def convert_data_to_dataframe(data):
    data_frame = pd.DataFrame(data)
    data_frame = data_frame.drop_duplicates(0)
    data_frame_date = data_frame[0]

    final_date = []

    for time in data_frame_date.unique():
        readable = datetime.strptime(time, '%Y-%m-%d-%H-%M')
        final_date.append(readable)

    data_frame.pop(0)
    dataframe_final_date = pd.DataFrame(final_date)

    dataframe_final_date.columns = ['Date']

    final_dataframe = data_frame.join(dataframe_final_date)

    final_dataframe.set_index('Date', inplace = True)

    final_dataframe.columns = ['Data']
    return final_dataframe

def process_websocket_data(raw_data):
    data_json = json.loads(raw_data)
    data = [data_json['Date'],data_json['Data']]
    datas.append(data)
    while len(datas) > MAX_SIZE:
        datas.pop(0)


def main():
    threads = []
    HOST = 'localhost'
    PORT = 8765
    clientSocket = socket(AF_INET, SOCK_STREAM)
    connected = False
    while not connected:
        try:
            print("Waiting connecting server...")
            time.sleep(1)
            clientSocket.connect((HOST, PORT))
            connected = True
        except :
            pass
    
    print('Client is connected to a chat sever!n')
    clientSocket.send('start'.encode())

    t_send = threading.Thread(target=send_to_server, args=(clientSocket,))
    t_rcv = threading.Thread(target=recv_from_server, args=(clientSocket,))


    t_send.start()
    t_rcv.start()



if __name__ == '__main__':
    main()
    app.run_server()
 

Что я сделал неправильно или пропустил?

Изменить : Это данные, которые я получаю из очереди в методе обратного вызова update_graph(n).

                 Data
Date
2019-09-16  10302.00
2019-09-23   8061.98
2019-09-30   8042.08
2019-10-07   7851.01
2019-10-14   8274.33
2019-10-21   8218.23
2019-10-28   9534.37
2019-11-04   9197.86
2019-11-11   9041.31
2019-11-18   8504.13
 

Я также попытался заменить свою ось X списком int вместо списка дат, но результат тот же. Не освежает.

Заранее спасибо.

Мне наконец удалось найти проблему, и, похоже, это очередь в обратном вызове. Поэтому я удалил механизм очереди и вместо этого использовал классический список с блокировкой.

 app = dash.Dash(__name__)
app.layout = html.Div([
    dcc.Graph(id="live_graph", animate=False),
    dcc.Interval(id='interval_component',
                interval=INTERVAL,
               
    )])


@app.callback(Output('live_graph', 'figure'),
    [Input('interval_component', 'n_intervals')])
def update_graph(num):
    global datas_dataframe
    
    lock.acquire()
    if len(datas_dataframe) == 0:
        print("No data")
        lock.release()
        return no_update

    try:
        df = datas_dataframe.pop(0)
        fig = go.Figure(data=[go.Scatter(x=df.index,y=df['Data'])],
                        layout=go.Layout(yaxis=dict(tickfont=dict(size=22))))

        lock.release()
        return fig
    except Exception as ex:
        print(str(ex))
        lock.release()
        return no_update