Python_ファイルを監視してグラフをライブ更新(したい)

やりたいことの詳細

データが書き込まれていくファイルを常時監視し、
ファイルが更新されたらその更新内容をグラフに反映(更新)したい

環境

python3.9
dash 2.17.0

完成イメージ

まずは環境構築

Dashのインストール

pip install dash

コーディングしていきます

モジュールインストール

import os
from glob import glob

import pandas as pd
from dash import Dash, dcc, html, Input, Output
import plotly.graph_objects as go

監視するcsvのパスを定義

# CSVのパス(テスト)
csvpath = "test.csv"

レイアウトを決めていく

# Dash
# レイアウトの設定
app = Dash()
app.layout = html.Div([
    html.H1("A粗材の判定モニター"),
    dcc.Graph(id="test_graph")
    ]
)
"""
レイアウトは、
H1タグとグラフのみに設定
"""

dcc.Graphの引数,idは任意の文字列でOK
更新させたいものにはidを決める必要がありそうなので、
まずは深く考えずテキトウに決めます

コールバック関数(今回の肝!)

@app.callback(Output("test_graph","figure"),
              Input("",""))            

Outputは「更新させたいデータ」、と解釈しました
今回はグラフなので,先ほどのレイアウト決定の際に、設定した「"test_graph"」を第一引数に渡します。

第二引数は,dcc.Graphの引数のどれかです
説明が難しいのですが、以下dcc.Graphの引数です

id,responsive,,,loading_stateとズラーっと並んでいるどれかを指定する感じみたいです。
今回はグラフを変えたいので「figure」になります

そしてInputですが、困りました。
何かボタンを押したときとか、アクションを起こしたときにその情報をもとに更新をかけるのであれば良いのですが、今回のケースには当てはまりそうもないです。
「ファイルを監視して更新されたら」をトリガーに、とも思いましたが無理そうでした。
少し調べた結果、「任意の秒数で更新をかける」でとりあえず代用できそうでしたのでその方向で書いていきます

レイアウト設定の修正

# Dash
# レイアウトの設定
app = Dash()
app.layout = html.Div([
    html.H1("A粗材の判定モニター"),
    dcc.Graph(id="test_graph"),
    # 追加
    dcc.Interval(id="interval-component",interval=5000,n_intervals=0)
    ]
)

dcc.Intervalを追加しました。
引数については、
id : Graphと同様任意の文字列
interval : 更新間隔 5000なら5秒
n_interval : 何回目の更新か的な。0なら0スタート※コールバック関数で受け取れるので何か意図があれば0以外も有り

コールバック関数の続き

@app.callback(Output("test_graph","figure"),
              Input("interval-component","n_intervals"))

Inputの中に、レイアウトで設定したインターバルを書きました
第一引数は、設定したid
第二引数は、0と設定した「n_intervals」にしました

コールバックの直下にグラフの描画を書いていく

def update_figure(n):
    # CSVをデータフレームで読み込む
    df = pd.read_csv(csvpath)
    # dfの中身は以下
    # ID | Detect
    # ---|--------
    # A_1|      0
    # ---|--------
    # A_2|      0
    # ---|--------
    # A_3|      1
    
    # IDとDetectの2列
    # IDはAワーク_連番
    # Detectは不良があれば1とかっていう感じです

    # IDをリストで取得
    id_list = df["ID"]
    # Detectをリストで取得
    Detect_list = df["Detect"]
    # グラフ (xが横軸, yが縦軸) -> 検出したIDは山になるようなグラフができる
    plot = go.Scatter(x=id_list, y=Detect_list)
    # グラフを描画
    figure = go.Figure(data=plot)
    return figure

グラフを描画した変数figureをreturnすることで、Outputで設定したid(今回だとdcc.Graph)が受け取ります

※Outputの第二引数をfigureとしたので、
変数figureが引数figureに渡される感じだと思います

動作させるための文言

if __name__ == "__main__":
    # 決まり文句
    app.run_server()

動かしてみる

127.0.0.1~~ってとこをCtrl+クリックするとWebアプリケーションとして立ち上がります

すると、とりあえず初期情報は問題無く表示されました

CSVファイルを更新してみる

更新はとりあえず手動でデータを追加します

A_4,1 という行を追加しました。

問題無く更新されました

最低限のことはこれで完成です

ただ、実業務で使おうとするともっと複雑になってくるので、
記事として挙げながら進めていきます

プログラム全文

import os

import pandas as pd
from dash import Dash, dcc, html, Input, Output
import plotly.graph_objects as go

# CSVのパス(テスト)
csvpath = "test.csv"

# Dash
# レイアウトの設定
app = Dash()
app.layout = html.Div([
    html.H1("A粗材の判定モニター"),
    dcc.Graph(id="test_graph"),
    dcc.Interval(id="interval-component",interval=5000,n_intervals=0)
    ]
)

# コールバック(グラフの更新)
@app.callback(Output("test_graph","figure"),
              Input("interval-component","n_intervals"))
def update_figure(n):
    df = pd.read_csv(csvpath)
    id_list = df["ID"]
    Detect_list = df["Detect"]
    plot = go.Scatter(x=id_list, y=Detect_list)
    figure = go.Figure(data=plot)
    return figure


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