""" run_all.py: run the core pipeline and the add-on modules in one command ========================================================================================= Runs, in order: 1. run_pipeline.py core pipeline and note extraction (Stage 7c) 2. note_dashboard.py note extraction summary 3. cohort_eda.py cohort summary report 4. device_extraction.py devices and equipment, with dashboard 5. exposure_extraction.py exposures, with dashboard 6. drug_repurposing.py drug repurposing cohort assembly (needs --cutoff) 7. phenopackets_etl.py GA4GH Phenopacket v2 export, de-identified 8. omop_etl.py OMOP CDM v5.4 tables (only with --vocab) and sorts what it writes into three places inside the run folder: RUN/ core pipeline outputs (dashboard_data.json and the rest): identified RUN/restricted/ row-level files with participant IDs or text excerpts: stay inside the approved environment RUN/reports/ counts-only reports with small cells held back: review in full before anything is shared RUN is the working directory: it holds "CCDA and FHIR data/" (ccda_chunks.csv and fhir_chunks.csv) and uuid_mapping.csv from the Databricks export, or files already split by patient under inputs/. Usage: python run_all.py --base-dir RUN --cutoff 2026-10-31 python run_all.py --base-dir RUN --cutoff 2026-10-31 --vital-status RUN/vital_status.csv --vocab ./vocab Registry Forge (Boyce D, Premasiri A, Sullivan S, Levine B, Vieira FG. JAMIA Open 2026;9(5):ooag200). MIT License. """ import argparse import os import shutil import sys HERE = os.path.dirname(os.path.abspath(__file__)) if HERE not in sys.path: sys.path.insert(0, HERE) def step(n, title): print(f'\n=== Step {n}: {title} ===', flush=True) def main(argv=None): ap = argparse.ArgumentParser(description='Registry Forge: run everything') ap.add_argument('--base-dir', required=True, help='working directory holding the chunk exports and uuid_mapping.csv, or inputs/') ap.add_argument('--cutoff', help='data cut-off date, YYYY-MM-DD (needed for drug repurposing)') ap.add_argument('--vital-status', help='vital_status.csv with death and last-confirmed-alive dates') ap.add_argument('--vocab', help='Athena vocabulary folder (adds OMOP tables)') ap.add_argument('--k', type=int, default=5, help='small-cell threshold for every report (default 5)') ap.add_argument('--cohort-name', default='ALS registry cohort') ap.add_argument('--no-name-dob-merge', action='store_true', help='never merge records on name and birth date') ap.add_argument('--with-excerpts', action='store_true', help='also write review dashboards with masked excerpts into restricted/') ap.add_argument('--skip', nargs='*', default=[], choices=['devices', 'exposures', 'repurposing', 'phenopackets', 'omop'], help='add-on modules to leave out') a = ap.parse_args(argv) base = os.path.abspath(a.base_dir) restricted = os.path.join(base, 'restricted') reports = os.path.join(base, 'reports') os.makedirs(restricted, exist_ok=True) os.makedirs(reports, exist_ok=True) bundle = os.path.join(base, 'dashboard_data.json') notes = os.path.join(base, 'note_extractions.csv') vital = a.vital_status or (os.path.join(base, 'vital_status.csv') if os.path.exists(os.path.join(base, 'vital_status.csv')) else None) def move_review(name): src = os.path.join(reports, name) if os.path.exists(src): shutil.move(src, os.path.join(restricted, name)) step(1, 'core pipeline and note extraction') import run_pipeline run_pipeline.main(['--base-dir', base] + (['--no-name-dob-merge'] if a.no_name_dob_merge else [])) step(2, 'note extraction summary') import note_dashboard note_dashboard.main(notes, os.path.join(reports, 'note_dashboard.html'), a.cohort_name, a.k, bundle, a.with_excerpts) move_review('note_dashboard_review.html') step(3, 'cohort summary report') import cohort_eda cohort_eda.generate_report(bundle, os.path.join(reports, 'cohort_eda_report.html'), a.k, a.cohort_name, a.cutoff) if 'devices' not in a.skip: step(4, 'devices and equipment') import device_extraction import device_dashboard device_extraction.main(bundle, restricted) device_dashboard.main(os.path.join(restricted, 'device_codes.csv'), os.path.join(restricted, 'device_extractions.csv'), os.path.join(reports, 'device_dashboard.html'), a.cohort_name, a.k, a.with_excerpts) move_review('device_dashboard_review.html') if 'exposures' not in a.skip: step(5, 'exposures') import exposure_extraction import exposure_dashboard exposure_extraction.main(bundle, restricted) exposure_dashboard.main(os.path.join(restricted, 'exposure_codes.csv'), os.path.join(restricted, 'exposure_extractions.csv'), os.path.join(reports, 'exposure_dashboard.html'), a.cohort_name, a.k, a.with_excerpts) move_review('exposure_dashboard_review.html') if 'repurposing' not in a.skip: step(6, 'drug repurposing cohort assembly') if not a.cutoff: print(' Skipped: --cutoff is needed so follow-up ends at a defined date.') else: import drug_repurposing out = os.path.join(restricted, 'drug_repurposing') drug_repurposing.main(bundle, out, min_exposed=a.k, k=a.k, cohort_name=a.cohort_name, cutoff=a.cutoff, vital_status_path=vital) for name in ('drug_repurposing_report.html', 'drug_repurposing_summary.csv'): shutil.copy(os.path.join(out, name), os.path.join(reports, name)) if 'phenopackets' not in a.skip: step(7, 'Phenopacket export (de-identified)') import phenopackets_etl phenopackets_etl.main(bundle, notes, a.vocab, restricted) if 'omop' not in a.skip: step(8, 'OMOP CDM tables') if not a.vocab: print(' Skipped: add --vocab with an Athena vocabulary folder to write OMOP tables.') else: import omop_etl omop_etl.main(bundle, a.vocab, os.path.join(restricted, 'omop_output')) print(f'\nDone.\n Identified core outputs: {base}\n Restricted row-level files: {restricted}\n' f' Counts-only reports (review before sharing): {reports}') if __name__ == '__main__': main()