HC900-Crawler/scripts/analysis/gen_temp_profiles.py

"""
컬럼 온도 프로파일 기준 산출 → c{prefix}_tempref.json (이격 모니터 레이어①).

근거: 메모리 §17 — 같은 제품·진공이면 단별 온도(reb-A>T_B>T_C>T_D)가 부하무관 불변.
  제품 식별 = 온도프로파일(reb_temp 클러스터). 진공 종속이라 vacuum 기준 동시 산출.

안정구간 PROD에서 제품별 단별 median/σ + 진공 median → 기준밴드(±2σ).
오퍼레이터 이격 모니터/품질 게이트의 reference.
"""
import argparse
import json
import os
import numpy as np
import pandas as pd
from sklearn.cluster import KMeans

BASE = os.path.dirname(os.path.abspath(__file__))
STAGES = ["reb_temp", "T_B", "T_C", "T_D"]   # 보텀(리보일러) → 탑상
PROD_MIN = 50                                 # feed 하한
MIN_CLUSTER = 0.05                            # 클러스터 최소 비중(5%)


def cluster_products(reb):
    """reb_temp 1D 클러스터 = 제품 식별. 인접 중심차 <2℃면 병합(과분할 방지)."""
    reb = reb.reshape(-1, 1)
    for k in (3, 2, 1):
        km = KMeans(k, n_init=10, random_state=0).fit(reb)
        cen = np.sort(km.cluster_centers_.ravel())
        sizes = np.bincount(km.labels_, minlength=k) / len(reb)
        if k == 1:
            return km
        if all(cen[i+1]-cen[i] >= 2.0 for i in range(k-1)) and sizes.min() >= MIN_CLUSTER:
            return km
    return KMeans(1, n_init=10, random_state=0).fit(reb)


def build(prefix, stable_from=None, stable_to=None):
    pkl = os.path.join(BASE, f"c{prefix}_data.pkl")
    if prefix == "61" and not os.path.exists(pkl):
        pkl = os.path.join(BASE, "c6111_data.pkl")
    if not os.path.exists(pkl):
        print(f"  [skip] {prefix}: {pkl} 없음")
        return None
    df = pd.read_pickle(pkl)
    df = df[df["mode"] == "PROD"].copy()
    df = df[(df["feed"] > PROD_MIN) & df[STAGES + ["vacuum"]].notna().all(axis=1)]
    if stable_from:
        df = df[df["dtat"] >= pd.Timestamp(stable_from)]
    if stable_to:
        df = df[df["dtat"] <= pd.Timestamp(stable_to)]
    if len(df) < 200:
        print(f"  [skip] {prefix}: 안정 PROD 부족 ({len(df)}행)")
        return None

    km = cluster_products(df["reb_temp"].values)
    df["prod"] = km.labels_
    order = df.groupby("prod")["reb_temp"].median().sort_values().index  # 저온→고온
    products = []
    for rank, pid in enumerate(order):
        g = df[df["prod"] == pid]
        stages = {s: {"median": round(float(g[s].median()), 2),
                      "std": round(float(g[s].std()), 2)} for s in STAGES}
        products.append({
            "label": f"P{rank}",                      # 추후 PM/PGMEA/EL 등 화학 라벨 매핑
            "n_rows": len(g),
            "span_AD": round(float((g["reb_temp"] - g["T_D"]).median()), 2),
            "vacuum": {"median": round(float(g["vacuum"].median()), 2),
                       "std": round(float(g["vacuum"].std()), 2)},
            "stages": stages,
        })
    ref = {"column": f"c{prefix}", "stages_order": STAGES,
           "n_products": len(products),
           "period": f"{df['dtat'].min():%Y-%m-%d}~{df['dtat'].max():%Y-%m-%d}",
           "products": products}
    out = os.path.join(BASE, f"c{prefix}_tempref.json")
    with open(out, "w") as f:
        json.dump(ref, f, indent=2, ensure_ascii=False)
    print(f"  c{prefix}: 제품 {len(products)}개  ", end="")
    for p in products:
        s = p["stages"]
        print(f"[{p['label']} reb{s['reb_temp']['median']:.1f}/Tc{s['T_C']['median']:.1f}/"
              f"Td{s['T_D']['median']:.1f} vac{p['vacuum']['median']:.0f} n{p['n_rows']}]", end=" ")
    print(f"→ {os.path.basename(out)}")
    return ref


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--prefix", help="단일 컬럼 (생략시 전체)")
    ap.add_argument("--from", dest="stable_from", help="안정구간 시작 YYYY-MM-DD")
    ap.add_argument("--to", dest="stable_to", help="안정구간 끝")
    args = ap.parse_args()
    prefixes = [args.prefix] if args.prefix else ["61", "62", "81", "91", "92", "101", "102"]
    for p in prefixes:
        build(p, args.stable_from, args.stable_to)


if __name__ == "__main__":
    main()