"""
make_synthetic_cohort.py: invented ALS participants for Registry Forge
=======================================================================
Writes a fully synthetic data set in the layout the Databricks export produces, so the
whole pipeline can be run end to end without any real record:
synthetic/
CCDA and FHIR data/
ccda_chunks.csv every document (C-CDA XML, RTF, PDF, HTML, plain text),
base64-encoded and split into chunks
fhir_chunks.csv FHIR R4 Bundles as JSON text, split into chunks
uuid_mapping.csv document_uuid -> patient_id, with demographics
vital_status.csv deaths and last confirmed alive dates from outside the records
README.txt
Every person is an invented Jane Doe or John Doe shell with a SYN- identifier. Names, dates
of birth and clinical histories are fabricated; no part of this data set comes from a real
record. The output is deterministic: the same seed produces the same files, so demos and
tests are stable.
Usage
-----
python make_synthetic_cohort.py # writes ./synthetic next to this script
python make_synthetic_cohort.py --out some/dir
"""
import argparse
import base64
import csv
import datetime as dt
import io
import json
import random
import shutil
import textwrap
from pathlib import Path
SEED = 20261109
TODAY_CUTOFF = dt.date(2026, 10, 31) # data cut-off used by the demos
CHUNK = 4000 # characters per chunk row
try:
from reportlab.lib.pagesizes import letter
from reportlab.pdfgen import canvas
HAVE_REPORTLAB = True
except ImportError: # PDFs fall back to text documents
HAVE_REPORTLAB = False
# RxNorm ingredient codes, for demonstration only
RX = {
# Codes left blank where they were not checked; the display name still identifies the medicine
'riluzole': '35623', 'edaravone': '', 'baclofen': '1292', 'tizanidine': '57258',
'glycopyrrolate': '', 'dextromethorphan-quinidine': '', 'simvastatin': '36567',
'atorvastatin': '83367', 'tamsulosin': '77492', 'sildenafil': '136411', 'cholecalciferol': '2418',
'metformin': '6809', 'sertraline': '36437', 'gabapentin': '25480', 'lisinopril': '29046',
'omeprazole': '7646', 'amitriptyline': '704', 'tofersen': '',
}
LOINC = {
'alsfrs': ('82953-1', 'Total score [ALSFRS-R]', '{score}'),
'fvc_pct': ('19872-1', 'FVC measured/predicted', '%'),
'weight': ('29463-7', 'Body weight', 'kg'),
'ck': ('2157-6', 'Creatine kinase [Enzymatic activity/volume] in Serum or Plasma', 'U/L'),
'creatinine': ('2160-0', 'Creatinine [Mass/volume] in Serum or Plasma', 'mg/dL'),
'albumin': ('1751-7', 'Albumin [Mass/volume] in Serum or Plasma', 'g/dL'),
}
DIAGNOSES = [ # (ICD-10-CM, SNOMED CT or None, display, how many)
('G12.21', '86044005', 'Amyotrophic lateral sclerosis', 46),
('G12.23', None, 'Primary lateral sclerosis', 5),
('G12.22', None, 'Progressive bulbar palsy', 4),
('G12.20', '230258005', 'Motor neuron disease, unspecified', 3),
]
COMORBID = [
('E78.5', '55822004', 'Hyperlipidemia', 'simvastatin'),
('E78.00', '55822004', 'Pure hypercholesterolemia', 'atorvastatin'),
('N40.0', '266569009', 'Benign prostatic hyperplasia', 'tamsulosin'),
('E11.9', '44054006', 'Type 2 diabetes mellitus', 'metformin'),
('I10', '38341003', 'Essential hypertension', 'lisinopril'),
('F32.9', '35489007', 'Depressive disorder', 'sertraline'),
('E55.9', '34713006', 'Vitamin D deficiency', 'cholecalciferol'),
('K21.9', '235595009', 'Gastro-esophageal reflux disease', 'omeprazole'),
]
HEALTH_SYSTEMS = ['Riverbend Health', 'Northgate Medical Center', 'Lakeside Neurology Associates',
'Hillcrest University Hospital']
ONSET = ['limb-onset', 'limb-onset', 'bulbar-onset', 'upper limb onset', 'lower limb onset', 'respiratory-onset']
ESCORIAL = ['clinically definite ALS', 'clinically probable ALS', 'lab-supported probable ALS', 'clinically possible ALS']
FAMILY = ['No family history of ALS or dementia.', 'No family history of ALS or dementia.',
'Father was diagnosed with ALS at age 61.', 'Family history of ALS in a paternal aunt.',
'Mother had frontotemporal dementia.', 'No known family history of neurodegenerative disease.']
GENETIC = ['Genetic testing: C9orf72 repeat expansion positive (pathogenic).',
'Genetic testing: SOD1 pathogenic variant (p.Ala5Val) detected.',
'Genetic testing: no pathogenic variants in C9orf72, SOD1, FUS or TARDBP.',
'Genetic testing: no pathogenic variants in C9orf72, SOD1, FUS or TARDBP.',
'Genetic testing: FUS variant of uncertain significance.',
'Genetic testing: no pathogenic variants in C9orf72, SOD1, FUS or TARDBP.']
EXPOSURES = ['Served in the US Army during the Gulf War, deployed 1990 to 1991.',
'Former smoker, 20 pack-years, quit 2010.', 'Worked as a farmer with pesticide exposure for 30 years.',
'Played college football for four years with several concussions.',
'Worked as an electrician for 25 years.', 'Occupational exposure to lead as a battery plant worker.',
'Lived near a lake with recurring cyanobacteria blooms.', 'Degreasing work with trichloroethylene exposure.',
'Never smoker. No military service.', '', '', '']
DEVICES = ['Uses a power wheelchair with tilt-in-space.', 'Uses a rollator walker for ambulation.',
'Uses a Tobii Dynavox eye-gaze speech-generating device.', 'Uses a cough assist device twice daily.',
'Hoyer lift and hospital bed at home.', 'Bilateral ankle-foot orthoses; uses a manual wheelchair outside the home.',
'Kangaroo feeding pump for overnight feeds.', 'Bedside commode and shower chair.', '', '']
def iso(d):
return d.isoformat()
def fmt_us(d):
return f'{d.month}/{d.day}/{d.year}'
def b64(text):
return base64.b64encode(text.encode('utf-8')).decode('ascii')
def pick(seq, i):
return seq[i % len(seq)] if seq else ''
# ---------------------------------------------------------------------------
# Participant histories
# ---------------------------------------------------------------------------
def build_participants(rng):
people = []
idx = 0
for icd, sct, display, n in DIAGNOSES:
for _ in range(n):
idx += 1
pid = f'SYN-{idx:04d}'
sex = rng.choice(['female', 'male', 'male'])
given = 'Jane' if sex == 'female' else 'John'
age = rng.randint(42, 79)
dx = dt.date(2021, 6, 1) + dt.timedelta(days=rng.randint(0, 1000))
dob = dt.date(dx.year - age, rng.randint(1, 12), rng.randint(1, 28))
symptom_onset = dx - dt.timedelta(days=rng.randint(180, 540))
als = icd == 'G12.21'
slope = rng.uniform(0.4, 1.4) if als else rng.uniform(0.1, 0.4) # ALSFRS-R points per month
frs0 = rng.randint(34, 44)
died = als and rng.random() < 0.42
death = None
if died:
death = dx + dt.timedelta(days=rng.randint(420, 1500))
if death > TODAY_CUTOFF:
death = None
last_alive = None if death else (TODAY_CUTOFF - dt.timedelta(days=rng.randint(0, 220)) if rng.random() < 0.7 else None)
people.append(dict(
pid=pid, icd=icd, sct=sct, display=display, als=als, sex=sex, given=given,
family=f'Doe-{idx:03d}', dob=dob, mrn=f'SYN{idx:05d}', dx=dx, symptom_onset=symptom_onset,
slope=slope, frs0=frs0, fvc0=rng.randint(72, 98), death=death, last_alive=last_alive,
onset=pick(ONSET, idx * 7), escorial=pick(ESCORIAL, idx * 3), family_hx=pick(FAMILY, idx * 5),
genetic=pick(GENETIC, idx * 11), exposure=pick(EXPOSURES, idx * 13), device=pick(DEVICES, idx * 17),
ecas=rng.randint(88, 120), ftd=(rng.random() < 0.15),
riluzole=rng.random() < 0.85, edaravone=rng.random() < 0.3 and als, nuedexta=rng.random() < 0.25,
peg=als and rng.random() < 0.45, niv=als and rng.random() < 0.6, trach=als and rng.random() < 0.08,
comorbid=rng.sample(COMORBID, rng.randint(0, 3)),
systems=rng.sample(HEALTH_SYSTEMS, 2 if rng.random() < 0.4 else 1),
race=rng.choice([('2106-3', 'White'), ('2106-3', 'White'), ('2054-5', 'Black or African American'),
('2028-9', 'Asian'), ('2131-1', 'Other Race')]),
))
return people
def cap(p, d):
limit = TODAY_CUTOFF
if p.get('death'):
limit = min(limit, p['death'] - dt.timedelta(days=7))
return min(d, limit)
def frs_at(p, when):
months = max(0, (when - p['dx']).days / 30.44)
return max(4, round(p['frs0'] - p['slope'] * months))
def fvc_at(p, when):
months = max(0, (when - p['dx']).days / 30.44)
return max(25, round(p['fvc0'] - p['slope'] * 1.6 * months))
# ---------------------------------------------------------------------------
# Note text, written the way the facts appear in ALS clinic notes
# ---------------------------------------------------------------------------
def clinic_note(p, when, rng_i):
frs = frs_at(p, when)
b = min(12, round(frs * 0.25 + (rng_i % 3) - 1))
fm = min(12, round(frs * 0.24))
gm = min(12, round(frs * 0.23))
resp = max(0, min(12, frs - b - fm - gm))
lines = [f'ALS MULTIDISCIPLINARY CLINIC NOTE Visit date: {iso(when)}',
f'Diagnosis: {p["display"].lower()}, diagnosed {p["dx"].strftime("%B %Y")}; {p["onset"]}, symptom onset {p["symptom_onset"].strftime("%B %Y")}.']
if p['als']:
lines.append(f'Meets revised El Escorial criteria for {p["escorial"]}.')
lines += [f'ALSFRS-R total score: {frs}/48. ALSFRS-R subscores: bulbar {b}, fine motor {fm}, gross motor {gm}, respiratory {resp}.',
f'Pulmonary function today: FVC {round(3.6 * fvc_at(p, when) / 100, 1)} L ({fvc_at(p, when)}% predicted), seated.']
if p['riluzole']:
lines.append(f'Medications: started riluzole 50 mg twice daily on {fmt_us(p["dx"] + dt.timedelta(days=21))}.')
if p['edaravone']:
lines.append(f'Edaravone started {iso(p["dx"] + dt.timedelta(days=90))}, oral suspension daily.')
if p['peg'] and when > p['dx'] + dt.timedelta(days=300):
lines.append(f'PEG tube placed on {iso(p["dx"] + dt.timedelta(days=300))}; tolerating feeds.')
if p['niv'] and when > p['dx'] + dt.timedelta(days=240):
lines.append(f'BiPAP initiated on {fmt_us(p["dx"] + dt.timedelta(days=240))} for nocturnal hypoventilation.')
if p['trach'] and when > p['dx'] + dt.timedelta(days=500):
lines.append(f'Tracheostomy placed on {iso(p["dx"] + dt.timedelta(days=500))}; home ventilator.')
lines.append(f'Cognition: ECAS total {p["ecas"]}/136.' + (' Behavioral variant FTD features noted by family.' if p['ftd'] else ''))
lines.append(p['family_hx'])
if p['exposure']:
lines.append(f'Social history: {p["exposure"]}')
if p['device']:
lines.append(f'Equipment: {p["device"]}')
if rng_i % 4 == 0:
lines.append('Plan: OT evaluation for wheelchair seating and positioning; SLP referral for AAC evaluation.')
return '\n'.join(x for x in lines if x)
def emg_text(p):
d = p['dx'] - dt.timedelta(days=14)
return (f'ELECTRODIAGNOSTIC REPORT (EMG/NCS) Study date: {iso(d)}\n'
'Findings: active denervation (fibrillation potentials and positive sharp waves) with chronic reinnervation in '
'cervical, thoracic and lumbosacral segments. Sensory responses normal.\n'
f'Impression: findings support a diagnosis of {p["display"].lower()}.')
def genetic_text(p):
d = p['dx'] + dt.timedelta(days=60)
return (f'NEUROGENETICS LABORATORY REPORT Report date: {iso(d)}\n'
'Test: ALS gene panel with C9orf72 repeat-primed PCR (synthetic laboratory).\n'
f'Result: {p["genetic"]}\n'
'Interpretation applies to the tested genes only; a negative result does not exclude a genetic cause.')
def hospice_text(p):
return (f'HOSPICE DISCHARGE SUMMARY\nPatient died on {iso(p["death"])} at home under hospice care. '
f'Cause: respiratory failure due to {p["display"].lower()}.')
# ---------------------------------------------------------------------------
# FHIR
# ---------------------------------------------------------------------------
def coding(system, code, display):
return {'system': system, 'code': code, 'display': display}
def fhir_bundle(p, system_name, sys_index, rng):
local_id = f"{system_name.split()[0].lower()}-{p['pid'][-4:]}"
patient = {'resourceType': 'Patient', 'id': local_id,
'identifier': [{'type': {'coding': [{'code': 'MR'}], 'text': 'MRN'}, 'value': f"{p['mrn']}{sys_index}"}],
'name': [{'use': 'official', 'family': p['family'], 'given': [p['given']]}],
'gender': p['sex'], 'birthDate': iso(p['dob']),
'extension': [{'url': 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-race',
'extension': [{'url': 'ombCategory', 'valueCoding': {'system': 'urn:oid:2.16.840.1.113883.6.238',
'code': p['race'][0], 'display': p['race'][1]}},
{'url': 'text', 'valueString': p['race'][1]}]}]}
if p['death'] and sys_index == 0:
patient['deceasedDateTime'] = iso(p['death'])
subj = {'reference': f'Patient/{local_id}'}
entries = [patient]
n = 0
def nid(kind):
nonlocal n
n += 1
return f'{kind.lower()}-{local_id}-{n}'
def med(resource_type, name, when, status='active', extra=None):
res = {'resourceType': resource_type, 'id': nid(resource_type), 'subject': subj, 'status': status,
'medicationCodeableConcept': {'coding': [coding('http://www.nlm.nih.gov/research/umls/rxnorm', RX.get(name, ''), name)],
'text': name}}
if resource_type == 'MedicationRequest':
res.update({'intent': 'order', 'authoredOn': iso(when)})
elif resource_type == 'MedicationDispense':
res['whenHandedOver'] = iso(when)
else:
res['effectiveDateTime'] = iso(when)
res.update(extra or {})
entries.append(res)
if sys_index == 0:
codes = [coding('http://hl7.org/fhir/sid/icd-10-cm', p['icd'], p['display'])]
if p['sct']:
codes.append(coding('http://snomed.info/sct', p['sct'], p['display']))
entries.append({'resourceType': 'Condition', 'id': nid('Condition'), 'subject': subj,
'clinicalStatus': {'coding': [{'code': 'active'}]},
'code': {'coding': codes, 'text': p['display']},
'onsetDateTime': iso(p['dx']), 'recordedDate': iso(p['dx'] + dt.timedelta(days=2))})
if p['ftd']:
entries.append({'resourceType': 'Condition', 'id': nid('Condition'), 'subject': subj,
'clinicalStatus': {'coding': [{'code': 'active'}]},
'code': {'coding': [coding('http://hl7.org/fhir/sid/icd-10-cm', 'G31.09', 'Other frontotemporal dementia')]},
'onsetDateTime': iso(p['dx'] + dt.timedelta(days=200))})
for icd, sct, disp, _ in p['comorbid']:
entries.append({'resourceType': 'Condition', 'id': nid('Condition'), 'subject': subj,
'clinicalStatus': {'coding': [{'code': 'active'}]},
'code': {'coding': [coding('http://hl7.org/fhir/sid/icd-10-cm', icd, disp),
coding('http://snomed.info/sct', sct, disp)], 'text': disp},
'onsetDateTime': iso(p['dx'] - dt.timedelta(days=rng.randint(300, 4000)))})
if 'Gulf War' in p['exposure']:
entries.append({'resourceType': 'Condition', 'id': nid('Condition'), 'subject': subj,
'code': {'coding': [coding('http://hl7.org/fhir/sid/icd-10-cm', 'Z91.82', 'Personal history of military deployment')]},
'onsetDateTime': iso(p['dx'])})
if 'pack-years' in p['exposure']:
entries.append({'resourceType': 'Condition', 'id': nid('Condition'), 'subject': subj,
'code': {'coding': [coding('http://hl7.org/fhir/sid/icd-10-cm', 'Z87.891', 'Personal history of nicotine dependence')]},
'onsetDateTime': iso(p['dx'])})
start = p['dx'] + dt.timedelta(days=21)
if p['riluzole']:
med('MedicationRequest', 'riluzole', start)
for r in range(3):
med('MedicationDispense', 'riluzole', start + dt.timedelta(days=30 * r), status='completed')
if p['edaravone']:
med('MedicationRequest', 'edaravone', p['dx'] + dt.timedelta(days=90))
if p['nuedexta']:
med('MedicationRequest', 'dextromethorphan-quinidine', p['dx'] + dt.timedelta(days=120))
med('MedicationRequest', 'baclofen' if rng.random() < 0.6 else 'tizanidine', p['dx'] + dt.timedelta(days=45))
if rng.random() < 0.5:
med('MedicationRequest', 'glycopyrrolate', p['dx'] + dt.timedelta(days=150))
for _, _, _, m in p['comorbid']:
when = p['dx'] - dt.timedelta(days=rng.randint(30, 400))
if m == 'cholecalciferol':
med('MedicationStatement', m, when)
else:
med('MedicationRequest', m, when)
med('MedicationDispense', m, p['dx'] + dt.timedelta(days=20), status='completed')
# coded ALSFRS-R and FVC at two visits
for k in range(2):
when = cap(p, p['dx'] + dt.timedelta(days=30 + 180 * k))
entries.append({'resourceType': 'Observation', 'id': nid('Obs'), 'subject': subj, 'status': 'final',
'category': [{'coding': [{'code': 'survey'}]}],
'code': {'coding': [coding('http://loinc.org', *LOINC['alsfrs'][:2])], 'text': 'ALSFRS-R total'},
'valueQuantity': {'value': frs_at(p, when), 'unit': '{score}'}, 'effectiveDateTime': iso(when)})
entries.append({'resourceType': 'Observation', 'id': nid('Obs'), 'subject': subj, 'status': 'final',
'category': [{'coding': [{'code': 'exam'}]}],
'code': {'coding': [coding('http://loinc.org', *LOINC['fvc_pct'][:2])], 'text': 'FVC % predicted'},
'valueQuantity': {'value': fvc_at(p, when), 'unit': '%'}, 'effectiveDateTime': iso(when)})
if p['peg']:
entries.append({'resourceType': 'Procedure', 'id': nid('Proc'), 'subject': subj, 'status': 'completed',
'code': {'coding': [coding('http://www.ama-assn.org/go/cpt', '43246',
'Upper GI endoscopy with directed placement of percutaneous gastrostomy tube')],
'text': 'PEG placement'},
'performedDateTime': iso(cap(p, p['dx'] + dt.timedelta(days=300)))})
if p['niv']:
entries.append({'resourceType': 'Procedure', 'id': nid('Proc'), 'subject': subj, 'status': 'completed',
'code': {'coding': [coding('https://www.cms.gov/Medicare/Coding/HCPCSReleaseCodeSets', 'E0471',
'Respiratory assist device, bi-level pressure capability, with backup rate')],
'text': 'BiPAP with backup rate'},
'performedDateTime': iso(cap(p, p['dx'] + dt.timedelta(days=240)))})
xr = random.Random(f"{SEED}-{p['pid']}-equipment")
if xr.random() < 0.45:
code, disp = xr.choice([('K0861', 'Power wheelchair, group 3, multiple power option'),
('E0482', 'Cough stimulating device, alternating positive and negative airway pressure'),
('E2510', 'Speech generating device, synthesized speech, multiple methods'),
('E0630', 'Patient lift, hydraulic or mechanical')])
entries.append({'resourceType': 'Procedure', 'id': nid('Proc'), 'subject': subj, 'status': 'completed',
'code': {'coding': [coding('https://www.cms.gov/Medicare/Coding/HCPCSReleaseCodeSets', code, disp)],
'text': disp},
'performedDateTime': iso(cap(p, p['dx'] + dt.timedelta(days=200 + xr.randint(0, 200))))})
for k in range(2 if sys_index == 0 else 1):
when = cap(p, p['dx'] + dt.timedelta(days=10 + 120 * k + 9 * sys_index))
for lab in ('ck', 'creatinine', 'albumin', 'weight'):
code, disp, unit = LOINC[lab]
val = {'ck': rng.uniform(120, 650), 'creatinine': rng.uniform(0.4, 1.0), 'albumin': rng.uniform(3.2, 4.6),
'weight': rng.uniform(52, 98)}[lab]
entries.append({'resourceType': 'Observation', 'id': nid('Obs'), 'subject': subj, 'status': 'final',
'category': [{'coding': [{'code': 'vital-signs' if lab == 'weight' else 'laboratory'}]}],
'code': {'coding': [coding('http://loinc.org', code, disp)], 'text': disp},
'valueQuantity': {'value': round(val, 2), 'unit': unit}, 'effectiveDateTime': iso(when)})
entries.append({'resourceType': 'Encounter', 'id': nid('Enc'), 'subject': subj, 'status': 'finished',
'class': {'code': 'AMB', 'display': 'ambulatory'},
'type': [{'coding': [coding('http://snomed.info/sct', '185349003', 'Encounter for check up')], 'text': 'ALS clinic'}],
'period': {'start': iso(cap(p, p['dx'] + dt.timedelta(days=90 + 14 * sys_index)))}})
if sys_index == 0:
entries.append({'resourceType': 'DiagnosticReport', 'id': nid('DxRpt'), 'subject': subj, 'status': 'final',
'category': [{'coding': [{'code': 'NEU', 'display': 'Neurology'}]}],
'code': {'coding': [coding('http://loinc.org', '11506-3', 'Progress note')], 'text': 'EMG and nerve conduction report'},
'effectiveDateTime': iso(p['dx'] - dt.timedelta(days=14)),
'presentedForm': [{'contentType': 'text/plain', 'data': b64(emg_text(p)), 'title': 'EMG report'}]})
note_day = cap(p, p['dx'] + dt.timedelta(days=330 + 40 * sys_index))
entries.append({'resourceType': 'DocumentReference', 'id': nid('DocRef'), 'subject': subj, 'status': 'current',
'type': {'coding': [coding('http://loinc.org', '11506-3', 'Progress note')], 'text': 'ALS clinic note'},
'date': iso(note_day),
'content': [{'attachment': {'contentType': 'text/plain', 'data': b64(clinic_note(p, note_day, 1 + sys_index)),
'title': 'ALS clinic note'}}]})
return {'resourceType': 'Bundle', 'type': 'collection', 'id': f'bundle-{local_id}',
'meta': {'source': system_name, 'tag': [{'code': 'synthetic', 'display': 'Fully synthetic demonstration data'}]},
'entry': [{'fullUrl': f"urn:uuid:{e['id']}", 'resource': e} for e in entries]}
# ---------------------------------------------------------------------------
# Documents
# ---------------------------------------------------------------------------
def xml_escape(s):
return (s or '').replace('&', '&').replace('<', '<').replace('>', '>')
def ccda(p, when):
ymd = when.strftime('%Y%m%d')
note = clinic_note(p, when, 0)
meds = (['riluzole'] if p['riluzole'] else []) + [m for _, _, _, m in p['comorbid']]
med_entries = ''.join(
f"""
{xml_escape(line)}
' for line in text.split('\n'))
return f'{xml_escape(title)}{xml_escape(title)}
{paras}'
def pdf_bytes(text):
if not HAVE_REPORTLAB:
return None
buf = io.BytesIO()
c = canvas.Canvas(buf, pagesize=letter, invariant=1)
c.setTitle('Synthetic report')
y = 740
for para in text.split('\n'):
for chunk in textwrap.wrap(para, 92) or ['']:
c.drawString(54, y, chunk)
y -= 14
if y < 60:
c.showPage()
y = 740
c.save()
return buf.getvalue()
def chunks(text):
return [text[i:i + CHUNK] for i in range(0, len(text), CHUNK)] or ['']
# ---------------------------------------------------------------------------
def main(out_dir):
rng = random.Random(SEED)
out = Path(out_dir)
if out.exists():
shutil.rmtree(out)
data_dir = out / 'CCDA and FHIR data'
data_dir.mkdir(parents=True)
people = build_participants(rng)
ccda_rows, fhir_rows, mapping_rows, vital_rows = [], [], [], []
def add_doc(p, doc_id, payload_bytes):
for i, part in enumerate(chunks(base64.b64encode(payload_bytes).decode('ascii'))):
ccda_rows.append({'id': doc_id, 'chunk_index': i, 'chunk_data': part})
mapping_rows.append({'document_uuid': doc_id, 'patient_id': p['pid'], 'first_name': p['given'],
'last_name': p['family'], 'mrn': p['mrn'], 'dob': iso(p['dob']), 'gender': p['sex']})
for p in people:
n = int(p['pid'][-4:])
for i, system in enumerate(p['systems']):
fid = f"fhir-{p['pid'].lower()}-{i + 1}"
text = json.dumps(fhir_bundle(p, system, i, rng), indent=1)
for k, part in enumerate(chunks(text)):
fhir_rows.append({'id': fid, 'chunk_index': k, 'chunk_data': part})
mapping_rows.append({'document_uuid': fid, 'patient_id': p['pid'], 'first_name': p['given'],
'last_name': p['family'], 'mrn': p['mrn'], 'dob': iso(p['dob']), 'gender': p['sex']})
visit = cap(p, p['dx'] + dt.timedelta(days=rng.randint(150, 400)))
add_doc(p, f"doc-{p['pid'].lower()}-ccda", ccda(p, visit).encode('utf-8'))
later = cap(p, visit + dt.timedelta(days=150))
note = clinic_note(p, later, n)
if n % 3 == 0:
add_doc(p, f"doc-{p['pid'].lower()}-note", rtf(note).encode('utf-8'))
else:
add_doc(p, f"doc-{p['pid'].lower()}-note", html_doc('ALS clinic note', note).encode('utf-8'))
pdf = pdf_bytes(genetic_text(p)) if n % 2 == 0 else None
add_doc(p, f"doc-{p['pid'].lower()}-genetics", pdf if pdf else genetic_text(p).encode('utf-8'))
if p['death']:
add_doc(p, f"doc-{p['pid'].lower()}-hospice", hospice_text(p).encode('utf-8'))
vr = random.Random(f"{SEED}-{p['pid']}-vital")
alive = p['last_alive']
method = 'synthetic linkage result'
if alive and vr.random() < 0.5:
alive = TODAY_CUTOFF + dt.timedelta(days=vr.randint(1, 30))
method = 'synthetic registry follow-up contact'
vital_rows.append({'participant_id': p['pid'],
'status': 'deceased' if p['death'] else ('alive' if alive else 'unknown'),
'death_date': iso(p['death']) if p['death'] else '',
'date_precision': 'day' if p['death'] else '',
'last_confirmed_alive_date': iso(alive) if alive else '',
'ascertainment_method': method if p['death'] or alive else '',
'data_cutoff': iso(TODAY_CUTOFF)})
def write(path, rows, fields):
with path.open('w', newline='', encoding='utf-8') as f:
w = csv.DictWriter(f, fieldnames=fields, quoting=csv.QUOTE_ALL, lineterminator='\n')
w.writeheader()
w.writerows(rows)
write(data_dir / 'ccda_chunks.csv', ccda_rows, ['id', 'chunk_index', 'chunk_data'])
write(data_dir / 'fhir_chunks.csv', fhir_rows, ['id', 'chunk_index', 'chunk_data'])
write(out / 'uuid_mapping.csv', mapping_rows, ['document_uuid', 'patient_id', 'first_name', 'last_name', 'mrn', 'dob', 'gender'])
write(out / 'vital_status.csv', vital_rows, list(vital_rows[0]))
(out / 'README.txt').write_text(
'Fully synthetic ALS demonstration data for Registry Forge, laid out like the Databricks export.\n'
'Every person is an invented Jane Doe or John Doe shell. Generated by make_synthetic_cohort.py '
f'(seed {SEED}); data cut-off {iso(TODAY_CUTOFF)}.\n', encoding='utf-8')
print(f'Wrote {len(people)} synthetic participants, {len(mapping_rows)} documents and bundles to {out}')
return people
if __name__ == '__main__':
ap = argparse.ArgumentParser(description=__doc__.split('\n')[1])
ap.add_argument('--out', default=str(Path(__file__).resolve().parent / 'synthetic'))
main(ap.parse_args().out)