"""GLP1AESignal-Kor — Streamlit pharmacovigilance dashboard. ⚠️ 참고용·연구용 — Not for clinical decision. Run: streamlit run app.py """ from __future__ import annotations import os import sys # Lazy imports so `python3 -c "import ast; ast.parse(open('app.py').read())"` passes # even without streamlit installed. try: import streamlit as st import pandas as pd except Exception: # pragma: no cover st = None # type: ignore pd = None # type: ignore HERE = os.path.dirname(os.path.abspath(__file__)) if HERE not in sys.path: sys.path.insert(0, HERE) DISCLAIMER = ( "⚠️ **참고용·연구용 — Not for clinical decision.** " "본 도구는 합성 데이터에 기반한 약물감시 워크플로 프로토타입입니다. " "실제 임상시험·규제 제출·시그널 판단에 그대로 사용할 수 없습니다." ) def _main() -> None: assert st is not None and pd is not None from modules.ingest import (load_ae, load_labs, load_meddra_mini, auto_map_ae_freetext, ingest_summary) from modules.panels import (compute_all_panels, panel_incidence, PANEL_LABEL_KO) from modules.subgroup import (all_panel_irrs, onset_distribution, dose_response, subgroup_incidence) from modules.disproportion import (load_faers, disproportionality_all, filter_glp1ra) from modules.report import (build_quarterly_summary, write_dsc_quarterly_json, write_module_docx) st.set_page_config(page_title="GLP1AESignal-Kor", layout="wide") st.title("GLP1AESignal-Kor — GLP-1RA AE Signal & Disproportionality") st.warning(DISCLAIMER) with st.sidebar: st.header("Data") ae_path = st.text_input("AE CSV", value=os.path.join(HERE, "data/synthetic_ae.csv")) lab_path = st.text_input("Labs CSV", value=os.path.join(HERE, "data/synthetic_labs.csv")) faers_path = st.text_input("FAERS CSV", value=os.path.join(HERE, "data/synthetic_faers.csv")) st.divider() ref_arm = st.selectbox("Reference arm (IRR)", ["placebo"]) drug_arms = st.multiselect( "Drug arms", ["semaglutide_2.4", "tirzepatide_15"], default=["semaglutide_2.4", "tirzepatide_15"]) early_cutoff = st.slider("Early onset cutoff (wk)", 4, 24, 12) st.divider() st.caption("MedDRA 라이센스 우회 — 합성 mini-사전 사용. FAERS는 합성 mini.") # Load ae = load_ae(ae_path) labs = load_labs(lab_path) meddra = load_meddra_mini() ae_mapped = auto_map_ae_freetext(ae, meddra) tabs = st.tabs([ "1. Ingest & MedDRA", "2. Panels (7 class-특이)", "3. IRR / Subgroup / Dose-response", "4. FAERS Disproportionality", "5. DSC Quarterly Report", ]) # ----- Tab 1: ingest ----- with tabs[0]: st.subheader("AE ingest summary") isum = ingest_summary(ae_mapped) c1, c2, c3 = st.columns(3) c1.metric("Total AE events", isum["n_events"]) c2.metric("Serious", isum["serious"]) c3.metric("Mapped (PT)", sum(1 for r in ae_mapped if r["suggested_pt_code"])) st.markdown("**By arm**") st.dataframe(pd.DataFrame(list(isum["by_arm"].items()), columns=["arm", "n_events"])) st.markdown("**By panel**") st.dataframe(pd.DataFrame(list(isum["by_panel"].items()), columns=["panel", "n_events"])) st.markdown("**Free-text → MedDRA mini 매핑 sample**") df = pd.DataFrame(ae_mapped[:50])[ ["ae_id", "arm", "panel", "freetext", "pt_term", "suggested_pt_term", "suggested_pt_conf"] ] st.dataframe(df, use_container_width=True) st.markdown(f"**MedDRA mini-dictionary** (N PT = {len(meddra)})") st.dataframe(pd.DataFrame( [{"pt_code": e.pt_code, "pt_term": e.pt_term, "soc": e.soc, "panel": e.panel} for e in meddra]), height=240) # ----- Tab 2: panels ----- with tabs[1]: st.subheader("Class-특이 signal panel 7개") panels = compute_all_panels(ae_mapped, labs) for key, pr in panels.items(): with st.expander(f"{PANEL_LABEL_KO[key]} ({key})", expanded=(key in ("pancreatitis", "thyroid"))): c1, c2 = st.columns(2) c1.markdown("**Counts by arm**") c1.dataframe(pd.DataFrame(list(pr.counts_by_arm.items()), columns=["arm", "n"])) c1.metric("Serious", pr.n_serious) c1.markdown(f"Early (<12w) / Late: **{pr.early_count} / {pr.late_count}**") c2.markdown("**Detail**") c2.json(pr.detail) inc = panel_incidence(ae_mapped, key) inc_df = pd.DataFrame([ {"arm": a, "n_subjects": v["n_subjects"], "n_cases": v["n_cases"], "incidence": v["incidence"]} for a, v in inc.items() ]) st.markdown("**Incidence per arm**") st.dataframe(inc_df, use_container_width=True) if not inc_df.empty: st.bar_chart(inc_df.set_index("arm")["incidence"]) # ----- Tab 3: IRR / subgroup / dose-response ----- with tabs[2]: st.subheader("IRR (drug vs placebo) — 모든 panel") irrs = all_panel_irrs(ae_mapped, list(PANEL_LABEL_KO.keys()), drug_arms, ref_arm) irr_df = pd.DataFrame([i.to_dict() for i in irrs]) st.dataframe(irr_df, use_container_width=True) st.divider() panel_pick = st.selectbox("Panel", list(PANEL_LABEL_KO.keys()), index=0) c1, c2 = st.columns(2) with c1: st.markdown("**Onset distribution (early vs late)**") od = onset_distribution(ae_mapped, panel_pick, early_cutoff_wk=early_cutoff) st.dataframe(pd.DataFrame(od).T) with c2: st.markdown("**Dose-response (per drug arm)**") for arm in drug_arms: dr = dose_response(ae_mapped, panel_pick, arm) if dr: df = pd.DataFrame([{"dose_mg": d, **v} for d, v in dr.items()]) st.markdown(f"_{arm}_") st.dataframe(df) st.divider() sg_key = st.selectbox("Subgroup key", ["baseline_bmi", "age", "sex", "t2dm", "prior_pancreatitis"]) sg = subgroup_incidence(ae_mapped, panel_pick, sg_key) st.dataframe(pd.DataFrame(sg).T, use_container_width=True) # ----- Tab 4: FAERS ----- with tabs[3]: st.subheader("FAERS-like background disproportionality") faers = load_faers(faers_path) disps = disproportionality_all(faers) glp1 = filter_glp1ra(disps) df = pd.DataFrame([d.to_dict() for d in glp1]) st.markdown("**GLP-1RA × Panel PT (signal = PRR≥2 + χ²≥4 + count≥3)**") st.dataframe(df.sort_values(["panel", "PRR"], ascending=[True, False]), use_container_width=True) signals = df[df["signal"]] st.metric("Signals detected", len(signals)) if not signals.empty: st.bar_chart(signals.set_index("drug")["PRR"]) # ----- Tab 5: DSC report ----- with tabs[4]: st.subheader("DSC quarterly report") quarter = st.text_input("Quarter label", value="") if st.button("Build summary"): with st.spinner("Computing..."): summary = build_quarterly_summary( ae_mapped, labs, faers_path=faers_path, drug_arms=tuple(drug_arms), ref_arm=ref_arm, quarter_label=(quarter or None), ) st.success("Summary built.") st.json({"quarter": summary["quarter"], "panels": [{"panel": p["panel"], "strength": p["signal_strength"], "susar7": p["susar_7day_candidate"], "susar15": p["susar_15day_candidate"]} for p in summary["panels"]]}) c1, c2, c3, c4 = st.columns(4) if c1.button("Export DSC JSON"): path = write_dsc_quarterly_json(summary) st.write(f"Saved: `{path}`") if c2.button("Draft RMP docx"): path = write_module_docx(summary, module="RMP") st.write(f"Saved: `{path}`") if c3.button("Draft PSUR docx"): path = write_module_docx(summary, module="PSUR") st.write(f"Saved: `{path}`") if c4.button("Draft DSUR docx"): path = write_module_docx(summary, module="DSUR") st.write(f"Saved: `{path}`") if __name__ == "__main__": if st is None: sys.stderr.write("streamlit이 설치되지 않았습니다. `pip install -r requirements.txt`\n") sys.exit(1) _main()