Convert FHIR resources into other types of FHIR resources or flatten into a table.
Please note:
- Data on the FHIR server might change (resources might disapear making the IDs invalid, resources might be updated and no longer pass validation ...)
- you can search for and modify resources through the HAPI FHIR swagger UI
- This is not meant to be a fully robust solution - we make a few simplifications (like only looking at the 1st item in a list when pulling attributes) - to avoid getting lost in the details ...
import requests
import json
from pathlib import Path
import pandas as pd
from fhirclient import client
from fhirclient.models.annotation import Annotation
from fhirclient.models.dosage import Dosage
from fhirclient.models.fhirreference import FHIRReference
from fhirclient.models.medication import Medication
from fhirclient.models.medicationadministration import MedicationAdministration
from fhirclient.models.medicationdispense import MedicationDispense
from fhirclient.models.medicationrequest import MedicationRequest
from fhirclient.models.medicationstatement import MedicationStatement
def resource_to_string(resource, indent=None, length=100):
if isinstance(resource, list): return [resource_to_string(r,indent,length) for r in resource]
s=f'{resource.__class__.__name__} {json.dumps(resource.as_json(), indent=indent)}'
return f'{s[:length-4]} ...' if len(s)>length else s
def print_resource(resource, indent=None, length=90):
if isinstance(resource, list): return [print_resource(r,indent,length) for r in resource]
print(resource_to_string(resource, indent, length))
settings = {
'app_id': 'my_web_app',
'api_base': 'http://hapi.fhir.org/baseR4'
}
smart = client.FHIRClient(settings=settings)
pull_attr allows us to pull values out of a resource using "attribute paths" that can
- contain
.s for nested attributes and - multiple paths separated by
OR
def noop(r): return r
def default_format(r): return r.as_json() if hasattr(r,'as_json') else r
def pull_attr(resouce,attr_path,fmt=default_format):
"Pull a value from `resource` if we can find the attribute specified"
for _attr_path in attr_path.split(' OR '):
r,found=resouce,True
for _attr in _attr_path.split('.'):
if not hasattr(r,_attr):
found=False; break
r=getattr(r,_attr)
if isinstance(r,list) and r: r=r[0]
if found:
return [fmt(_r) for _r in r] if isinstance(r,list) else fmt(r)
medication_request=MedicationRequest.read('2087020', smart.server)
print_resource(medication_request)
assert '2087020'==pull_attr(medication_request,'id')
assert '2021-05-11T12:51:16.534+00:00'==pull_attr(medication_request,'meta.lastUpdated')
assert '2087020'==pull_attr(medication_request,'id OR meta.lastUpdated')
assert '2087020'==pull_attr(medication_request,'doesNotExist OR doesNotExist2 OR id')
assert '2021-05-11T12:51:16.534+00:00'==pull_attr(medication_request,'doesNotExist OR meta.lastUpdated')
assert '2021-05-11T12:51:16.534+00:00'==pull_attr(medication_request,'meta.lastUpdated OR id')
When an attribute is a list (like "coding" below),
"medicationCodeableConcept": {
"coding": [
{
"code": "630208",
"display": "Albuterol 0.83 MG/ML Inhalant Solution",
"system": "http://www.nlm.nih.gov/research/umls/rxnorm"
},
{
"system": "urn:oid:"
}
],
"text": "Albuterol Sulfate (Albuterol Sulfate 2.5MG/3ML) 0.083 % Neb Neb, 2.5 Mg Neb"
},we want to use the 1st item in the list ↓
pull_attr(medication_request,'medicationCodeableConcept.coding.code')
medication_request_to_medication_statement=dict(
id='id', meta='meta', implicitRules='implicitRules', language='language', text='text', contained='contained', extension='extension', modifierExtension='modifierExtension',
identifier='identifier',
# basedOn='id', #TODO: make this a reference
partOf='partOf',
status='status',
statusReason='statusReason',
category='category',
medicationCodeableConcept='medicationCodeableConcept',
medicationReference='medicationReference',
subject='subject',
context='encounter',
# effectiveDateTime # might be better to leave this blank as we have dosage
effectivePeriod='dosageInstruction.timing', # might be better to leave this blank as we have dosage
dateAsserted='authoredOn',
informationSource='requester',
# derivedFrom='id', #TODO: make this a reference
reasonCode='reasonCode',
reasonReference='reasonReference',
note='note',
dosage='dosageInstruction'
)
medication_dispense_to_medication_statement=dict(
id='id', meta='meta', implicitRules='implicitRules', language='language', text='text', contained='contained', extension='extension', modifierExtension='modifierExtension',
identifier='identifier',
basedOn='authorizingPrescription',
partOf='partOf',
status='status',
statusReason='statusReason',
category='category',
medicationCodeableConcept='medicationCodeableConcept',
medicationReference='medicationReference',
subject='subject',
context='context',
# effectiveDateTime # might be better to leave this blank as we have dosage
effectivePeriod='dosageInstruction.timing', # might be better to leave this blank as we have dosage
# dateAsserted # pull from event history?
informationSource='performer',
# derivedFrom='id', #TODO: make this a reference
# reasonCode
# reasonReference
note='note',
dosage='dosageInstruction'
)
medication_administration_dose_to_dose=dict(
text='text',
site='site',
route='route',
method='method',
doseQuantity='dose',
rateRatio='rateRatio',
rateQuantity='rateQuantity'
)
medication_administration_to_medication_statement=dict(
id='id', meta='meta', implicitRules='implicitRules', language='language', text='text', contained='contained', extension='extension', modifierExtension='modifierExtension',
identifier='identifier',
basedOn='request',
partOf='partOf',
status='status',
statusReason='statusReason',
category='category',
medicationCodeableConcept='medicationCodeableConcept',
medicationReference='medicationReference',
subject='subject',
context='context',
effectiveDateTime='effectiveDateTime',
effectivePeriod='effectivePeriod',
# dateAsserted # pull from event history?
informationSource='performer',
# derivedFrom='id', #TODO: make this a reference
# reasonCode
reasonReference='reasonReference',
note='note',
# dosage='dosage' # need to map MedicationAdministrationDosage to Dosage
)
def transform(resource,new_type,mapping):
"Pull data from `resource` to create an instance of `new_type` using `mapping`"
result=new_type()
for k in mapping:
setattr(result,k,pull_attr(resource,mapping[k],noop))
if hasattr(result,'note'):
if not result.note: result.note=[Annotation()]
if result.note[0].text: result.note[0].text+='\nCreated by py transform'
else: result.note[0].text='Created by py transform' # TODO: timestamp
return result
tfm_request=transform(medication_request,MedicationStatement,medication_request_to_medication_statement)
ref=FHIRReference()
ref.reference=f'MedicationRequest/{medication_request.id}'
tfm_request.derivedFrom=[ref]
tfm_request.basedOn=[ref]
tfm_request.identifier=[tfm_request.identifier]
# print_resource(tfm_request)
print_resource(tfm_request,2,9999)
medication_dispense=MedicationDispense.read('2040745', smart.server)
tfm_dispense=transform(medication_dispense, MedicationStatement, medication_dispense_to_medication_statement)
ref=FHIRReference()
ref.reference=f'MedicationDispense/{medication_dispense.id}'
tfm_dispense.derivedFrom=[ref]
tfm_dispense.identifier=[tfm_dispense.identifier]
print_resource(tfm_dispense)
# print_resource(tfm_dispense,2,9999)
medication_administration=MedicationAdministration.read('2086901', smart.server)
tfm_administration=transform(medication_administration, MedicationStatement, medication_administration_to_medication_statement)
tfm_administration.dosage=[transform(medication_administration.dosage,Dosage,medication_administration_dose_to_dose)]
ref=FHIRReference()
ref.reference=f'MedicationAdministration/{medication_administration.id}'
tfm_administration.derivedFrom=[ref]
tfm_administration.identifier=[tfm_administration.identifier]
print_resource(tfm_administration)
# print_resource(tfm_administration,2,9999)
medication_statement=MedicationStatement.read(2086902, smart.server)
print_resource(medication_statement)
# print_resource(medication_statement,2,9999)
This bit ↓ uses the mapping dictionaries we created earlier to build a list of "attribute paths" that we can use to create a data frame of all 4 mediction types
attr_paths=[]
for k in set([*medication_request_to_medication_statement.keys(),
*medication_dispense_to_medication_statement.keys(),
*medication_administration_to_medication_statement.keys()]):
attr_paths.append(set([
medication_request_to_medication_statement.get(k,None),
medication_dispense_to_medication_statement.get(k,None),
medication_administration_to_medication_statement.get(k,None)
]))
def set_to_attr_path(s):
return ' OR '.join([a for a in s if a])
attr_paths=[set_to_attr_path(s) for s in attr_paths]
attr_paths
def to_df(resources,attr_paths):
"Create a data frame of `resources` - one row per resource - one column per attribute path"
d=dict(resourceType=[r.__class__.__name__ for r in resources])
for attr_path in attr_paths:
d[attr_path]=[]
for r in resources:
d[attr_path].append(pull_attr(r,attr_path))
return pd.DataFrame(d)
create a data frame of all 4 mediction types
df=to_df([medication_request,medication_dispense,medication_administration,medication_statement],attr_paths)
df.to_csv('_tmp_a.csv',index=False) # In colab, use the "Files" sidebar to open or download this file
df
create a dataframe of the same 4 resources - after transforming them to medication statements
statement_attr_paths=[
'id', 'meta', 'implicitRules', 'text', 'contained', 'extension',
'modifierExtension', 'identifier', 'basedOn', 'partOf', 'status',
'statusReason', 'category', 'medicationCodeableConcept', 'medicationReference',
'subject', 'context', 'effectiveDateTime', 'effectivePeriod', 'dateAsserted',
'informationSource', 'derivedFrom', 'reasonCode', 'reasonReference', 'note', 'dosage']
df=to_df([tfm_request,tfm_dispense,tfm_administration,medication_statement],statement_attr_paths)
df.to_csv('_tmp_b.csv',index=False)
df