Writing a REST api service for exporting the generated FHIR bundle in JSON

 

Hi all,

 

Let's do some more work about the testing data generation and export the result by REST API.😁

Here, I would like to reuse the datagen.restservice class which built in the pervious article Writing a REST api service for exporting the generated patient data in .csv

This time, we are planning to generate a FHIR bundle include multiple resources for testing the FHIR repository.

Here is some reference for you, if you want to know mare about FHIR The Concept of FHIR: A Healthcare Data Standard Designed for the Future

 

OK... Let's start😆

1. Create a new utility class datagen.utli.genfhirjson.cls for hosting the FHIR data generation functions

Class datagen.utli.genfhirjson Extends %RegisteredObject
{
}

 

2. Write a Python function genfhirbundle to generate FHIR bundle in JSON string (this part I asked Chat-GPT to help me😂🤫)

- here, I added an argument isirisfhir (default value is 0) to indicate if this FHIR bundle is sending to IRIS FHIR repository or not.

The reason of doing this is the different syntax of resource reference while using uuid 

For example some of the FHIR repository by referencing a patient resource using uuid looks like the following

 "subject": {
          "reference": "Patient/3ce18a5b-b904-4c77-8a33-b09c3f8c79cc"
        }

 While IRIS FHIR repository by referencing a patient resource using uuid looks like the following

 "subject": {
          "reference": "3ce18a5b-b904-4c77-8a33-b09c3f8c79cc"
        }

As a result, an argument is added as a selector

Class datagen.utli.genfhirjson Extends %RegisteredObject
{

ClassMethod genfhirbundle(isirisfhir = 0) As %String [ Language = python ]
{
    # w ##class(datagen.utli.genfhirjson).genfhirbundle(0)
    # w ##class(datagen.utli.genfhirjson).genfhirbundle(1)
    from faker import Faker
    import random
    import json
    import uuid
    from datetime import datetime
    import base64

    fake = Faker()


    # ----------------------
    # Patient
    # ----------------------
    def generate_fhir_patient():
        patient_id = str(uuid.uuid4())
        return {
            "resourceType": "Patient",
            "id": patient_id,
            "identifier": [{
                "use": "usual",
                "system": "http://hospital.smarthealth.org/patient-ids",
                "value": str(random.randint(100000, 999999))
            }],
            "name": [{
                "use": "official",
                "family": fake.last_name(),
                "given": [fake.first_name()]
            }],
            "gender": random.choice(["male", "female"]),
            "birthDate": fake.date_of_birth(minimum_age=0, maximum_age=100).strftime("%Y-%m-%d"),
            "address": [{
                "use": "home",
                "line": [fake.street_address()],
                "city": fake.city(),
                "state": fake.state(),
                "postalCode": fake.postcode(),
                "country": fake.country()
            }],
            "telecom": [{
                "system": "phone",
                "value": fake.phone_number(),
                "use": "mobile"
            }],
            "meta": {"lastUpdated": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")}
        }


    # ----------------------
    # Practitioner
    # ----------------------
    def generate_fhir_practitioner():
        practitioner_id = str(uuid.uuid4())
        return {
            "resourceType": "Practitioner",
            "id": practitioner_id,
            "identifier": [{
                "system": "http://hospital.smarthealth.org/staff-ids",
                "value": str(random.randint(1000, 9999))
            }],
            "name": [{
                "family": fake.last_name(),
                "given": [fake.first_name()],
                "prefix": ["Dr."]
            }],
            "telecom": [{
                "system": "phone",
                "value": fake.phone_number(),
                "use": "work"
            }],
            "meta": {"lastUpdated": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")}
        }


    # ----------------------
    # Encounter
    # ----------------------
    def generate_fhir_encounter(patient_id, practitioner_id):
        encounter_id = str(uuid.uuid4())
        if isirisfhir==0:
            patient_ref=f"Patient/{patient_id}"
            practitioner_ref=f"Practitioner/{practitioner_id}"
        else:
            patient_ref=patient_id
            practitioner_ref=practitioner_id
        return {
            "resourceType": "Encounter",
            "id": encounter_id,
            "status": "finished",
            "class": {
                "system": "http://terminology.hl7.org/CodeSystem/v3-ActCode",
                "code": "AMB",
                "display": "ambulatory"
            },
            "subject": {"reference": patient_ref},
            "participant": [{
                "individual": {"reference": practitioner_ref}
            }],
            "period": {
                "start": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
                "end": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
            }
        }


    # ----------------------
    # Condition
    # ----------------------
    def generate_fhir_condition(patient_id, encounter_id, practitioner_id):
        condition_id = str(uuid.uuid4())
        code_choice = random.choice([
            ("44054006", "Diabetes mellitus"),
            ("195967001", "Asthma"),
            ("38341003", "Hypertension"),
            ("254637007", "Viral upper respiratory tract infection")
        ])
        if isirisfhir==0:
            patient_ref=f"Patient/{patient_id}"
            practitioner_ref=f"Practitioner/{practitioner_id}"
            encounter_ref=f"Encounter/{encounter_id}"
        else:
            patient_ref=patient_id
            practitioner_ref=practitioner_id
            encounter_ref=encounter_id
        return {
            "resourceType": "Condition",
            "id": condition_id,
            "clinicalStatus": {
                "coding": [{
                    "system": "http://terminology.hl7.org/CodeSystem/condition-clinical",
                    "code": "active"
                }]
            },
            "code": {
                "coding": [{
                    "system": "http://snomed.info/sct",
                    "code": code_choice[0],
                    "display": code_choice[1]
                }]
            },
            "subject": {"reference": patient_ref},
            "encounter": {"reference": encounter_ref},
            "asserter": {"reference": practitioner_ref},
            "onsetDateTime": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
        }


    # ----------------------
    # Vitals Observation
    # ----------------------
    def generate_fhir_vitals(patient_id, encounter_id, practitioner_id):
        observation_id = str(uuid.uuid4())
        if isirisfhir==0:
            patient_ref=f"Patient/{patient_id}"
            practitioner_ref=f"Practitioner/{practitioner_id}"
            encounter_ref=f"Encounter/{encounter_id}"
        else:
            patient_ref=patient_id
            practitioner_ref=practitioner_id
            encounter_ref=encounter_id
        return {
            "resourceType": "Observation",
            "id": observation_id,
            "status": "final",
            "category": [{
                "coding": [{
                    "system": "http://terminology.hl7.org/CodeSystem/observation-category",
                    "code": "vital-signs",
                    "display": "Vital Signs"
                }]
            }],
            "code": {
                "coding": [{
                    "system": "http://loinc.org",
                    "code": "85354-9",
                    "display": "Blood pressure panel"
                }]
            },
            "subject": {"reference": patient_ref},
            "encounter": {"reference": encounter_ref},
            "performer": [{"reference": practitioner_ref}],
            "effectiveDateTime": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
            "component": [
                {
                    "code": {
                        "coding": [{
                            "system": "http://loinc.org",
                            "code": "8480-6",
                            "display": "Systolic blood pressure"
                        }]
                    },
                    "valueQuantity": {
                        "value": random.randint(100, 140),
                        "unit": "mmHg",
                        "system": "http://unitsofmeasure.org",
                        "code": "mm[Hg]"
                    }
                },
                {
                    "code": {
                        "coding": [{
                            "system": "http://loinc.org",
                            "code": "8462-4",
                            "display": "Diastolic blood pressure"
                        }]
                    },
                    "valueQuantity": {
                        "value": random.randint(60, 90),
                        "unit": "mmHg",
                        "system": "http://unitsofmeasure.org",
                        "code": "mm[Hg]"
                    }
                }
            ]
        }


    # ----------------------
    # Lab Observations
    # ----------------------
    def generate_fhir_lab(patient_id, encounter_id, practitioner_id):
        labs = [
            ("2339-0", "Glucose [Mass/volume] in Blood", "mg/dL", random.randint(70, 200)),
            ("2093-3", "Cholesterol [Mass/volume] in Serum", "mg/dL", random.randint(150, 250)),
            ("4548-4", "Hemoglobin A1c/Hemoglobin.total in Blood", "%", round(random.uniform(4.5, 10.0), 1))
        ]
        lab_obs = []
        if isirisfhir==0:
            patient_ref=f"Patient/{patient_id}"
            practitioner_ref=f"Practitioner/{practitioner_id}"
            encounter_ref=f"Encounter/{encounter_id}"
        else:
            patient_ref=patient_id
            practitioner_ref=practitioner_id
            encounter_ref=encounter_id
        for loinc, display, unit, value in labs:
            obs_id = str(uuid.uuid4())
            lab_obs.append({
                "resourceType": "Observation",
                "id": obs_id,
                "status": "final",
                "category": [{
                    "coding": [{
                        "system": "http://terminology.hl7.org/CodeSystem/observation-category",
                        "code": "laboratory",
                        "display": "Laboratory"
                    }]
                }],
                "code": {
                    "coding": [{
                        "system": "http://loinc.org",
                        "code": loinc,
                        "display": display
                    }]
                },
                "subject": {"reference": patient_ref},
                "encounter": {"reference": encounter_ref},
                "performer": [{"reference": practitioner_ref}],
                "effectiveDateTime": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
                "valueQuantity": {
                    "value": value,
                    "unit": unit,
                    "system": "http://unitsofmeasure.org",
                    "code": unit
                }
            })
        return lab_obs


    # ----------------------
    # Medication + MedicationRequest
    # ----------------------
    def generate_fhir_medication(patient_id, encounter_id, practitioner_id):
        meds = [
            ("860975", "Metformin 500mg tablet"),
            ("860976", "Lisinopril 10mg tablet"),
            ("860977", "Salbutamol inhaler"),
            ("860978", "Atorvastatin 20mg tablet")
        ]
        med_code, med_name = random.choice(meds)
        med_id = str(uuid.uuid4())
        med_request_id = str(uuid.uuid4())
        if isirisfhir==0:
            patient_ref=f"Patient/{patient_id}"
            practitioner_ref=f"Practitioner/{practitioner_id}"
            encounter_ref=f"Encounter/{encounter_id}"
        else:
            patient_ref=patient_id
            practitioner_ref=practitioner_id
            encounter_ref=encounter_id

        medication = {
            "resourceType": "Medication",
            "id": med_id,
            "code": {
                "coding": [{
                    "system": "http://www.nlm.nih.gov/research/umls/rxnorm",
                    "code": med_code,
                    "display": med_name
                }]
            }
        }

        medication_request = {
            "resourceType": "MedicationRequest",
            "id": med_request_id,
            "status": "active",
            "intent": "order",
            "medicationReference": {"reference": f"Medication/{med_id}"},
            "subject": {"reference": patient_ref},
            "encounter": {"reference": encounter_ref},
            "requester": {"reference": practitioner_ref},
            "authoredOn": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
            "dosageInstruction": [{
                "text": f"Take {random.randint(1,2)} tablet(s) daily"
            }]
        }

        return [medication, medication_request]


    # ----------------------
    # DocumentReference (Discharge Summary)
    # ----------------------
    def generate_fhir_documentreference(patient_id, encounter_id, practitioner_id):
        doc_id = str(uuid.uuid4())
        note_text = f"Discharge summary for patient {patient_id}. Diagnosis: Stable. Follow-up in 2 weeks."
        encoded_note = base64.b64encode(note_text.encode("utf-8")).decode("utf-8")
        if isirisfhir==0:
            patient_ref=f"Patient/{patient_id}"
            practitioner_ref=f"Practitioner/{practitioner_id}"
            encounter_ref=f"Encounter/{encounter_id}"
        else:
            patient_ref=patient_id
            practitioner_ref=practitioner_id
            encounter_ref=encounter_id

        return {
            "resourceType": "DocumentReference",
            "id": doc_id,
            "status": "current",
            "type": {
                "coding": [{
                    "system": "http://loinc.org",
                    "code": "18842-5",
                    "display": "Discharge summary"
                }]
            },
            "subject": {"reference": patient_ref},
            "author": [{"reference": practitioner_ref}],
            "context": {"encounter": [{"reference": encounter_ref}]},
            "content": [{
                "attachment": {
                    "contentType": "text/plain",
                    "language": "en",
                    "data": encoded_note,
                    "title": "Discharge Summary Note"
                }
            }]
        }


    # ----------------------
    # Bundle Generator (transaction)
    # ----------------------
    def generate_patient_bundle(n=1):
        bundle = {
            "resourceType": "Bundle",
            "type": "transaction",
            "id": str(uuid.uuid4()),
            "meta": {"lastUpdated": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")},
            "entry": []
        }

        for _ in range(n):
            patient = generate_fhir_patient()
            practitioner = generate_fhir_practitioner()
            encounter = generate_fhir_encounter(patient["id"], practitioner["id"])
            condition = generate_fhir_condition(patient["id"], encounter["id"], practitioner["id"])
            vitals = generate_fhir_vitals(patient["id"], encounter["id"], practitioner["id"])
            labs = generate_fhir_lab(patient["id"], encounter["id"], practitioner["id"])
            meds = generate_fhir_medication(patient["id"], encounter["id"], practitioner["id"])
            docref = generate_fhir_documentreference(patient["id"], encounter["id"], practitioner["id"])

            all_resources = [patient, practitioner, encounter, condition, vitals] + labs + meds + [docref]

            for resource in all_resources:
                bundle["entry"].append({
                    "fullUrl": f"urn:uuid:{resource['id']}",
                    "resource": resource,
                    "request": {
                        "method": "POST",
                        "url": resource["resourceType"]
                    }
                })

        return bundle


    # ----------------------
    # Example Run
    # ----------------------
    fhir_bundle = generate_patient_bundle(1)

    return json.dumps(fhir_bundle, indent=2)
}

}

Let's test the function by the following script, for generating FHIR bundle for non IRIS FHIR repository

w ##class(datagen.utli.genfhirjson).genfhirbundle(0)

and check the reference syntax

 

Now test the function by the following script, for generating FHIR bundle for IRIS FHIR repository😁

w ##class(datagen.utli.genfhirjson).genfhirbundle(1)

and check the reference syntax

 

---

Great! Now go back the pervious article Writing a REST api service for exporting the generated patient data in .csv

We would like to add a new function and update the route for the datagen.restservice class.

 

1. Add a new function GenFHIRBundle, with an parameter isirisfhir for indicating which kind of reference syntax we prefer.

If no parameters are input, it will out put the reference syntax for non IRIS FHIR repository

ClassMethod GenFHIRBundle() As %Status
{
    // get the parameter isirisfhir
    set isirisfhir=$GET(%request.Data("isirisfhir",1),"")
    // default value for non - IRIS repository
    if isirisfhir="" set isirisfhir=0
    // gen the FHIR bundle
    w ##class(datagen.utli.genfhirjson).genfhirbundle(isirisfhir)
    
    return $$$OK
}

2. Then we add a route for the REST service and compile😀

<Route Url="/gen/fhirbundle" Method="GET" Call="GenFHIRBundle"/>

the updated  datagen.restservice class will look like the following.

 

---

Great!! That's all we need to do!😁

Let's test in postman!!

Try to GET by the following URL for generating FHIR bundle for non IRIS FHIR repository

localhost/irishealth/csp/mpapp/gen/fhirbundle

 

 

Looks good!!

Now Try to GET by the following URL for generating FHIR bundle for IRIS FHIR repository
 

localhost/irishealth/csp/mpapp/gen/fhirbundle?isirisfhir=1

 

Works well!!!😆😉

Thank you very much for reading. 😘