HC900-Crawler/scripts/analysis/run_column.py

"""
형제 컬럼 확장(작업1) 일괄 실행 래퍼.

사용법:
  python3 run_column.py --prefix 62               # 6-2차 단독
  python3 run_column.py --prefix 81 --asset /ASSETS/P8
  python3 run_column.py --all                     # 모든 형제 컬럼
"""
import argparse
import subprocess
import sys
import os
import psycopg
import pandas as pd

BASE = os.path.dirname(os.path.abspath(__file__))

COLUMNS = [
    ("51", "/ASSETS/P5", "C-5111 (5차)"),          # 측류 솔벤트, T_C 센서 부재(대체)
    ("61", "/ASSETS/P6", "C-6111 (6-1차)"),
    ("62", "/ASSETS/P6", "C-6211 (6-2차)"),
    ("81", "/ASSETS/P8", "C-8111 (8차, 단일 train)"),
    ("91", "/ASSETS/P9", "C-9111 (9-1차)"),
    ("92", "/ASSETS/P9", "C-9211 (9-2차)"),
    ("101", "/ASSETS/P10", "C-10111 (10-1차) ※본 dump엔 시운전기간 多, 생산샘플 적음"),
    ("102", "/ASSETS/P10", "C-10211 (10-2차) ※첫생산 5/26~, 이후 데이터 유효"),
]
# ※ 10차는 본 dump 창(2026-02~06) 대부분이 시운전이라 PROD 샘플이 적음(10-1 2%, 10-2 5/26~).
#    단 2026-06 현재 정상 생산 진입 — 데이터 누적되면(또는 최신 dump 재반입) 정상 분석 대상.
#    현재는 생산구간만 추려 쓰되 샘플 부족에 주의(맵 R²·트리거 변동성 클 수 있음).

PREFIX_ASSET = {p: a for p, a, _ in COLUMNS}

DSN = "host=localhost port=5432 dbname=field_hist user=postgres password=postgres"
PY = sys.executable


def extract(prefix, asset):
    """추출 + 운전모드 분류. c{prefix}_data.pkl 저장."""
    from c6111_extract import roles_for, tag_frame, classify_phases

    with psycopg.connect(DSN) as conn:
        roles = roles_for(prefix, asset)
        print(f"\n  ROLES ({len(roles)}):")
        for k, v in roles.items():
            print(f"    {k:15s} -> {v}")
        df = tag_frame(conn, roles, asset)

    df["mode"] = classify_phases(df)
    out = os.path.join(BASE, f"c{prefix}_data.pkl")
    df.to_pickle(out)

    print(f"\n=== {prefix} ({asset}) ===")
    print(f"  행수={len(df)}  기간={df.dtat.min()} ~ {df.dtat.max()}")
    vc = df["mode"].value_counts()
    for m, n in vc.items():
        print(f"  {m:9s} {n:7d}  {100*n/len(df):5.1f}%  ≈ {n*30/3600:.1f}h")
    print(f"  저장: {out}")
    return out


def run_analysis(script, prefix):
    """분석 스크립트 1개 실행 (subprocess)."""
    data = os.path.join(BASE, f"c{prefix}_data.pkl")
    cmd = [PY, os.path.join(BASE, script), "--data", data, "--prefix", f"c{prefix}"]
    print(f"\n>>> {' '.join(cmd)}")
    r = subprocess.run(cmd)
    return r.returncode


def run_column(prefix, asset, label):
    """컬럼 1개 전체 파이프라인."""
    print(f"\n{'='*60}")
    print(f"  {label}  (prefix={prefix}, asset={asset})")
    print(f"{'='*60}")
    extract(prefix, asset)
    for script in ["c6111_prodmap.py", "c6111_shadow.py", "c6111_rolling.py", "c6111_startup.py", "c6111_shutdown.py", "c6111_operator_assist.py", "c6111_export_model.py"]:
        rc = run_analysis(script, prefix)
        if rc != 0:
            print(f"  [WARN] {script} → exit {rc}")


def compare():
    """모든 컬럼 결과 취합 → 비교표 (prodmap + shadow + startup)."""
    import numpy as np

    rows = []
    for prefix, asset, label in COLUMNS:
        pkl = os.path.join(BASE, f"c{prefix}_data.pkl")
        # 6-1 legacy: c6111_data.pkl (not c61_data.pkl)
        if prefix == "61" and not os.path.exists(pkl):
            alt = os.path.join(BASE, "c6111_data.pkl")
            if os.path.exists(alt):
                pkl = alt
        if not os.path.exists(pkl):
            print(f"  [skip] {label}: {pkl} 없음")
            continue
        df = pd.read_pickle(pkl)
        prod = df[df["mode"] == "PROD"]
        steam_feed = prod["steam_flow"].median() / prod["feed"].median() if len(prod) else float("nan")
        total_h = len(df) * 30 / 3600
        prod_h = len(prod) * 30 / 3600

        # 컷인 탐지 (startup.py detect_cutins 로직 인라인)
        prod_arr = df["product"].values
        reb_arr = df["reb_temp"].values
        dtat_vals = df["dtat"].values
        cutins = []
        i = 60
        n = len(df)
        while i < n:
            if prod_arr[i] > 100 and prod_arr[i-1] <= 100:
                pre = prod_arr[max(0, i-60):i]
                if np.nanmedian(pre) < 50 and reb_arr[i] > 75:
                    cutins.append(i)
                    i += 720
                    continue
            i += 1

        row = {"컬럼": label,
               "기간": f"{df['dtat'].min():%m-%d}~{df['dtat'].max():%m-%d}",
               "전체(h)": f"{total_h:.0f}",
               "PROD%": f"{100*len(prod)/len(df):.1f}",
               "생산(h)": f"{prod_h:.0f}",
               "steam/feed": f"{steam_feed:.3f}",
               "컷인": str(len(cutins))}

        if cutins:
            cutin_data = []
            for ci in cutins:
                cutin_data.append({"reb": df.loc[ci, "reb_temp"],
                                   "tc": df.loc[ci, "T_C"],
                                   "dT": df.loc[ci, "reb_temp"] - df.loc[ci, "T_D"]})
            cdf = pd.DataFrame(cutin_data)
            row["컷인_reb-A"] = f"{cdf['reb'].median():.1f}±{cdf['reb'].std():.1f}"
            row["컷인_dT_AD"] = f"{cdf['dT'].median():.1f}±{cdf['dT'].std():.1f}"
        else:
            row["컷인_reb-A"] = ""
            row["컷인_dT_AD"] = ""

        rows.append(row)

    pd.set_option("display.width", 300)
    pd.set_option("display.max_columns", 20)
    print("\n\n" + "="*120)
    print("  형제 컬럼 비교표")
    print("="*120)
    tbl = pd.DataFrame(rows).set_index("컬럼")
    print(tbl.to_string())


def main():
    parser = argparse.ArgumentParser(description="형제 컬럼 확장 일괄 실행")
    parser.add_argument("--prefix", help="컬럼 prefix (61, 62, 81, 91, 101)")
    parser.add_argument("--asset", help="asset 경로 (예: /ASSETS/P6)")
    parser.add_argument("--all", action="store_true", help="모든 형제 컬럼 실행")
    parser.add_argument("--compare", action="store_true", help="기존 pkl로 비교표만 출력")
    args = parser.parse_args()

    if args.compare:
        compare()
    elif args.all:
        for prefix, asset, label in COLUMNS:
            run_column(prefix, asset, label)
        compare()
    elif args.prefix:
        asset = args.asset or PREFIX_ASSET.get(args.prefix, f"/ASSETS/P{args.prefix[0]}")
        label = f"C-{args.prefix}11"
        for p, a, l in COLUMNS:
            if p == args.prefix:
                label = l
                break
        run_column(args.prefix, asset, label)
    else:
        parser.print_help()


if __name__ == "__main__":
    main()