""" exports.py — REDCap data dictionary, CDISC SDTM-QS mapping, DSMB(한국 IRB·MFDS) 보고서, 영문 sponsor 보고서 생성. """ from __future__ import annotations import csv import io import json from datetime import date from scales import ALL_SCALES # ------------------------------------------------------------------- # REDCap data dictionary CSV # ------------------------------------------------------------------- REDCAP_HEADER = [ "Variable / Field Name", "Form Name", "Section Header", "Field Type", "Field Label", "Choices, Calculations, OR Slider Labels", "Field Note", "Text Validation Type OR Show Slider Number", "Text Validation Min", "Text Validation Max", "Identifier?", "Branching Logic", "Required Field?", "Custom Alignment", "Question Number", "Matrix Group Name", "Matrix Ranking?", "Field Annotation", ] def build_redcap_dict() -> str: out = io.StringIO() w = csv.writer(out) w.writerow(REDCAP_HEADER) # 환자 코어 w.writerow(["record_id", "demographics", "", "text", "Subject ID", "", "site-prefix-NNN", "", "", "", "y", "", "y", "", "", "", "", ""]) w.writerow(["site_id", "demographics", "", "text", "Site ID", "", "", "", "", "", "n", "", "y", "", "", "", "", ""]) w.writerow(["enrolled_date", "demographics", "", "text", "등록일", "", "", "date_ymd", "", "", "n", "", "y", "", "", "", "", ""]) w.writerow(["visit_week", "epro_visit", "", "text", "방문 주차 (W0/W4/...)", "", "", "integer", "0", "104", "n", "", "y", "", "", "", "", ""]) for scale in ALL_SCALES: form = scale["id"].lower() w.writerow([f"{form}_meta", form, scale["name_kr"], "descriptive", f"{scale['name_kr']} (검증: {scale['validation_status']['level']})", "", scale["validation_status"]["label"], "", "", "", "n", "", "n", "", "", "", "", ""]) for it in scale["items"]: rng = it.get("range", scale.get("response_range", (0, 10))) ftype = "text" choices = "" if scale["id"] in ("FACIT_FATIGUE_KOR", "PROMIS_FATIGUE_KOR_SF8A", "EQ5D5L_KOR") and it.get("type") != "vas": ftype = "radio" if scale["id"] == "FACIT_FATIGUE_KOR": choices = "0, 전혀 그렇지 않다 | 1, 약간 | 2, 보통 | 3, 상당히 | 4, 매우 그렇다" elif scale["id"] == "PROMIS_FATIGUE_KOR_SF8A": choices = "1, 전혀 | 2, 거의 없음 | 3, 가끔 | 4, 자주 | 5, 매우" elif scale["id"] == "EQ5D5L_KOR": choices = "1, 문제없음 | 2, 약간 | 3, 보통 | 4, 심함 | 5, 극심" elif scale["id"] == "CLDQ_NASH_KOR_DEMO": ftype = "radio" choices = "1, 항상 | 2, 대부분 | 3, 자주 | 4, 보통 | 5, 가끔 | 6, 드물게 | 7, 전혀 없음" else: ftype = "text" w.writerow([ it["id"], form, "", ftype, it["text_kr"], choices, "", "integer" if ftype == "text" else "", rng[0] if ftype == "text" else "", rng[1] if ftype == "text" else "", "n", "", "y", "", "", "", "", f"@SCALE={scale['id']}", ]) return out.getvalue() # ------------------------------------------------------------------- # CDISC SDTM-QS domain mapping # ------------------------------------------------------------------- SDTM_QS_HEADER = [ "STUDYID", "DOMAIN", "USUBJID", "QSSEQ", "QSTESTCD", "QSTEST", "QSCAT", "QSSCAT", "QSORRES", "QSORRESU", "QSSTRESC", "QSSTRESN", "VISITNUM", "VISIT", "QSDTC", ] # QSCAT 매핑 _QSCAT = { "CLDQ_NASH_KOR_DEMO": "CLDQ-NASH", "FACIT_FATIGUE_KOR": "FACIT-FATIGUE", "PROMIS_FATIGUE_KOR_SF8A": "PROMIS-FATIGUE", "HEPATIC_ITCH_NRS": "ITCH-NRS", "EQ5D5L_KOR": "EQ-5D-5L", } def build_sdtm_qs(entries: list, study_id: str = "MASHePROKOR-001") -> str: """entries: data/patients.json 의 ePRO entries → SDTM QS rows.""" out = io.StringIO() w = csv.writer(out) w.writerow(SDTM_QS_HEADER) seq = {} for e in entries: usubj = f"{study_id}-{e['patient_id']}" for scale_id, ans in e.get("answers", {}).items(): cat = _QSCAT.get(scale_id, scale_id) for qcd, val in ans.items(): seq[usubj] = seq.get(usubj, 0) + 1 w.writerow([ study_id, "QS", usubj, seq[usubj], qcd.upper()[:8], qcd, cat, "", str(val), "", str(val), val if isinstance(val, (int, float)) else "", e.get("week", ""), f"WEEK {e.get('week','')}", e.get("date", ""), ]) return out.getvalue() # ------------------------------------------------------------------- # DSMB 보고서 (한국 IRB·MFDS 양식 시연) # ------------------------------------------------------------------- def build_dsmb_kr_report(stats: dict, today: str | None = None) -> str: today = today or date.today().isoformat() s = stats md = [] md.append(f"# 데이터 안전성 모니터링 위원회(DSMB) 보고서\n") md.append(f"**시험명**: MASHePRO-Kor 한국어 ePRO 통합 키트 시연 시험 ") md.append(f"**보고일**: {today} ") md.append(f"**보고 형식**: 한국 IRB·MFDS DSMB 권고 양식 (시연)\n") md.append("## 1. 시험 진행 요약\n") md.append(f"- 총 등록 환자: **{s.get('n_enrolled', 0)}명**") md.append(f"- 활성 환자: {s.get('n_active', 0)}명") md.append(f"- 중도탈락(dropout): {s.get('n_dropout', 0)}명 ({s.get('dropout_rate', 0):.1%})") md.append(f"- 평균 추적 기간: {s.get('mean_followup_weeks', 0):.1f} 주\n") md.append("## 2. ePRO 컴플라이언스\n") md.append(f"- 사이트 평균 입력률: **{s.get('mean_compliance', 0):.1%}**") md.append(f"- 직전 4주 평균: {s.get('mean_compliance_4w', 0):.1%}") md.append(f"- 컴플라이언스 ≥80% 환자: {s.get('n_compliance_80plus', 0)}명") md.append(f"- 컴플라이언스 <50% 환자(고위험): {s.get('n_compliance_below50', 0)}명\n") md.append("## 3. 안전성 신호 (PRO 기반 시연)\n") md.append("- 본 시연 데이터에서는 SAE 별도 보고 없음 (합성 데이터)") md.append(f"- PROMIS-Fatigue T-score 평균: {s.get('promis_t_mean', 'NA')}") md.append(f"- 가려움 NRS 평균(최근 4주): {s.get('itch_mean_4w', 'NA')}\n") md.append("## 4. dropout 위험 환자 (HIGH risk)\n") for p in s.get("high_risk_patients", [])[:10]: md.append(f"- {p['patient_id']} (사이트 {p.get('site','?')}, " f"4주 입력률 {p.get('compliance_4w',0):.0%}, risk={p.get('risk',0):.2f})") md.append("\n## 5. 권고 사항\n") md.append("- 컴플라이언스 <50% 환자에 대해 사이트 코디네이터 직접 연락 권고") md.append("- 다음 DSMB 회의: 6주 후 예정\n") md.append("---") md.append("*본 보고서는 시연용입니다. 실제 IRB·MFDS 제출용은 별도 양식 검증 필요.*") return "\n".join(md) def build_sponsor_en_report(stats: dict, today: str | None = None) -> str: today = today or date.today().isoformat() s = stats md = [] md.append(f"# Sponsor Compliance & Power Report — MASHePRO-Kor (Demo)\n") md.append(f"**Reporting date**: {today}\n") md.append("## Enrollment\n") md.append(f"- Enrolled (n): **{s.get('n_enrolled', 0)}**") md.append(f"- Active: {s.get('n_active', 0)}") md.append(f"- Discontinued: {s.get('n_dropout', 0)} ({s.get('dropout_rate',0):.1%})\n") md.append("## ePRO Compliance\n") md.append(f"- Mean compliance (entire follow-up): **{s.get('mean_compliance', 0):.1%}**") md.append(f"- Mean compliance (last 4 weeks): {s.get('mean_compliance_4w', 0):.1%}") md.append(f"- Subjects ≥80%: {s.get('n_compliance_80plus', 0)}") md.append(f"- Subjects <50%: {s.get('n_compliance_below50', 0)} (action required)\n") md.append("## Power Implications\n") for row in s.get("power_table", []): md.append(f"- compliance {row['compliance']:.0%}: " f"required n/arm = {row['n_per_arm_required']}, " f"total = {row['total_required']}") md.append("\n## Recommended Actions\n") md.append("- Site outreach for subjects with last-4w compliance <50%.") md.append("- Re-evaluate sample size if mean compliance falls below 75%.") md.append("- Continue ePRO push-time A/B optimization at chronotype level.\n") md.append("---") md.append("*Demo prototype. Not for regulatory submission without further validation.*") return "\n".join(md) def export_all(entries: list, stats: dict, out_dir: str) -> dict: import os os.makedirs(out_dir, exist_ok=True) paths = {} p = os.path.join(out_dir, "redcap_dictionary.csv") with open(p, "w", encoding="utf-8") as f: f.write(build_redcap_dict()) paths["redcap"] = p p = os.path.join(out_dir, "sdtm_qs.csv") with open(p, "w", encoding="utf-8") as f: f.write(build_sdtm_qs(entries)) paths["sdtm"] = p p = os.path.join(out_dir, "dsmb_kr_report.md") with open(p, "w", encoding="utf-8") as f: f.write(build_dsmb_kr_report(stats)) paths["dsmb_kr"] = p p = os.path.join(out_dir, "sponsor_en_report.md") with open(p, "w", encoding="utf-8") as f: f.write(build_sponsor_en_report(stats)) paths["sponsor_en"] = p return paths if __name__ == "__main__": print(build_redcap_dict()[:600])