app.py

# 統合版 Gradio アプリ (閾値診断 + 傾向検出 + 予兆解析 + 変動解析, Supabase固定ファイル)
import gradio as gr
import pandas as pd
import numpy as np
import json
from dotenv import load_dotenv
import os
from supabase import create_client, Client
from sklearn.linear_model import LinearRegression

# .envファイルを読み込む
load_dotenv()

# --- Supabase 設定 ---
SUPABASE_URL = os.getenv("SUPABASE_URL")
SUPABASE_KEY = os.getenv("SUPABASE_KEY")
SUPABASE_BUCKET = os.getenv("SUPABASE_BUCKET")

CSV_FILE = os.getenv("CSV_PATH")
EXCEL_FILE = os.getenv("EXCEL_PATH")
LAG_FILE = os.getenv("LAG_PATH")
TMP_DIR = "./tmp"

def _create_supabase_client() -> Client | None:
    """
    Create Supabase client if credentials are available; return None otherwise.
    """
    if not SUPABASE_URL or not SUPABASE_KEY:
        return None
    try:
        return create_client(SUPABASE_URL, SUPABASE_KEY)
    except Exception:
        return None

supabase: Client | None = _create_supabase_client()

# --- 共通ユーティリティ ---
def convert_value(v) -> int | float | str:
    """
    numpy型を Python 標準の型に変換する。
    Args:
        v: numpyの値またはPythonの値
    Returns:
        int | float | str: Python標準型に変換した値
    """
    if hasattr(v, "item"):
        return v.item()
    if isinstance(v, (np.integer, int)):
        return int(v)
    if isinstance(v, (np.floating, float)):
        return float(v)
    return v

def normalize(s: str) -> str:
    """
    文字列を正規化（全角スペース・改行削除）する。
    Args:
        s (str): 入力文字列
    Returns:
        str: 正規化後の文字列
    """
    return str(s).replace("\u3000", " ").replace("\n", "").replace("\r", "").strip()

def col_to_str(col) -> str:
    """
    Convert multi-index column tuple to a single string identifier.
    """
    return "_".join([str(c) for c in col if c]) if isinstance(col, tuple) else str(col)

def find_matching_column(df: pd.DataFrame, col_id: str, item_name: str, process_name: str) -> str | None:
    """
    DataFrame内で指定条件に一致する列を探索する。
    Args:
        df (pd.DataFrame): データフレーム
        col_id (str): カラムID
        item_name (str): 項目名
        process_name (str): プロセス名
    Returns:
        str | None: 一致するカラム名、見つからなければ None
    """
    norm_item = normalize(item_name)
    candidates = [
        c for c in df.columns
        if isinstance(c, str)
        and col_id in c
        and process_name in c
        and norm_item in normalize(c)
    ]
    return candidates[0] if candidates else None

# --- グローバル変数 ---
DF = None
THRESHOLDS_DF = None
LAG_MATRIX = None
INITIAL_CHOICES = []

# --- Supabaseからファイルをロード ----
def load_from_supabase() -> str:
    """
    Supabase Storage から CSV・Excel・ラグファイルを読み込み、
    グローバル変数 (DF, THRESHOLDS_DF, LAG_MATRIX, INITIAL_CHOICES) を更新する。
    Returns:
        str: 読み込み結果のメッセージ
    """
    global DF, THRESHOLDS_DF, LAG_MATRIX, INITIAL_CHOICES
    try:
        if supabase is None:
            return "❌ Supabase クライアント未初期化: 環境変数 SUPABASE_URL/SUPABASE_KEY を確認してください"
        if not SUPABASE_BUCKET:
            return "❌ SUPABASE_BUCKET が未設定です"

        # --- CSV ---
        csv_path = CSV_FILE
        if not csv_path:
            return "❌ CSV_PATH が未設定です"
        os.makedirs(TMP_DIR, exist_ok=True)
        csv_bytes = supabase.storage.from_(SUPABASE_BUCKET).download(csv_path)
        with open(os.path.join(TMP_DIR, "data.csv"), "wb") as f:
            f.write(csv_bytes)
        DF = pd.read_csv(os.path.join(TMP_DIR, "data.csv"), header=[0, 1, 2])
        timestamp_col = pd.to_datetime(DF.iloc[:, 0], errors="coerce")
        DF = DF.drop(DF.columns[0], axis=1)
        DF.insert(0, "timestamp", timestamp_col)

        DF.columns = [
            "timestamp" if (isinstance(c, str) and c == "timestamp") else col_to_str(c)
            for c in DF.columns
        ]

        # --- Excel 閾値 ---
        excel_path = EXCEL_FILE
        if not excel_path:
            return "❌ EXCEL_PATH が未設定です"
        thr_bytes = supabase.storage.from_(SUPABASE_BUCKET).download(excel_path)
        with open(os.path.join(TMP_DIR, "thresholds.xlsx"), "wb") as f:
            f.write(thr_bytes)
        THRESHOLDS_DF = pd.read_excel(os.path.join(TMP_DIR, "thresholds.xlsx"))
        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_path = LAG_FILE
        if not lag_path:
            return "❌ LAG_PATH が未設定です"
        lag_bytes = supabase.storage.from_(SUPABASE_BUCKET).download(lag_path)
        with open(os.path.join(TMP_DIR, "lag.xlsx"), "wb") as f:
            f.write(lag_bytes)
        LAG_MATRIX = pd.read_excel(os.path.join(TMP_DIR, "lag.xlsx"), index_col=0)

        print(50*'=')
        print('INITIAL_CHOICES')
        print(50*'-')
        # INITIAL_CHOICES を更新
        if THRESHOLDS_DF is not None and "ProcessNo_ProcessName" in THRESHOLDS_DF.columns:
            INITIAL_CHOICES = sorted(THRESHOLDS_DF["ProcessNo_ProcessName"].dropna().unique().tolist())
        else:
            INITIAL_CHOICES = []
        print('INITIAL_CHOICES: ', INITIAL_CHOICES)
        print(50*'=')

        return "✅ Supabase からファイルを読み込み成功"

    except Exception as e:
        return f"❌ Supabase からの読み込み失敗: {e}"

# --- Tab1: 閾値診断 ---
def judge_status(value: float, ll: float, l: float, h: float, hh: float) -> str:
    """
    測定値を閾値と比較してステータスを判定する。
    Returns:
        str: 判定結果 ("LOW-LOW", "LOW", "OK", "HIGH", "HIGH-HIGH")
    """
    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 parse_datetime_safe(s: str) -> pd.Timestamp | None:
    """
    / や - 区切り、ゼロ埋めの有無などを吸収して datetime に変換する。
    """
    try:
        return pd.to_datetime(s, errors="coerce")
    except Exception:
        return None

def _diagnose_process_range_core(process_name: str, datetime_str: str, window_minutes: int
                                ) -> tuple[pd.DataFrame | None, pd.DataFrame | None,
                                           pd.DataFrame | None, str, str]:
    """
    指定プロセスに対して閾値診断を行い、テーブル・サマリー・JSONを生成するコア処理。
    """
    global DF, THRESHOLDS_DF
    if DF is None or THRESHOLDS_DF is None:
        return None, None, None, "⚠ ファイル未読み込み", None

    # ✅ 日時を安全に解釈
    target_time = parse_datetime_safe(datetime_str)
    if target_time is None or pd.isna(target_time):
        return None, None, None, f"⚠ 入力した日時 {datetime_str} が無効です。", 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)]
    if df_window.empty:
        return None, None, None, "⚠ 指定した時間幅にデータが見つかりません。", None

    # ✅ プロセス名を normalize して突き合わせる
    proc_thresholds = THRESHOLDS_DF[
        THRESHOLDS_DF["ProcessNo_ProcessName"].apply(normalize) == normalize(process_name)
    ]
    if proc_thresholds.empty:
        return None, None, None, f"⚠ プロセス {process_name} の閾値が設定されていません。", None

    all_results = []
    for _, row in df_window.iterrows():
        for _, thr in proc_thresholds.iterrows():
            col_name = f"{thr['ColumnID']}_{thr['ItemName']}_{thr['ProcessNo_ProcessName']}"
            if col_name not in DF.columns:
                continue
            value = row[col_name]
            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"], "判定": 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=["状態", "件数", "割合(%)"])
        status_ratio_imp = pd.Series(dtype=float)

    result_per_item = []
    for item in set([r["ItemName"] for r in important_results]):
        item_results = [r for r in important_results if r["ItemName"] == item]
        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, "件数": int(c), "割合(%)": float(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 "対象データなし")
    )

    json_data = {
        "集計結果": {
            "全項目割合": {k: float(v) for k, v in status_ratio.to_dict().items()},
            "重要項目全体割合": {k: float(v) for k, v in status_ratio_imp.to_dict().items()} if not result_df_imp.empty else {},
            "重要項目ごと割合": [dict(row) for _, row in result_df_imp_items.iterrows()]
        }
    }
    result_json = json.dumps(json_data, ensure_ascii=False, indent=2)

    return result_df_all, result_df_imp, result_df_imp_items, summary, result_json

def diagnose_process_range_ui(
    process_name: str,
    datetime_str: str,
    window_minutes: int
) -> tuple[pd.DataFrame | None, pd.DataFrame | None, pd.DataFrame | None, str]:
    """
    UI 用: Dropdown で選択されたプロセス名を受け取り、
    閾値診断を実行してテーブルとサマリーを返す。
    Args:
        process_name (str): UI の Dropdown から選択されたプロセス名
        datetime_str (str): 基準日時（例: "2024-06-01 12:00"）
        window_minutes (int): 遡る時間幅（分）
    Returns:
        tuple:
            result_df_all (pd.DataFrame | None): 全項目の状態集計
            result_df_imp (pd.DataFrame | None): 重要項目全体の状態集計
            result_df_imp_items (pd.DataFrame | None): 重要項目ごとの状態集計
            summary (str): サマリーテキスト
    """
    result_df_all, result_df_imp, result_df_imp_items, summary, _ = _diagnose_process_range_core(
        process_name, datetime_str, window_minutes
    )
    return result_df_all, result_df_imp, result_df_imp_items, summary

def diagnose_process_range_json(
    process_name: str,
    datetime_str: str,
    window_minutes: int
) -> dict:
    """
    指定されたプロセス名、基準日時、さかのぼる時間幅（分）を受け取り、閾値診断を実行し、その結果をJSON形式（dict）で返す関数。

    Args:
        process_name (str): 診断対象となるプロセス名。UIのドロップダウン等で選択された値を想定。
        datetime_str (str): 診断の基準となる日時（例: "2024-06-01 12:00"）。この日時を基準にデータを抽出する。
        window_minutes (int): 基準日時からさかのぼる時間幅（分単位）。この範囲のデータで診断を行う。

    Returns:
        dict: 閾値診断の結果を含むJSONオブジェクト。
            - "meta": 入力パラメータ情報（process_name, datetime_str, window_minutes）
            - "tab1": 
                - "summary": 診断のサマリーテキスト（例: 正常/異常の割合や診断完了メッセージ等）
                - "results": 
                    - "全項目集計": 全項目の状態集計（リスト形式）
                    - "重要項目全体集計": 重要項目全体の状態集計（リスト形式）
                    - "重要項目ごと集計": 重要項目ごとの状態集計（リスト形式）
            例外発生時は、"summary"にエラーメッセージ、各集計は空リストとなる。

    注意:
        - 内部で _diagnose_process_range_core を呼び出し、DataFrameやサマリー、JSON文字列を取得する。
        - 取得したJSON文字列が有効な場合はそれをパースして返す。無効な場合や例外時はDataFrameをdict化して返す。
        - MCP等のAPI連携や外部システム連携用途を想定。
    """
    print(50*'=')
    print('in diagnose_process_range_json()')
    print('process_name: ', process_name)
    print('datetime_str: ', datetime_str)
    print('window_minutes: ', window_minutes)
    print(50*'=')
    try:
        result_df_all, result_df_imp, result_df_imp_items, summary, result_json = _diagnose_process_range_core(
            process_name, datetime_str, window_minutes
        )

        # --- case1: core が JSON を返している場合 ---
        if result_json and isinstance(result_json, str):
            try:
                parsed = json.loads(result_json)
                return parsed
            except Exception:
                # JSON が壊れている場合 → fallback
                pass

        # --- case2: fallback (DataFrameをdictに変換) ---
        def df_to_records(df):
            return df.to_dict(orient="records") if df is not None else []

        return {
            "meta": {
                "process_name": process_name,
                "datetime_str": datetime_str,
                "window_minutes": window_minutes,
            },
            "tab1": {
                "summary": summary if summary else "診断結果は得られませんでした。",
                "results": {
                    "全項目集計": df_to_records(result_df_all),
                    "重要項目全体集計": df_to_records(result_df_imp),
                    "重要項目ごと集計": df_to_records(result_df_imp_items),
                },
            },
        }

    except Exception as e:
        # --- case3: 例外発生時 ---
        return {
            "meta": {
                "process_name": process_name,
                "datetime_str": datetime_str,
                "window_minutes": window_minutes,
            },
            "tab1": {
                "summary": f"診断エラー: {str(e)}",
                "results": {
                    "全項目集計": [],
                    "重要項目全体集計": [],
                    "重要項目ごと集計": [],
                },
            },
        }

# --- Tab2: 傾向検出 ---
def detect_trends_with_forecast(process_name: str, datetime_str: str, window_minutes: int,
                                forecast_minutes: int, downsample_factor: int
                                ) -> tuple[pd.DataFrame | None, str, str]:
    """
    線形回帰を用いて傾向検出と未来予測を行う。
    Args:
        process_name (str): プロセス名
        datetime_str (str): 基準日時
        window_minutes (int): 遡る時間幅
        forecast_minutes (int): 予測時間幅
        downsample_factor (int): サンプリング粗化倍率
    Returns:
        tuple: (予測結果DataFrame, サマリーテキスト, JSON文字列)
    """
    global DF, THRESHOLDS_DF
    if DF is None or THRESHOLDS_DF is None:
        return None, "⚠ ファイル未読み込み", None
    target_time = pd.to_datetime(datetime_str)
    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 None, "⚠ データなし", None
    interval = df_window["timestamp"].diff().median()
    if pd.isna(interval):
        return None, "⚠ サンプリング間隔検出失敗", None
    interval_minutes = interval.total_seconds() / 60
    df_window = df_window.iloc[::int(downsample_factor), :].reset_index(drop=True)
    effective_interval = interval_minutes * int(downsample_factor)
    proc_thresholds = THRESHOLDS_DF[
        (THRESHOLDS_DF["ProcessNo_ProcessName"].apply(normalize) == normalize(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 = f"{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]
        forecast_steps = int(forecast_minutes / effective_interval)
        forecast_val = model.predict([[len(series) + forecast_steps]])[0][0]
        forecast_time = target_time + pd.Timedelta(minutes=forecast_minutes)
        risk = "安定"
        if pd.notna(thr.get("LL")) and forecast_val <= thr["LL"]:
            risk = "LL逸脱予測"
        elif pd.notna(thr.get("HH")) and forecast_val >= thr["HH"]:
            risk = "HH逸脱予測"
        results.append({
            "ItemName": thr["ItemName"], "傾き": round(float(slope), 4),
            "最終値": round(float(last_val), 3), "予測値": round(float(forecast_val), 3),
            "予測時刻": str(forecast_time), "予測リスク": risk, "粗化間隔(分)": round(effective_interval, 2)
        })
    result_df = pd.DataFrame(results)
    result_json = json.dumps(results, ensure_ascii=False, indent=2)
    return result_df, "✅ 傾向検出完了", result_json

# --- Tab2: 傾向検出 JSON 版 ---
def detect_trends_with_forecast_json(
    process_name: str,
    datetime_str: str,
    window_minutes: int,
    forecast_minutes: int,
    downsample_factor: int
) -> dict:
    """
    MCP 用: 傾向検出を実行して JSON(dict) を返す
    """
    result_df, summary, result_json = detect_trends_with_forecast(
        process_name, datetime_str, window_minutes, forecast_minutes, downsample_factor
    )

    # JSON文字列を優先的にパース
    if result_json and isinstance(result_json, str):
        try:
            parsed = json.loads(result_json)
            return {
                "meta": {
                    "process_name": process_name,
                    "datetime_str": datetime_str,
                    "window_minutes": window_minutes,
                    "forecast_minutes": forecast_minutes,
                },
                "tab2": {
                    "summary": summary,
                    "results": parsed
                }
            }
        except Exception:
            pass

    # fallback
    return {
        "meta": {
            "process_name": process_name,
            "datetime_str": datetime_str,
            "window_minutes": window_minutes,
            "forecast_minutes": forecast_minutes,
        },
        "tab2": {
            "summary": summary if summary else "診断結果なし",
            "results": result_df.to_dict(orient="records") if result_df is not None else [],
        },
    }

# --- Tab3: 予兆解析 ---
def forecast_process_with_lag(process_name: str, datetime_str: str, forecast_minutes: int,
                              downsample_factor: int = 3
                              ) -> tuple[pd.DataFrame | None, str, str]:
    """
    ラグ行列を利用して予兆解析を行う。
    Args:
        process_name (str): プロセス名
        datetime_str (str): 基準日時
        forecast_minutes (int): 予測時間幅
        downsample_factor (int): サンプリング粗化倍率
    Returns:
        tuple: (解析結果DataFrame, サマリーテキスト, JSON文字列)
    """
    global DF, THRESHOLDS_DF, LAG_MATRIX
    if DF is None or THRESHOLDS_DF is None or LAG_MATRIX is None:
        return None, "⚠ ファイル未読み込み", None

    target_time = pd.to_datetime(datetime_str)
    forecast_time = target_time + pd.Timedelta(minutes=forecast_minutes)
    
    proc_thresholds = THRESHOLDS_DF[
        (THRESHOLDS_DF["ProcessNo_ProcessName"].apply(normalize) == normalize(process_name)) &
        (THRESHOLDS_DF["Important"] == True)
    ]
    if proc_thresholds.empty:
        return None, f"⚠ {process_name} の重要項目なし", None
    
    # ✅ normalize対応のインデックス探索
    matched_idx = None
    for idx in LAG_MATRIX.index:
        if normalize(idx) == normalize(process_name):
            matched_idx = idx
            break
    
    if matched_idx is None:
        return None, f"⚠ {process_name} のラグ行なし", None
    
    lag_row = LAG_MATRIX.loc[matched_idx].dropna()
    lag_row = lag_row[lag_row > 0]  # 正のラグのみ
    if lag_row.empty:
        return None, f"⚠ {process_name} に正のラグなし", None

    results = []
    for _, thr in proc_thresholds.iterrows():
        y_col = find_matching_column(DF, thr["ColumnID"], thr["ItemName"], thr["ProcessNo_ProcessName"])
        if y_col is None:
            continue

        # 学習データ（直近24h）
        df_window = DF[(DF["timestamp"] >= target_time - pd.Timedelta(hours=24)) &
                       (DF["timestamp"] <= target_time)].copy()
        if df_window.empty:
            continue

        base_df = df_window[["timestamp", y_col]].rename(columns={y_col: "y"})

        # === 粗化処理 ===
        dt = df_window["timestamp"].diff().median()
        base_min = max(int(dt.total_seconds() // 60), 1) if pd.notna(dt) else 1
        new_interval = max(base_min * int(downsample_factor), 1)

        df_down = (base_df.set_index("timestamp")
                           .resample(f"{new_interval}min").mean() 
                           .dropna()
                           .reset_index())

        merged_df = df_down.copy()
        for up_proc, lag_min in lag_row.items():
            up_cols = [c for c in DF.columns if isinstance(c, str) and up_proc in c]
            for x_col in up_cols:
                shifted = df_window.loc[:, ["timestamp", x_col]].copy()
                shifted["timestamp"] = shifted["timestamp"] + pd.Timedelta(minutes=lag_min)
                shifted = shifted.rename(columns={x_col: f"{x_col}_lag{lag_min}"})
                merged_df = pd.merge_asof(
                    merged_df.sort_values("timestamp"),
                    shifted.sort_values("timestamp"),
                    on="timestamp",
                    direction="nearest"
                )

        X_all = merged_df.drop(columns=["timestamp", "y"], errors="ignore").values
        Y_all = merged_df["y"].values
        if X_all.shape[1] == 0 or len(Y_all) < 5:
            continue

        model = LinearRegression().fit(X_all, Y_all)

        # 未来予測
        X_pred = []
        for up_proc, lag_min in lag_row.items():
            up_cols = [c for c in DF.columns if isinstance(c, str) and up_proc in c]
            for x_col in up_cols:
                ref_time = forecast_time - pd.Timedelta(minutes=lag_min)
                idx = (DF["timestamp"] - ref_time).abs().idxmin()
                X_pred.append(DF.loc[idx, x_col])
        if not X_pred:
            continue

        pred_val = model.predict([X_pred])[0]

        # 閾値リスク判定
        ll, l, h, hh = thr.get("LL"), thr.get("L"), thr.get("H"), thr.get("HH")
        risk = "OK"
        if pd.notna(ll) and pred_val <= ll:
            risk = "LOW-LOW"
        elif pd.notna(l) and pred_val <= l:
            risk = "LOW"
        elif pd.notna(hh) and pred_val >= hh:
            risk = "HIGH-HIGH"
        elif pd.notna(h) and pred_val >= h:
            risk = "HIGH"

        results.append({
            "ItemName": thr["ItemName"],
            "予測値": round(float(pred_val), 3),
            "予測時刻": str(forecast_time),
            "予測リスク": risk,
            "粗化間隔(分)": new_interval
        })

    result_df = pd.DataFrame(results)
    result_json = json.dumps(results, ensure_ascii=False, indent=2)
    return result_df, f"✅ {process_name} の予兆解析完了", result_json

# --- Tab3: 予兆解析 JSON 版 ---
def forecast_process_with_lag_json(
    process_name: str,
    datetime_str: str,
    forecast_minutes: int,
    downsample_factor: int = 3
) -> dict:
    """
    MCP 用: 予兆解析を実行して JSON(dict) を返す
    """
    result_df, summary, result_json = forecast_process_with_lag(
        process_name, datetime_str, forecast_minutes, downsample_factor
    )

    if result_json and isinstance(result_json, str):
        try:
            return json.loads(result_json)
        except Exception:
            pass

    return {
        "meta": {
            "process_name": process_name,
            "datetime_str": datetime_str,
            "forecast_minutes": forecast_minutes,
        },
        "tab3": {
            "summary": summary if summary else "診断結果なし",
            "results": result_df.to_dict(orient="records") if result_df is not None else [],
        },
    }

# --- Tab4: 変動解析 ---
def robust_mad(x: pd.Series) -> float:
    """
    Median Absolute Deviation (MAD) を計算する。
    Args:
        x (pd.Series): 数値系列
    Returns:
        float: MAD値
    """
    if len(x) == 0:
        return np.nan
    med = np.median(x)
    mad = np.median(np.abs(x - med))
    return 1.4826 * mad

def analyze_variability(datetime_str: str, window_minutes: int,
                        cv_threshold_pct: float = 10.0,
                        jump_pct_threshold: float = 10.0,
                        mad_sigma: float = 3.0
                        ) -> tuple[pd.DataFrame | None, str, str]:
    """
    変動解析を行い、不安定項目を検出する。
    Args:
        datetime_str (str): 基準日時
        window_minutes (int): 遡る時間幅
        cv_threshold_pct (float): CVしきい値 (%)
        jump_pct_threshold (float): ジャンプ率しきい値 (%)
        mad_sigma (float): MAD倍率
    Returns:
        tuple: (解析結果DataFrame, サマリーテキスト, JSON文字列)
    """
    global DF
    if DF is None:
        return None, "⚠ ファイル未読み込み", None

    target_time = pd.to_datetime(datetime_str)
    start_time = target_time - pd.Timedelta(minutes=window_minutes)
    end_time = target_time

    dfw = DF[(DF["timestamp"] >= start_time) & (DF["timestamp"] <= end_time)].copy()
    if dfw.empty:
        return None, f"⚠ 指定時間幅（{start_time}～{end_time}）にデータが見つかりません。", None

    data_cols = [c for c in dfw.columns if c != "timestamp" and pd.api.types.is_numeric_dtype(dfw[c])]
    results = []
    unstable_count = 0

    for col in data_cols:
        s = dfw[col].dropna()
        n = len(s)
        if n < 3:
            continue

        mean = float(np.mean(s))
        std = float(np.std(s, ddof=1)) if n >= 2 else 0.0
        cv_pct = abs(std / mean) * 100.0 if mean != 0 else np.nan

        diffs = s.diff().dropna()
        mad_scale = robust_mad(diffs)
        ref = max(1e-9, abs(float(np.median(s))))
        rel_jump = diffs.abs() / ref * 100.0

        abs_cond = diffs.abs() > (mad_sigma * mad_scale if not np.isnan(mad_scale) and mad_scale > 0 else np.inf)
        pct_cond = rel_jump >= jump_pct_threshold
        spike_mask = abs_cond | pct_cond

        spike_count = int(spike_mask.sum())
        spike_up_count = int((diffs[spike_mask] > 0).sum())
        spike_down_count = spike_count - spike_up_count
        max_step = float(diffs.abs().max()) if len(diffs) else np.nan

        last_val, first_val = float(s.iloc[-1]), float(s.iloc[0])
        unstable = (not np.isnan(cv_pct) and cv_pct >= cv_threshold_pct) or (spike_count > 0)
        if unstable:
            unstable_count += 1

        results.append({
            "項目名": col,
            "サンプル数": n,
            "平均": mean,
            "標準偏差": std,
            "CV(%)": None if np.isnan(cv_pct) else round(cv_pct, 3),
            "スパイク数": spike_count,
            "スパイク上昇数": spike_up_count,
            "スパイク下降数": spike_down_count,
            "最大|ステップ|": None if np.isnan(max_step) else round(max_step, 6),
            "最初の値": first_val,
            "最後の値": last_val,
            "不安定判定": bool(unstable)
        })

    result_df = pd.DataFrame(results)
    if not result_df.empty:
        result_df = result_df.sort_values(
            by=["不安定判定", "CV(%)", "スパイク数"],
            ascending=[False, False, False],
            na_position="last"
        ).reset_index(drop=True)

    summary = (
        f"✅ 変動解析完了（{start_time} ～ {end_time}）\n"
        f"- 対象項目数: {len(results)}\n"
        f"- 不安定と判定: {unstable_count} 項目（CV≥{cv_threshold_pct:.1f}% または スパイクあり）\n"
        f"- スパイク条件: |diff| > {mad_sigma:.1f}×MAD  または  相対変化 ≥ {jump_pct_threshold:.1f}%"
    )

    # NaN / inf を安全に処理して JSON 化
    safe_df = result_df.replace([np.inf, -np.inf], np.nan).fillna("null")
    result_json = json.dumps(safe_df.to_dict(orient="records"), ensure_ascii=False, indent=2)

    return result_df, summary, result_json

# --- Tab4: 変動解析 JSON 版 ---
def analyze_variability_json(
    datetime_str: str,
    window_minutes: int,
    cv_threshold_pct: float = 10.0,
    jump_pct_threshold: float = 10.0,
    mad_sigma: float = 3.0
) -> dict:
    """
    MCP 用: 変動解析を実行して JSON(dict) を返す。
    返すのは 項目名, 平均, CV(%), スパイク数, 不安定判定 のみ（上位10件）。
    """
    result_df, summary, _ = analyze_variability(
        datetime_str, window_minutes, cv_threshold_pct, jump_pct_threshold, mad_sigma
    )

    if result_df is not None and not result_df.empty:
        # 必要なカラムだけに絞り込んで上位10件
        top10 = (
            result_df.dropna(subset=["CV(%)"])
            .sort_values(by="CV(%)", ascending=False)
            .head(10)[["項目名", "平均", "CV(%)", "スパイク数", "不安定判定"]]
        )
        results = top10.to_dict(orient="records")
    else:
        results = []

    return result_df, summary, {
        "meta": {
            "datetime_str": datetime_str,
            "window_minutes": window_minutes,
            "cv_threshold_pct": cv_threshold_pct,
            "jump_pct_threshold": jump_pct_threshold,
            "mad_sigma": mad_sigma,
        },
        "tab4": {
            "summary": summary if summary else "診断結果なし",
            "top10_by_cv": results,
        },
    }

# UI初期化前にload_from_supabaseを呼び出し
load_from_supabase()

# --- Gradio UI ---
with gr.Blocks(css=".gradio-container {overflow: auto !important;}") as demo:
    gr.Markdown("## 統合トレンド解析アプリ (MCP対応, Supabase固定ファイル)")

    # --- プロセス入力欄 ---
    process_selector = gr.Dropdown(
        label="プロセス名 (UI + MCP)",
        choices=INITIAL_CHOICES,                # ← グローバル変数を参照
        allow_custom_value=True,
        interactive=True
    )

    # 共通入力欄
    with gr.Row():
        datetime_str_global = gr.DateTime(label="診断基準日時", value='2025-08-02 12:00:00', type='string')
        window_minutes_global = gr.Number(label="さかのぼる時間幅（分）", value=60)
        forecast_minutes_global = gr.Number(label="未来予測時間幅（分）", value=60)

    # Tabs
    with gr.Tabs():
        with gr.Tab("閾値診断"):
            run_btn1 = gr.Button("診断実行")
            result_df_all = gr.Dataframe(label="全項目の状態集計")
            result_df_imp = gr.Dataframe(label="重要項目全体の状態集計")
            result_df_imp_items = gr.Dataframe(label="重要項目ごとの状態集計")
            summary_output = gr.Textbox(label="サマリー")
            json_output = gr.Json(label="JSON集計結果")

            # ✅ UI用
            run_btn1.click(
                diagnose_process_range_ui,
                inputs=[process_selector, datetime_str_global, window_minutes_global],
                outputs=[result_df_all, result_df_imp, result_df_imp_items, summary_output]
            )

            # ✅ MCP用 (UIへの出力は不要にする)
            run_btn1.click(
                diagnose_process_range_json,
                inputs=[process_selector, datetime_str_global, window_minutes_global],
                outputs=[json_output]
            )

        with gr.Tab("傾向検出"):
            downsample_factor2 = gr.Slider(label="粗化倍率", minimum=1, maximum=10, step=1, value=3)
            run_btn2 = gr.Button("傾向検出実行")
            result_df2 = gr.Dataframe(label="傾向＋予測結果")
            summary_output2 = gr.Textbox(label="サマリー")
            json_output2 = gr.Json(label="JSON結果")

            # ✅ UI用
            run_btn2.click(
                detect_trends_with_forecast,
                inputs=[process_selector, datetime_str_global, window_minutes_global, forecast_minutes_global, downsample_factor2],
                outputs=[result_df2, summary_output2, json_output2]
            )

            # ✅ MCP用
            run_btn2.click(
                detect_trends_with_forecast_json,
                inputs=[process_selector, datetime_str_global, window_minutes_global, forecast_minutes_global, downsample_factor2],
                outputs=[json_output2]
            )

        with gr.Tab("予兆解析"):
            downsample_factor3 = gr.Slider(1, 10, value=3, step=1, label="粗化倍率（サンプリング間引き）")
            run_btn3 = gr.Button("予兆解析実行")
            result_df3 = gr.Dataframe(label="予兆解析結果")
            summary_output3 = gr.Textbox(label="サマリー")
            json_output3 = gr.Json(label="JSON結果")

            # ✅ UI用
            run_btn3.click(
                forecast_process_with_lag,
                inputs=[process_selector, datetime_str_global, forecast_minutes_global, downsample_factor3],
                outputs=[result_df3, summary_output3, json_output3]
            )

            # ✅ MCP用
            run_btn3.click(
                forecast_process_with_lag_json,
                inputs=[process_selector, datetime_str_global, forecast_minutes_global, downsample_factor3],
                outputs=[json_output3]
            )

        with gr.Tab("変動解析"):
            cv_threshold4 = gr.Number(label="CVしきい値(%)", value=10.0)
            jump_threshold4 = gr.Number(label="ジャンプ率しきい値(%)", value=10.0)
            mad_sigma4 = gr.Number(label="MAD倍率", value=3.0)
            run_btn4 = gr.Button("変動解析実行")

            result_df4 = gr.Dataframe(label="変動解析結果")
            summary_output4 = gr.Textbox(label="サマリー")
            json_output4 = gr.Json(label="JSON結果")

            # ✅ UI + MCP 共通
            run_btn4.click(
                analyze_variability_json,
                inputs=[datetime_str_global, window_minutes_global, cv_threshold4, jump_threshold4, mad_sigma4],
                outputs=[result_df4, summary_output4, json_output4]
            )

if __name__ == "__main__":
    demo.launch(mcp_server=True)