#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ DiaRescueMiner-Kor — Streamlit UI ================================= main.py 의 핵심 로직(표준 라이브러리)을 그대로 재사용해 UI 로 래핑한다. 네트워크 호출 없음 — 로컬 data/terminated_programs.json 만 사용. 실행: streamlit run app.py (streamlit 미설치 시 main.py 의 CLI 가 모든 기능을 동일하게 제공한다.) [디스클레이머] 참고용·연구용 가설 생성 도구. 임상 의사결정 도구 아님. """ import json import os # networkx 는 선택적 의존성. 없으면 그래프 시각화만 비활성화하고 나머지는 동작. try: import networkx as nx # noqa: F401 HAS_NX = True except Exception: HAS_NX = False import streamlit as st # main.py 의 로직 재사용 (같은 폴더) import main as engine st.set_page_config(page_title="DiaRescueMiner-Kor", layout="wide") @st.cache_data def get_data(): return engine.load_data() def render_disclaimer(): st.warning("참고용·연구용 가설 생성 도구입니다. **임상 의사결정 도구가 아닙니다.** " "약물명/중단사유는 공개 사실 기반이며, 모든 점수는 합성 데모값입니다.") def page_hypotheses(data): st.subheader("Rescue 연구 가설 (ranked)") top_n = st.slider("상위 N개", min_value=3, max_value=40, value=10) verbose = st.checkbox("근거추적(audit) 상세 표시", value=False) hyps = engine.all_hypotheses(data) st.caption("전체 %d개 가설 중 상위 %d개" % (len(hyps), top_n)) # 표로 한눈에 table = [{ "drug": h["drug"], "lever": h["lever_label"], "score": h["rescue_score"], "plaus": h["plausibility"], "feas": h["feasibility"], "impact": h["impact"], "지지 실패축": ", ".join(engine.AXIS_LABELS.get(a, a) for a in h["supporting_axes"]), } for h in hyps[:top_n]] st.dataframe(table, use_container_width=True) for i, h in enumerate(hyps[:top_n], 1): with st.expander("#%d %s -> %s (score=%.3f)" % (i, h["drug"], h["lever_label"], h["rescue_score"])): st.markdown("**표적 / class**: %s / %s" % (h["target"], h["class"])) st.markdown("**부활레버**: %s" % h["lever_label"]) st.markdown("**지지 실패축**: %s" % ", ".join(engine.AXIS_LABELS.get(a, a) for a in h["supporting_axes"])) st.markdown("**레버 상세**: %s" % h["detail"]) st.markdown("**점수내역**: plaus=%.2f · feas=%.2f · impact=%.2f · " "tgtbonus=%+.2f · source=%s" % (h["plausibility"], h["feasibility"], h["impact"], h["target_validity_bonus"], h["source"])) ms = h["min_study"] st.markdown("**최소 confirmatory 연구 초안**") st.markdown("- design: %s" % ms["design"]) st.markdown("- N: %s" % ms["suggested_N"]) st.markdown("- primary endpoint: %s" % ms["primary_endpoint"]) st.markdown("- 반증조건: %s" % ms["falsifier"]) st.markdown("- 입증조건: %s" % ms["confirmer"]) if verbose: st.markdown("**[근거 추적 / audit]**") st.json(h["audit"]) def page_registry(data): st.subheader("중단 프로그램 레지스트리") rows = [{ "drug": p["drug"], "target": p["target"], "class": p["class"], "indication": p["indication"], "phase": p["max_phase"], "year": p.get("year_stopped"), "ctgov": p.get("ctgov_status"), "why_stopped": p["why_stopped_text"], } for p in data["programs"]] st.dataframe(rows, use_container_width=True) def page_drug(data): st.subheader("약물/표적별 상세") names = [p["drug"] for p in data["programs"]] pick = st.selectbox("약물 선택", names) prog = next(p for p in data["programs"] if p["drug"] == pick) c1, c2 = st.columns(2) with c1: st.markdown("**표적**: %s" % prog["target"]) st.markdown("**class**: %s" % prog["class"]) st.markdown("**적응증**: %s" % prog["indication"]) st.markdown("**phase**: %s" % prog["max_phase"]) with c2: st.markdown("**year_stopped**: %s" % prog.get("year_stopped")) st.markdown("**ct.gov status**: %s" % prog.get("ctgov_status")) tv = prog.get("target_validity", {}) st.markdown("**표적 verdict**: %s" % tv.get("verdict", "")) st.info("why_stopped: %s" % prog["why_stopped_text"]) axes, curated, nlp = engine.derive_axes(prog) st.markdown("**실패원인 축(병합)**: %s" % ", ".join(engine.AXIS_LABELS.get(a, a) for a in axes)) st.caption("curated=%s | NLP교차검증=%s" % (curated, nlp)) st.markdown("### rescue 가설") for i, h in enumerate(engine.build_hypotheses(prog), 1): with st.expander("#%d %s (score=%.3f)" % (i, h["lever_label"], h["rescue_score"])): st.markdown("**레버 상세**: %s" % h["detail"]) ms = h["min_study"] st.markdown("- design: %s" % ms["design"]) st.markdown("- N: %s" % ms["suggested_N"]) st.markdown("- endpoint: %s" % ms["primary_endpoint"]) st.json(h["audit"]) # 온톨로지 그래프 (networkx 있을 때만) st.markdown("### 실패원인 -> 부활레버 온톨로지") if HAS_NX: try: G = nx.DiGraph() for a in axes: G.add_edge(engine.AXIS_LABELS.get(a, a), "[%s]" % prog["drug"]) for h in engine.build_hypotheses(prog): G.add_edge("[%s]" % prog["drug"], h["lever_label"]) st.caption("노드 %d · 엣지 %d (networkx)" % (G.number_of_nodes(), G.number_of_edges())) st.write({"edges": list(G.edges())}) except Exception as e: st.caption("그래프 생성 생략: %s" % e) else: # networkx 없이도 텍스트로 온톨로지 경로 표시 st.caption("(networkx 미설치 — 텍스트 경로로 표시)") for a in axes: for lv in engine.AXIS_TO_LEVER.get(a, []): st.text("%s -> [%s] -> %s" % (engine.AXIS_LABELS.get(a, a), prog["drug"], engine.LEVER_LABELS.get(lv, lv))) def page_summary(data): st.subheader("요약 통계") from collections import Counter progs = data["programs"] st.metric("중단 프로그램 수", len(progs)) st.metric("생성 rescue 가설 수", len(engine.all_hypotheses(data))) classes = Counter(p["class"] for p in progs) st.markdown("**클래스 분포**") st.bar_chart({c: n for c, n in classes.items()}) axis_counter = Counter() for p in progs: for a in p.get("failure_attribution", {}).get("primary", []): axis_counter[engine.AXIS_LABELS.get(a, a)] += 1 st.markdown("**실패원인 축 분포 (primary)**") st.bar_chart(dict(axis_counter)) def main(): st.title("DiaRescueMiner-Kor (디아레스큐마이너코어)") st.caption("당뇨·합병증 중단 신약 -> rescue 연구 가설 생성기 · 오프라인 데모 · 도메인 DM") render_disclaimer() data = get_data() page = st.sidebar.radio( "메뉴", ["Rescue 가설(ranked)", "중단 레지스트리", "약물/표적 상세", "요약 통계"], ) st.sidebar.caption("출처(무료 공개): CT.gov v2 · Inxight · Open Targets · PubMed") st.sidebar.caption("networkx: %s" % ("사용가능" if HAS_NX else "미설치(텍스트 대체)")) if page == "Rescue 가설(ranked)": page_hypotheses(data) elif page == "중단 레지스트리": page_registry(data) elif page == "약물/표적 상세": page_drug(data) else: page_summary(data) if __name__ == "__main__": main()