app.py

# 閾値診断アプリ Gradio + MCP対応版

import gradio as gr
import pandas as pd
import numpy as np
import json
import os

# --- ユーティリティ ---
def judge_status(value, ll, l, h, hh):
    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 convert_value(v):
    if hasattr(v, "item"):
        return v.item()
    return float(v) if isinstance(v, (np.floating, float)) else int(v) if isinstance(v, (np.integer, int)) else v

# --- 診断関数 ---
def diagnose_process_range(csv_file, excel_file, process_name, datetime_str, window_minutes):
    try:
        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")
    except Exception as e:
        return None, None, None, f"❌ 入力ファイルの読み込みに失敗しました: {e}", None

    try:
        target_time = pd.to_datetime(datetime_str)
    except Exception:
        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

    proc_thresholds = thresholds_df[thresholds_df["ProcessNo_ProcessName"] == 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_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"],
                "判定": 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 [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 "対象データなし"
        )
    )

    json_data = {
        "集計結果": {
            "全項目割合": {k: convert_value(v) for k, v in status_ratio.to_dict().items()},
            "重要項目全体割合": {k: convert_value(v) for k, v in status_ratio_imp.to_dict().items()} if not result_df_imp.empty else {},
            "重要項目ごと割合": [
                {k: convert_value(v) for k, v in row.items()} 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

# --- Gradio UI ---
with gr.Blocks() as demo:
    gr.Markdown("## 閾値診断アプリ (MCP対応)")

    with gr.Row():
        csv_input = gr.File(label="CSVファイルをアップロード", file_types=[".csv"], type="filepath")
        excel_input = gr.File(label="Excel閾値ファイルをアップロード", file_types=[".xlsx"], type="filepath")

    process_name = gr.Textbox(label="プロセス名", value="E018-A012_除害RO")
    datetime_str = gr.Textbox(label="診断基準日時", value="2025/8/1 1:05")
    window_minutes = gr.Number(label="さかのぼる時間幅（分）", value=60)

    run_btn = 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集計結果")

    run_btn.click(
        diagnose_process_range,
        inputs=[csv_input, excel_input, process_name, datetime_str, window_minutes],
        outputs=[result_df_all, result_df_imp, result_df_imp_items, summary_output, json_output]
    )

if __name__ == "__main__":
    use_mcp = os.getenv("USE_MCP", "0") == "1"
    if use_mcp:
        demo.launch(mcp_server=True)
    else:
        demo.launch(server_name="0.0.0.0", share=False)