Refactor diagnosis function to separate processing logic and improve Gradio UI for user input handling

app.py

@@ -1,6 +1,6 @@
-import gradio as gr
 import pandas as pd
 import json
+import gradio as gr
 
 # --- 状態判定関数 ---
 def judge_status(value, ll, l, h, hh):
@@ -16,41 +16,26 @@ def judge_status(value, ll, l, h, hh):
         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, None, None
-
-    # CSV読み込み（3行ヘッダー）
-    df = pd.read_csv(csv_file, 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)
-    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")
-
-    # 入力日時
+def diagnose_process_range(process_name, datetime_str, window_minutes, df, thresholds_df):
     try:
         target_time = pd.to_datetime(datetime_str)
     except Exception:
-        return f"⚠ 入力した日時 {datetime_str} が無効です", None, None, None, None
+        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 "⚠ 指定時間幅にデータなし", None, None, None, None
+        return (pd.DataFrame(), pd.DataFrame(), pd.DataFrame(),
+                "⚠ 指定した時間幅にデータが見つかりません。", "")
 
     proc_thresholds = thresholds_df[thresholds_df["ProcessNo_ProcessName"] == process_name]
     if proc_thresholds.empty:
-        return f"⚠ プロセス {process_name} の閾値が見つかりません", None, None, None, None
+        return (pd.DataFrame(), pd.DataFrame(), pd.DataFrame(),
+                f"⚠ プロセス {process_name} の閾値が設定されていません。", "")
 
-    # 全結果リスト
     all_results = []
     for _, row in df_window.iterrows():
         for _, thr in proc_thresholds.iterrows():
@@ -72,10 +57,7 @@ def diagnose_app(csv_file, excel_file, process_name, datetime_str, window_minute
                 "時刻": str(row["timestamp"])
             })
 
-    if not all_results:
-        return f"⚠ 診断結果なし", None, None, None, None
-
-    # --- 全項目集計 ---
+    # 集計（全項目）
     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
@@ -83,7 +65,7 @@ def diagnose_app(csv_file, excel_file, process_name, datetime_str, window_minute
     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)
@@ -95,10 +77,12 @@ def diagnose_app(csv_file, excel_file, process_name, datetime_str, window_minute
     else:
         result_df_imp = pd.DataFrame(columns=["状態", "件数", "割合(%)"])
 
-    # --- 重要項目ごと ---
+    # 集計（重要項目ごと）
     result_per_item = []
-    for item in {r["ItemName"] for r in important_results}:
+    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
@@ -108,7 +92,7 @@ def diagnose_app(csv_file, excel_file, process_name, datetime_str, window_minute
             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"
@@ -118,9 +102,26 @@ def diagnose_app(csv_file, excel_file, process_name, datetime_str, window_minute
         )
     )
 
-    result_json = json.dumps(all_results, ensure_ascii=False, indent=2, default=str)
+    result_json = json.dumps(all_results, ensure_ascii=False, indent=2,
+                             default=lambda x: x.item() if hasattr(x, "item") else x)
 
-    return summary, result_df_all, result_df_imp, result_df_imp_items, 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")
+
+    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:
@@ -128,29 +129,24 @@ with gr.Blocks() as demo:
 
     with gr.Row():
         csv_input = gr.File(label="CSVファイル", type="filepath")
-        excel_input = gr.File(label="閾値テーブル", type="filepath")
-
-    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)
+        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)
 
     run_btn = gr.Button("診断実行")
 
-    summary_out = gr.Textbox(label="サマリー")
-    table_all_out = gr.Dataframe(label="全項目の状態集計結果")
-    table_imp_out = gr.Dataframe(label="重要項目全体の状態集計結果")
-    table_imp_items_out = gr.Dataframe(label="重要項目ごとの状態集計結果")
-    json_out = gr.JSON(label="JSON出力")
+    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")
 
     run_btn.click(
-        diagnose_app,
-        inputs=[csv_input, excel_input, process_name, datetime_str, window_minutes],
-        outputs=[summary_out, table_all_out, table_imp_out, table_imp_items_out, json_out]
+        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],
     )
 
 if __name__ == "__main__":
-    import os
-    if os.getenv("SPACE_ID"):  # Hugging Face環境
-        demo.launch(server_name="0.0.0.0")
-    else:  # ローカル
-        demo.launch()
+    demo.launch(server_name="0.0.0.0", share=False, ssr_mode=False)