18963455f2
- 제품 라벨(P0/P1)에 MSDS 화학물질명(PMA/PGMEA/EL) 매핑 레이어 추가
(product_labels.json, 플랜트=컬럼별). 통계 기준밴드는 데이터 산출 유지
- 운전원용 ⚙제품명 설정 모달: GET/POST /api/steam/productlabels/{col}
- 온도기반 자동매칭 폐지 → 운전원이 고른 제품만 기준밴드로 비교, 미선택 시작
- 드롭다운: 탭 진입/컬럼변경 시 즉시 채움, 선택은 플랜트별 localStorage 유지
- 중복 '제품(수동)' 배지 제거(헤더 줄바꿈 해소)
- 설계·검수 문서 추가
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
101 lines
4.2 KiB
Python
101 lines
4.2 KiB
Python
"""
|
||
컬럼 온도 프로파일 기준 산출 → 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"{prefix}_data.pkl")
|
||
if prefix == "C-6111" 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}",
|
||
"name": f"P{rank}", # 기본값=label. product_labels.json에서 후처리 채움
|
||
"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": 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"{prefix}_tempref.json")
|
||
with open(out, "w") as f:
|
||
json.dump(ref, f, indent=2, ensure_ascii=False)
|
||
print(f" {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 ["C-6111", "C-6211", "C-8111", "C-9111", "C-9211", "C-10111", "C-10211"]
|
||
for p in prefixes:
|
||
build(p, args.stable_from, args.stable_to)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|