Refactor diagnosis logic and enhance Gradio UI for threshold evaluation application

app.py

@@ -1,125 +1,130 @@
-import os
 import gradio as gr
 import pandas as pd
-import numpy as np
 import json
-from sklearn.linear_model import LinearRegression
-
-# =========================================================
-# データロード部分（共通）
-# =========================================================
-csv_path = r"sample.csv"  # あなたのCSVパスに置き換え
-excel_path = r"thresholds.xlsx"  # あなたの閾値テーブルExcelパスに置き換え
-lag_excel_path = r"lag_table.xlsx"  # タイムラグ表Excelパスに置き換え
-
-# CSV（3行ヘッダー）
-df = pd.read_csv(csv_path, header=[0, 1, 2])
-timestamp_col = df.iloc[:, 0]
-df = df.drop(df.columns[0], axis=1)
-df.insert(0, "timestamp", timestamp_col)
-df["timestamp"] = pd.to_datetime(df["timestamp"], errors="coerce")
-
-# 閾値テーブル
-thresholds_df = pd.read_excel(excel_path)
-thresholds_df["Important"] = thresholds_df["Important"].astype(str).str.upper().map({"TRUE": True, "FALSE": False})
-for col in ["LL", "L", "H", "HH"]:
-    if col in thresholds_df.columns:
-        thresholds_df[col] = pd.to_numeric(thresholds_df[col], errors="coerce")
-
-# タイムラグ表
-lag_matrix = pd.read_excel(lag_excel_path, index_col=0)
-
-
-# =========================================================
-# 閾値診断関数
-# =========================================================
-def diagnose(process_name, datetime_str, window_minutes):
+
+# --- 状態判定関数 ---
+def judge_status(value, ll, l, h, hh):
+    if pd.notna(ll) and value < ll:
+        return "LOW-LOW"
+    elif pd.notna(l) and value < l:
+        return "LOW"
+    elif pd.notna(hh) and value > hh:
+        return "HIGH-HIGH"
+    elif pd.notna(h) and value > h:
+        return "HIGH"
+    else:
+        return "OK"
+
+# --- 診断処理 ---
+def diagnose_app(csv_file, excel_file, process_name, datetime_str, window_minutes):
+    if csv_file is None or excel_file is None:
+        return "⚠ CSVとExcelファイルを両方アップロードしてください", None, None
+
+    # CSV読み込み（3行ヘッダー維持）
+    df = pd.read_csv(csv_file.name, header=[0, 1, 2])
+    timestamp_col = df.iloc[:, 0]
+    df = df.drop(df.columns[0], axis=1)
+    df.insert(0, "timestamp", timestamp_col)
+    df["timestamp"] = pd.to_datetime(df["timestamp"], errors="coerce")
+
+    # 閾値テーブル読み込み
+    thresholds_df = pd.read_excel(excel_file.name)
+    thresholds_df["Important"] = thresholds_df["Important"].astype(str).str.upper().map({"TRUE": True, "FALSE": False})
+    for col in ["LL", "L", "H", "HH"]:
+        if col in thresholds_df.columns:
+            thresholds_df[col] = pd.to_numeric(thresholds_df[col], errors="coerce")
+
+    # 入力日時
     try:
         target_time = pd.to_datetime(datetime_str)
     except Exception:
-        return None, "⚠ 日時が不正です", None
+        return f"⚠ 入力した日時 {datetime_str} が無効です", None, None
+
+    # 時間幅のデータ抽出
+    start_time = target_time - pd.Timedelta(minutes=window_minutes)
+    end_time = target_time
+    df_window = df[(df["timestamp"] >= start_time) & (df["timestamp"] <= end_time)]
 
-    start_time = target_time - pd.Timedelta(minutes=int(window_minutes))
-    df_window = df[(df["timestamp"] >= start_time) & (df["timestamp"] <= target_time)]
     if df_window.empty:
-        return None, "⚠ データなし", None
+        return "⚠ 指定した時間幅にデータがありません", None, None
 
+    # プロセスの閾値行を抽出
     proc_thresholds = thresholds_df[thresholds_df["ProcessNo_ProcessName"] == process_name]
-    results = []
-    for _, thr in proc_thresholds.iterrows():
-        col_tuple = (thr["ColumnID"], thr["ItemName"], thr["ProcessNo_ProcessName"])
-        if col_tuple not in df.columns:
-            continue
-        series = df_window[col_tuple].dropna()
-        if series.empty:
-            continue
-
-        value = series.iloc[-1]
-        status = "OK"
-        if pd.notna(thr["LL"]) and value < thr["LL"]:
-            status = "LOW-LOW"
-        elif pd.notna(thr["L"]) and value < thr["L"]:
-            status = "LOW"
-        elif pd.notna(thr["HH"]) and value > thr["HH"]:
-            status = "HIGH-HIGH"
-        elif pd.notna(thr["H"]) and value > thr["H"]:
-            status = "HIGH"
-
-        results.append({
-            "ItemName": thr["ItemName"],
-            "値": value,
-            "判定": status,
-            "重要項目": bool(thr.get("Important", False))
-        })
-
-    result_df = pd.DataFrame(results)
-    result_json = json.dumps(results, ensure_ascii=False, indent=2, default=lambda x: x.item() if hasattr(x, "item") else x)
-    summary = f"✅ {process_name} の診断完了（{start_time}～{target_time}）"
-    return result_df, summary, result_json
-
-
-# =========================================================
-# Gradio UI 構築
-# =========================================================
+    if proc_thresholds.empty:
+        return f"⚠ プロセス {process_name} の閾��が見つかりません", None, None
+
+    # 判定実行
+    all_results = []
+    for _, row in df_window.iterrows():
+        for _, thr in proc_thresholds.iterrows():
+            col_tuple = (thr["ColumnID"], thr["ItemName"], thr["ProcessNo_ProcessName"])
+            if col_tuple not in df.columns:
+                continue
+            value = row[col_tuple]
+            status = judge_status(value, thr.get("LL"), thr.get("L"), thr.get("H"), thr.get("HH"))
+            all_results.append({
+                "ColumnID": thr["ColumnID"],
+                "ItemName": thr["ItemName"],
+                "値": value,
+                "判定": status,
+                "重要項目": bool(thr.get("Important", False)),
+                "時刻": row["timestamp"]
+            })
+
+    # 集計（件数と割合）
+    total = len(all_results)
+    if total == 0:
+        return f"⚠ プロセス {process_name} の診断結果なし", None, None
+
+    status_counts = pd.Series([r["判定"] for r in all_results]).value_counts().reindex(
+        ["LOW-LOW", "LOW", "OK", "HIGH", "HIGH-HIGH"], fill_value=0
+    )
+    status_ratio = (status_counts / total * 100).round(1)
+
+    result_df = pd.DataFrame({
+        "状態": status_counts.index,
+        "件数": status_counts.values,
+        "割合(%)": status_ratio.values
+    })
+
+    summary = (
+        f"✅ {process_name} の診断完了（{start_time} ～ {end_time}）\n"
+        + " / ".join([f"{s}:{r:.1f}%" for s, r in status_ratio.items()])
+    )
+
+    result_json = json.dumps(all_results, ensure_ascii=False, indent=2, default=str)
+
+    return summary, result_df, result_json
+
+
+# --- Gradio UI ---
 with gr.Blocks() as demo:
-    gr.Markdown("## 水処理データ解析アプリ")
+    gr.Markdown("## 閾値診断アプリ")
 
-    with gr.Tab("閾値診断アプリ"):
-        process_in = gr.Textbox(label="プロセス名", value="E018-A012_除害RO")
-        datetime_in = gr.Textbox(label="基準日時 (例: 2025/8/1 0:05)", value="2025/8/1 0:05")
-        window_in = gr.Number(label="時間幅（分）", value=60)
+    with gr.Row():
+        csv_input = gr.File(label="CSVファイルをアップロード", type="file")
+        excel_input = gr.File(label="閾値テーブルをアップロード", type="file")
 
-        run_btn = gr.Button("診断実行")
-        summary_out = gr.Textbox(label="サマリー")
-        table_out = gr.Dataframe(label="診断結果", interactive=False)
-        json_out = gr.Textbox(label="JSON出力")
+    process_name = gr.Textbox(label="プロセス名", placeholder="例: E018-A012_除害RO")
+    datetime_str = gr.Textbox(label="基準日時 (例: 2025/8/1 0:05)")
+    window_minutes = gr.Number(label="さかのぼり時間幅 (分)", value=60)
 
-        def run_diagnose(process, dt, win):
-            df_out, summary, js = diagnose(process, dt, win)
-            return summary, df_out, js
+    run_btn = gr.Button("診断実行")
 
-        run_btn.click(
-            fn=run_diagnose,
-            inputs=[process_in, datetime_in, window_in],
-            outputs=[summary_out, table_out, json_out]
-        )
+    summary_out = gr.Textbox(label="サマリー")
+    table_out = gr.Dataframe(label="診断結果（集計表）")
+    json_out = gr.JSON(label="JSON出力")
 
+    run_btn.click(
+        diagnose_app,
+        inputs=[csv_input, excel_input, process_name, datetime_str, window_minutes],
+        outputs=[summary_out, table_out, json_out]
+    )
 
-# =========================================================
-# 実行部分
-# =========================================================
 if __name__ == "__main__":
-    if os.getenv("USE_MCP", "0") == "1":
-        # Hugging Face 用: MCP + UI
-        demo.launch(
-            mcp_server=True,
-            server_name="0.0.0.0",
-            share=False
-        )
+    import os
+    # Hugging Face / ローカル両対応
+    if os.getenv("SPACE_ID"):
+        demo.launch(server_name="0.0.0.0")
     else:
-        # ローカル用: UI のみ
-        demo.launch(
-            server_name="0.0.0.0",
-            share=False,
-            ssr_mode=False
-        )
+        demo.launch()