""" DMTrialSupplyChain-Kor (디엠트라이얼서플라이체인코어) 당뇨 cold-chain 주사제 IMP 임상시험 약물 공급망 시뮬레이터 & drug accountability 추적기. 도메인: DM (당뇨) 카테고리: 인체실험 도구 (임상시험 운영 물류) 기술스택: Python + Streamlit + pandas + simpy(선택) + plotly(선택) 실행: streamlit run app.py 데모 데이터로 CSV 업로드 없이 즉시 동작 (data/ 폴더 합성 데이터 자동 로드). """ import os import io import pandas as pd import streamlit as st import engine as E # plotly는 선택 의존성 — 미설치 시 streamlit 기본 차트로 graceful fallback try: import plotly.express as px HAS_PLOTLY = True except ImportError: HAS_PLOTLY = False HERE = os.path.dirname(os.path.abspath(__file__)) DATA_DIR = os.path.join(HERE, "data") DISCLAIMER = ( "⚠️ **디스클레이머**: 본 도구는 연구·운영 보조용 참고 도구이며 " "실제 임상시험 규제 의사결정을 대체하지 않는다. " "모든 데이터는 합성/사용자입력이며 외부 네트워크·API를 사용하지 않는다." ) # --------------------------------------------------------------------------- # 데이터 로딩 (데모 데이터 또는 업로드 CSV) # --------------------------------------------------------------------------- @st.cache_data def load_demo(name: str) -> pd.DataFrame: return pd.read_csv(os.path.join(DATA_DIR, name)) def load_or_upload(label: str, demo_name: str, key: str) -> pd.DataFrame: up = st.sidebar.file_uploader(f"{label} CSV 업로드 (선택)", type="csv", key=key) if up is not None: return pd.read_csv(up) return load_demo(demo_name) def _bar(df, x, y, color=None, title=""): if df.empty: st.info("표시할 데이터가 없습니다.") return if HAS_PLOTLY: fig = px.bar(df, x=x, y=y, color=color, title=title, barmode="group") st.plotly_chart(fig, use_container_width=True) else: st.caption(title) idx = [x] + ([color] if color else []) st.bar_chart(df.set_index(x)[y] if not color else df.pivot_table(index=x, columns=color, values=y, aggfunc="sum")) def _line(df, x, y, color=None, title=""): if df.empty: st.info("표시할 데이터가 없습니다.") return if HAS_PLOTLY: fig = px.line(df, x=x, y=y, color=color, title=title, markers=True) st.plotly_chart(fig, use_container_width=True) else: st.caption(title) st.line_chart(df.set_index(x)[y] if not color else df.pivot_table(index=x, columns=color, values=y, aggfunc="sum")) # --------------------------------------------------------------------------- # 앱 본체 # --------------------------------------------------------------------------- def main(): st.set_page_config(page_title="DMTrialSupplyChain-Kor", layout="wide") st.title("DMTrialSupplyChain-Kor") st.caption("당뇨 cold-chain 주사제 IMP(인슐린·GLP-1RA·tirzepatide) 임상시험 " "약물 공급망 시뮬레이터 & drug accountability 추적기") st.warning(DISCLAIMER) if not HAS_PLOTLY: st.sidebar.info("plotly 미설치 — streamlit 기본 차트로 표시됩니다. " "(requirements.txt 설치 시 plotly 차트 사용)") st.sidebar.header("데이터 소스") st.sidebar.caption("업로드하지 않으면 합성 데모 데이터가 사용됩니다.") lots = load_or_upload("Lot 재고", "lots.csv", "u_lots") temp = load_or_upload("온도 로그", "temperature_log.csv", "u_temp") enr = load_or_upload("등록 로그", "enrollment_log.csv", "u_enr") acct = load_or_upload("Accountability", "accountability.csv", "u_acct") sites = load_or_upload("Site 마스터", "sites.csv", "u_sites") dpw, kd, budget = E.imp_dicts() tabs = st.tabs([ "1. Resupply 예측", "2. FEFO 만료관리", "3. 온도일탈 추적", "4. Accountability 원장", "5. 배분 What-if", ]) # -------------------- 기능 1 -------------------- with tabs[0]: st.subheader("1. 무작위배정 연동 resupply forecasting") st.markdown("등록 속도·arm별 dose·titration → site별 주차별 약물 소요량·부족 " "위험일 예측·par level 보급 trigger 제안.") c1, c2 = st.columns(2) titration = c1.slider("Titration 가중 (적정기 dose 배수)", 0.8, 1.5, 1.0, 0.05) horizon = c2.slider("예측 horizon (주)", 4, 24, 12, 1) fc = E.forecast_resupply(enr, dpw, kd, titration_factor=titration, horizon_weeks=horizon) triggers = E.resupply_triggers(fc, sites) st.markdown("**site별 보급 trigger / 부족 위험**") if not triggers.empty: def _hl(row): color = {"HIGH": "background-color:#ffd6d6", "WATCH": "background-color:#fff3cd", "OK": "background-color:#d4edda"}.get(row["risk"], "") return [color] * len(row) st.dataframe(triggers.style.apply(_hl, axis=1), use_container_width=True) n_high = (triggers["risk"] == "HIGH").sum() n_watch = (triggers["risk"] == "WATCH").sum() m1, m2, m3 = st.columns(3) m1.metric("HIGH (부족 위험)", int(n_high)) m2.metric("WATCH (par 도달)", int(n_watch)) m3.metric("제안 발주 합계 (kits)", int(triggers["suggested_order_kits"].sum())) else: st.info("trigger 계산 결과 없음.") st.markdown("**주차별 누적 약물 소요량 (kits)**") site_sel = st.selectbox("Site 선택", sorted(fc["site_id"].unique()) if not fc.empty else []) if site_sel: sub = fc[fc["site_id"] == site_sel] _line(sub, "week_index", "kits_needed", color="arm", title=f"{site_sel} — arm별 주차별 누적 kit 소요") # -------------------- 기능 2 -------------------- with tabs[1]: st.subheader("2. FEFO 만료관리 엔진") st.markdown("lot별 만료일·재고 → FEFO(First-Expiry-First-Out) 배분 순서·" "예상 폐기량·폐기 비용 산출.") c1, c2, c3 = st.columns(3) d_ins = c1.number_input("인슐린 글라진 수요(kits)", 0, 5000, 200, 10) d_sem = c2.number_input("세마글루타이드 수요(kits)", 0, 5000, 150, 10) d_tir = c3.number_input("터제파타이드 수요(kits)", 0, 5000, 120, 10) c4, c5 = st.columns(2) kit_cost = c4.number_input("kit당 비용 (원)", 0, 10_000_000, 250_000, 10_000) horizon_days = c5.slider("폐기 평가 horizon (일)", 30, 400, 180, 10) demand = {"insulin_glargine": d_ins, "semaglutide": d_sem, "tirzepatide": d_tir} fefo = E.fefo_allocation(lots, demand) st.markdown("**FEFO 배분 결과 (만료일 오름차순)**") st.dataframe(fefo, use_container_width=True) waste = E.expiry_waste(lots, demand, horizon_days=horizon_days, kit_cost=kit_cost) st.markdown(f"**horizon {horizon_days}일 내 예상 폐기 lot**") if not waste.empty: st.dataframe(waste, use_container_width=True) total_waste_kits = int(waste["remaining_kits"].sum()) total_waste_cost = float(waste["est_waste_cost"].sum()) m1, m2 = st.columns(2) m1.metric("예상 폐기량 (kits)", total_waste_kits) m2.metric("예상 폐기 비용 (원)", f"{total_waste_cost:,.0f}") _bar(waste, "lot_id", "remaining_kits", color="imp", title="lot별 예상 폐기량") else: st.success("해당 horizon 내 예상 폐기 lot 없음 (수요로 모두 소진).") # -------------------- 기능 3 -------------------- with tabs[2]: st.subheader("3. 온도일탈(excursion) 영향 추적") st.markdown("temperature 로그 → 일탈 구간·영향 lot·안정성 budget 기반 " "격리(QUARANTINE)/사용가능(USABLE) 자동 판정.") c1, c2 = st.columns(2) low_c = c1.number_input("하한 온도 (°C)", -20.0, 8.0, 2.0, 0.5) high_c = c2.number_input("상한 온도 (°C)", 2.0, 40.0, 8.0, 0.5) with st.expander("IMP별 안정성 budget (누적 실온 허용 시간, h) — 데모 템플릿"): bud_df = pd.DataFrame([ {"imp": k, "label": v["label"], "budget_hours": v["excursion_budget_h"]} for k, v in E.IMP_MASTER.items()]) st.dataframe(bud_df, use_container_width=True) exc = E.detect_excursions(temp, low_c=low_c, high_c=high_c, budget_hours_by_imp=budget) if not exc.empty: def _hl_exc(row): c = "background-color:#ffd6d6" if row["disposition"] == "QUARANTINE" else "background-color:#d4edda" return [c] * len(row) st.dataframe(exc.style.apply(_hl_exc, axis=1), use_container_width=True) m1, m2 = st.columns(2) m1.metric("격리(QUARANTINE) shipment", int((exc["disposition"] == "QUARANTINE").sum())) m2.metric("사용가능(USABLE) shipment", int((exc["disposition"] == "USABLE").sum())) _bar(exc, "shipment_id", "cumulative_excursion_hours", color="disposition", title="shipment별 누적 일탈 시간") sel_ship = st.selectbox("온도 곡선 상세 shipment", sorted(temp["shipment_id"].unique())) tsub = temp[temp["shipment_id"] == sel_ship].copy() tsub["timestamp"] = pd.to_datetime(tsub["timestamp"]) _line(tsub, "timestamp", "temp_c", title=f"{sel_ship} 온도 곡선 (°C)") else: st.info("온도 로그가 비어있거나 일탈 분석 결과 없음.") # -------------------- 기능 4 -------------------- with tabs[3]: st.subheader("4. drug accountability 원장") st.markdown("ICH GCP E6(R3) 요건 기반 dispensed/returned/destroyed 균형 대장 " "및 site별 불일치(MISMATCH) alert.") bal = E.accountability_balance(acct) if not bal.empty: def _hl_bal(row): c = "background-color:#ffd6d6" if row["alert"] == "MISMATCH" else "background-color:#d4edda" return [c] * len(row) st.dataframe(bal.style.apply(_hl_bal, axis=1), use_container_width=True) n_mis = int((bal["alert"] == "MISMATCH").sum()) m1, m2, m3 = st.columns(3) m1.metric("불일치(MISMATCH) 건", n_mis) m2.metric("총 dispensed (kits)", int(bal["dispensed"].sum())) m3.metric("총 destroyed (kits)", int(bal["destroyed"].sum())) if n_mis > 0: st.error(f"{n_mis}건의 accountability 불일치 발견 — site 재고 실사·" "원자료 reconciliation 필요 (GCP E6(R3) §8 참조).") _bar(bal, "site_id", "dispensed", color="imp", title="site별 dispensed (kits)") else: st.info("accountability 데이터 없음.") # -------------------- 기능 5 -------------------- with tabs[4]: st.subheader("5. depot→site 배분 시나리오 (what-if)") st.markdown("등록 가속·site 추가·공급 지연 what-if 시뮬레이션 " "(simpy 이산사건 시뮬레이션, 미설치 시 결정론적 모델).") c1, c2, c3 = st.columns(3) accel = c1.slider("등록 가속 배수", 0.5, 3.0, 1.5, 0.1) delay = c2.slider("공급 지연 (주)", 0, 8, 2, 1) extra = c3.slider("추가 site 수", 0, 10, 2, 1) horizon_w = st.slider("시뮬레이션 horizon (주)", 8, 36, 24, 1) use_simpy = st.checkbox("simpy 이산사건 시뮬레이션 사용 (미설치 시 자동 fallback)", True) sim = E.simulate_distribution(enr, sites, dpw, kd, enroll_accel=accel, supply_delay_weeks=delay, extra_sites=extra, horizon_weeks=horizon_w, use_simpy=use_simpy) if not sim.empty: stockout_weeks = int(sim["stockout_flag"].sum()) peak_backorder = float(sim["backorder_kits"].max()) m1, m2 = st.columns(2) m1.metric("stockout 발생 주차 수", stockout_weeks) m2.metric("최대 backorder (kits)", f"{peak_backorder:,.1f}") melt = sim.melt(id_vars="week_index", value_vars=["total_demand_kits", "total_resupplied_kits", "backorder_kits"], var_name="metric", value_name="kits") _line(melt, "week_index", "kits", color="metric", title="주차별 수요 vs 보급 vs backorder") st.dataframe(sim, use_container_width=True) st.caption("backorder가 0보다 크면 해당 주 공급 부족 — 보급 전진 배치 또는 " "depot par level 상향 검토.") else: st.info("시뮬레이션 결과 없음.") st.divider() st.caption("출처/근거: ICH GCP E6(R3) §8 (IMP accountability) · GMP cold-chain " "2-8°C 보관 · IATA PCR cold-chain · 주사제 안정성 budget(라벨 기반 데모 템플릿). " "본 도구는 참고용이며 규제 의사결정을 대체하지 않는다.") if __name__ == "__main__": main()