HL7v2 to FHIR transformation

Transform ADT, ORM, ORU and other HL7v2 messages into FHIR resources with plain Python, automatically or through manual mapping.

Transform ADT, ORM, ORU, and other HL7v2 message types into FHIR resources with regular Python services. A message arrives over MLLP, your service turns it into typed FHIR resources and saves them to a FHIR server - there is no separate mapping language to learn, the transformation is plain Python from end to end.

There are two ways to transform, and they combine freely:

The automatic converter - one call, msg.to_fhir(), maps the whole message to a FHIR bundle, the right choice when the standard mapping covers your data
Manual mapping - your service reads fields by their semantic names and builds each FHIR resource itself, for business rules applied mid-mapping or resources the automatic converter does not produce

The automatic conversion

For the common cases, the entire transformation is one line:

bundle = msg.to_fhir()

The bundle contains a Patient built from PID, an Encounter from PV1, Observations from OBX and so on, with all cross-references wired up, ready to post to a FHIR server. To see a complete service, the resources produced per message family and the configuration options, see Automatic HL7v2 to FHIR conversion.

The overall flow

Manual mapping builds each resource in your service, one element at a time. The pipeline consists of three parts:

An MLLP channel receives an HL7v2 message, e.g. an ADT^A01 admission, parses it and invokes your service with a typed HL7Message object
The service maps HL7v2 fields to FHIR resources - msg.pid becomes a Patient, msg.pv1 becomes an Encounter
An outgoing FHIR connection, available in code as self.fhir, saves the resources to the FHIR server

Create the MLLP channel in the Dashboard under Channels > HL7 > MLLP, pointing to the service you will write below:

Then create the FHIR connection under Connections > Outgoing > HL7 > FHIR:

Neither step requires a server restart - both are ready the moment you click OK.

Field mapping

The core of the transformation is deciding which HL7v2 field feeds which FHIR element. For an ADT admission, the essential mappings are:

HL7v2 field	Accessed as	FHIR element
PID-3	`msg.pid.patient_identifier_list`	`Patient.identifier`
PID-5	`msg.pid.patient_name`	`Patient.name`
PID-7	`msg.pid.date_time_of_birth`	`Patient.birthDate`
PID-8	`msg.pid.administrative_sex`	`Patient.gender`
PV1-2	`msg.pv1.patient_class`	`Encounter.class`
PV1-44	`msg.pv1.admit_date_time`	`Encounter.period.start`

For the complete segment-to-resource and field-to-field tables, including ORU results and the datatype conversions, see the HL7 v2 to FHIR mapping reference.

The transformation service

The service below is complete - it receives a parsed ADT message from the MLLP channel, builds a typed Patient and Encounter, wraps both in a transaction bundle so the server stores them atomically, and posts the bundle through self.fhir:

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

# Zato
from zato.fhir import Encounter, Patient
from zato.fhir.bundle import TransactionBuilder
from zato.server.service import Service

# #########################################################################################
# #########################################################################################

if 0:
    from zato.hl7v2.base import HL7Message

# #########################################################################################
# #########################################################################################

# PID-8 admin sex codes to FHIR administrative gender
_gender_map = {'M': 'male', 'F': 'female', 'O': 'other', 'U': 'unknown'}

# PV1-2 patient class codes to FHIR v3-ActCode encounter classes
_class_map = {
    'I': {'code': 'IMP', 'display': 'inpatient encounter'},
    'O': {'code': 'AMB', 'display': 'ambulatory'},
    'E': {'code': 'EMER', 'display': 'emergency'},
}

_act_code_system = 'http://terminology.hl7.org/CodeSystem/v3-ActCode'

# #########################################################################################
# #########################################################################################

def _to_fhir_date(value:'str') -> 'str':
    """ Turns an HL7 DTM value like 19800115 into a FHIR date like 1980-01-15.
    """
    return f'{value[:4]}-{value[4:6]}-{value[6:8]}'

# #########################################################################################
# #########################################################################################

class ADTToFHIR(Service):
    """ Receives ADT admissions over MLLP and stores them in a FHIR server.
    """
    name = 'hl7-api.adt-to-fhir'

    def handle(self) -> 'None':

        # The MLLP channel already parsed the raw ER7 bytes for us
        msg:'HL7Message' = self.request.input

        # Build the Patient from the PID segment ..
        mrn = msg.pid.patient_identifier_list[0].id_number

        patient = Patient()
        patient.identifier = [{'system': 'urn:oid:2.16.840.1.113883.19.3', 'value': mrn}]
        patient.name = [{
            'family': msg.pid.patient_name.family_name,
            'given': [msg.pid.patient_name.given_name],
        }]
        patient.birthDate = _to_fhir_date(msg.pid.date_time_of_birth)
        patient.gender = _gender_map[msg.pid.administrative_sex]

        # .. and the Encounter from the PV1 segment ..
        encounter_class = _class_map[msg.pv1.patient_class]

        encounter = Encounter()
        encounter.status = 'in-progress'
        encounter.class_ = {
            'system': _act_code_system,
            'code': encounter_class['code'],
            'display': encounter_class['display'],
        }
        encounter.period = {'start': _to_fhir_date(msg.pv1.admit_date_time)}

        # .. wrap both in a transaction bundle so the server stores them atomically ..
        transaction = TransactionBuilder()
        _ = transaction.create(patient)
        _ = transaction.create(encounter)
        bundle = transaction.build()

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

        self.logger.info('Stored %s in FHIR server -> %s',
            msg.msh.message_control_id, response['type'])

Send a test ADT^A01 message to the channel with any MLLP client and the final log line reports the outcome - the message's control ID and the server's transaction-response reply.

Because class is a reserved word in Python, the typed Encounter exposes the FHIR Encounter.class element as class_ - the serialized output still says class.

Validation before sending

The typed resources can be validated against the FHIR R4 schema before anything leaves your service, which catches mapping mistakes early instead of as server-side rejections:

# Zato
from zato.fhir import validate

# Validate each built resource before it goes into the bundle
for resource in [patient, encounter]:
    result = validate(resource)

    if not result.is_valid:
        self.logger.warning('Validation errors: %s', result.errors)
        return

result.errors lists each problem with the path to the offending element, e.g. a gender outside the allowed value set or a missing required field.

Page	What it covers
Automatic conversion	The whole message mapped with one call, msg.to_fhir()
FHIR resources	Creating, reading and updating typed resources on a server
MLLP channels	The channel that receives and parses the incoming messages
The HL7 MLLP tutorial	Route a feed end to end, with test messages and the audit log