""" app.py — HepatoLipidFlux (헤파토리피드플럭스) MASLD/MASH 동물모델의 in vivo 간 지질 플럭스 정량 도구 (Streamlit MVP). 5개 핵심 기능: 1. VLDL-TG 분비율 (tyloxapol/poloxamer 시계열 + 선형구간 자동탐지) 2. DNL 플럭스 엔진 (MIDA) 3. 지방산 산화 플럭스 4. MASH 모델 참조범위 + QC 5. 코호트 통계 + 기전 분류 리포트 연구용·참고용 (for research / reference use only) — 임상 진단 목적 사용 금지. 실행: pip install -r requirements.txt && streamlit run app.py """ from __future__ import annotations import io import os import numpy as np import pandas as pd import streamlit as st import matplotlib.pyplot as plt import flux_core as fc HERE = os.path.dirname(os.path.abspath(__file__)) DATA_DIR = os.path.join(HERE, "data") st.set_page_config(page_title="HepatoLipidFlux", layout="wide") # --------------------------------------------------------------------------- # 데이터 로딩 헬퍼 # --------------------------------------------------------------------------- def _read_csv(uploaded, demo_name: str) -> pd.DataFrame | None: """업로드 파일이 있으면 사용, 없고 demo 모드면 data/ 폴더 CSV 로드.""" if uploaded is not None: return pd.read_csv(uploaded) if st.session_state.get("use_demo"): path = os.path.join(DATA_DIR, demo_name) if os.path.exists(path): return pd.read_csv(path) return None def _demo_df(name: str) -> pd.DataFrame | None: path = os.path.join(DATA_DIR, name) if os.path.exists(path): return pd.read_csv(path) return None # --------------------------------------------------------------------------- # 사이드바 # --------------------------------------------------------------------------- st.sidebar.title("HepatoLipidFlux") st.sidebar.caption("간 지질 플럭스 정량 · MASLD 동물실험 도구") if st.sidebar.button("데모 데이터 로드 (Load demo data)"): st.session_state["use_demo"] = True if st.session_state.get("use_demo"): st.sidebar.success("데모 데이터 사용 중 (data/ 폴더 합성 CSV).") if not os.path.exists(os.path.join(DATA_DIR, "plasma_tg_timeseries.csv")): st.sidebar.warning( "data/ 폴더에 CSV가 없습니다. 먼저 `python3 generate_sample_data.py` 실행." ) else: st.sidebar.info("데모 데이터를 로드하거나 각 탭에서 CSV를 업로드하세요.") r2_threshold = st.sidebar.slider( "VLDL 선형구간 r² 임계값", 0.80, 0.999, 0.95, 0.005, help="이 r² 이상을 유지하는 가장 긴 초기 구간을 분비율 회귀에 사용.", ) st.sidebar.markdown("---") st.sidebar.warning( "본 도구는 연구용·참고용입니다. 임상 진단·치료 의사결정에 사용하지 마십시오." ) # --------------------------------------------------------------------------- # 헤더 # --------------------------------------------------------------------------- st.title("HepatoLipidFlux — 간 지질 플럭스 정량") st.markdown( "**도메인**: MASLD (대사성간질환) · **카테고리**: 동물실험 도구 \n" "tyloxapol/poloxamer VLDL-TG 분비, 2H2O/13C DNL(MIDA), 지방산 산화 추적자를 " "표준화된 플럭스 지표로 정량합니다." ) st.info( "참고용·연구용 도구입니다. 합성 데모 데이터로 오프라인 동작하며, " "외부 네트워크/API 호출이 없습니다.", icon="ℹ️", ) tab1, tab2, tab3, tab4, tab5 = st.tabs([ "1. VLDL-TG 분비율", "2. DNL 플럭스 (MIDA)", "3. 지방산 산화", "4. MASH 참조 + QC", "5. 코호트 통계 + 리포트", ]) # =========================================================================== # 탭 1 — VLDL-TG 분비율 # =========================================================================== with tab1: st.header("VLDL-TG 분비율") st.markdown( "tyloxapol / poloxamer-407 주사 후 혈장 TG 시계열을 업로드하면 " "초기 **선형 구간을 자동 탐지**하여 회귀, 분비율을 계산합니다. " "후기 포화(saturation) 구간은 자동 제외하고 경고합니다." ) c1, c2 = st.columns(2) ts_up = c1.file_uploader( "혈장 TG 시계열 CSV (animal_id, group, time_min, plasma_tg_mg_dl)", type="csv", key="ts_up", ) meta_up = c2.file_uploader( "동물 메타 CSV (animal_id, group, body_weight_g, lean_mass_g, batch)", type="csv", key="meta_up", ) ts_df = _read_csv(ts_up, "plasma_tg_timeseries.csv") meta_df = _read_csv(meta_up, "animal_meta.csv") if ts_df is not None and meta_df is not None: st.session_state["ts_df"] = ts_df st.session_state["meta_df"] = meta_df try: vldl_table = fc.batch_vldl(ts_df, meta_df, r2_threshold=r2_threshold) st.session_state["vldl_table"] = vldl_table st.subheader("동물별 VLDL-TG 분비율") st.dataframe(vldl_table, use_container_width=True) n_nonlin = int(vldl_table["nonlinear_warning"].sum()) if n_nonlin: st.warning( f"{n_nonlin} 마리에서 비선형 포화 구간이 감지되어 초기 구간만 사용됨." ) animals = sorted(ts_df["animal_id"].unique()) sel = st.selectbox("개별 동물 곡선 보기", animals, key="vldl_sel") sub = ts_df[ts_df["animal_id"] == sel].sort_values("time_min") row = vldl_table[vldl_table["animal_id"] == sel].iloc[0] fig, ax = plt.subplots(figsize=(6, 3.5)) ax.scatter(sub["time_min"], sub["plasma_tg_mg_dl"], color="#444", label="측정 TG", zorder=3) w0, w1 = row["window_start_min"], row["window_end_min"] res = fc.compute_vldl_secretion( sub, animal_id=sel, r2_threshold=r2_threshold, body_weight_g=float( meta_df.set_index("animal_id").loc[sel, "body_weight_g"] ) if sel in meta_df["animal_id"].values else None, ) xs = np.linspace(w0, w1, 50) ys = res.secretion_rate_raw / 60.0 * xs + res.intercept ax.plot(xs, ys, color="#c0392b", lw=2, label=f"선형회귀 (r²={res.r_squared:.3f})") ax.axvspan(w0, w1, color="#f1c40f", alpha=0.15, label="회귀 사용 구간") ax.set_xlabel("시간 (분)") ax.set_ylabel("혈장 TG (mg/dL)") ax.set_title(f"{sel} — VLDL-TG 분비 곡선") ax.legend(fontsize=8) st.pyplot(fig) plt.close(fig) m1, m2, m3 = st.columns(3) m1.metric("분비율 (체중정규화)", f"{res.secretion_rate_norm:.1f} mg/kg/h" if not np.isnan(res.secretion_rate_norm) else "N/A") m2.metric("분비율 (raw)", f"{res.secretion_rate_raw:.2f} mg/dL/h") m3.metric("회귀 r²", f"{res.r_squared:.3f}") for w in res.warnings: st.warning(w) except Exception as exc: st.error(f"VLDL 계산 오류: {exc}") else: st.info("시계열 + 메타 CSV를 업로드하거나 사이드바에서 데모 데이터를 로드하세요.") # =========================================================================== # 탭 2 — DNL 플럭스 (MIDA) # =========================================================================== with tab2: st.header("DNL 플럭스 엔진 (MIDA)") st.markdown( "2H2O body-water enrichment 또는 [13C]전구체 enrichment 와 " "간 TG 지방산 **mass isotopomer 분포**를 입력하면, MIDA(이항모델)로 " "**fractional DNL(신생합성 분율)** 을 계산합니다. " "전구체 풀 enrichment 가정을 명시하고 ±20% 민감도를 함께 표시합니다." ) dnl_up = st.file_uploader( "DNL isotopomer CSV (animal_id, group, precursor_enrichment, " "n_label_sites, M0, M1, M2, ...)", type="csv", key="dnl_up", ) dnl_df = _read_csv(dnl_up, "dnl_isotopomer.csv") if dnl_df is not None: st.session_state["dnl_df"] = dnl_df try: dnl_table = fc.batch_dnl(dnl_df) st.session_state["dnl_table"] = dnl_table st.subheader("동물별 fractional DNL") st.dataframe(dnl_table, use_container_width=True) st.caption( "sens_minus20 / sens_plus20: 전구체 풀 enrichment 를 ±20% " "변동시켰을 때의 fractional DNL — 전구체 가정 민감도." ) animals = sorted(dnl_df["animal_id"].astype(str).unique()) sel = st.selectbox("개별 동물 isotopomer 보기", animals, key="dnl_sel") r = dnl_df[dnl_df["animal_id"].astype(str) == sel].iloc[0] m_cols = sorted( [c for c in dnl_df.columns if c.upper().startswith("M") and c[1:].isdigit()], key=lambda c: int(c[1:]), ) dist = [float(r[c]) for c in m_cols] n_sites = int(r["n_label_sites"]) if "n_label_sites" in dnl_df.columns else 22 res = fc.compute_dnl( dist, precursor_enrichment=float(r["precursor_enrichment"]), animal_id=sel, group=str(r["group"]), n_label_sites=n_sites, ) f_dnl, fitted = fc.compute_fractional_dnl( dist, float(r["precursor_enrichment"]), n_sites ) fig, ax = plt.subplots(figsize=(6, 3.5)) x = np.arange(len(m_cols)) ax.bar(x - 0.2, np.array(dist) / sum(dist), width=0.4, label="측정", color="#444") ax.bar(x + 0.2, fitted, width=0.4, label="MIDA 적합", color="#2980b9") ax.set_xticks(x) ax.set_xticklabels(m_cols) ax.set_ylabel("상대 비율") ax.set_title(f"{sel} — isotopomer 분포 (fractional DNL={f_dnl:.3f})") ax.legend(fontsize=8) st.pyplot(fig) plt.close(fig) m1, m2, m3 = st.columns(3) m1.metric("Fractional DNL", f"{res.fractional_dnl:.3f}") m2.metric("전구체 enrichment 가정", f"{res.precursor_enrichment:.4f}") m3.metric("라벨 위치 수 (n)", f"{res.n_label_assumed}") st.write("**민감도 (전구체 enrichment ±20%)**:", res.sensitivity) for w in res.warnings: st.warning(w) except Exception as exc: st.error(f"DNL 계산 오류: {exc}") else: st.info("isotopomer CSV를 업로드하거나 사이드바에서 데모 데이터를 로드하세요.") # =========================================================================== # 탭 3 — 지방산 산화 플럭스 # =========================================================================== with tab3: st.header("지방산 산화 플럭스") st.markdown( "[13C]palmitate 추적자의 호기 13CO2 회수분율 및 케톤체 표지분율로부터 " "**회수율 보정 산화 플럭스**를 계산합니다. 호기 13CO2가 체내 " "bicarbonate pool 에 일부 격리되는 점을 bicarbonate 회수율로 보정합니다." ) ox_up = st.file_uploader( "산화 추적자 CSV (animal_id, group, tracer_dose_umol, " "co2_recovery_fraction, ketone_label_fraction, body_weight_g, " "bicarbonate_recovery, duration_h)", type="csv", key="ox_up", ) ox_df = _read_csv(ox_up, "oxidation_tracer.csv") if ox_df is not None: st.session_state["ox_df"] = ox_df try: ox_table = fc.batch_oxidation(ox_df) st.session_state["ox_table"] = ox_table st.subheader("동물별 산화 플럭스 (회수율 보정)") st.dataframe(ox_table, use_container_width=True) fig, ax = plt.subplots(figsize=(6, 3.5)) for g, sub in ox_table.groupby("group"): ax.scatter([g] * len(sub), sub["oxidation_flux"], alpha=0.7, s=40) ax.set_ylabel("산화 플럭스 (µmol/kg/h)") ax.set_title("그룹별 지방산 산화 플럭스") plt.setp(ax.get_xticklabels(), rotation=15) st.pyplot(fig) plt.close(fig) except Exception as exc: st.error(f"산화 플럭스 계산 오류: {exc}") else: st.info("산화 추적자 CSV를 업로드하거나 사이드바에서 데모 데이터를 로드하세요.") # =========================================================================== # 탭 4 — MASH 모델 참조 + QC # =========================================================================== with tab4: st.header("MASH 모델 참조범위 + QC") st.markdown( "GAN diet, CDAHFD, MCD, AMLN, HFD, db/db, ob/ob 등 대표적 MASH " "동물모델의 내장 참조범위와 측정값을 비교하여 자동 경고를 생성합니다." ) ref_rows = [] for model, metrics in fc.MASH_MODEL_REFERENCE.items(): ref_rows.append({ "model": model, "VLDL분비율 (mg/kg/h)": f"{metrics['vldl_secretion'][0]:g}–{metrics['vldl_secretion'][1]:g}", "DNL 분율": f"{metrics['dnl_fraction'][0]:g}–{metrics['dnl_fraction'][1]:g}", "산화 (µmol/kg/h)": f"{metrics['oxidation'][0]:g}–{metrics['oxidation'][1]:g}", }) st.subheader("내장 참조범위 (연구용 참고치)") st.dataframe(pd.DataFrame(ref_rows), use_container_width=True) st.subheader("측정값 QC 점검") model_sel = st.selectbox( "비교할 MASH 모델", list(fc.MASH_MODEL_REFERENCE.keys()), key="qc_model" ) vldl_table = st.session_state.get("vldl_table") dnl_table = st.session_state.get("dnl_table") ox_table = st.session_state.get("ox_table") if vldl_table is None and dnl_table is None and ox_table is None: st.info("탭 1~3에서 데이터를 먼저 처리하면 QC 경고가 표시됩니다.") else: all_msgs = [] if vldl_table is not None: mean_v = float(np.nanmean(vldl_table["secretion_rate_norm"])) all_msgs += [("VLDL", m) for m in fc.qc_against_reference("vldl_secretion", mean_v, model_sel)] st.write(f"코호트 평균 VLDL 분비율 = {mean_v:.1f} mg/kg/h") if dnl_table is not None: mean_d = float(np.nanmean(dnl_table["fractional_dnl"])) all_msgs += [("DNL", m) for m in fc.qc_against_reference("dnl_fraction", mean_d, model_sel)] st.write(f"코호트 평균 fractional DNL = {mean_d:.3f}") if ox_table is not None: mean_o = float(np.nanmean(ox_table["oxidation_flux"])) all_msgs += [("산화", m) for m in fc.qc_against_reference("oxidation", mean_o, model_sel)] st.write(f"코호트 평균 산화 플럭스 = {mean_o:.1f} µmol/kg/h") # 추가 데이터-품질 경고 if vldl_table is not None: n_nl = int(vldl_table["nonlinear_warning"].sum()) if n_nl: all_msgs.append(("VLDL", f"{n_nl} 마리 비선형 포화 — 채혈 시점 재검토.")) low_r2 = vldl_table[vldl_table["r_squared"] < r2_threshold] if len(low_r2): all_msgs.append(("VLDL", f"{len(low_r2)} 마리 회귀 r² 미달 — enrichment 불안정 의심.")) dnl_df_now = st.session_state.get("dnl_df") if dnl_df_now is not None and "precursor_enrichment" in dnl_df_now.columns: if (dnl_df_now["precursor_enrichment"] <= 0).any(): all_msgs.append(("DNL", "전구체 풀 enrichment 누락/0 — MIDA 가정 불가.")) if all_msgs: for tag, m in all_msgs: st.warning(f"[{tag}] {m}") else: st.success("모든 측정값이 선택 모델의 참조범위 내에 있으며 QC 경고 없음.") # =========================================================================== # 탭 5 — 코호트 통계 + 리포트 # =========================================================================== with tab5: st.header("코호트 통계 + 기전 분류 리포트") st.markdown( "그룹별 VLDL 분비율 / DNL 분율 / 산화 플럭스의 ANOVA(및 가능 시 " "혼합효과 모델)를 수행하고, 개입의 작용 기전을 " "**합성억제형 / 분비변화형 / 산화촉진형**으로 분류합니다." ) vldl_table = st.session_state.get("vldl_table") dnl_table = st.session_state.get("dnl_table") ox_table = st.session_state.get("ox_table") meta_df = st.session_state.get("meta_df") if vldl_table is None or dnl_table is None or ox_table is None: st.info("탭 1~3을 모두 처리하면 코호트 통계와 리포트가 생성됩니다.") else: # batch 임의효과 병합 if meta_df is not None and "batch" in meta_df.columns: batch_map = meta_df.set_index("animal_id")["batch"].to_dict() for t in (vldl_table, dnl_table, ox_table): t["batch"] = t["animal_id"].map(batch_map) report_lines = [] def _analyze(label, table, col): st.subheader(label) summ = fc.group_summary(table, col) st.dataframe(summ, use_container_width=True) anova = fc.one_way_anova(table, col) pval = anova["p_value"] ptxt = f"{pval:.4f}" if not np.isnan(pval) else "N/A (scipy 필요)" st.write( f"일원배치 ANOVA: F={anova['F']:.3f}, p={ptxt} " f"(df {anova['df_between']:.0f}, {anova['df_within']:.0f})" ) mm = fc.mixed_effects_model(table, col) if mm and "error" not in mm: st.caption("혼합효과 모델 (고정=group, 임의=batch) 적합 완료.") with st.expander(f"{label} 혼합효과 모델 요약"): st.text(mm["summary"]) elif mm and "error" in mm: st.caption(f"혼합효과 모델 미적합: {mm['error']}") else: st.caption("혼합효과 모델: statsmodels 미설치 또는 batch 정보 부족 — 생략.") report_lines.append( f"- {label}: F={anova['F']:.3f}, p={ptxt}" ) return summ sv = _analyze("VLDL-TG 분비율 (mg/kg/h)", vldl_table, "secretion_rate_norm") sd = _analyze("Fractional DNL", dnl_table, "fractional_dnl") so = _analyze("지방산 산화 플럭스 (µmol/kg/h)", ox_table, "oxidation_flux") # 기전 분류 — 대조군(첫 그룹) 대비 변화율 st.subheader("작용 기전 분류") groups = list(sv["group"]) ctrl = st.selectbox("대조군 선택", groups, key="ctrl_grp") targets = [g for g in groups if g != ctrl] tgt = st.selectbox("비교 대상군 선택", targets, key="tgt_grp") if targets else None def _pct(summ, grp): row = summ[summ["group"] == grp] return float(row["mean"].iloc[0]) if len(row) else float("nan") if tgt: def chg(summ): c = _pct(summ, ctrl) t = _pct(summ, tgt) return (t - c) / c * 100.0 if c not in (0, float("nan")) else float("nan") v_chg = chg(sv) d_chg = chg(sd) o_chg = chg(so) mech = fc.classify_mechanism(v_chg, d_chg, o_chg) c1, c2, c3 = st.columns(3) c1.metric("VLDL 분비율 변화", f"{v_chg:+.1f}%") c2.metric("DNL 분율 변화", f"{d_chg:+.1f}%") c3.metric("산화 플럭스 변화", f"{o_chg:+.1f}%") st.success( f"기전 분류: **{mech['classification']}** \n{mech['rationale']}" ) report_lines.append("") report_lines.append(f"기전 분류 ({ctrl} → {tgt}): {mech['classification']}") report_lines.append(f"근거: {mech['rationale']}") report_lines.append( f"변화율 — VLDL {v_chg:+.1f}%, DNL {d_chg:+.1f}%, 산화 {o_chg:+.1f}%" ) # Method + Result 섹션 export st.subheader("Method + Result 섹션 export (KOR/ENG)") method_kor = ( "## 방법 (Methods)\n" "VLDL-TG 분비율은 tyloxapol/poloxamer-407 정맥 주사 후 채혈한 " "혈장 TG 시계열에서 초기 선형 구간(r² 임계 기반 자동 탐지)을 " "선형회귀하여 산출하고 체중으로 정규화하였다. De novo lipogenesis는 " "2H2O body-water(또는 13C 전구체) enrichment 와 간 TG 지방산 " "mass isotopomer 분포에 MIDA(이항모델)를 적용하여 fractional DNL을 " "추정하였으며 전구체 풀 enrichment 가정에 대한 민감도를 평가하였다. " "지방산 산화는 [13C]palmitate 추적자의 호기 13CO2 및 케톤체 표지 " "회수율로부터 bicarbonate 회수율 보정 플럭스로 산출하였다. " "그룹 간 비교는 일원배치 ANOVA(가능 시 batch 임의효과 혼합모델)로 " "수행하였다.\n" ) method_eng = ( "## Methods\n" "Hepatic VLDL-TG secretion rate was determined by linear regression " "of plasma TG over the auto-detected early linear window after " "tyloxapol/poloxamer-407 injection, normalized to body weight. " "De novo lipogenesis was estimated as fractional DNL by MIDA " "(binomial model) from 2H2O body-water (or 13C precursor) enrichment " "and hepatic TG fatty-acid mass isotopomer distributions, with " "sensitivity analysis on the precursor pool enrichment assumption. " "Fatty-acid oxidation flux was computed from recovery-corrected " "13CO2 / ketone label recovery of a [13C]palmitate tracer. Groups " "were compared by one-way ANOVA (mixed-effects with batch random " "effect where available).\n" ) result_txt = "## 결과 (Results)\n" + "\n".join(report_lines) + "\n" full_report = method_kor + "\n" + method_eng + "\n" + result_txt full_report += ( "\n---\n*본 리포트는 HepatoLipidFlux로 생성되었으며 " "연구용·참고용입니다. 임상 사용 금지.*\n" ) st.text_area("리포트 미리보기", full_report, height=320) st.download_button( "리포트 다운로드 (Markdown)", data=full_report.encode("utf-8"), file_name="hepatolipidflux_report.md", mime="text/markdown", ) st.markdown("---") st.caption( "HepatoLipidFlux MVP · 연구용·참고용 (for research/reference use only) · " "오프라인 동작 · 외부 API 없음" )