Implement initial project structure and setup

app.py

@@ -0,0 +1,478 @@
+import gradio as gr
+import pandas as pd
+import numpy as np
+import os
+import re
+from typing import Dict, Tuple, List, Optional
+import plotly.graph_objects as go
+
+# ======================================
+# 設定（添付CSVの既定パス：必要に応じて変更可）
+# ======================================
+DEFAULT_CSV_PATH = "/mnt/data/mock_data_id_9999.csv"
+
+# ======================================
+# ユーティリティ
+# ======================================
+def normalize(s: str) -> str:
+    return str(s).replace("\u3000", " ").replace("\n", "").replace("\r", "").strip()
+
+def try_read_csv_3header(path_or_file) -> pd.DataFrame:
+    """
+    3行ヘッダーCSVを読み込む（cp932/utf-8-sig フォールバック）。
+    1列目は timestamp として datetime 変換。
+    2列目以降は (ID, ItemName, ProcessName) の3段。
+    """
+    last_err = None
+    for enc in ["cp932", "utf-8-sig", "utf-8"]:
+        try:
+            df = pd.read_csv(path_or_file, header=[0, 1, 2], encoding=enc)
+            break
+        except Exception as e:
+            last_err = e
+            df = None
+    if df is None:
+        raise last_err
+
+    # 先頭列を timestamp に
+    ts = pd.to_datetime(df.iloc[:, 0], errors="coerce")
+    df = df.drop(df.columns[0], axis=1)
+    df.insert(0, "timestamp", ts)
+
+    # 列名はタプルのまま保持（timestampは str）
+    # ただし内部処理用に文字列連結も作成できるように関数を用意
+    return df
+
+def col_tuple_to_str(col) -> str:
+    if isinstance(col, tuple):
+        return "_".join([str(x) for x in col if x])
+    return str(col)
+
+def build_index_maps(df: pd.DataFrame):
+    """
+    プロセス（3行目＝タプルの3つ目）→ 該当列情報 の辞書を作る。
+    各列は (col_tuple, id, item, process, col_str)
+    """
+    process_map = {}
+    for col in df.columns:
+        if col == "timestamp":
+            continue
+        if isinstance(col, tuple) and len(col) >= 3:
+            col_id, item_name, process_name = str(col[0]), str(col[1]), str(col[2])
+        else:
+            # 非タプル（安全策）
+            parts = str(col).split("_")
+            if len(parts) >= 3:
+                col_id, item_name, process_name = parts[0], "_".join(parts[1:-1]), parts[-1]
+            else:
+                # プロセスが分からない列はスキップ
+                continue
+        rec = {
+            "col_tuple": col,
+            "id": col_id,
+            "item": item_name,
+            "process": process_name,
+            "col_str": col_tuple_to_str(col),
+        }
+        process_map.setdefault(process_name, []).append(rec)
+    # プロセス候補・アイテム候補を返すために使う
+    processes = sorted(list(process_map.keys()), key=lambda x: normalize(x))
+    return process_map, processes
+
+def extract_measure_tag(item_name: str) -> str:
+    """
+    項目名末尾の計測項目タグを抽出。
+    例:
+      "処理水 有機物 分析値 [mg/L]" → "mg/L"
+      "原水 TOC" → "TOC"
+      "導電率（電気伝導度） [mS/cm]" → "mS/cm"
+    優先順:
+      1) [...] の中身
+      2) 全角/半角スペース区切りの末尾語（英字混在や記号含む）
+    """
+    s = normalize(item_name)
+    m = re.search(r"\[([^\[\]]+)\]\s*$", s)
+    if m:
+        return m.group(1).strip()
+    # 角括弧がなければ末尾語
+    tokens = re.split(r"\s+", s)
+    if tokens:
+        return tokens[-1]
+    return s
+
+# ======================================
+# しきい値ハンドリング
+# ======================================
+def try_read_thresholds_excel(file) -> Optional[pd.DataFrame]:
+    """
+    しきい値Excel（任意）を読み込み。
+    想定カラム: ColumnID, ItemName, ProcessNo_ProcessName, LL, L, H, HH, Important(任意)
+    """
+    if file is None:
+        return None
+    df = pd.read_excel(file)
+    df.columns = [normalize(c) for c in df.columns]
+    # 必須カラム確認（最低限）
+    needed = {"ColumnID", "ItemName", "ProcessNo_ProcessName"}
+    if not needed.issubset(set(df.columns)):
+        # 列名が違う場合の簡易吸収
+        rename_map = {}
+        for k in list(df.columns):
+            nk = normalize(str(k))
+            if nk.lower() in ["columnid", "colid", "id"]:
+                rename_map[k] = "ColumnID"
+            elif nk.lower() in ["itemname", "item", "name"]:
+                rename_map[k] = "ItemName"
+            elif nk.lower() in ["processno_processname", "process", "processname"]:
+                rename_map[k] = "ProcessNo_ProcessName"
+        if rename_map:
+            df = df.rename(columns=rename_map)
+    # 数値化
+    for c in ["LL", "L", "H", "HH"]:
+        if c in df.columns:
+            df[c] = pd.to_numeric(df[c], errors="coerce")
+    if "Important" in df.columns:
+        df["Important"] = (
+            df["Important"].astype(str).str.upper().map({"TRUE": True, "FALSE": False})
+        )
+    return df
+
+def build_threshold_lookup(thr_df: Optional[pd.DataFrame]) -> Dict[Tuple[str, str, str], Tuple[float, float, float, float]]:
+    """
+    キー: (ColumnID, ItemName, ProcessNo_ProcessName) → (LL, L, H, HH)
+    """
+    lookup = {}
+    if thr_df is None or thr_df.empty:
+        return lookup
+    for _, r in thr_df.iterrows():
+        colid = normalize(str(r.get("ColumnID", "")))
+        item = normalize(str(r.get("ItemName", "")))
+        proc = normalize(str(r.get("ProcessNo_ProcessName", "")))
+        LL = r.get("LL", np.nan)
+        L  = r.get("L",  np.nan)
+        H  = r.get("H",  np.nan)
+        HH = r.get("HH", np.nan)
+        lookup[(colid, item, proc)] = (LL, L, H, HH)
+    return lookup
+
+def auto_threshold(series: pd.Series) -> Tuple[float, float, float, float]:
+    """
+    自動しきい値: mean ± std（LL/L/H/HH の2段に同じ幅を割当）
+    例: L=mean-std, LL=mean-2std, H=mean+std, HH=mean+2std
+    """
+    s = series.dropna()
+    if len(s) < 5:
+        return (np.nan, np.nan, np.nan, np.nan)
+    m = float(s.mean())
+    sd = float(s.std(ddof=1)) if len(s) >= 2 else 0.0
+    return (m - 2*sd, m - sd, m + sd, m + 2*sd)
+
+def judge_status(value, LL, L, H, HH) -> str:
+    if pd.notna(LL) and value <= LL:
+        return "LL"
+    if pd.notna(L) and value <= L:
+        return "L"
+    if pd.notna(HH) and value >= HH:
+        return "HH"
+    if pd.notna(H) and value >= H:
+        return "H"
+    return "OK"
+
+# カラー（点の色）：閾値逸脱を強調
+STATUS_COLOR = {
+    "LL": "#2b6cb0",   # 青系
+    "L":  "#63b3ed",   # 水色
+    "OK": "#a0aec0",   # グレー
+    "H":  "#f6ad55",   # 橙
+    "HH": "#e53e3e",   # 赤
+}
+
+# 線色（系列ライン）：列ごとに安定色
+LINE_COLOR = "#4a5568"  # 濃いグレー
+
+# ======================================
+# 図作成
+# ======================================
+def make_trend_figs(
+    df: pd.DataFrame,
+    process_map: Dict[str, List[dict]],
+    process_name: str,
+    selected_items: List[str],
+    thr_df: Optional[pd.DataFrame],
+    thr_mode: str,  # "excel" or "auto"
+    date_min: Optional[str] = None,
+    date_max: Optional[str] = None,
+) -> List[go.Figure]:
+    """
+    計測項目タグごと（extract_measure_tag）に図を分けて生成。
+    selected_items は「2行目（ItemName）」の値。
+    """
+    if df is None or process_name is None or process_name == "":
+        return []
+
+    # 対象プロセスの列レコード
+    recs = process_map.get(process_name, [])
+    if not recs:
+        return []
+
+    # 2行目（ItemName）で絞り込み
+    selected_items_set = set([normalize(x) for x in (selected_items or [])])
+    recs = [r for r in recs if normalize(r["item"]) in selected_items_set]
+    if not recs:
+        return []
+
+    # 日付範囲フィルタ
+    dfw = df.copy()
+    if date_min:
+        dfw = dfw[dfw["timestamp"] >= pd.to_datetime(date_min)]
+    if date_max:
+        dfw = dfw[dfw["timestamp"] <= pd.to_datetime(date_max)]
+    if dfw.empty:
+        return []
+
+    # しきい値参照
+    thr_lookup = build_threshold_lookup(thr_df) if thr_mode == "excel" else {}
+
+    # 測定項目タグごとにグループ化
+    groups: Dict[str, List[dict]] = {}
+    for r in recs:
+        tag = extract_measure_tag(r["item"])
+        groups.setdefault(tag, []).append(r)
+
+    figs = []
+    for tag, cols in groups.items():
+        fig = go.Figure()
+        # 各列を描画
+        for r in cols:
+            col = r["col_tuple"]
+            col_str = r["col_str"]
+            if col not in dfw.columns:
+                # まれにヘッダー崩れなど
+                if col_str in dfw.columns:
+                    series = dfw[col_str]
+                else:
+                    continue
+            else:
+                series = dfw[col]
+
+            # 値
+            x = dfw["timestamp"]
+            y = pd.to_numeric(series, errors="coerce")
+
+            # しきい値決定
+            if thr_mode == "excel":
+                key = (normalize(r["id"]), normalize(r["item"]), normalize(r["process"]))
+                LL, L, H, HH = thr_lookup.get(key, (np.nan, np.nan, np.nan, np.nan))
+                # Excelに見つからない場合は自動にフォールバック
+                if all(pd.isna(v) for v in [LL, L, H, HH]):
+                    LL, L, H, HH = auto_threshold(y)
+            else:
+                LL, L, H, HH = auto_threshold(y)
+
+            # 状態ごとに点色を決める
+            colors = []
+            for v in y:
+                if pd.isna(v):
+                    colors.append("rgba(0,0,0,0)")
+                else:
+                    st = judge_status(v, LL, L, H, HH)
+                    colors.append(STATUS_COLOR.get(st, STATUS_COLOR["OK"]))
+
+            # 下地のライン（視認性のため薄色）
+            fig.add_trace(go.Scatter(
+                x=x, y=y, mode="lines",
+                name=f"{r['item']} ({r['id']})",
+                line=dict(color=LINE_COLOR, width=1.5),
+                hovertemplate="%{x}<br>%{y}<extra>"+f"{r['item']} ({r['id']})"+"</extra>"
+            ))
+            # 色付きマーカーで逸脱強調
+            fig.add_trace(go.Scatter(
+                x=x, y=y, mode="markers",
+                name=f"{r['item']} markers",
+                marker=dict(size=6, color=colors),
+                showlegend=False,
+                hovertemplate="%{x}<br>%{y}<extra></extra>"
+            ))
+
+            # しきい値ガイド（あれば）
+            def add_hline(val, label):
+                if pd.notna(val):
+                    fig.add_hline(y=float(val), line=dict(width=1, dash="dot"),
+                                  annotation_text=label, annotation_position="top left")
+
+            add_hline(LL, "LL")
+            add_hline(L,  "L")
+            add_hline(H,  "H")
+            add_hline(HH, "HH")
+
+        fig.update_layout(
+            title=f"{process_name} ｜ 計測項目: {tag}",
+            xaxis_title="timestamp",
+            yaxis_title=tag,
+            legend_title="系列",
+            margin=dict(l=10, r=10, t=40, b=10),
+            hovermode="x unified",
+        )
+        figs.append(fig)
+
+    return figs
+
+# ======================================
+# グローバル状態（UI間共有）
+# ======================================
+G_DF: Optional[pd.DataFrame] = None
+G_PROCESS_MAP = {}
+G_PROCESSES = []
+G_THRESHOLDS_DF: Optional[pd.DataFrame] = None
+
+# ======================================
+# コールバック
+# ======================================
+def initialize_default_csv():
+    """
+    起動時にデフォルトCSVが存在すれば読み込む。
+    """
+    global G_DF, G_PROCESS_MAP, G_PROCESSES
+    if os.path.exists(DEFAULT_CSV_PATH):
+        try:
+            df = try_read_csv_3header(DEFAULT_CSV_PATH)
+            G_DF = df
+            G_PROCESS_MAP, G_PROCESSES = build_index_maps(df)
+            return f"✅ 既定CSVを読み込みました: {DEFAULT_CSV_PATH}", gr.update(choices=G_PROCESSES, value=(G_PROCESSES[0] if G_PROCESSES else None)), G_PROCESSES
+        except Exception as e:
+            return f"⚠ 既定CSV読み込み失敗: {e}", gr.update(), []
+    return "ℹ CSVをアップロードしてください。", gr.update(), []
+
+def on_csv_upload(file):
+    """
+    CSVアップロード → パース → プロセス候補更新
+    """
+    global G_DF, G_PROCESS_MAP, G_PROCESSES
+    if file is None:
+        return "⚠ ファイルが選択されていません。", gr.update(choices=[]), []
+    try:
+        df = try_read_csv_3header(file.name if hasattr(file, "name") else file)
+        G_DF = df
+        G_PROCESS_MAP, G_PROCESSES = build_index_maps(df)
+        return f"✅ CSV読み込み: {df.shape[0]}行 × {df.shape[1]}列", gr.update(choices=G_PROCESSES, value=(G_PROCESSES[0] if G_PROCESSES else None)), G_PROCESSES
+    except Exception as e:
+        return f"❌ 読み込みエラー: {e}", gr.update(choices=[]), []
+
+def on_thr_upload(file):
+    """
+    しきい値Excelアップロード → メモリ更新
+    """
+    global G_THRESHOLDS_DF
+    if file is None:
+        G_THRESHOLDS_DF = None
+        return "ℹ しきい値ファイルなし（自動しきい値が使われます）"
+    try:
+        thr = try_read_thresholds_excel(file.name if hasattr(file, "name") else file)
+        G_THRESHOLDS_DF = thr
+        return f"✅ しきい値を読み込みました（{thr.shape[0]}件）"
+    except Exception as e:
+        G_THRESHOLDS_DF = None
+        return f"❌ しきい値読み込みエラー: {e}"
+
+def update_items(process_name: str):
+    """
+    プロセス選択に応じて、項目（2行目）候補を返す。
+    """
+    if not process_name or process_name not in G_PROCESS_MAP:
+        return gr.update(choices=[], value=[])
+    items = sorted(list({rec["item"] for rec in G_PROCESS_MAP[process_name]}), key=lambda x: normalize(x))
+    # デフォルトは全選択
+    return gr.update(choices=items, value=items)
+
+def render_figs(process_name: str, items: List[str], thr_mode: str, date_min, date_max):
+    """
+    図を生成して返す（複数図）。GradioではList[plotly.Figure]を直接返せる。
+    """
+    if G_DF is None:
+        return "⚠ データ未読み込み", []
+    if not process_name:
+        return "⚠ プロセスを選択してください", []
+    if not items:
+        return "⚠ 項目を選択してください", []
+
+    figs = make_trend_figs(
+        G_DF, G_PROCESS_MAP, process_name, items, G_THRESHOLDS_DF, thr_mode, date_min, date_max
+    )
+    if not figs:
+        return "⚠ 図を生成できませんでした（データ無し or 条件不一致）", []
+    return f"✅ {process_name}: {len(figs)}枚のトレンド図を生成しました（計測項目タグごと）", figs
+
+# ======================================
+# UI
+# ======================================
+with gr.Blocks(css="""
+.gradio-container {overflow: auto !important;}
+""") as demo:
+    gr.Markdown("## トレンドグラフ専用アプリ（3行ヘッダー対応・プロセス別・計測項目タグ別・閾値色分け）")
+
+    with gr.Row():
+        csv_uploader = gr.File(label="① 時系列CSV（3行ヘッダー）", file_count="single", file_types=[".csv"])
+        thr_uploader = gr.File(label="② 閾値Excel（任意: LL/L/H/HH）", file_count="single", file_types=[".xlsx", ".xls"])
+
+    with gr.Row():
+        thr_mode = gr.Radio(
+            ["excel（アップロード優先・無ければ自動）", "自動（平均±標準偏差）"],
+            value="excel（アップロード優先・無ければ自動）",
+            label="しきい値モード"
+        )
+        date_min = gr.Textbox(label="抽出開始日時（任意）例: 2024-07-01 00:00")
+        date_max = gr.Textbox(label="抽出終了日時（任意）例: 2024-07-31 23:59")
+
+    status_csv = gr.Markdown()
+    status_thr = gr.Markdown()
+
+    process_dd = gr.Dropdown(label="対象プロセス（3行ヘッダーの3行目）", choices=[])
+    items_cb = gr.CheckboxGroup(label="表示する項目（3行ヘッダーの2行目）", choices=[], value=[])
+
+    with gr.Row():
+        btn_render = gr.Button("トレンド図を生成", variant="primary")
+
+    msg = gr.Markdown()
+    plots = gr.Plotly(label="トレンド図（計測項目タグごとに複数枚）", height=540, every=1, interactive=True, show_label=True, scale=100, container=True, visible=True, elem_id="plot_container", elem_classes=["w-full"], )
+
+    # コールバック接続
+    # 1) 既定CSVの自動ロード
+    init_msg, init_proc_update, _ = initialize_default_csv()
+    status_csv.value = init_msg
+    process_dd.value = init_proc_update.value
+    process_dd.choices = init_proc_update.choices
+
+    # 2) CSVアップロードで更新
+    csv_uploader.change(
+        on_csv_upload,
+        inputs=[csv_uploader],
+        outputs=[status_csv, process_dd, gr.State()],
+    )
+
+    # 3) 閾値アップロードで更新
+    thr_uploader.change(
+        on_thr_upload,
+        inputs=[thr_uploader],
+        outputs=[status_thr],
+    )
+
+    # 4) プロセス選択で項目候補更新
+    process_dd.change(
+        update_items,
+        inputs=[process_dd],
+        outputs=[items_cb],
+    )
+
+    # 5) 図生成
+    def _thr_mode_key(s):
+        return "excel" if s.startswith("excel") else "auto"
+
+    btn_render.click(
+        fn=lambda proc, items, mode, dmin, dmax: render_figs(proc, items, _thr_mode_key(mode), dmin, dmax),
+        inputs=[process_dd, items_cb, thr_mode, date_min, date_max],
+        outputs=[msg, plots],
+    )
+
+if __name__ == "__main__":
+    # gradio>=4 で Plotly がそのままレンダリング可能
+    demo.launch()