# -*- coding: utf-8 -*- """HepatoMRQuant — Streamlit 앱. rodent in vivo 간 MRI(multi-echo PDFF / ¹H-MRS / MRE·SWE) 정량 도구. MASLD 도메인 / 동물실험 도구 (in vivo 이미지 정량). 본 도구는 연구용·참고용이며, 정량 결과는 사용자 검증이 필요합니다. 원시 k-space/DICOM 재구성은 범위 외(정량 CSV ingest 전제)입니다. 샘플 데이터는 합성(synthetic) 데이터입니다. 핵심 계산 로직은 main.py(표준 라이브러리)와 공유한다. """ import io import os import pandas as pd import streamlit as st import main as core # 계산 엔진 재사용 (T2* 적합·MRS·stiffness·통계·상관) DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data") st.set_page_config(page_title="HepatoMRQuant", layout="wide") st.title("HepatoMRQuant — rodent in vivo 간 MRI 정량") st.caption("MASLD 도메인 · 동물실험 도구 (in vivo 이미지 정량) · standalone") st.warning( "본 도구는 연구용·참고용이며, 정량 결과는 사용자 검증이 필요합니다. " "원시 k-space/DICOM 재구성은 범위 외(정량 CSV ingest 전제)이며, " "샘플 데이터는 합성 데이터입니다." ) # -------------------------------------------------------------------------- # 데이터 입력: 업로드 또는 샘플 로드 # -------------------------------------------------------------------------- def _read_uploaded(uploaded): return list(pd.read_csv(uploaded).to_dict("records")) if uploaded else None def _read_sample(name): return core.load_csv(os.path.join(DATA_DIR, name)) def _to_str_rows(rows): """core 함수는 csv.DictReader(문자열) 기준이므로 값 문자열화.""" out = [] for r in rows: out.append({k: ("" if v is None else str(v)) for k, v in r.items()}) return out with st.sidebar: st.header("1. 데이터 입력") use_sample = st.button("샘플 데이터 로드 (합성)") st.markdown("**또는 CSV 업로드:**") up_me = st.file_uploader("multi-echo (animal_id,group,timepoint,te_ms,signal)", type="csv", key="me") up_mrs = st.file_uploader("MRS/MRE (animal_id,group,timepoint,mrs_lipid_area," "mrs_water_area,mre_freq_hz,shear_velocity_ms)", type="csv", key="mrs") up_histo = st.file_uploader("histology (animal_id,group,timepoint,nas_score," "sirius_red_pct,cpa_pct)", type="csv", key="histo") lang = st.radio("리포트 언어", ["ko", "en"], horizontal=True) # 세션 상태로 데이터 보관 if use_sample: st.session_state["me"] = _read_sample("sample_multiecho.csv") st.session_state["mrs_data"] = _read_sample("sample_mrs_mre.csv") st.session_state["histo"] = _read_sample("sample_histology.csv") st.session_state["loaded"] = "sample" if up_me is not None: st.session_state["me"] = _to_str_rows(_read_uploaded(up_me)) st.session_state["loaded"] = "upload" if up_mrs is not None: st.session_state["mrs_data"] = _to_str_rows(_read_uploaded(up_mrs)) if up_histo is not None: st.session_state["histo"] = _to_str_rows(_read_uploaded(up_histo)) me_rows = st.session_state.get("me") mrs_rows = st.session_state.get("mrs_data") histo_rows = st.session_state.get("histo", []) if not me_rows or not mrs_rows: st.info("좌측에서 '샘플 데이터 로드'를 누르거나 multi-echo + MRS/MRE CSV를 " "업로드하세요.") st.stop() src = st.session_state.get("loaded", "") st.success(f"데이터 로드됨 ({'합성 샘플' if src == 'sample' else '업로드'}): " f"multi-echo {len(me_rows)} rows, MRS/MRE {len(mrs_rows)} rows, " f"histology {len(histo_rows)} rows") # -------------------------------------------------------------------------- # 분석 실행 (core 엔진) # -------------------------------------------------------------------------- table = core.build_quant_table(me_rows, mrs_rows) changes = core.longitudinal_changes(table) corr = core.histology_correlation(table, histo_rows) if histo_rows else None summary, tests = core.cohort_stats(table) tab1, tab2, tab3, tab4, tab5 = st.tabs([ "1. Ingest+ROI/QC", "2. PDFF T2* 보정", "3. MRS+MRE 통합", "4. 종단+histology", "5. cohort 통계+리포트"]) # --- Tab 1: ingest + QC --- with tab1: st.subheader("다중 readout ingest + ROI/QC flag") st.markdown("간 ROI multi-echo 신호·MRS lipid/water·MRE 전단파 속도 import. " "QC: MRS SNR 미달, T2* 미보정, ROI 위치/종단 매핑 불일치 flag.") qc_df = pd.DataFrame([{ "animal": r["animal_id"], "group": r["group"], "timepoint": r["timepoint"], "n_echoes": r["n_echoes"], "MRS_total": r["mrs_total_signal"], "flags": ",".join(r["flags"]) or "-", } for r in table]) st.dataframe(qc_df, use_container_width=True) flagged = qc_df[qc_df["flags"] != "-"] if len(flagged): st.warning(f"QC flag {len(flagged)}건: {', '.join(set(qc_df['flags']) - {'-'})}") else: st.info("QC flag 없음.") # --- Tab 2: PDFF T2* 보정 --- with tab2: st.subheader("PDFF T2* 보정 엔진") st.markdown("multi-echo magnitude S(TE)=S0·exp(-TE/T2*) 적합 → T2* 보정 PDFF. " "single-echo 비보정 대비 효과 표시.") pdff_df = pd.DataFrame([{ "animal": r["animal_id"], "group": r["group"], "timepoint": r["timepoint"], "S0": r["S0"], "T2*(ms)": r["T2star_ms"], "PDFF_corrected(%)": r["PDFF_pct"], "PDFF_uncorrected(%)": r["PDFF_uncorrected_pct"], } for r in table]) st.dataframe(pdff_df.round(2), use_container_width=True) try: import matplotlib.pyplot as plt first = table[0] rows = [r for r in me_rows if r["animal_id"] == first["animal_id"] and r["timepoint"] == first["timepoint"]] te = [float(r["te_ms"]) for r in rows] sig = [float(r["signal"]) for r in rows] fig, ax = plt.subplots(figsize=(5, 3)) ax.plot(te, sig, "o-", label="observed") if first["S0"] and first["T2star_ms"]: import math fit = [first["S0"] * math.exp(-t / first["T2star_ms"]) for t in te] ax.plot(te, fit, "--", label="T2* fit") ax.set_xlabel("TE (ms)"); ax.set_ylabel("signal") ax.set_title(f"{first['animal_id']} {first['timepoint']} multi-echo decay") ax.legend() st.pyplot(fig) except Exception as e: st.caption(f"(plot skipped: {e})") # --- Tab 3: MRS + MRE 통합 --- with tab3: st.subheader("MRS line fitting + MRE stiffness 통합") st.markdown("MRS PDFF = lipid/(lipid+water). stiffness μ=ρ·c² (ρ=1000 kg/m³). " "PDFF·MRS·stiffness 3 readout 통합 테이블.") tri_df = pd.DataFrame([{ "animal": r["animal_id"], "group": r["group"], "timepoint": r["timepoint"], "PDFF(%)": r["PDFF_pct"], "MRS_PDFF(%)": r["mrs_pdff_pct"], "MRE_freq(Hz)": r["mre_freq_hz"], "shear_v(m/s)": r["shear_velocity_ms"], "stiffness(kPa)": r["stiffness_kpa"], } for r in table]) st.dataframe(tri_df.round(3), use_container_width=True) # --- Tab 4: 종단 + histology + 모델 reference --- with tab4: st.subheader("종단 within-animal 변화 + histology 상관 + 모델 reference") chg_df = pd.DataFrame([{ "animal": c["animal_id"], "group": c["group"], "dPDFF": c["dPDFF"], "dMRS": c["dMRS"], "dStiffness": c["dStiffness"], "class": core.classify_response(c), } for c in changes]) st.markdown("**종단(baseline→post) 변화 + 반응 분류**") st.dataframe(chg_df.round(3), use_container_width=True) st.markdown("**동물 모델별 전형 참고 범위 (합성 가정)**") ref_df = pd.DataFrame([{ "model": k, "PDFF(%)": f"{v['pdff_pct'][0]}-{v['pdff_pct'][1]}", "stiffness(kPa)": f"{v['stiffness_kpa'][0]}-{v['stiffness_kpa'][1]}", "note": v["note"], } for k, v in core.MODEL_REFERENCE.items()]) st.dataframe(ref_df, use_container_width=True) if corr: st.markdown("**영상-조직(terminal histology) 상관 (post 매칭)**") st.write({ "matched": corr["matched"], "PDFF vs NAS (Pearson)": round(corr["PDFF_vs_NAS_pearson"], 3) if corr["PDFF_vs_NAS_pearson"] is not None else None, "PDFF vs NAS (Spearman)": round(corr["PDFF_vs_NAS_spearman"], 3) if corr["PDFF_vs_NAS_spearman"] is not None else None, "stiffness vs SiriusRed (Pearson)": round(corr["stiffness_vs_SiriusRed_pearson"], 3) if corr["stiffness_vs_SiriusRed_pearson"] is not None else None, "stiffness vs CPA (Pearson)": round(corr["stiffness_vs_CPA_pearson"], 3) if corr["stiffness_vs_CPA_pearson"] is not None else None, }) else: st.info("histology CSV 미입력 — 상관 생략.") # --- Tab 5: cohort 통계 + 리포트 --- with tab5: st.subheader("cohort 통계 + 리포트") rows = [] for g in sorted(summary): for tp in ("baseline", "post"): s = summary[g][tp] rows.append({ "group": g, "timepoint": tp, "n": s["n"], "PDFF_mean(%)": s["PDFF_mean"], "PDFF_sd": s["PDFF_sd"], "MRS_mean(%)": s["MRS_mean"], "stiffness_mean(kPa)": s["stiffness_mean"], "stiffness_sd": s["stiffness_sd"], }) st.dataframe(pd.DataFrame(rows).round(3), use_container_width=True) st.markdown("**paired t-test (baseline vs post)**") trows = [] for g, t in tests.items(): for key, label in (("PDFF_ttest", "PDFF"), ("stiffness_ttest", "stiffness")): tt = t[key] if tt[0] is not None: trows.append({"group": g, "readout": label, "t": round(tt[0], 3), "df": tt[1], "p": round(tt[2], 4)}) st.dataframe(pd.DataFrame(trows), use_container_width=True) st.markdown("**텍스트 리포트**") buf = io.StringIO() import contextlib with contextlib.redirect_stdout(buf): core._print_report(table, changes, corr or core.histology_correlation(table, []), summary, tests, lang) report_txt = buf.getvalue() st.code(report_txt) st.download_button("리포트 다운로드 (.txt)", report_txt, file_name="hepatomrquant_report.txt") st.divider() st.caption("출처 맥락: AASLD MASLD 2023 / EASL-EASD-EASO MASLD 2024. " "모든 데이터는 합성. 정량 결과는 사용자 검증 필요.")