#!/usr/bin/env python3 """ main.py — PETGlucoFlux 오프라인 CLI 시연. 표준 라이브러리만으로 동작한다 (numpy/scipy/pandas/streamlit 불필요). data/ 의 합성 샘플을 읽어 SUV / Patlak Ki / MRGlu / 종단 변화 / cohort 요약 / mechanism 분류를 stdout 에 출력한다. 사용법: python3 main.py --help python3 main.py --demo python3 main.py --demo --model db/db python3 main.py --tac data/sample_tac.csv --meta data/sample_meta.csv \ --input data/sample_input_function.csv 본 도구는 연구용·참고용이며, 정량 결과는 사용자 검증이 필요하다. 원시 DICOM 재구성은 범위 밖(정량 CSV ingest 전제). 샘플은 합성(synthetic) 데이터. """ import argparse import os import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import petgluco_core as core HERE = os.path.dirname(os.path.abspath(__file__)) DATA = os.path.join(HERE, "data") DISCLAIMER = ( "[디스클레이머] 본 도구는 연구용·참고용이며, 정량 결과는 사용자 검증이 필요합니다. " "원시 DICOM 재구성은 범위 밖(정량 CSV ingest 전제)이며, 기본 샘플은 합성 데이터입니다." ) def _fmt(v, nd=3): return " n/a" if v is None else ("%.*f" % (nd, v)) def hr(c="-", n=72): print(c * n) def run(tac_path, meta_path, input_path, model): print("=" * 72) print("PETGlucoFlux — rodent micro-PET 18F-FDG 정량 (오프라인 CLI 시연)") print("=" * 72) print(DISCLAIMER) print() tac = core.load_tac(tac_path) meta = core.load_meta(meta_path) inp = core.load_input_function(input_path) core.decay_correct_tac(tac) print("[1] Ingest + decay 보정 (18F t1/2 = %.2f min, lambda=%.5f /min)" % (core.F18_HALF_LIFE_MIN, core.F18_DECAY_LAMBDA)) print(" TAC 행수=%d meta 항목=%d input function 곡선=%d 모델=%s" % (len(tac), len(meta), len(inp), model)) # decay 보정 예시 1건 sample = next((r for r in tac if r["time_min"] and r["time_min"] > 30), tac[0]) print(" 예시: %s/%s/%s t=%.1f min raw=%.1f -> decay-corr=%.1f Bq/mL" % (sample["animal_id"], sample["timepoint"], sample["tissue"], sample["time_min"], sample["activity_bqml"], sample["decay_corrected_bqml"])) qc = core.qc_flags(tac, inp, meta) if qc: print(" QC 플래그:") for f in qc: print(" - " + f) else: print(" QC 플래그: 없음") print() # ---- SUV ---- suv_rows = core.suv_table(tac, meta) print("[2] SUV 정량 (BW / lean / glucose-corrected)") hr() print("%-6s %-9s %-9s %-16s %8s %8s %8s" % ("animal", "group", "timept", "tissue", "SUV_bw", "SUV_lean", "SUV_glu")) hr() for r in suv_rows: print("%-6s %-9s %-9s %-16s %8s %8s %8s" % (r["animal_id"], r["group"], r["timepoint"], r["tissue"], _fmt(r["SUV_bw"]), _fmt(r["SUV_lean"]), _fmt(r["SUV_glu"]))) print() # ---- Patlak Ki + MRGlu ---- kin_rows = core.kinetic_table(tac, inp, meta) print("[3] Patlak Ki + MRGlu (LC 보정)") hr() print("%-6s %-9s %-9s %-16s %9s %6s %6s %5s %8s" % ("animal", "group", "timept", "tissue", "Ki", "r2", "t*", "LC", "MRGlu")) hr() for r in kin_rows: print("%-6s %-9s %-9s %-16s %9s %6s %6s %5s %8s" % (r["animal_id"], r["group"], r["timepoint"], r["tissue"], _fmt(r["Ki"], 4), _fmt(r["patlak_r2"], 3), _fmt(r["tstar"], 1), _fmt(r["lumped_constant"], 2), _fmt(r["MRGlu"], 3))) print() # ---- 종단 within-animal ---- long_ki = core.longitudinal_changes(kin_rows, "Ki") long_suv = core.longitudinal_changes(suv_rows, "SUV_bw") print("[4] 종단 within-animal 변화 (paired baseline -> post)") hr() print("%-6s %-9s %-16s %10s %10s %9s %9s (metric: Ki)" % ("animal", "group", "tissue", "baseline", "post", "delta", "pct%")) hr() for r in long_ki: print("%-6s %-9s %-16s %10s %10s %9s %8s%%" % (r["animal_id"], r["group"], r["tissue"], _fmt(r["baseline"], 4), _fmt(r["post"], 4), _fmt(r["delta"], 4), _fmt(r["pct_change"], 1))) print() # ---- 참고 범위 / 누락 경고 ---- ref_warn = core.reference_check(suv_rows, model=model) meta_warn = core.metadata_warnings(meta, inp, core.DEFAULT_LUMPED_CONSTANT, tac) print("[4b] 모델 참고 범위 / 메타데이터 경고 (model=%s)" % model) if ref_warn: for w in ref_warn: print(" - " + w) else: print(" 참고 범위 이탈 없음") if meta_warn: for w in meta_warn[:6]: print(" - " + w) if len(meta_warn) > 6: print(" - ... (+%d 추가 경고)" % (len(meta_warn) - 6)) else: print(" 메타데이터 누락 경고 없음") print() # ---- cohort 통계 ---- print("[5] Cohort 통계 (Ki paired t-test: baseline vs post)") cs = core.cohort_summary(kin_rows, "Ki") hr() print("%-9s %-16s %4s %12s %12s" % ("group", "tissue", "n", "mean_dKi", "p(paired)")) hr() for t in cs["paired_tests"]: print("%-9s %-16s %4d %12s %12s" % (t["group"], t["tissue"], t["n"], _fmt(t["mean_delta"], 4), _fmt(t["p"], 4))) print() # ---- mechanism 분류 ---- mech = core.classify_mechanism(long_ki) print("[5b] Mechanism 분류 (treatment 군 종단 Ki 변화 기반)") print(" => %s" % mech["label"]) for k, v in mech["scores"].items(): print(" %-45s 평균 pct_change = %s" % (k, _fmt(v, 1))) print() # ---- 텍스트 리포트 ---- print("[6] 요약 리포트 (국문/영문)") hr("=") _report(suv_rows, kin_rows, cs, mech, model) print() print(DISCLAIMER) def _report(suv_rows, kin_rows, cs, mech, model): n_animals = len({r["animal_id"] for r in suv_rows}) n_tissues = len({r["tissue"] for r in suv_rows}) # treatment muscle Ki paired test tx_muscle = next((t for t in cs["paired_tests"] if t["group"] == "treatment" and t["tissue"] == "skeletal_muscle"), None) pstr = _fmt(tx_muscle["p"], 4) if tx_muscle else "n/a" dstr = _fmt(tx_muscle["mean_delta"], 4) if tx_muscle else "n/a" print("[국문] %d마리·%d조직 micro-PET 18F-FDG TAC 를 18F decay 보정 후 SUV(BW/lean/" "glucose) 와 Patlak Ki·LC 보정 MRGlu 로 정량하였다. treatment 군 골격근 " "Ki 의 종단 변화량은 평균 %s (paired p=%s) 였다. 본 cohort 의 주된 기전은 " "'%s' 로 분류되었다 (모델 참고: %s)." % (n_animals, n_tissues, dstr, pstr, mech["label"], model)) print() print("[EN] Rodent micro-PET 18F-FDG TACs from %d animals across %d tissues were " "decay-corrected (18F) and quantified as SUV (BW/lean/glucose) and Patlak Ki " "with lumped-constant-corrected MRGlu. The longitudinal change in skeletal " "muscle Ki for the treatment group was %s on average (paired p=%s). The " "dominant mechanism of this cohort was classified as '%s' (reference model: " "%s)." % (n_animals, n_tissues, dstr, pstr, mech["label"], model)) def main(argv=None): p = argparse.ArgumentParser( prog="main.py", description="PETGlucoFlux — rodent micro-PET 18F-FDG 정량 오프라인 CLI 시연 " "(표준 라이브러리만으로 동작)", epilog="예: python3 main.py --demo | python3 main.py --demo --model db/db") p.add_argument("--demo", action="store_true", help="data/ 의 합성 샘플로 전체 파이프라인 시연 출력") p.add_argument("--tac", default=os.path.join(DATA, "sample_tac.csv"), help="동적 TAC CSV 경로") p.add_argument("--meta", default=os.path.join(DATA, "sample_meta.csv"), help="meta CSV 경로 (dose/weight/lean/glucose/temp/anesthesia)") p.add_argument("--input", default=os.path.join(DATA, "sample_input_function.csv"), help="input function CSV 경로 (plasma activity)") p.add_argument("--model", default="C57BL/6", choices=list(core.MODEL_REFERENCE.keys()), help="참고 범위 모델 (기본 C57BL/6)") args = p.parse_args(argv) if not (args.demo or os.path.exists(args.tac)): p.print_help() return 0 for path, label in [(args.tac, "TAC"), (args.meta, "meta"), (args.input, "input function")]: if not os.path.exists(path): print("ERROR: %s CSV 를 찾을 수 없습니다: %s" % (label, path), file=sys.stderr) return 2 run(args.tac, args.meta, args.input, args.model) return 0 if __name__ == "__main__": sys.exit(main())