"""app.py — MetaCalorimetria-Kor (메타칼로리메트리아). 도메인: Obesity | 카테고리: 인체실험 도구(in-clinic 간접열량측정 분석 계산기) 메타볼릭카트/ventilated hood 의 분당 VO2·VCO2 raw export 를 올리면 안정상태 자동검출 -> Weir REE·RQ·Frayn 기질산화·예측식 대비·DIT 를 산출하는 standalone Streamlit 간접열량측정 분석기. 완전 오프라인. ⚠️ 본 도구는 연구용·참고용이며 임상 의사결정에 직접 사용 금지. 실행: 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 matplotlib.use("Agg") import matplotlib.pyplot as plt import calorimetry as cal import steady_state as ss import predictive as pred import importers import demo_data HERE = os.path.dirname(os.path.abspath(__file__)) DATA_DIR = os.path.join(HERE, "data") st.set_page_config(page_title="MetaCalorimetria-Kor", layout="wide") DISCLAIMER = ( "⚠️ **본 도구는 연구용·참고용이며 임상 의사결정에 직접 사용 금지.** " "산출값은 측정 정확도·교정·환자 상태에 좌우되며 의료적 판단을 대체하지 않습니다." ) # ---------- 공통 헬퍼 ---------- def ensure_demo(): needed = ["demo_cosmed_steady.csv", "demo_unstable.csv", "demo_dit_postprandial.csv"] if not all(os.path.exists(os.path.join(DATA_DIR, n)) for n in needed): demo_data.write_all() @st.cache_data def load_reference(): return importers.load_reference() def normalized_from_upload(file, device, vo2_unit, user_mapping): raw = pd.read_csv(file) norm = importers.normalize(raw, device=None if device == "(자동/혼합)" else device, user_mapping=user_mapping or None, vo2_unit=vo2_unit) return raw, norm def plot_gas(norm, steady=None): fig, ax = plt.subplots(2, 1, figsize=(9, 5), sharex=True) t = norm["time_min"].values ax[0].plot(t, norm["VO2_Lmin"], label="VO2 (L/min)", color="tab:blue") ax[0].plot(t, norm["VCO2_Lmin"], label="VCO2 (L/min)", color="tab:red") ax[0].set_ylabel("L/min") ax[0].legend(loc="upper right", fontsize=8) rqv = np.where(norm["VO2_Lmin"] > 0, norm["VCO2_Lmin"] / norm["VO2_Lmin"], np.nan) ax[1].plot(t, rqv, label="RQ", color="tab:green") ax[1].axhline(cal.RQ_PHYS_MIN, ls="--", color="gray", lw=0.7) ax[1].axhline(cal.RQ_PHYS_MAX, ls="--", color="gray", lw=0.7) ax[1].set_ylabel("RQ") ax[1].set_xlabel("time (min)") if steady and steady.get("found"): for a in ax: a.axvspan(steady["start_min"], steady["end_min"], color="yellow", alpha=0.3) fig.tight_layout() return fig # ---------- UI ---------- st.title("MetaCalorimetria-Kor 🫁🔥") st.caption("간접열량측정 분석기 · 도메인: Obesity · in-clinic 인체실험 도구") st.info(DISCLAIMER) ensure_demo() reference = load_reference() with st.sidebar: st.header("1. 데이터 입력") source = st.radio("데이터 소스", ["합성 데모", "CSV 업로드"]) device_options = ["(자동/혼합)"] + list(reference["device_presets"].keys()) device = st.selectbox("기종 프리셋", device_options) vo2_unit = st.selectbox("VO2/VCO2 단위", ["auto", "mL/min", "L/min"]) st.header("2. 환자/단백 보정") sex = st.selectbox("성별", ["M", "F"]) age = st.number_input("나이(년)", 1, 120, 40) weight = st.number_input("체중(kg)", 1.0, 300.0, 80.0) height = st.number_input("신장(cm)", 50.0, 230.0, 170.0) ffm = st.number_input("제지방량 FFM(kg, 0=미입력)", 0.0, 200.0, 0.0) protein_corr = st.checkbox("단백 보정(UN) 적용", value=False) un = st.number_input("24h 요소질소 UN(g/day)", 0.0, 50.0, 0.0) if protein_corr else 0.0 st.header("3. 안정상태 파라미터") warmup = st.number_input("warm-up 제외(분)", 0.0, 60.0, 5.0) window = st.number_input("윈도우(분)", 1.0, 30.0, 5.0) cv_thr = st.number_input("CV 임계(%)", 1.0, 50.0, 10.0) / 100.0 # 데이터 로드 norm = None raw = None if source == "합성 데모": demo_choice = st.selectbox( "데모 파일", ["demo_cosmed_steady.csv (안정 REE)", "demo_unstable.csv (불안정)", "demo_dit_postprandial.csv (DIT 식후)"]) fname = demo_choice.split()[0] raw = pd.read_csv(os.path.join(DATA_DIR, fname)) norm = importers.normalize(raw, device=None, vo2_unit=vo2_unit) else: up = st.file_uploader("카트 raw CSV 업로드", type=["csv"]) user_map = {} if up is not None: peek = pd.read_csv(up) up.seek(0) st.write("감지된 컬럼:", list(peek.columns)) cols = ["(자동)"] + list(peek.columns) c1, c2, c3 = st.columns(3) tmap = c1.selectbox("time", cols) vmap = c2.selectbox("VO2", cols) cmap = c3.selectbox("VCO2", cols) for canon, sel in [("time", tmap), ("vo2", vmap), ("vco2", cmap)]: if sel != "(자동)": user_map[canon] = sel try: raw, norm = normalized_from_upload(up, device, vo2_unit, user_map) except Exception as e: st.error(f"정규화 실패: {e}") if norm is None: st.stop() st.success(f"정규화 완료 · 감지 단위: {norm.attrs.get('vo2_unit_detected')} · " f"행 수: {len(norm)}") tab1, tab2, tab3, tab4 = st.tabs( ["📈 안정상태/REE", "🧪 기질산화/예측식", "🍽️ DIT 식후", "📋 QC·Export"]) # 안정상태 검출 steady = ss.detect_steady_state(norm, warmup_min=warmup, window_min=window, cv_threshold=cv_thr) with tab1: st.subheader("안정상태 자동검출 + REE") st.pyplot(plot_gas(norm, steady)) if steady["found"]: st.write(f"✅ 추천 안정구간: **{steady['start_min']:.0f}–{steady['end_min']:.0f} 분** " f"(CV ≤ {cv_thr*100:.0f}%)") else: st.warning("CV 임계 내 안정구간 자동검출 실패 — 수동 구간을 지정하세요.") tmin = float(norm["time_min"].min()) tmax = float(norm["time_min"].max()) default_lo = steady["start_min"] if steady["found"] else tmin default_hi = steady["end_min"] if steady["found"] else tmax lo, hi = st.slider("분석 구간(수동 override)", tmin, tmax, (float(default_lo), float(default_hi))) sel = norm[(norm["time_min"] >= lo) & (norm["time_min"] <= hi)] if len(sel) == 0: st.error("선택 구간에 데이터가 없습니다.") st.stop() summ = cal.summarize_steady(sel, un_g_day=un, protein_correction=protein_corr) st.session_state["summary"] = summ st.session_state["sel_range"] = (lo, hi) c1, c2, c3, c4 = st.columns(4) c1.metric("REE (Weir)", f"{summ['ree_kcal_day']:.0f} kcal/day") c2.metric("RQ", f"{summ['rq']:.3f}") c3.metric("VO2", f"{summ['vo2_lmin']:.3f} L/min") c4.metric("VCO2", f"{summ['vco2_lmin']:.3f} L/min") if summ["qc_flags"]: st.warning("QC 플래그: " + ", ".join(summ["qc_flags"])) with tab2: st.subheader("Frayn 기질산화 & 예측식 대비") summ = st.session_state.get("summary") if summ is None: st.info("먼저 안정상태 탭에서 구간을 확정하세요.") else: c1, c2, c3 = st.columns(3) c1.metric("탄수화물 산화", f"{summ['cho_g_min']*60:.1f} g/h", f"{summ['cho_kcal_day']:.0f} kcal/day") c2.metric("지방 산화", f"{summ['fat_g_min']*60:.1f} g/h", f"{summ['fat_kcal_day']:.0f} kcal/day") c3.metric("비단백 RQ (npRQ)", f"{summ['np_rq']:.3f}") if summ["cho_g_min"] < 0 or summ["fat_g_min"] < 0: st.warning("⚠️ 음수 산화율 — RQ 가 생리범위 끝(0.7/1.0) 부근일 수 있음. 해석 주의.") st.markdown("**예측식 대비 측정 REE (대사적응 지표)**") ffm_val = ffm if ffm > 0 else None preds = pred.all_predictions(weight, height, age, sex, ffm_kg=ffm_val) thr = reference["reference_ranges"]["metabolic_adaptation_ratio_threshold"] adapt = pred.metabolic_adaptation(summ["ree_kcal_day"], preds, thr) rows = [] for name, d in adapt.items(): rows.append({ "예측식": name, "예측 REE": round(d["predicted"], 0) if d["predicted"] else None, "측정/예측": round(d["ratio"], 3) if d["ratio"] else None, "대사적응 의심": "예" if d["adapted"] else "아니오", }) st.dataframe(pd.DataFrame(rows), use_container_width=True) st.session_state["pred_table"] = pd.DataFrame(rows) st.caption(f"측정/예측 < {thr} 이면 대사적응(metabolic adaptation) 의심. 임의 임계.") with tab3: st.subheader("식이유발열생성(DIT) / postprandial") c1, c2 = st.columns(2) base_end = c1.number_input("기저 종료 시점(분)", 0.0, float(norm["time_min"].max()), min(30.0, float(norm["time_min"].max()))) meal_kcal = c2.number_input("섭취 에너지(kcal, 0=미입력)", 0.0, 5000.0, 500.0) dit = cal.dit_analysis(norm, baseline_end_min=base_end, meal_energy_kcal=(meal_kcal or None), un_g_day=un, protein_correction=protein_corr) st.session_state["dit"] = dit fig, ax = plt.subplots(figsize=(9, 3.5)) t = norm["time_min"].values ax.plot(t, dit["ee_series"], color="tab:purple", label="EE (kcal/day)") ax.axhline(dit["baseline_ee_kcal_day"], ls="--", color="gray", label=f"baseline {dit['baseline_ee_kcal_day']:.0f}") ax.axvline(base_end, ls=":", color="black", lw=0.8) ax.set_xlabel("time (min)") ax.set_ylabel("EE (kcal/day)") ax.legend(fontsize=8) fig.tight_layout() st.pyplot(fig) c1, c2, c3 = st.columns(3) c1.metric("기저 EE", f"{dit['baseline_ee_kcal_day']:.0f} kcal/day") c2.metric("DIT incremental AUC", f"{dit['dit_auc_kcal']:.1f} kcal") if dit["pct_thermogenesis"] is not None: c3.metric("% thermogenesis", f"{dit['pct_thermogenesis']:.1f} %") st.caption(f"피크 증분 {dit['peak_incr_kcal_day']:.0f} kcal/day @ {dit['peak_time_min']:.0f} 분. " "incremental AUC = 식후 (EE−baseline) 사다리꼴 적분(kcal).") with tab4: st.subheader("QC · 코호트 · Export") summ = st.session_state.get("summary") if summ: export = { "VO2_Lmin": summ["vo2_lmin"], "VCO2_Lmin": summ["vco2_lmin"], "RQ": summ["rq"], "REE_kcal_day": summ["ree_kcal_day"], "CHO_g_min": summ["cho_g_min"], "FAT_g_min": summ["fat_g_min"], "npRQ": summ["np_rq"], "QC_flags": ";".join(summ["qc_flags"]) or "OK", "sel_start_min": st.session_state.get("sel_range", (None, None))[0], "sel_end_min": st.session_state.get("sel_range", (None, None))[1], } dit = st.session_state.get("dit") if dit: export["DIT_AUC_kcal"] = dit["dit_auc_kcal"] export["pct_thermogenesis"] = dit["pct_thermogenesis"] edf = pd.DataFrame([export]) st.dataframe(edf, use_container_width=True) st.download_button("결과 CSV 다운로드", edf.to_csv(index=False).encode("utf-8"), "metacalorimetria_result.csv", "text/csv") # 텍스트 리포트 rep = io.StringIO() rep.write("MetaCalorimetria-Kor 리포트 (연구용·참고용)\n") rep.write("=" * 48 + "\n") for k, v in export.items(): rep.write(f"{k}: {v}\n") rep.write("\n⚠️ 본 도구는 연구용·참고용이며 임상 의사결정에 직접 사용 금지.\n") st.download_button("텍스트 리포트 다운로드", rep.getvalue().encode("utf-8"), "metacalorimetria_report.txt", "text/plain") else: st.info("안정상태 탭에서 구간을 확정하면 export 가 활성화됩니다.") st.markdown("---") st.markdown("**코호트 비교** — 여러 결과 CSV 를 업로드해 REE/RQ 분포 비교") cohort = st.file_uploader("결과 CSV 다중 업로드", type=["csv"], accept_multiple_files=True, key="cohort") if cohort: frames = [] for f in cohort: try: d = pd.read_csv(f) d["source"] = f.name frames.append(d) except Exception as e: st.error(f"{f.name}: {e}") if frames: cdf = pd.concat(frames, ignore_index=True) st.dataframe(cdf, use_container_width=True) if "REE_kcal_day" in cdf and "RQ" in cdf: fig, ax = plt.subplots(1, 2, figsize=(9, 3)) ax[0].bar(range(len(cdf)), cdf["REE_kcal_day"]) ax[0].set_title("REE (kcal/day)") ax[1].bar(range(len(cdf)), cdf["RQ"], color="tab:green") ax[1].set_title("RQ") fig.tight_layout() st.pyplot(fig) st.markdown("---") st.caption(DISCLAIMER)