synthea_webservice/webservice/apps/synthea/lib/utils.py

from pathlib import Path
import pandas as pd
import shlex
import subprocess
from zipfile import ZipFile
import json

from django.conf import settings
from uuid import uuid4


def available_states():
    states = []
    # Read the timezones.csv file from the Synthea resources. This should give us all the 'state' on the first column
    df = pd.read_csv(settings.SYNTHEA_STATES_DIR / 'timezones.csv', index_col=False)
    for state in df[df.columns[0]].to_list():
        states.append({'id' : state , 'name' : state})

    #states = df[df.columns[0]].to_list()
    # Sort on name
    states = sorted(states, key=lambda k: k['name'].lower())
    #states.sort()
    return states

def available_modules():
    # Assumption here: Only .json files in the main folder are modules. The rest are submodules...
    modules = []
    for module in settings.SYNTHEA_MODULE_DIR.iterdir():
        if module.is_file() and module.suffix == '.json':
            data = json.loads(module.read_text())
            modules.append({'id' : module.name.replace('.json',''), 'name' : data['name']})

    modules = sorted(modules, key=lambda k: k['name'].lower())
    return modules

def run_synthea(state = None, population = None, gender = None, age = None, module = None):
    # Add a unique dir to the output, so multiple Synthea processes can run parallel
    temp_id = uuid4().hex
    output_folder = settings.SYNTHEA_OUTPUT_DIR / temp_id

    synthea_cmd = [settings.SYNTHEA_BASE_DIR / 'run_synthea','--exporter.baseDirectory',output_folder]
    zip_file = 'Synthea_'

    if population:
        synthea_cmd.append('-p')
        synthea_cmd.append(str(population))
        zip_file += f'population_{population}_'

    if gender:
        synthea_cmd.append('-g')
        synthea_cmd.append(gender.upper())
        zip_file += f'gender_{gender}_'

    if age:
        synthea_cmd.append('-a')
        synthea_cmd.append(age)
        zip_file += f'age_{age}_'

    if module:
        synthea_cmd.append('-m')
        synthea_cmd.append(module)
        zip_file += f'module_{module}_'

    if state:
        synthea_cmd.append(state)
        zip_file += f'state_{state}'

    process_ok = False
    log = ''
    with subprocess.Popen(synthea_cmd,cwd=settings.SYNTHEA_BASE_DIR, stdout=subprocess.PIPE,stderr=subprocess.PIPE) as process:
        for line in process.stdout:
            line = line.decode('utf8')
            log += line
            if not process_ok:
                process_ok = line.find('BUILD SUCCESSFUL') >= 0

    if process_ok:
        with ZipFile(f'{output_folder}/{zip_file}_{temp_id}.zip', 'w') as export:
            for file in (output_folder / settings.SYNTHEA_EXPORT_TYPE).iterdir():
                export.write(file,file.name)

        return (log,Path(f'{output_folder}/{zip_file}_{temp_id}.zip'))
    else:
        raise Exception(log)