HL7 v2 to FHIR mapping reference

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class ADTToFHIRAuto(Service):
    name = 'demo.mapping.adt-to-fhir-auto'

    def handle(self) -> 'None':

        # An ADT^A01, parsed by the MLLP channel ..
        msg = self.request.input

        # .. one call maps the whole message to a FHIR bundle ..
        bundle = msg.to_fhir()

        # .. and the bundle is ready to post to a FHIR server.
        client = self.fhir['FHIR.Sample']
        client.execute('', method='post', data=bundle.to_dict())

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class ADTToPatientEncounter(Service):
    name = 'demo.mapping.adt-to-patient-encounter'

    def handle(self) -> 'None':

        # The MLLP channel already parsed the ADT^A01 ..
        msg = self.request.input

        # .. PID fields, by their semantic names ..
        mrn = msg.pid.patient_identifier_list.id_number
        family_name = msg.pid.patient_name.family_name
        given_name = msg.pid.patient_name.given_name
        date_of_birth = msg.pid.date_time_of_birth
        sex = msg.pid.administrative_sex

        # .. translate the coded values to FHIR ..
        gender_map = {'M': 'male', 'F': 'female', 'O': 'other', 'U': 'unknown'}
        gender = gender_map[sex]

        birth_date = f'{date_of_birth[0:4]}-{date_of_birth[4:6]}-{date_of_birth[6:8]}'

        # .. build and save the Patient ..
        client = self.fhir['FHIR.Sample']

        patient = client.resource('Patient',
            identifier = [{'system': 'urn:mrn', 'value': mrn}],
            name = [{'family': family_name, 'given': [given_name]}],
            birthDate = birth_date,
            gender = gender,
        )

        patient.save()

        # .. and the Encounter that points back at the Patient.
        class_map = {'I': 'IMP', 'O': 'AMB', 'E': 'EMER'}
        encounter_class = class_map[msg.pv1.patient_class]

        encounter = client.resource('Encounter',
            status = 'in-progress',
            subject = patient,
        )

        # "class" is a Python keyword, dict syntax sets the field
        encounter['class'] = {'code': encounter_class}

        encounter.save()

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class ORUToObservation(Service):
    name = 'demo.mapping.oru-to-observation'

    def handle(self) -> 'None':

        msg = self.request.input
        client = self.fhir['FHIR.Sample']

        # OBX-11 status codes translated to FHIR
        status_map = {'F': 'final', 'P': 'preliminary', 'C': 'corrected'}

        # OBX-3, the observation code, e.g. a LOINC code
        code = msg.get('OBX.3.1')
        code_text = msg.get('OBX.3.2')

        # OBX-5 value and OBX-6 units
        value = msg.get('OBX.5')
        unit = msg.get('OBX.6.1')

        observation = client.resource('Observation',
            status = status_map[msg.get('OBX.11')],
            code = {
                'coding': [{'system': 'http://loinc.org', 'code': code}],
                'text': code_text,
            },
            valueQuantity = {
                'value': float(value),
                'unit': unit,
            },
        )

        observation.save()

# PID-3 with two repetitions:
#     12345^^^HOSP^MR~67890^^^CLINIC^PI

# The first repetition, then the second
mrn = msg.get('PID.3[0]')
clinic_id = msg.get('PID.3[1]')

# A component of the second repetition - the identifier type
id_type = msg.get('PID.3[1].5')

# A subcomponent - for PID-3 of PT1^^^ACME&1.2.3&ISO^MR,
# the universal ID inside the assigning authority
universal_id = msg.get('PID.3.4.2')

# Every repetition of an identifier becomes one FHIR identifier
identifiers = []

for repetition in msg.pid.patient_identifier_list:
    identifiers.append({
        'system': 'urn:mrn',
        'value': repetition.id_number,
    })