-
Notifications
You must be signed in to change notification settings - Fork 0
/
fhir.py
151 lines (130 loc) · 4.19 KB
/
fhir.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
# -*- coding: utf-8 -*-
from fhirclient import client
import fhirclient.models.patient as p
import fhirclient.models.medicationstatement as cm
import fhirclient.models.observation as ob
class FHIRConnection(object):
def __init__(self, base_url):
# reuse the client-py library
self.smart = client.FHIRClient(settings=dict(api_base=base_url, app_id='tornado-id'))
@property
def patients(self):
"""
Get the Patients resources
:return:
"""
search = p.Patient.where(struct=dict())
patients = []
for patient in search.perform_resources(self.smart.server):
patients.append(patient.as_json())
return patients
def get_patient(self, patient_id):
"""
Get a Patient resources
:arg str patient_id: the id for the patient resource
:return:
"""
patient = p.Patient.read(patient_id, self.smart.server)
return patient.as_json()
def get_patient_medications(self, patient_id):
"""
Get MedicationStatement for a Patient
:param str patient_id: Patient Identifier
:return:
"""
search_meds = cm.MedicationStatement.where(struct=dict(patient=patient_id))
meds = []
for meds in search_meds.perform_resources(self.smart.server):
meds.append(meds.as_json())
return meds
def get_medication(self, medication_id):
"""
Get an individual MedicationStatement Resource
:param str medication_id: Medication Statement resource ID
:return:
"""
medication = cm.MedicationStatement.read(medication_id, self.smart.server)
return medication.as_json()
def get_obs(self, patient_id, loinc_code):
search_obs = ob.Observation.where(struct=dict(patient=patient_id, code=loinc_code))
obs = ""
if len(search_obs.perform_resources(self.smart.server)) > 0:
sysbp = search_obs.perform_resources(self.smart.server).pop().as_json()
return obs
def get_vitals_sysbp(self, patient_id):
"""
Get the Systolic BP for the Patient
:param patient_id:
:return:
"""
return self.get_obs(patient_id, "8480-6")
def get_vitals_diabp(self, patient_id):
"""
Get the Diastolic BP for the Patient
:param patient_id:
:return:
"""
return self.get_obs(patient_id, "55284-4")
def get_vitals_bmi(self, patient_id):
"""
Get the BMI for the Patient
:param patient_id:
:return:
"""
return self.get_obs(patient_id, "39156-5")
def get_vitals_hr(self, patient_id):
"""
Get the HR for the Patient
:param patient_id:
:return:
"""
return self.get_obs(patient_id, "8867-4")
def get_vitals_temp(self, patient_id):
"""
Get the Temp for the Patient
:param patient_id:
:return:
"""
return self.get_obs(patient_id, "8310-5")
def get_vitals_weight(self, patient_id):
"""
Get the Weight for the Patient
:param patient_id:
:return:
"""
return self.get_obs(patient_id, "3141-9")
def get_vitals_height(self, patient_id):
"""
Get the Height for the Patient
:param patient_id:
:return:
"""
return self.get_obs(patient_id, "8302-2")
def get_labs_wbc(self, patient_id):
"""
Get the White Blood Count for the Patient
:param patient_id:
:return:
"""
return self.get_obs(patient_id, "6690-2")
def get_labs_rbc(self, patient_id):
"""
Get the Red Blood Count for the Patient
:param patient_id:
:return:
"""
return self.get_obs(patient_id, "789-8")
def get_labs_creat(self, patient_id):
"""
Get the Creatinine for the Patient
:param patient_id:
:return:
"""
return self.get_obs(patient_id, "14682-9")
def labs_alb(self, patient_id):
"""
Get the Albumin for the Patient
:param patient_id:
:return:
"""
return self.get_obs(patient_id, "1751-7")