Bonjour à tous,
Continuons à travailler sur la génération de données de test et l'exportation des résultats via une API REST. 😁
Ici, je souhaite réutiliser la classe `datagen.restservice` créée dans l'article précédent : « Écriture d'un service API REST pour exporter les données patient générées au format .csv ».
Cette fois-ci, nous prévoyons de générer un bundle FHIR incluant plusieurs ressources pour tester le référentiel FHIR.
Voici une référence si vous souhaitez en savoir plus sur FHIR : « The Concept of FHIR: A Healthcare Data Standard Designed for the Future ».
C'est parti ! 😆
1. Créez une nouvelle classe utilitaire `datagen.util.genfhirjson.cls` pour héberger les fonctions de génération de données FHIR.
Class datagen.utli.genfhirjson Extends %RegisteredObject
{
}2. Écrivez une fonction Python `genfhirbundle` pour générer un bundle FHIR au format JSON (j'ai demandé de l'aide au Chat-GPT pour cette partie 😂🤫).
J'ai ajouté ici un argument `isirisfhir` (valeur par défaut : 0) pour indiquer si ce bundle FHIR est destiné au référentiel FHIR IRIS.
Ceci est nécessaire en raison de la syntaxe différente des références de ressources utilisant un UUID.
Par exemple, voici à quoi ressemble une référence FHIR pour une ressource patient utilisant un UUID.
"subject": {
"reference": "Patient/3ce18a5b-b904-4c77-8a33-b09c3f8c79cc"
}Le référentiel IRIS FHIR, qui référence une ressource patient à l'aide de son UUID, ressemble à ceci :
"subject": {
"reference": "3ce18a5b-b904-4c77-8a33-b09c3f8c79cc"
}Par conséquent, un argument est ajouté en tant que sélecteur.
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)
}
}
Testons la fonction à l'aide du script suivant, pour générer un bundle FHIR pour un dépôt FHIR autre qu'IRIS.
w ##class(datagen.utli.genfhirjson).genfhirbundle(0).png)
et vérifiez la syntaxe de référence
.png)
Testez maintenant la fonction avec le script suivant, pour générer un bundle FHIR pour le dépôt IRIS FHIR 😁
w ##class(datagen.utli.genfhirjson).genfhirbundle(1).png)
et vérifiez la syntaxe de référence
.png)
Parfait ! Revenons à l'article précédent : « Écriture d'un service API REST pour exporter les données patient générées au format .csv ».
Nous souhaitons ajouter une nouvelle fonction et mettre à jour la route de la classe `datagen.restservice`.
1. Ajoutez une nouvelle fonction `GenFHIRBundle`, avec un paramètre `isirisfhir` indiquant la syntaxe de référence souhaitée.
Si aucun paramètre n'est fourni, la fonction renverra la syntaxe de référence pour un référentiel FHIR autre qu'IRIS.
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. Ensuite, nous ajoutons une route pour le service REST et nous compilons 😀
<Route Url="/gen/fhirbundle" Method="GET" Call="GenFHIRBundle"/>La classe datagen.restservice mise à jour ressemblera à ceci :
.png)
Super ! C'est tout ce qu'il nous reste à faire ! 😁
Testons avec Postman !
Essayez d'effectuer une requête GET à l'URL suivante pour générer un bundle FHIR pour un dépôt FHIR autre qu'IRIS.
localhost/irishealth/csp/mpapp/gen/fhirbundle .png)
Parfait !
Essayez maintenant d'effectuer une requête GET à l'URL suivante pour générer un bundle FHIR pour le dépôt IRIS FHIR.
localhost/irishealth/csp/mpapp/gen/fhirbundle?isirisfhir=1.png)
Ça marche super bien !!! 😆😉
Merci beaucoup d'avoir lu. 😘
