"""app.py — HepatoLipidClear-Kor (헤파토리피드클리어) 도메인: MASLD | 카테고리: 인체실험 도구 경구 지방부하검사(OFTT)와 ¹³C 호기검사 원시 시점값 → 식후 지질 동태·호기 산화 동태 분석. standalone Streamlit. 오프라인. 외부 네트워크 호출 0. 면책: 본 도구는 연구용·참고용이며 임상 의사결정에 직접 사용 금지. 실행: streamlit run app.py """ from __future__ import annotations import io import os from datetime import date import numpy as np import pandas as pd import auc as auc_mod import breath as breath_mod import oftt as oftt_mod import units as units_mod import demo_data HERE = os.path.dirname(os.path.abspath(__file__)) DATA_DIR = os.path.join(HERE, "data") DISCLAIMER = ( "⚠️ 본 도구는 **연구용·교육용·참고용**입니다. " "임상 진단·치료 의사결정에 직접 사용하지 마십시오. 모든 데모 데이터는 합성입니다." ) # ───────────────────────── reference 로드 ───────────────────────── def load_reference() -> dict: """reference.yaml 로드. 실패 시 내장 기본값(dict) 폴백.""" path = os.path.join(HERE, "reference.yaml") try: import yaml with open(path, "r", encoding="utf-8") as f: return yaml.safe_load(f) except Exception: return { "isotope": {"R_PDB": 0.0112372}, "co2_production": {"default_vco2_per_bsa_mmol_m2_h": 300.0}, "oftt": {"sampling_times_h": [0, 2, 4, 6, 8], "fasting_hours_required": 12}, "disclaimer": DISCLAIMER, } # ──────────────── 비-Streamlit 호출 가능 헬퍼(검수용) ──────────────── def run_oftt_table(df: pd.DataFrame, value_col: str = "tg", analyte: str = "TG", units_label: str = "mg/dL", fat_load_g: float | None = None, expected_times=(0, 2, 4, 6, 8)) -> pd.DataFrame: """OFTT long DataFrame → 피험자별 결과 DataFrame. app/검수 공용.""" rows = oftt_mod.analyze_oftt_long( df, value_col=value_col, analyte=analyte, units=units_label, fat_load_g=fat_load_g, expected_times=list(expected_times), ) return pd.DataFrame(rows) def run_breath_table(df: pd.DataFrame, vco2_per_bsa: float = 300.0, use_measured_vco2: bool = False, use_nonlinear: bool = True, expected_times_min=(0, 10, 20, 30, 40, 60, 90, 120, 150, 180, 240)) -> pd.DataFrame: """호기 long DataFrame → 피험자별 결과 DataFrame. app/검수 공용.""" out = [] for sid, sub in df.groupby("subject_id"): sub = sub.sort_values("time_min") kwargs = dict( dose_mmol_13c=float(sub["dose_mmol_13c"].iloc[0]), vco2_per_bsa=vco2_per_bsa, use_nonlinear=use_nonlinear, expected_times_min=list(expected_times_min), ) if use_measured_vco2 and "vco2_mmol_h" in sub.columns: kwargs["vco2_mmol_h"] = float(sub["vco2_mmol_h"].iloc[0]) else: kwargs["weight_kg"] = float(sub["weight_kg"].iloc[0]) kwargs["height_cm"] = float(sub["height_cm"].iloc[0]) res = breath_mod.analyze_breath( sub["time_min"].tolist(), sub["delta13c"].tolist(), **kwargs ) d = res.as_dict() d["subject_id"] = sid if "group" in sub.columns: d["group"] = sub["group"].iloc[0] out.append(d) return pd.DataFrame(out) # ───────────────────────── Streamlit UI ───────────────────────── def main(): # pragma: no cover (UI) import streamlit as st import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt ref = load_reference() st.set_page_config(page_title="HepatoLipidClear-Kor", layout="wide") st.title("HepatoLipidClear-Kor — 간 대사 동적검사 분석기") st.caption("도메인: MASLD · 카테고리: 인체실험 도구 (OFTT + ¹³C 호기검사)") st.warning(DISCLAIMER) tabs = st.tabs([ "① OFTT (식후 지질)", "② ¹³C 호기검사", "③ 표준화 체크리스트", "④ 단위·동위원소 변환", "ⓘ Reference / 출처", ]) # ─────── ① OFTT ─────── with tabs[0]: st.subheader("경구 지방부하검사 (Oral Fat Tolerance Test)") c1, c2 = st.columns([1, 1]) with c1: src = st.radio("데이터 소스", ["합성 데모: 정상", "합성 데모: 지연청소", "CSV 업로드"], key="oftt_src") with c2: analyte = st.selectbox("분석 analyte", ["tg", "apob48", "retinyl"], key="oftt_an") unit_label = st.text_input("단위 라벨", "mg/dL" if analyte == "tg" else "μg/mL", key="oftt_unit") fat_g = st.number_input("섭취 총 지방량 (g, normalize용; 0=미사용)", min_value=0.0, value=65.0, key="oftt_fat") tg_unit_toggle = st.checkbox("TG 입력을 mmol/L → mg/dL 변환", value=False, key="oftt_tgconv") if src == "합성 데모: 정상": df = demo_data.make_oftt_normal() elif src == "합성 데모: 지연청소": df = demo_data.make_oftt_delayed() else: up = st.file_uploader("OFTT CSV (subject_id,time_h,tg[,apob48,retinyl])", type=["csv"], key="oftt_up") df = pd.read_csv(up) if up is not None else None if df is not None and len(df): work = df.copy() if tg_unit_toggle and analyte == "tg": work["tg"] = work["tg"].apply(units_mod.tg_mmoll_to_mgdl) st.dataframe(work, use_container_width=True, height=200) res_df = run_oftt_table(work, value_col=analyte, analyte=analyte.upper(), units_label=unit_label, fat_load_g=(fat_g if fat_g > 0 else None)) st.markdown("**피험자별 결과**") st.dataframe(res_df, use_container_width=True) # 곡선 오버레이 fig, ax = plt.subplots(figsize=(7, 4)) for sid, sub in work.groupby("subject_id"): sub = sub.sort_values("time_h") ax.plot(sub["time_h"], sub[analyte], marker="o", alpha=0.7, label=str(sid)) ax.set_xlabel("time (h)"); ax.set_ylabel(f"{analyte} ({unit_label})") ax.set_title("OFTT 곡선 오버레이"); ax.legend(fontsize=7, ncol=2) st.pyplot(fig) flagged = res_df[res_df["flags"] != ""] if len(flagged): st.error(f"QC 플래그 {len(flagged)}건"); st.dataframe(flagged[["subject_id", "flags"]]) st.download_button("결과 CSV 다운로드", res_df.to_csv(index=False).encode("utf-8"), "oftt_results.csv", "text/csv") st.caption("공식: iAUC = ∫(C(t)-baseline)dt (사다리꼴). " "iAUC/g fat = iAUC ÷ 섭취 지방량. 출처: Frayn; Cobelli; Mihas 2011.") # ─────── ② 호기검사 ─────── with tabs[1]: st.subheader("¹³C 호기검사 (¹³C Breath Test)") c1, c2 = st.columns([1, 1]) with c1: bsrc = st.radio("데이터 소스", ["합성 데모", "CSV 업로드"], key="br_src") vco2_per_bsa = st.number_input("CO₂ 생성률 가정 (mmol/m²/h)", min_value=100.0, value=300.0, key="br_vco2") use_meas = st.checkbox("VCO₂ 측정값 컬럼(vco2_mmol_h) 우선 사용", value=False, key="br_meas") with c2: use_nl = st.checkbox("비선형(단일지수) kel fit 사용 (실패 시 선형 폴백)", value=True, key="br_nl") st.caption("PDR(t)=DOB/1000 × VCO₂ × R_PDB / Dose × 100 (R_PDB=0.0112372)") if bsrc == "합성 데모": bdf = demo_data.make_breath_curves() else: up = st.file_uploader( "호기 CSV (subject_id,time_min,delta13c,weight_kg,height_cm,dose_mmol_13c[,group,vco2_mmol_h])", type=["csv"], key="br_up") bdf = pd.read_csv(up) if up is not None else None if bdf is not None and len(bdf): st.dataframe(bdf, use_container_width=True, height=200) bres = run_breath_table(bdf, vco2_per_bsa=vco2_per_bsa, use_measured_vco2=use_meas, use_nonlinear=use_nl) st.markdown("**피험자별 결과 (PDR/cPDR/반감기)**") st.dataframe(bres, use_container_width=True) fig, ax = plt.subplots(figsize=(7, 4)) for sid, sub in bdf.groupby("subject_id"): sub = sub.sort_values("time_min") t_h, pdr = breath_mod.delta_to_pdr( sub["time_min"].tolist(), sub["delta13c"].tolist(), float(sub["dose_mmol_13c"].iloc[0]), breath_mod.estimate_vco2(float(sub["weight_kg"].iloc[0]), float(sub["height_cm"].iloc[0]), vco2_per_bsa)) ax.plot(np.array(t_h) * 60, pdr, marker="o", alpha=0.7, label=str(sid)) ax.set_xlabel("time (min)"); ax.set_ylabel("PDR (%/h)") ax.set_title("¹³C PDR 곡선 오버레이"); ax.legend(fontsize=7, ncol=2) st.pyplot(fig) if "group" in bres.columns: st.markdown("**군간 비교 (평균)**") num = bres.select_dtypes(include=[np.number]) num["group"] = bres["group"].values st.dataframe(num.groupby("group").mean(numeric_only=True)) flagged = bres[bres["flags"] != ""] if len(flagged): st.error(f"QC 플래그 {len(flagged)}건") st.dataframe(flagged[["subject_id", "flags"]]) st.download_button("결과 CSV 다운로드", bres.to_csv(index=False).encode("utf-8"), "breath_results.csv", "text/csv") st.caption("cPDR[%]=∫PDR dt(h). t½=ln2/kel. VCO₂ 추정: 300 mmol/m²/h × Du Bois BSA. " "출처: Ghoos 1993; Schadewaldt; Sanaka.") # ─────── ③ 체크리스트 ─────── with tabs[2]: st.subheader("검사 표준화 체크리스트 (재현성 메타데이터)") st.caption("아래 항목은 결과 메타데이터로 강제 기록을 권장합니다.") with st.form("checklist"): meal = st.text_input("고지방식 조성/명칭", "고지방식 A (지방 65g, ~900kcal)") fat_g = st.number_input("지방량 (g)", value=65.0) kcal = st.number_input("총 열량 (kcal)", value=900.0) fast_h = st.number_input("금식 시간 (h)", value=12.0) dose = st.number_input("¹³C 기질 용량 (mmol)", value=4.0) vco2_src = st.selectbox("VCO₂ 출처", ["bsa_estimated", "measured"]) vco2_bsa = st.number_input("VCO₂ 가정값 (mmol/m²/h)", value=300.0) tg_units = st.selectbox("TG 단위", ["mg/dL", "mmol/L"]) operator = st.text_input("검사자", "") tdate = st.date_input("검사일", value=date.today()) submitted = st.form_submit_button("메타데이터 기록 생성") if submitted: meta = { "meal_name": meal, "meal_fat_g": fat_g, "meal_kcal": kcal, "fasting_hours": fast_h, "substrate_dose_mmol_13c": dose, "vco2_source": vco2_src, "vco2_per_bsa": vco2_bsa, "tg_units": tg_units, "operator": operator, "test_date": str(tdate), } warns = [] if fast_h < ref.get("oftt", {}).get("fasting_hours_required", 12): warns.append("금식 시간 권고(12h) 미달") if vco2_src == "bsa_estimated": warns.append("VCO₂ 추정값 사용 — 결과 해석 시 가정값 명시 필요") st.json(meta) for w in warns: st.warning(w) st.download_button("메타데이터 CSV", pd.DataFrame([meta]).to_csv(index=False).encode("utf-8"), "test_metadata.csv", "text/csv") # ─────── ④ 단위 변환 ─────── with tabs[3]: st.subheader("단위·동위원소 변환") c1, c2, c3 = st.columns(3) with c1: st.markdown("**TG**") v = st.number_input("값", value=150.0, key="u_tg") d = st.radio("방향", ["mg/dL→mmol/L", "mmol/L→mg/dL"], key="u_tgdir") out = units_mod.tg_mgdl_to_mmoll(v) if d.startswith("mg") else units_mod.tg_mmoll_to_mgdl(v) st.metric("결과", f"{out:.3f}") with c2: st.markdown("**δ¹³C ↔ 비**") dv = st.number_input("δ¹³C (‰)", value=-21.0, key="u_d") st.metric("¹³C/¹²C 비", f"{units_mod.delta_to_ratio(dv):.6f}") rv = st.number_input("¹³C/¹²C 비", value=0.0112372, format="%.7f", key="u_r") st.metric("δ¹³C (‰)", f"{units_mod.ratio_to_delta(rv):.3f}") with c3: st.markdown("**apoB48**") av = st.number_input("값", value=5.0, key="u_a") ad = st.radio("방향", ["μg/mL→nmol/L", "nmol/L→μg/mL"], key="u_adir") ao = (units_mod.apob48_ugml_to_nmoll(av) if ad.startswith("μg") else units_mod.apob48_nmoll_to_ugml(av)) st.metric("결과", f"{ao:.4f}") # ─────── ⓘ Reference ─────── with tabs[4]: st.subheader("내장 Reference / 산출 공식 / 출처") st.json(ref) st.markdown( "**주요 공식 footnote**\n\n" "- iAUC = ∫(C(t)−baseline) dt (사다리꼴)\n" "- PDR(t)[%/h] = DOB(t)/1000 × VCO₂[mmol/h] × R_PDB / Dose[mmol] × 100\n" "- R_PDB = 0.0112372 (PDB 표준)\n" "- VCO₂(추정) = 300 mmol/m²/h × BSA; BSA = 0.007184·W^0.425·H^0.725 (Du Bois)\n" "- cPDR[%] = ∫PDR dt; t½ = ln(2)/kel (단일지수 또는 로그-선형 fit)\n\n" "**출처**: Frayn KN, Metabolic Regulation; Cobelli/Toffolo minimal-model; " "Mihas C et al. Curr Vasc Pharmacol 2011 (OFTT); Ghoos YF et al. " "Gastroenterology 1993; Schadewaldt; Sanaka (¹³C breath test)." ) st.warning(DISCLAIMER) if __name__ == "__main__": # pragma: no cover main()