app.py

# Gradio application to predict ESI triag level based on observation
# author: tbartolo
import math
import joblib
import gradio as gr

# START models
symptom_code_map = joblib.load('resources/etc/symptom_code_map.joblib')
age_min_max_scaler = joblib.load('resources/normalizers/AGE_min_max_scaler.joblib')
bpdias_min_max_scaler = joblib.load('resources/normalizers/BPDIAS_min_max_scaler.joblib')
bpsys_min_max_scaler = joblib.load('resources/normalizers/BPSYS_min_max_scaler.joblib')
lov_min_max_scaler = joblib.load('resources/normalizers/LOV_min_max_scaler.joblib')
painscale_min_max_scaler = joblib.load('resources/normalizers/PAINSCALE_min_max_scaler.joblib')
respr_min_max_scaler = joblib.load('resources/normalizers/RESPR_min_max_scaler.joblib')
tempf_discretizer_model = joblib.load('resources/normalizers/TEMPF_discretizer_model.joblib')
waittime_min_max_scaler = joblib.load('resources/normalizers/WAITTIME_min_max_scaler.joblib')
pulse_min_max_scaler = joblib.load('resources/normalizers/PULSE_min_max_scaler.joblib')
popct_min_max_scaler = joblib.load('resources/normalizers/POPCT_min_max_scaler.joblib')
gb_classifier_model = joblib.load('resources/algorithms/gb_classifier_model.joblib')
# END models

waittime_mean = 30.86
lov_mean = 236.02
age_mean = 37.71
tempf_mean = 94.95
pulse_mean = 89.86
respr_mean = 19.89
bpsys_mean = 119.74
bpdias_mean = 69.47
popct_mean = 94.52


def master_fn(waittime, lov, age, newborn, private_residnce, sex, arrems, ambtransfer, pulse, respr, bpsys, bpdias,
              popct, painscale, seen72, initial_visit, injury, injury72, etohab, alzhd, asthma, cancer, cebvd, ckd,
              copd, chf, cad, deprn, diabtyp1, diabtyp2, diabtyp0, esrd, hpe, edhiv, hyplipid, htn, obesity, osa,
              ostprsis, substab, racer, tempf, rfv1, rfv2, rfv3, rfv4, rfv5, trauma, overdose_poison, medical_surgical):
    def waittime_normalizer(value):
        if not value or value < 0:
            value = waittime_mean

        log10_value = math.log10(value)
        return waittime_min_max_scaler.transform([[log10_value]])[0][0]

    def lov_normalizer(value):
        if not value or value <= 0:
            value = lov_mean

        log10_value = math.log10(value)
        return lov_min_max_scaler.transform([[log10_value]])[0][0]

    def age_normalizer(value):
        if not value or value < 0:
            value = age_mean

        return age_min_max_scaler.transform([[value]])[0][0]

    def newborn_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def private_residnce_normalizer(value):
        if not value:
            value = "yes"

        return yes_no_to_binary(value)

    def sex_normalizer(value):
        if not value:
            value = "male"

        if value.lower() == "female":
            return 1
        elif value.lower() == "male":
            return 0
        else:
            raise ValueError("Illegal argument: SEX must \"female\" or \"male\"")

    def arrems_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def ambtransfer_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def pulse_normalizer(value):
        if not value or value <= 0:
            value = pulse_mean

        return pulse_min_max_scaler.transform([[value]])[0][0]

    def respr_normalizer(value):
        if not value or value <= 0:
            value = respr_mean

        if value > 35:
            value = 35  # clip

        return respr_min_max_scaler.transform([[value]])[0][0]

    def bpsys_normalizer(value):
        if not value or value <= 0:
            value = bpsys_mean

        return bpsys_min_max_scaler.transform([[value]])[0][0]

    def bpdias_normalizer(value):
        if not value or value <= 0:
            value = bpdias_mean

        return bpdias_min_max_scaler.transform([[value]])[0][0]

    def popct_normalizer(value):
        if not value or value <= 0:
            value = popct_mean

        log10_value = math.log10(value)
        return popct_min_max_scaler.transform([[log10_value]])[0][0]

    def painscale_normalizer(value):
        if not value:
            value = 0

        return painscale_min_max_scaler.transform([[value]])[0][0]

    def seen72_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def initial_visit_normalizer(value):
        if not value:
            value = "yes"

        return yes_no_to_binary(value)

    def injury_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def injury72_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def etohab_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def alzhd_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def asthma_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def cancer_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def cebvd_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def ckd_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def copd_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def chf_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def cad_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def deprn_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def diabtyp2_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def diabtyp1_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def diabtyp0_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def esrd_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def hpe_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def edhiv_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def hyplipid_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def htn_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def obesity_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def osa_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def ostprsis_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def substab_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def racer_normalizer(value):
        if not value:
            value = "other"

        if value.lower() == "white":
            return [0, 1]
        elif value.lower() == "black":
            return [1, 0]
        elif value.lower() == "other":
            return [0, 0]
        else:
            raise ValueError("Illegal argument: RACER must \"white\" or \"black\" or \"other\"")

    def tempf_normalizer(value):
        if not value or value < 70:
            value = tempf_mean

        return tempf_discretizer_model.transform([[value * 10]]).flatten()

    def rfv1_normalizer(value):
        if not value:
            value = 0

        return rfv_to_mapping(value)

    def rfv2_normalizer(value):
        if not value:
            value = 0

        return rfv_to_mapping(value)

    def rfv3_normalizer(value):
        if not value:
            value = 0

        return rfv_to_mapping(value)

    def rfv4_normalizer(value):
        if not value:
            value = 0

        return rfv_to_mapping(value)

    def rfv5_normalizer(value):
        if not value:
            value = 0

        return rfv_to_mapping(value)

    def trauma_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def overdose_poison_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def medical_surgical_normalizer(value):
        if not value:
            value = "no"

        return yes_no_to_binary(value)

    def yes_no_to_binary(value):
        if value.lower() == "yes":
            return 1
        elif value.lower() == "no":
            return 0
        else:
            raise ValueError("Illegal argument: Value must \"yes\" or \"no\"")

    def rfv_to_mapping(value):
        value = symptom_code_map.get(value * 10)

        if value or value == 0:
            return decimal_to_binary_array(value, 13)
        else:
            raise ValueError("Illegal argument: Value must be valid symptom code")

    def decimal_to_binary_array(number, array_size):
        binary_representation = bin(number)[2:]  # Convert to binary and remove '0b' prefix
        binary_array = [int(bit) for bit in binary_representation.zfill(array_size)[-array_size:]]
        return binary_array

    observation = []

    observation.append(waittime_normalizer(waittime))
    observation.append(lov_normalizer(lov))
    observation.append(age_normalizer(age))
    observation.append(newborn_normalizer(newborn))
    observation.append(private_residnce_normalizer(private_residnce))
    observation.append(sex_normalizer(sex))
    observation.append((arrems_normalizer(arrems)))
    observation.append((ambtransfer_normalizer(ambtransfer)))
    observation.append(pulse_normalizer(pulse))
    observation.append(respr_normalizer(respr))
    observation.append(bpsys_normalizer(bpsys))
    observation.append(bpdias_normalizer(bpdias))
    observation.append(popct_normalizer(popct))
    observation.append(painscale_normalizer(painscale))
    observation.append(seen72_normalizer(seen72))
    observation.append(initial_visit_normalizer(initial_visit))
    observation.append(injury_normalizer(injury))
    observation.append(injury72_normalizer(injury72))
    observation.append(etohab_normalizer(etohab))
    observation.append(alzhd_normalizer(alzhd))
    observation.append(asthma_normalizer(asthma))
    observation.append(cancer_normalizer(cancer))
    observation.append(cebvd_normalizer(cebvd))
    observation.append(ckd_normalizer(ckd))
    observation.append(copd_normalizer(copd))
    observation.append(chf_normalizer(chf))
    observation.append(cad_normalizer(cad))
    observation.append(deprn_normalizer(deprn))
    observation.append(diabtyp1_normalizer(diabtyp1))
    observation.append(diabtyp2_normalizer(diabtyp2))
    observation.append(diabtyp0_normalizer(diabtyp0))
    observation.append(esrd_normalizer(esrd))
    observation.append(hpe_normalizer(hpe))
    observation.append(edhiv_normalizer(edhiv))
    observation.append(hyplipid_normalizer(hyplipid))
    observation.append(htn_normalizer(htn))
    observation.append(obesity_normalizer(obesity))
    observation.append(osa_normalizer(osa))
    observation.append(ostprsis_normalizer(ostprsis))
    observation.append(substab_normalizer(substab))

    observation = (observation + racer_normalizer(racer) + tempf_normalizer(tempf).tolist() + rfv1_normalizer(rfv1)
                   + rfv2_normalizer(rfv2)) + rfv3_normalizer(rfv3) + rfv4_normalizer(rfv4) + rfv5_normalizer(rfv5)

    observation.append(trauma_normalizer(trauma))
    observation.append(overdose_poison_normalizer(overdose_poison))
    observation.append(medical_surgical_normalizer(medical_surgical))

    return gb_classifier_model.predict([observation])[0]


demo = gr.Interface(
    master_fn,
    [
        "number",  # waittime
        "number",  # lov
        "number",  # age
        gr.Radio(["yes", "no"]),  # newborn
        gr.Radio(["yes", "no"]),  # private_residnce
        gr.Radio(["male", "female"]),  # sex
        gr.Radio(["yes", "no"]),  # arrems
        gr.Radio(["yes", "no"]),  # ambtransfer
        "number",  # pulse
        "number",  # respr
        "number",  # bpsys
        "number",  # bpdias
        "number",  # popct
        "number",  # painscale
        gr.Radio(["yes", "no"]),  # seen72
        gr.Radio(["yes", "no"]),  # initial_visit
        gr.Radio(["yes", "no"]),  # injury
        gr.Radio(["yes", "no"]),  # injury72
        gr.Radio(["yes", "no"]),  # etohab
        gr.Radio(["yes", "no"]),  # alzhd
        gr.Radio(["yes", "no"]),  # asthma
        gr.Radio(["yes", "no"]),  # cancer
        gr.Radio(["yes", "no"]),  # cebvd
        gr.Radio(["yes", "no"]),  # ckd
        gr.Radio(["yes", "no"]),  # copd
        gr.Radio(["yes", "no"]),  # chf
        gr.Radio(["yes", "no"]),  # cad
        gr.Radio(["yes", "no"]),  # deprn
        gr.Radio(["yes", "no"]),  # diabtyp1
        gr.Radio(["yes", "no"]),  # diabtyp2
        gr.Radio(["yes", "no"]),  # diabtyp0
        gr.Radio(["yes", "no"]),  # esrd
        gr.Radio(["yes", "no"]),  # hpe
        gr.Radio(["yes", "no"]),  # edhiv
        gr.Radio(["yes", "no"]),  # hyplipid
        gr.Radio(["yes", "no"]),  # htn
        gr.Radio(["yes", "no"]),  # obesity
        gr.Radio(["yes", "no"]),  # osa
        gr.Radio(["yes", "no"]),  # ostprsis
        gr.Radio(["yes", "no"]),  # substab
        gr.Radio(["white", "black", "other"]),  # racer
        "number",  # tempf
        "number",  # rfv1
        "number",  # rfv2
        "number",  # rfv3
        "number",  # rfv4
        "number",  # rfv5
        gr.Radio(["yes", "no"]),  # trauma
        gr.Radio(["yes", "no"]),  # overdose_poison
        gr.Radio(["yes", "no"])  # medical_surgical
    ],
    "number",
    title="ESI Triage Level Predictor",
    description="Enter the patient attributes and symptoms to perform the prediction. Note this application is a POC "
                "and by no means claims any sort of accuracy in its predictions.",
)

if __name__ == "__main__":
    demo.launch(show_api=False)