""" runtime.py ========== Web 서비스용 런타임 래퍼. webtool_runtime.py 를 HF Space 에 맞게 이식했다. 로드 대상 (기동 시 1회): webtool_core.py 모델 정의 + 상수 + feature + ensemble webtool_baseline.json 동결된 risk -> S(H) 매핑 (+ bootstrap CI 계수) weights/seed_*/fold_*.pt 31 seed x 10 fold 앙상블 (310 checkpoints) 입력 : T stage(1..6), station별 전이 count, station별 harvest(절제 node 수) 출력 : risk_mean, risk_std, SCR_3, entropy, total_meta, surv_60m = 기대 5년 생존확률(+95% CI), extrapolated flag """ from __future__ import annotations import json import os import threading from pathlib import Path from typing import Any import numpy as np import pandas as pd import torch from .webtool_core import ( STATIONS, T_MAP, load_canonical, build_ln_features, build_t_onehot, scr_and_entropy, MultiSeedEnsemble, ) def _n_label(total_meta: float) -> str: """전이 node 총수 -> AJCC N stage (N0/N1/N2/N3a/N3b).""" n = int(round(total_meta or 0)) if n == 0: return "N0" if n <= 2: return "N1" if n <= 6: return "N2" if n <= 15: return "N3a" return "N3b" # --------------------------------------------------------------------------- # # 경로 / 설정 # --------------------------------------------------------------------------- # BUNDLE_ROOT = Path(__file__).resolve().parents[1] CONFIG = { "WEIGHTS_ROOT": os.environ.get("WEIGHTS_ROOT", str(BUNDLE_ROOT / "weights")), "CANON_SEED": int(os.environ.get("CANON_SEED", "9999")), "SEEDS": None, # None 이면 seed_* 전부 자동 탐색 (baseline 동결 시점과 동일해야 함) "BATCH": int(os.environ.get("BATCH", "256")), "DEVICE": "cuda" if torch.cuda.is_available() else "cpu", "ARTIFACT_PATH": os.environ.get( "ARTIFACT_PATH", str(BUNDLE_ROOT / "webtool_baseline.json") ), } DEVICE = torch.device(CONFIG["DEVICE"]) # T stage 코드(1..6) -> 라벨. UI/스키마에서 사용. T_OPTIONS = [{"value": k, "label": v} for k, v in sorted(T_MAP.items())] # 지연 초기화 상태 (thread-safe) _STATE: dict[str, Any] = {} _INIT_LOCK = threading.Lock() def _ensure_baseline() -> dict[str, Any]: """baseline json + canonical config 만 로드한다(저렴). 앙상블은 로드하지 않음. schema/health 가 앙상블 로딩을 기다리지 않도록 분리한다.""" if "art" in _STATE: return _STATE with _INIT_LOCK: if "art" in _STATE: return _STATE print("[runtime] loading baseline + canonical config …", flush=True) with open(CONFIG["ARTIFACT_PATH"]) as f: art = json.load(f) model_config, t_cols = load_canonical(CONFIG["WEIGHTS_ROOT"], CONFIG["CANON_SEED"]) # 동결 시점과 런타임의 구조 및 T 인코딩 정합을 강제한다. assert art["t_cols"] == t_cols, "runtime t_cols 가 baseline 과 불일치" assert art["model_config"] == model_config, "runtime model_config 가 baseline 과 불일치" _STATE.update( art=art, model_config=model_config, t_cols=t_cols, horizon=art["horizon_months"], ) print(f"[runtime] baseline OK. horizon={art['horizon_months']}mo", flush=True) return _STATE def init() -> dict[str, Any]: """전체 앙상블(310 checkpoints)까지 로드한다. idempotent, thread-safe. 기동 지연을 피하려고 서버는 이 함수를 백그라운드 스레드에서 호출한다.""" if _STATE.get("ready"): return _STATE _ensure_baseline() with _INIT_LOCK: if _STATE.get("ready"): return _STATE print("[runtime] building ensemble (310 checkpoints) …", flush=True) ensemble = MultiSeedEnsemble( weights_root=CONFIG["WEIGHTS_ROOT"], model_config=_STATE["model_config"], batch=CONFIG["BATCH"], device=DEVICE, seeds=CONFIG["SEEDS"], ) _STATE.update(ensemble=ensemble, ready=True, n_seeds=len(ensemble.seeds)) print(f"[runtime] ensemble ready: {len(ensemble.seeds)} seeds", flush=True) return _STATE # --------------------------------------------------------------------------- # # risk -> 기대 survival (+ CI). webtool_runtime.py 수식 그대로. # --------------------------------------------------------------------------- # def _expected_survival(risk_series: pd.Series, art: dict) -> tuple[pd.Series, pd.Series]: r = risk_series.astype(float) s = art["S0_H"] ** np.exp(art["beta"] * (r - art["risk_center"])) oor = (r < art["risk_ref_min"]) | (r > art["risk_ref_max"]) return ( pd.Series(s, index=r.index, name=f"surv_{art['horizon_months']}m"), pd.Series(oor, index=r.index, name="extrapolated"), ) def _expected_survival_ci( risk_series: pd.Series, per_seed: pd.DataFrame | None, art: dict, mode: str = "cox" ) -> pd.DataFrame: r = risk_series.astype(float) boot = art["boot"] B, N = len(boot), len(r) M = np.empty((B, N)) seeds_arr = None if mode == "full": assert per_seed is not None, "mode='full' 에는 per_seed DataFrame 필요" seeds_arr = per_seed.reindex(r.index).values # [N, n_seeds] for j, bp in enumerate(boot): if mode == "full": rng = np.random.default_rng(1000 + j) n_seed = seeds_arr.shape[1] r_eval = np.array( [ rng.choice(seeds_arr[i], size=n_seed, replace=True).mean() for i in range(N) ] ) else: r_eval = r.values M[j] = bp["S0_H"] ** np.exp(bp["beta"] * (r_eval - bp["risk_center"])) a = art["ci_alpha"] lo = np.percentile(M, 100 * a / 2, axis=0) hi = np.percentile(M, 100 * (1 - a / 2), axis=0) return pd.DataFrame({"lo": lo, "hi": hi}, index=r.index) # --------------------------------------------------------------------------- # # End-to-end 채점 (DataFrame 반환) # --------------------------------------------------------------------------- # def score_frame( tstage: pd.Series, count_df: pd.DataFrame, harvest_df: pd.DataFrame, ci: bool = True, ci_mode: str = "cox", ) -> pd.DataFrame: """다수 환자 채점. index 보존. 컬럼 순서/의미는 webtool_runtime 과 동일.""" st = init() art, ensemble, t_cols = st["art"], st["ensemble"], st["t_cols"] ln = build_ln_features(count_df, harvest_df) t_df = build_t_onehot(tstage.reindex(count_df.index), t_cols) risk, per_seed = ensemble.predict(ln, t_df, return_per_seed=True) se = scr_and_entropy(count_df) s_H, oor = _expected_survival(risk, art) H = art["horizon_months"] out = pd.DataFrame(index=count_df.index) out["Tstage"] = tstage.reindex(count_df.index).astype(int) out["total_meta"] = se["total_meta"] out["risk_mean"] = risk out["risk_std"] = per_seed.std(axis=1) out["SCR_3"] = se["SCR_3"] out["entropy"] = se["entropy"] out[f"surv_{H}m"] = s_H if ci: cid = _expected_survival_ci(risk, per_seed=per_seed, art=art, mode=ci_mode) out[f"surv_{H}m_lo"] = cid["lo"] out[f"surv_{H}m_hi"] = cid["hi"] out["ci_mode"] = ci_mode out["extrapolated"] = oor return out # --------------------------------------------------------------------------- # # 입력 정규화 헬퍼 (dict 형태 -> STATIONS 순서 DataFrame) # --------------------------------------------------------------------------- # def _station_df(mapping: dict[str, float], index) -> pd.DataFrame: """{station: 값} -> [1,16] DataFrame. 미기재 station 은 0.""" mapping = mapping or {} return pd.DataFrame( [[float(mapping.get(s, 0) or 0) for s in STATIONS]], columns=STATIONS, index=index, ) def _clean(v): """numpy/pandas 스칼라 -> JSON 직렬화 가능한 파이썬 값. NaN -> None.""" if v is None: return None if isinstance(v, (np.floating, float)): v = float(v) return None if np.isnan(v) else v if isinstance(v, (np.integer,)): return int(v) if isinstance(v, (np.bool_, bool)): return bool(v) return v def _row_to_result(row: pd.Series, patient_id: str, H: int) -> dict[str, Any]: """score_frame 한 행 -> API 응답 dict.""" return { "id": patient_id, "Tstage": _clean(row["Tstage"]), "T_label": T_MAP.get(int(row["Tstage"])), "total_meta": _clean(row["total_meta"]), "risk_mean": _clean(row["risk_mean"]), "risk_std": _clean(row["risk_std"]), "SCR_3": _clean(row["SCR_3"]), "entropy": _clean(row["entropy"]), "surv": _clean(row[f"surv_{H}m"]), "surv_lo": _clean(row.get(f"surv_{H}m_lo")), "surv_hi": _clean(row.get(f"surv_{H}m_hi")), "extrapolated": _clean(row["extrapolated"]), "horizon_months": H, } def patient_curve(risk: float) -> dict[str, Any] | None: """환자 위험점수 -> 시간축 생존곡선 S(t|risk) + 95% CI band + 참조 KM. S(t|risk) = S_center(t) ** exp(beta*(risk - risk_center)). S_center(60)=S0_H 이므로 곡선의 60개월 값은 point surv 와 정확히 일치한다. baseline 아티팩트가 curve 를 포함하지 않으면 None.""" art = init()["art"] if "curve_base" not in art: return None beta, rc = art["beta"], art["risk_center"] t = np.asarray(art["curve_months"], dtype=int) base = np.asarray(art["curve_base"], dtype=float) s = base ** np.exp(beta * (risk - rc)) bb = np.asarray(art["curve_boot_base"], dtype=float) # [B, T] bbeta = np.asarray(art["curve_boot_beta"], dtype=float)[:, None] bcen = np.asarray(art["curve_boot_center"], dtype=float)[:, None] sb = bb ** np.exp(bbeta * (risk - bcen)) # [B, T] a = art.get("ci_alpha", 0.05) lo = np.percentile(sb, 100 * a / 2, axis=0) hi = np.percentile(sb, 100 * (1 - a / 2), axis=0) # 생존함수는 수술 시점(t=0)에서 정의상 1. baseline curve 의 0개월 값이 정확히 # 1 이 아닐 수 있으므로 그래프와 RMST 적분 모두에서 강제로 보정한다. s[0] = 1.0 lo[0] = 1.0 hi[0] = 1.0 sb[:, 0] = 1.0 # RMST_H = ∫_0^H S(t) dt (trapezoidal, 월 단위). 항상 0..H 범위. # CI 는 최종 lo/hi 곡선을 적분하는 게 아니라, bootstrap replicate 곡선별 # RMST 분포의 2.5/97.5 percentile 을 쓴다. tf = t.astype(float) _trapz = getattr(np, "trapezoid", np.trapz) rmst = float(_trapz(s, tf)) rmst_boot = _trapz(sb, tf, axis=1) rmst_lo, rmst_hi = ( float(x) for x in np.percentile(rmst_boot, [100 * a / 2, 100 * (1 - a / 2)]) ) r5 = lambda arr: [round(float(x), 5) for x in arr] return { "t": t.tolist(), "s": r5(s), "lo": r5(lo), "hi": r5(hi), "ref_km": r5(art["ref_km"]) if "ref_km" in art else None, "rmst": round(rmst, 2), "rmst_lo": round(rmst_lo, 2), "rmst_hi": round(rmst_hi, 2), } def score_single( tstage: int, counts: dict[str, float], harvests: dict[str, float], patient_id: str = "patient", ci_mode: str = "cox", ) -> dict[str, Any]: """단일 환자 채점 -> API 응답 dict (+ 시간축 생존곡선).""" st = init() idx = [patient_id] count_df = _station_df(counts, idx) harvest_df = _station_df(harvests, idx) t_series = pd.Series([int(tstage)], index=idx) out = score_frame(t_series, count_df, harvest_df, ci=True, ci_mode=ci_mode) res = _row_to_result(out.iloc[0], patient_id, st["horizon"]) res["curve"] = patient_curve(float(res["risk_mean"])) # RMST (H-개월 restricted mean survival time) 를 상위 레벨로 승격. # 고정 키(rmst*)는 프론트/일반 소비자용, rmst_{H}m* 는 horizon 명시용. H = st["horizon"] cur = res.get("curve") if cur and cur.get("rmst") is not None: res["rmst"] = cur["rmst"] res["rmst_lo"] = cur["rmst_lo"] res["rmst_hi"] = cur["rmst_hi"] res[f"rmst_{H}m"] = cur["rmst"] res[f"rmst_{H}m_lo"] = cur["rmst_lo"] res[f"rmst_{H}m_hi"] = cur["rmst_hi"] res["rmst_horizon_months"] = H art = st["art"] # TNM 분류 (T + total_meta 로부터 N 유도 -> TNM 그룹/stage) t_clin = T_MAP.get(int(res["Tstage"])) t_grp = art.get("tmap_tnm", {}).get(str(int(res["Tstage"])), t_clin) n_lab = _n_label(res["total_meta"]) group = art.get("tnm_dict", {}).get(f"{t_grp}|{n_lab}") stage = art.get("tnm_label_map", {}).get(str(group)) if group else None res["tnm"] = {"t": t_clin, "n": n_lab, "tn": f"{t_clin}{n_lab}", "group": group, "stage": stage} # SCR: 절대 기준 0.75 (high/low). entropy: ln2/ln4 로 low/intermediate/high. import math scr, ent = res["SCR_3"], res["entropy"] scr_thr = 0.75 res["scr_flag"] = None if scr is None else ("high" if scr >= scr_thr else "low") res["scr_threshold"] = scr_thr if ent is None: res["entropy_level"] = None elif ent < math.log(2): res["entropy_level"] = "low" elif ent < math.log(4): res["entropy_level"] = "intermediate" else: res["entropy_level"] = "high" res["entropy_thresholds"] = [round(math.log(2), 4), round(math.log(4), 4)] return res # --------------------------------------------------------------------------- # # CSV 배치 채점 # --------------------------------------------------------------------------- # def csv_columns() -> list[str]: """배치 CSV 표준 컬럼: id, Tstage, meta_, harv_ x 16.""" cols = ["id", "Tstage"] for s in STATIONS: cols += [f"meta_{s}", f"harv_{s}"] return cols def score_csv(df: pd.DataFrame, ci_mode: str = "cox") -> list[dict[str, Any]]: """표준 CSV DataFrame 채점. 누락 station 컬럼은 0 으로 채운다.""" st = init() H = st["horizon"] df = df.copy() if "Tstage" not in df.columns: raise ValueError("CSV 에 'Tstage' 컬럼이 필요합니다 (값 1..6).") if "id" not in df.columns: df["id"] = [f"row_{i:04d}" for i in range(len(df))] ids = df["id"].astype(str).tolist() index = pd.RangeIndex(len(df)) count_df = pd.DataFrame(index=index, columns=STATIONS, dtype=float) harvest_df = pd.DataFrame(index=index, columns=STATIONS, dtype=float) for s in STATIONS: count_df[s] = pd.to_numeric(df.get(f"meta_{s}", 0), errors="coerce").fillna(0).values harvest_df[s] = pd.to_numeric(df.get(f"harv_{s}", 0), errors="coerce").fillna(0).values t_series = pd.Series( pd.to_numeric(df["Tstage"], errors="coerce").fillna(1).astype(int).values, index=index, ) out = score_frame(t_series, count_df, harvest_df, ci=True, ci_mode=ci_mode) return [_row_to_result(out.iloc[i], ids[i], H) for i in range(len(out))]