""" app.py — RodentMealScope (로덴트밀스코프) Streamlit 앱 설치류 섭식 행동을 식사(meal) 미세구조로 분해하는 비만/신경내분비 동물실험용 도구. 자동 급이기(BioDAQ/FED3/lickometer/PhenoMaster 등)의 원자료 섭식 이벤트 로그를 표준화하여 식사 단위 지표를 산출한다. 도메인: 비만 (Obesity) | 카테고리: 동물실험 도구 용도: 참고용·연구용 (for reference / research use only) 실행: streamlit run app.py """ import io import os import numpy as np import pandas as pd import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import streamlit as st import meal_core as mc st.set_page_config(page_title="RodentMealScope", page_icon="🐭", layout="wide") DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data") DEMO_CSV = os.path.join(DATA_DIR, "sample_feeder_events.csv") # ---------------------------------------------------------------------- # 데이터 로딩 헬퍼 # ---------------------------------------------------------------------- @st.cache_data(show_spinner=False) def load_demo(): return pd.read_csv(DEMO_CSV) def synth_fallback(): """data/ CSV가 없을 때를 위한 즉석 합성 데이터.""" rng = np.random.default_rng(20260518) rows = [] start = pd.Timestamp("2026-05-12 07:00:00") for grp, n, sm, fm in [("Control", 4, 1.0, 1.0), ("Drug-A", 4, 0.6, 1.0), ("Drug-B", 4, 1.0, 0.6)]: for a in range(n): aid = f"{grp[:1]}{a + 1:02d}" for d in range(3): day = start + pd.Timedelta(days=d) nm = max(1, rng.poisson(8 * fm)) for _ in range(nm): zt = rng.uniform(12, 24) if rng.random() < 0.78 else rng.uniform(0, 12) t = day + pd.Timedelta(hours=zt) for e in range(max(1, rng.poisson(5))): rows.append({"animal_id": aid, "group": grp, "timestamp": t, "intake_amount": round(rng.uniform(0.005, 0.045) * sm, 5), "bout": f"B{len(rows)}"}) t = t + pd.Timedelta(seconds=rng.uniform(3, 45)) df = pd.DataFrame(rows) df["timestamp"] = df["timestamp"].astype(str) return df def get_fig(plot_fn): fig, ax = plt.subplots(figsize=(7, 3.6)) plot_fn(ax) fig.tight_layout() return fig # ---------------------------------------------------------------------- # 사이드바 — 데이터 입력 & 파라미터 # ---------------------------------------------------------------------- st.sidebar.title("🐭 RodentMealScope") st.sidebar.caption("설치류 식사 미세구조 분석 · 비만 동물실험 도구") st.sidebar.markdown("### 1. 데이터") source = st.sidebar.radio("데이터 소스", ["데모 데이터 불러오기", "CSV 업로드"], index=0) raw_df = None if source == "데모 데이터 불러오기": if os.path.exists(DEMO_CSV): raw_df = load_demo() st.sidebar.success(f"데모 데이터 로드 ({len(raw_df):,} events)") else: raw_df = synth_fallback() st.sidebar.warning("data/ CSV 없음 — 즉석 합성 데이터 사용") else: up = st.sidebar.file_uploader("섭식 이벤트 CSV", type=["csv"]) if up is not None: raw_df = pd.read_csv(up) st.sidebar.success(f"업로드 완료 ({len(raw_df):,} rows)") else: st.sidebar.info("CSV를 업로드하거나 데모 데이터를 선택하세요.") st.sidebar.markdown("### 2. 하드웨어") hardware = st.sidebar.selectbox("하드웨어 포맷", mc.SUPPORTED_HARDWARE, index=0) st.sidebar.markdown("### 3. 일주기 설정") lights_on = st.sidebar.slider("점등 시각 (ZT0, 시)", 0, 12, 7) st.sidebar.markdown("### 4. 식사 기준") use_override = st.sidebar.checkbox("식사 기준 수동 지정", value=False) manual_crit = st.sidebar.number_input("식사 분리 간격 (분)", 1.0, 60.0, 10.0, 0.5) st.sidebar.markdown("---") st.sidebar.caption("⚠️ 참고용·연구용 도구입니다. 임상 판단 근거로 " "사용할 수 없습니다.") # ---------------------------------------------------------------------- # 메인 # ---------------------------------------------------------------------- st.title("RodentMealScope — 설치류 식사 미세구조 분석") st.markdown( "자동 급이기 원자료를 표준화하여 **식사 크기·빈도·간격·섭취속도·포만비**와 " "**일주기 분해**, **코호트 통계**를 산출합니다.") st.info("⚠️ **참고용·연구용**(for reference / research use only) — " "본 도구의 결과는 연구 보조 목적이며 임상적 진단·치료 판단의 " "근거로 사용할 수 없습니다.", icon="⚠️") if raw_df is None: st.stop() # 파이프라인 실행 try: crit_override = manual_crit if use_override else None res = mc.run_pipeline(raw_df, hardware=hardware, criterion_override=crit_override, lights_on_hour=lights_on) except Exception as e: # noqa: BLE001 st.error(f"분석 실패: {e}") st.stop() norm = res["normalized"] crit = res["criterion"] tabs = st.tabs([ "1️⃣ 이벤트 정규화 & QC", "2️⃣ 식사 기준 도출", "3️⃣ 식사 미세구조", "4️⃣ 일주기 분해", "5️⃣ 코호트 통계 & 리포트", ]) # ---- Tab 1: 정규화 & QC ---- with tabs[0]: st.subheader("다중 하드웨어 이벤트 정규화") st.write(f"감지/지정된 하드웨어: **{norm['_hardware'].iloc[0]}** · " f"총 **{len(norm):,}** 이벤트 · " f"개체 **{norm['animal_id'].nunique()}** · " f"군 **{norm['group'].nunique()}**") c1, c2 = st.columns([2, 1]) with c1: st.markdown("**표준 스키마 (상위 200행)**") st.dataframe(norm.drop(columns=["_hardware"]).head(200), use_container_width=True, height=320) with c2: st.markdown("**QC 요약**") st.dataframe(res["qc_summary"], use_container_width=True, hide_index=True) flagged = norm[norm["qc_flag"] != ""] st.caption(f"QC 플래그 부여 이벤트: {len(flagged):,}건 " f"(SPILLAGE=흘림, DOUBLE_COUNT=이중계수, DATA_GAP=장비누락)") st.markdown("**개체별 이벤트/섭취량**") by_animal = norm.groupby(["group", "animal_id"]).agg( events=("intake_amount", "size"), total_intake_g=("intake_amount", "sum")).round(3).reset_index() st.dataframe(by_animal, use_container_width=True, hide_index=True, height=240) # ---- Tab 2: 식사 기준 도출 ---- with tabs[1]: st.subheader("객관적 식사 기준 도출 — Tolkamp 로그-생존곡선 변곡점법") c1, c2, c3 = st.columns(3) c1.metric("도출된 식사 기준", f"{crit['criterion_min']:.2f} 분") c2.metric("실제 적용 기준", f"{crit['criterion_used']:.2f} 분", delta="수동 지정" if use_override else "데이터 기반") c3.metric("판정 방법", crit["method"].split(" ")[0]) st.caption(f"방법 상세: {crit['method']} · 검출된 봉우리(log10분): {crit['peaks']}") ivs = res["intervals_min"] if crit["log_hist"].size: fig1 = get_fig(lambda ax: ( ax.bar(crit["log_centers"], crit["log_hist"], width=(crit["log_centers"][1] - crit["log_centers"][0]) if len(crit["log_centers"]) > 1 else 0.1, color="#5B8FF9", alpha=0.85), ax.axvline(np.log10(crit["criterion_used"]), color="crimson", ls="--", lw=2, label=f"식사 기준 {crit['criterion_used']:.1f}분"), ax.set_xlabel("log10(이벤트 간격, 분)"), ax.set_ylabel("빈도"), ax.set_title("이벤트 간격 분포 (로그 스케일) — 이봉성 → 변곡점"), ax.legend())) st.pyplot(fig1) if crit["surv_x"].size: fig2 = get_fig(lambda ax: ( ax.loglog(crit["surv_x"], np.clip(crit["surv_y"], 1e-9, None), color="#36B37E"), ax.axvline(crit["criterion_used"], color="crimson", ls="--", lw=2, label=f"식사 기준 {crit['criterion_used']:.1f}분"), ax.set_xlabel("이벤트 간격 (분)"), ax.set_ylabel("생존확률 P(X>t)"), ax.set_title("로그-생존곡선 (log-survivor)"), ax.legend())) st.pyplot(fig2) st.info("이봉 분포의 두 봉우리 사이 골(antimode)이 식사 내부 간격과 " "식사 사이 간격을 가르는 객관적 경계입니다. 사이드바에서 수동 " "지정도 가능합니다.") # ---- Tab 3: 식사 미세구조 ---- with tabs[2]: st.subheader("식사 미세구조 지표") per_animal = res["per_animal"] per_group = res["per_group"] st.markdown("**개체별 지표**") st.dataframe(per_animal, use_container_width=True, hide_index=True, height=300) st.markdown("**군별 요약 (평균 ± SEM)**") st.dataframe(per_group, use_container_width=True, hide_index=True) metric_labels = { "meal_size_mean_g": "평균 식사 크기 (g)", "meal_freq_per_day": "식사 빈도 (회/일)", "imi_mean_min": "평균 식사간격 IMI (분)", "ingestion_rate_g_per_min": "섭취속도 (g/분)", "satiety_ratio_min_per_g": "포만비 (분/g)", "total_intake_g": "총 섭취량 (g)", } sel = st.selectbox("막대그래프 지표", list(metric_labels.keys()), format_func=lambda k: metric_labels[k]) gp = per_group.copy() fig3 = get_fig(lambda ax: ( ax.bar(gp["group"], gp[f"{sel}_mean"], yerr=gp[f"{sel}_sem"], capsize=5, color=["#5B8FF9", "#F6BD16", "#E8684A"][:len(gp)]), ax.set_ylabel(metric_labels[sel]), ax.set_title(f"군별 {metric_labels[sel]}"))) st.pyplot(fig3) st.markdown("**식사 단위 테이블 (상위 200행)**") st.dataframe(res["meals"].head(200), use_container_width=True, hide_index=True, height=260) # ---- Tab 4: 일주기 분해 ---- with tabs[3]: st.subheader("일주기(circadian) 분해") circ = res["circadian"] st.markdown("**위상별 섭취량 (dark active / light rest)**") st.dataframe(circ, use_container_width=True, hide_index=True) ztp = res["zt_profile"] fig4 = get_fig(lambda ax: [ ax.plot(sub["zt_bin"], sub["intake_g_per_animal_day"], marker="o", label=g) for g, sub in ztp.groupby("group") ] and ( ax.axvspan(12, 24, color="0.85", alpha=0.6, label="dark active"), ax.set_xlabel("Zeitgeber Time (ZT, 시)"), ax.set_ylabel("섭취량 (g / 개체 / 일)"), ax.set_title("ZT 시간대별 섭식 프로파일"), ax.set_xlim(0, 24), ax.legend(fontsize=8))) st.pyplot(fig4) st.caption("회색 영역 = 암기(dark active phase). 정상 설치류는 야간에 " "섭식이 집중됩니다.") st.markdown("**급성 vs 만성 약물효과 분리 (투약 시각 기준)**") cc1, cc2 = st.columns(2) default_dose = pd.to_datetime(norm["timestamp"]).min() + pd.Timedelta(days=1) dose_date = cc1.date_input("투약 날짜", value=default_dose.date()) dose_time = cc2.time_input("투약 시각", value=default_dose.time()) acute_h = st.slider("급성 효과 구간 (시간)", 1.0, 24.0, 6.0, 0.5) try: dosing = pd.Timestamp.combine(pd.Timestamp(dose_date), dose_time) av = mc.acute_vs_chronic(res["zt_events"], dosing, acute_hours=acute_h) st.dataframe(av, use_container_width=True, hide_index=True) st.caption("baseline=투약 전, acute=투약 후 지정 구간, chronic=그 이후.") except Exception as e: # noqa: BLE001 st.warning(f"급성/만성 분리 계산 불가: {e}") # ---- Tab 5: 코호트 통계 & 리포트 ---- with tabs[4]: st.subheader("코호트 통계 + 기전 분류") anv = res["anova"] st.markdown("**일원배치 ANOVA (군간 비교)**") st.dataframe(anv, use_container_width=True, hide_index=True) st.markdown("**혼합효과모형 — meal_size ~ group, 개체=임의효과**") mtxt, mparams = mc.mixed_effects_intake(res["meals"]) with st.expander("MixedLM 요약 보기"): st.text(mtxt) if mparams: st.json(mparams) st.markdown("**기전 분류 — 식사 크기 감소형 vs 식사 빈도 감소형**") mech = res["mechanism"] st.dataframe(mech, use_container_width=True, hide_index=True) st.caption("식욕억제 약물이 식사 '크기'를 줄이는지 '빈도'를 줄이는지는 " "포만(satiation) vs 포만감(satiety) 기전 해석의 핵심입니다.") st.markdown("### 📄 Method + Result 리포트") lang = st.radio("언어", ["한국어", "English"], horizontal=True) report = mc.build_report(crit, res["per_group"], anv, mech, lang="ko" if lang == "한국어" else "en") st.code(report, language="markdown") st.download_button("리포트 다운로드 (.md)", report, file_name="rodentmealscope_report.md") # 결과 CSV 다운로드 st.markdown("### 결과 내보내기") d1, d2, d3 = st.columns(3) d1.download_button("개체별 지표 CSV", res["per_animal"].to_csv(index=False), file_name="per_animal_metrics.csv") d2.download_button("군별 요약 CSV", res["per_group"].to_csv(index=False), file_name="per_group_summary.csv") d3.download_button("식사 테이블 CSV", res["meals"].to_csv(index=False), file_name="meal_table.csv") st.markdown("---") st.caption("RodentMealScope · 비만 동물실험 도구 · 참고용·연구용 · " "오프라인 동작 · 합성 데이터 기반 데모")