Refactor trend detection application to improve CSV loading and streamline trend analysis. Update Gradio UI for better user interaction and error handling.

app.py

@@ -1,132 +1,122 @@
-# app_trend.py
-import os
-import json
+# 傾向検出アプリ（Gradio 単独版）
+
+import gradio as gr
 import pandas as pd
 import numpy as np
 from sklearn.linear_model import LinearRegression
-import gradio as gr
+import json
 
-# --- 傾向検出関数 ---
-def detect_trends(process_name, datetime_str, window_minutes, csv_file, excel_file):
+# ===============================
+# データ読み込み関数
+# ===============================
+def load_csv(file):
     try:
-        # CSV読み込み（3行ヘッダー維持）
-        df = pd.read_csv(csv_file.name, header=[0, 1, 2])
+        df = pd.read_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)
-        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(), f"⚠ 指定時間幅にデータなし", "{}"
-
-        proc_thresholds = thresholds_df[(thresholds_df["ProcessNo_ProcessName"] == process_name) &
-                                        (thresholds_df["Important"] == True)]
-        if proc_thresholds.empty:
-            return pd.DataFrame(), f"⚠ プロセス {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 len(series) < 3:
-                continue
-
-            # 線形回帰
-            x = np.arange(len(series)).reshape(-1, 1)
-            y = series.values.reshape(-1, 1)
-            model = LinearRegression().fit(x, y)
-            slope = model.coef_[0][0]
-
-            last_val = series.iloc[-1]
-            l, ll, h, hh = thr.get("L"), thr.get("LL"), thr.get("H"), thr.get("HH")
-
-            status = "安定"
-            if slope < 0 and pd.notna(ll):
-                if last_val > ll:
-                    status = "LL接近下降傾向"
-                elif last_val <= ll:
-                    status = "LL逸脱下降傾向"
-            if slope > 0 and pd.notna(hh):
-                if last_val < hh:
-                    status = "HH接近上昇傾向"
-                elif last_val >= hh:
-                    status = "HH逸脱上昇傾向"
-
-            results.append({
-                "ItemName": thr["ItemName"],
-                "傾向": status,
-                "傾き": round(float(slope), 4),
-                "最終値": round(float(last_val), 3) if pd.notna(last_val) else None,
-                "LL": ll, "L": l, "H": h, "HH": hh
-            })
-
-        if not results:
-            return pd.DataFrame(), f"⚠ データ不足のため傾向検出不可", "{}"
-
-        result_df = pd.DataFrame(results)
-        result_json = json.dumps(results, ensure_ascii=False, indent=2)
-
-        # JSONファイル保存（ダウンロード用）
-        json_path = "trend_result.json"
-        with open(json_path, "w", encoding="utf-8") as f:
-            f.write(result_json)
-
-        summary = f"✅ {process_name} の傾向検出完了（{start_time} ～ {end_time}）"
-        return result_df, summary, json_path
-
+        df.insert(0, "timestamp", pd.to_datetime(timestamp_col, errors="coerce"))
+        return df, f"✅ CSVを読み込みました（{df.shape[0]}行 × {df.shape[1]}列）"
     except Exception as e:
-        return pd.DataFrame(), f"❌ エラー: {str(e)}", "{}"
-
+        return None, f"❌ 読み込みエラー: {str(e)}"
 
-# --- Gradio UI ---
-def create_demo():
-    with gr.Blocks(css=".gradio-container {overflow-y: auto;}") as demo:
-        gr.Markdown("## 📈 傾向検出アプリ")
+# ===============================
+# 傾向検出関数
+# ===============================
+def detect_trend(csv_file, process_name, datetime_str, window_minutes=60):
+    # データ読み込み
+    df, msg = load_csv(csv_file)
+    if df is None:
+        return msg, None, None, None
 
-        with gr.Row():
-            csv_input = gr.File(label="CSVファイルをアップロード", type="filepath")
-            excel_input = gr.File(label="閾値テーブルExcelをアップロード", type="filepath")
-
-        process_name = gr.Textbox(label="プロセス名 (例: E018-A012_除害RO)")
-        datetime_input = gr.Textbox(label="基準日時 (例: 2025/8/1 0:05)")
-        window_minutes = gr.Number(label="さかのぼる時間幅 (分)", value=60)
-
-        run_btn = gr.Button("傾向検出を実行")
-
-        result_table = gr.Dataframe(label="傾向検出結果", interactive=False)
-        summary_out = gr.Textbox(label="サマリー")
-        json_file = gr.File(label="結果JSONダウンロード", interactive=False)
-
-        run_btn.click(
-            detect_trends,
-            inputs=[process_name, datetime_input, window_minutes, csv_input, excel_input],
-            outputs=[result_table, summary_out, json_file]
-        )
-
-    return demo
-
-
-# --- 起動設定 ---
+    try:
+        target_time = pd.to_datetime(datetime_str)
+    except Exception:
+        return f"⚠ 入力した日時 {datetime_str} が無効です。", None, None, None
+
+    start_time = target_time - pd.Timedelta(minutes=window_minutes)
+    df_window = df[(df["timestamp"] >= start_time) & (df["timestamp"] <= target_time)]
+
+    if df_window.empty:
+        return f"⚠ 指定した時間幅にデータが見つかりません。", None, None, None
+
+    # 対象プロセスのカラムを抽出
+    proc_cols = [col for col in df_window.columns if isinstance(col, tuple) and col[2] == process_name]
+    if not proc_cols:
+        return f"⚠ プロセス {process_name} のカラムが見つかりません。", None, None, None
+
+    results = []
+    for col in proc_cols:
+        series = df_window[col].dropna()
+        if len(series) < 3:
+            continue
+
+        X = np.arange(len(series)).reshape(-1, 1)
+        y = series.values
+        model = LinearRegression().fit(X, y)
+        slope = model.coef_[0]
+
+        trend = "上昇" if slope > 0 else "下降" if slope < 0 else "横ばい"
+        results.append({"ItemName": col[1], "傾き": slope, "傾向": trend})
+
+    if not results:
+        return f"⚠ 傾向を検出できるデータがありません。", None, None, None
+
+    df_results = pd.DataFrame(results)
+
+    # サマリー
+    trend_counts = df_results["傾向"].value_counts().to_dict()
+    summary = f"✅ {process_name} の傾向検出完了（{start_time} ～ {target_time}）\n"
+    summary += " / ".join([f"{k}:{v}" for k, v in trend_counts.items()])
+
+    # JSON
+    result_json = json.dumps(results, ensure_ascii=False, indent=2)
+
+    return "✅ 診断完了", df_results, summary, result_json
+
+
+# ===============================
+# Gradio アプリ UI
+# ===============================
+custom_css = """
+body, html, #root, [data-testid="block-container"] {
+    height: auto !important;
+    overflow-y: auto !important;
+}
+"""
+
+with gr.Blocks(css=custom_css) as demo:
+    gr.Markdown("## 📈 傾向検出アプリ")
+
+    with gr.Row():
+        csv_input = gr.File(label="CSVファイルをアップロード", type="filepath")
+        process_input = gr.Textbox(label="プロセス名 (例: E018-A012_除害RO)")
+        datetime_input = gr.Textbox(label="診断基準日時 (例: 2025/8/1 1:05)")
+        window_input = gr.Number(label="時間幅 (分)", value=60)
+
+    run_button = gr.Button("傾向検出を実行")
+
+    with gr.Row():
+        msg_output = gr.Textbox(label="メッセージ")
+    with gr.Row():
+        table_output = gr.Dataframe(label="傾向検出結果", wrap=True)
+    with gr.Row():
+        summary_output = gr.Textbox(label="サマリー")
+    with gr.Row():
+        json_output = gr.JSON(label="JSON出力")
+
+    run_button.click(
+        detect_trend,
+        inputs=[csv_input, process_input, datetime_input, window_input],
+        outputs=[msg_output, table_output, summary_output, json_output]
+    )
+
+# ===============================
+# 実行
+# ===============================
 if __name__ == "__main__":
-    demo = create_demo()
-    if os.getenv("USE_MCP") == "1":
-        demo.launch(server_name="0.0.0.0", mcp_server=True, ssr_mode=False)
+    import os
+    use_mcp = os.getenv("USE_MCP", "0") == "1"
+    if use_mcp:
+        demo.launch(server_name="0.0.0.0", mcp_server=True)
     else:
-        demo.launch(server_name="0.0.0.0", share=False, ssr_mode=False)
-        
+        demo.launch(server_name="0.0.0.0", share=False)