Refactor diagnosis function to streamline data processing and enhance Gradio UI for improved user experience

app.py

@@ -1,8 +1,8 @@
+import gradio as gr
 import pandas as pd
 import json
-import gradio as gr
 
-# --- 状態判定関数 ---
+# --- 閾値診断関数 ---
 def judge_status(value, ll, l, h, hh):
     if pd.notna(ll) and value < ll:
         return "LOW-LOW"
@@ -15,137 +15,136 @@ def judge_status(value, ll, l, h, hh):
     else:
         return "OK"
 
-# --- 診断関数 ---
-def diagnose_process_range(process_name, datetime_str, window_minutes, df, thresholds_df):
+def diagnose_process_range(csv_file, excel_file, process_name, datetime_str, window_minutes):
     try:
+        # 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")
+
+        # 入力日時
         target_time = pd.to_datetime(datetime_str)
-    except Exception:
-        return (pd.DataFrame(), pd.DataFrame(), pd.DataFrame(),
-                f"⚠ 入力した日時 {datetime_str} が無効です。", "")
-
-    start_time = target_time - pd.Timedelta(minutes=window_minutes)
-    end_time = target_time
-    df_window = df[(df["timestamp"] >= start_time) & (df["timestamp"] <= end_time)]
-
-    if df_window.empty:
-        return (pd.DataFrame(), pd.DataFrame(), pd.DataFrame(),
-                "⚠ 指定した時間幅にデータが見つかりません。", "")
-
-    proc_thresholds = thresholds_df[thresholds_df["ProcessNo_ProcessName"] == process_name]
-    if proc_thresholds.empty:
-        return (pd.DataFrame(), pd.DataFrame(), pd.DataFrame(),
-                f"⚠ プロセス {process_name} の閾値が設定されていません。", "")
-
-    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,
-                "LL": thr.get("LL"),
-                "L": thr.get("L"),
-                "H": thr.get("H"),
-                "HH": thr.get("HH"),
-                "判定": status,
-                "重要項目": bool(thr.get("Important", False)),
-                "時刻": str(row["timestamp"])
-            })
-
-    # 集計（全項目）
-    total = len(all_results)
-    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_all = pd.DataFrame({"状態": status_counts.index, "件数": status_counts.values, "割合(%)": status_ratio.values})
-
-    # 集計（重要項目全体）
-    important_results = [r for r in all_results if r["重要項目"]]
-    if important_results:
-        total_imp = len(important_results)
-        status_counts_imp = pd.Series([r["判定"] for r in important_results]).value_counts().reindex(
-            ["LOW-LOW", "LOW", "OK", "HIGH", "HIGH-HIGH"], fill_value=0
-        )
-        status_ratio_imp = (status_counts_imp / total_imp * 100).round(1)
-        result_df_imp = pd.DataFrame({"状態": status_counts_imp.index, "件数": status_counts_imp.values, "割合(%)": status_ratio_imp.values})
-    else:
-        result_df_imp = pd.DataFrame(columns=["状態", "件数", "割合(%)"])
-
-    # 集計（重要項目ごと）
-    result_per_item = []
-    for item in [r["ItemName"] for r in important_results]:
-        item_results = [r for r in important_results if r["ItemName"] == item]
-        if not item_results:
-            continue
-        total_item = len(item_results)
-        status_counts_item = pd.Series([r["判定"] for r in item_results]).value_counts().reindex(
+        start_time = target_time - pd.Timedelta(minutes=window_minutes)
+        end_time = target_time
+
+        # 対象期間の抽出
+        df_window = df[(df["timestamp"] >= start_time) & (df["timestamp"] <= end_time)]
+        if df_window.empty:
+            return None, None, None, f"⚠ 指定時間幅にデータが見つかりません。", "{}"
+
+        # 閾値対象
+        proc_thresholds = thresholds_df[thresholds_df["ProcessNo_ProcessName"] == process_name]
+        if proc_thresholds.empty:
+            return None, None, None, f"⚠ プロセス {process_name} の閾値が設定されていません。", "{}"
+
+        # 判定処理
+        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({
+                    "ItemName": thr["ItemName"],
+                    "判定": status,
+                    "重要項目": bool(thr.get("Important", False))
+                })
+
+        # 全項目集計
+        total = len(all_results)
+        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_item = (status_counts_item / total_item * 100).round(1)
-        for s, c, r in zip(status_counts_item.index, status_counts_item.values, status_ratio_item.values):
-            result_per_item.append({"ItemName": item, "状態": s, "件数": c, "割合(%)": r})
-    result_df_imp_items = pd.DataFrame(result_per_item)
-
-    # サマリー
-    summary = (
-        f"✅ {process_name} の診断完了（{start_time} ～ {end_time}）\n"
-        + "[全項目] " + " / ".join([f"{s}:{r:.1f}%" for s, r in status_ratio.items()]) + "\n"
-        + "[重要項目全体] " + (
-            " / ".join([f"{s}:{r:.1f}%" for s, r in status_ratio_imp.items()])
-            if not result_df_imp.empty else "対象データなし"
+        status_ratio = (status_counts / total * 100).round(1)
+        result_df_all = pd.DataFrame({"状態": status_counts.index, "件数": status_counts.values, "割合(%)": status_ratio.values})
+
+        # 重要項目全体集計
+        important_results = [r for r in all_results if r["重要項目"]]
+        if important_results:
+            total_imp = len(important_results)
+            status_counts_imp = pd.Series([r["判定"] for r in important_results]).value_counts().reindex(
+                ["LOW-LOW", "LOW", "OK", "HIGH", "HIGH-HIGH"], fill_value=0
+            )
+            status_ratio_imp = (status_counts_imp / total_imp * 100).round(1)
+            result_df_imp = pd.DataFrame({"状態": status_counts_imp.index, "件数": status_counts_imp.values, "割合(%)": status_ratio_imp.values})
+        else:
+            result_df_imp = pd.DataFrame(columns=["状態", "件数", "割合(%)"])
+
+        # 重要項目ごと集計
+        result_per_item = []
+        for item in [r["ItemName"] for r in important_results]:
+            item_results = [r for r in important_results if r["ItemName"] == item]
+            if not item_results:
+                continue
+            total_item = len(item_results)
+            status_counts_item = pd.Series([r["判定"] for r in item_results]).value_counts().reindex(
+                ["LOW-LOW", "LOW", "OK", "HIGH", "HIGH-HIGH"], fill_value=0
+            )
+            status_ratio_item = (status_counts_item / total_item * 100).round(1)
+            for s, c, r in zip(status_counts_item.index, status_counts_item.values, status_ratio_item.values):
+                result_per_item.append({"ItemName": item, "状態": s, "件数": c, "割合(%)": r})
+        result_df_imp_items = pd.DataFrame(result_per_item)
+
+        # サマリー
+        summary = (
+            f"✅ {process_name} の診断完了（{start_time} ～ {end_time}）\n"
+            + "[全項目] " + " / ".join([f"{s}:{r:.1f}%" for s, r in status_ratio.items()]) + "\n"
+            + "[重要項目全体] " + (
+                " / ".join([f"{s}:{r:.1f}%" for s, r in status_ratio_imp.items()])
+                if not result_df_imp.empty else "対象データなし"
+            )
         )
-    )
 
-    result_json = json.dumps(all_results, ensure_ascii=False, indent=2,
-                             default=lambda x: x.item() if hasattr(x, "item") else x)
+        # --- 軽量 JSON（集計結果のみ） ---
+        summary_stats = {
+            "全項目割合": status_ratio.to_dict(),
+            "重要項目全体割合": status_ratio_imp.to_dict() if not result_df_imp.empty else {},
+            "重要項目ごと割合": result_per_item
+        }
+        result_json = json.dumps({"集計結果": summary_stats}, ensure_ascii=False, indent=2)
 
-    return result_df_all, result_df_imp, result_df_imp_items, summary, result_json
+        return result_df_all, result_df_imp, result_df_imp_items, summary, result_json
 
-# --- Gradio用ラッパ ---
-def run_diagnosis(csv_file, excel_file, process_name, datetime_str, window_minutes):
-    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")
+    except Exception as e:
+        return None, None, None, f"❌ エラー: {str(e)}", "{}"
 
-    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")
-
-    return diagnose_process_range(process_name, datetime_str, int(window_minutes), df, thresholds_df)
 
 # --- Gradio UI ---
 with gr.Blocks() as demo:
     gr.Markdown("## 閾値診断アプリ")
 
     with gr.Row():
-        csv_input = gr.File(label="CSVファイル", type="filepath")
-        excel_input = gr.File(label="Excel閾値テーブル", type="filepath")
-    process_name_input = gr.Textbox(label="プロセス名", value="E018-A012_除害RO")
-    datetime_input = gr.Textbox(label="基準日時 (例: 2025/8/1 1:05)")
-    window_input = gr.Number(label="さかのぼり時間幅（分）", value=60)
+        csv_input = gr.File(label="CSVファイルをアップロード", type="filepath")
+        excel_input = gr.File(label="閾値テーブル (Excel)", type="filepath")
+
+    with gr.Row():
+        process_input = gr.Textbox(label="工程名（例: E018-A012_除害RO）")
+        datetime_input = gr.Textbox(label="診断日時 (YYYY/MM/DD HH:MM)")
+        window_input = gr.Number(label="時間幅 (分)", value=60)
 
     run_btn = gr.Button("診断実行")
 
+    result_all = gr.Dataframe(label="全項目の状態集計結果")
+    result_imp = gr.Dataframe(label="重要項目全体の状態集計結果")
+    result_imp_items = gr.Dataframe(label="重要項目ごとの状態集計結果")
     summary_out = gr.Textbox(label="診断サマリー")
-    all_df_out = gr.DataFrame(label="全項目の状態集計結果")
-    imp_df_out = gr.DataFrame(label="重要項目全体の状態集計結果")
-    imp_items_out = gr.DataFrame(label="重要項目ごとの状態集計結果")
-    json_out = gr.Code(label="JSON出力", language="json")
+    json_out = gr.JSON(label="集計結果 JSON")
 
     run_btn.click(
-        fn=run_diagnosis,
-        inputs=[csv_input, excel_input, process_name_input, datetime_input, window_input],
-        outputs=[all_df_out, imp_df_out, imp_items_out, summary_out, json_out],
+        diagnose_process_range,
+        inputs=[csv_input, excel_input, process_input, datetime_input, window_input],
+        outputs=[result_all, result_imp, result_imp_items, summary_out, json_out]
     )
 
 if __name__ == "__main__":