Enhance trend detection functionality by integrating future forecasting and improving error handling. Update Gradio UI for clearer input prompts and output display.

app.py

@@ -1,122 +1,153 @@
-# 傾向検出アプリ（Gradio 単独版）
+# 傾向検出アプリ（Gradio版）
 
 import gradio as gr
 import pandas as pd
 import numpy as np
 from sklearn.linear_model import LinearRegression
 import json
+import io
 
-# ===============================
-# データ読み込み関数
-# ===============================
-def load_csv(file):
+# --- 状態判定＆未来予測関数 ---
+def detect_trends_with_forecast(process_name, datetime_str, window_minutes, forecast_minutes, csv_file, excel_file):
     try:
-        df = pd.read_csv(file.name, header=[0, 1, 2])
+        # 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", pd.to_datetime(timestamp_col, errors="coerce"))
-        return df, f"✅ CSVを読み込みました（{df.shape[0]}行 × {df.shape[1]}列）"
-    except Exception as e:
-        return None, f"❌ 読み込みエラー: {str(e)}"
+        df.insert(0, "timestamp", timestamp_col)
+        df["timestamp"] = pd.to_datetime(df["timestamp"], errors="coerce")
 
-# ===============================
-# 傾向検出関数
-# ===============================
-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
+        # 閾値テーブル読み込み
+        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 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
+        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
+
+        # サンプリング間隔
+        interval = df_window["timestamp"].diff().median()
+        if pd.isna(interval):
+            return None, "⚠ サンプリング間隔を検出できません", None
+        interval_minutes = interval.total_seconds() / 60
+
+        # 閾値：対象プロセスかつ重要項目
+        proc_thresholds = thresholds_df[(thresholds_df["ProcessNo_ProcessName"] == process_name) &
+                                        (thresholds_df["Important"] == True)]
+        if proc_thresholds.empty:
+            return None, f"⚠ プロセス {process_name} の重要項目なし", None
+
+        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]
+            n = len(series)
+
+            # 未来予測
+            forecast_steps = int(forecast_minutes / interval_minutes)
+            forecast_index = n + forecast_steps
+            forecast_val = model.predict([[forecast_index]])[0][0]
+            forecast_time = target_time + pd.Timedelta(minutes=forecast_minutes)
+
+            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逸脱上昇傾向"
+
+            # 未来予測の逸脱
+            forecast_status = "安定"
+            if pd.notna(ll) and forecast_val <= ll:
+                forecast_status = "LL逸脱予測"
+            elif pd.notna(hh) and forecast_val >= hh:
+                forecast_status = "HH逸脱予測"
+
+            results.append({
+                "ItemName": thr["ItemName"],
+                "傾向": status,
+                "傾き": round(slope, 4),
+                "最終値": round(float(last_val), 3) if pd.notna(last_val) else None,
+                "予測値": round(float(forecast_val), 3),
+                "予測時刻": str(forecast_time),
+                "予測傾向": forecast_status,
+                "サンプリング間隔(分)": interval_minutes,
+                "LL": ll, "L": l, "H": h, "HH": hh
+            })
+
+        result_df = pd.DataFrame(results)
+        result_json = json.dumps(results, ensure_ascii=False, indent=2)
+
+        # JSONをファイルとして出力
+        json_bytes = io.BytesIO(result_json.encode("utf-8"))
+
+        return result_df, "✅ 傾向検出＋未来予測完了", json_bytes
 
+    except Exception as e:
+        return None, f"❌ エラー: {str(e)}", None
 
-# ===============================
-# 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("## 📈 傾向検出アプリ")
+# --- Gradio UI ---
+with gr.Blocks() 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("傾向検出を実行")
+        excel_input = gr.File(label="閾値テーブル (Excel)", type="filepath")
 
     with gr.Row():
-        msg_output = gr.Textbox(label="メッセージ")
+        process_name = gr.Textbox(label="プロセス名", placeholder="例: E018-A012_除害RO")
+        datetime_str = gr.Textbox(label="日時 (例: 2025/8/1 1:05)")
     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出力")
+        window_minutes = gr.Number(label="参照時間幅（分）", value=60)
+        forecast_minutes = gr.Number(label="未来予測時間幅（分）", value=60)
 
-    run_button.click(
-        detect_trend,
-        inputs=[csv_input, process_input, datetime_input, window_input],
-        outputs=[msg_output, table_output, summary_output, json_output]
+    run_btn = gr.Button("実行")
+
+    with gr.Row():
+        result_table = gr.Dataframe(label="傾向＋未来予測結果")
+    summary_output = gr.Textbox(label="サマリー")
+    json_download = gr.File(label="JSON結果ダウンロード")
+
+    run_btn.click(
+        fn=detect_trends_with_forecast,
+        inputs=[process_name, datetime_str, window_minutes, forecast_minutes, csv_input, excel_input],
+        outputs=[result_table, summary_output, json_download]
     )
 
-# ===============================
-# 実行
-# ===============================
 if __name__ == "__main__":
     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)
+        demo.launch(server_name="0.0.0.0")