Results over HL7 v2

Receive ORU messages from a laboratory information system and write each result to DHIS2.

This page follows Write a result in the Lab results series. Laboratory information systems report results as HL7 v2 ORU^R01 messages over MLLP. An MLLP channel receives each message, parses it and invokes the service below.

The message

One ORU^R01 message carries one result. The fields the service reads are the test code in OBX-3, the result code in OBX-5, the specimen ID in SPM-2 and the result date in OBR-22:

MSH|^~\&|LIS|LAB_CENTRAL|ZATO|MOH|20261005090000||ORU^R01^ORU_R01|LAB-000917|P|2.5.1
PID|1||PT-55120^^^LAB_CENTRAL^MR
OBR|1|||94500-6^SARS-CoV-2 RNA^LN|||20261003080000|||||||||||||||20261005090000|||F
OBX|1|CWE|94500-6^SARS-CoV-2 RNA^LN||260373001^Detected^SCT||||||F
SPM|1|SP-2026-004417||258500001^Nasopharyngeal swab^SCT

Translate the codes

The service translates the LOINC and SNOMED codes to DHIS2 option codes with the config tables from Setup. Segments inside ORU groups are read with path expressions:

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

# stdlib
from datetime import datetime

# Zato
from zato.common.hl7.exception import HL7ApplicationError
from zato.server.service import Service

if 0:
    from zato.hl7v2.base import HL7Message

class ReceiveLabResultHL7(Service):
    name = 'dhis2.lab.from-hl7'

    def translate_codes(self, message:'HL7Message') -> 'tuple[str, str]':

        loinc_code = message.get('OBX.3.1')
        snomed_code = message.get('OBX.5.1')

        test = self.config.lab_tests.translate(source='LOINC', code=loinc_code)
        result = self.config.lab_results.translate(source='SNOMED', code=snomed_code)

        # An unmapped code is reported to the laboratory as an application error
        if test is None or result is None:
            error = f'No DHIS2 option code for {loinc_code} or {snomed_code}'
            raise HL7ApplicationError(error)

        return test, result

Write the result

With the codes translated, the service converts the HL7 timestamp in OBR-22 to ISO 8601 and invokes dhis2.lab.write-result:

    def handle(self) -> 'None':

        message:'HL7Message' = self.request.input
        test, result = self.translate_codes(message)

        specimen_id = message.get('SPM.2')

        result_date = message.get('OBR.22')
        result_date = datetime.strptime(result_date, '%Y%m%d%H%M%S')
        result_date = result_date.isoformat()

        request = {
            'specimen_id': specimen_id,
            'test': test,
            'result': result,
            'result_date': result_date,
        }

        self.invoke('dhis2.lab.write-result', request)

Acknowledgements

The channel acknowledges each message after the service returns. A service that completes normally produces an AA acknowledgement. HL7ApplicationError produces AE, which tells the laboratory that the message was received and rejected, and any other exception produces AR. The laboratory information system resends AR messages according to its own retry policy.

When the case does not exist yet, dhis2.lab.write-result returns is_found as False. The service above still returns normally, so the laboratory receives AA and does not resend a message that will not succeed until the case is registered. Held results describes how such results are kept and retried.

Results from FHIR covers laboratories that publish results as FHIR Observations instead of HL7 v2 messages.

Feature	What it does
HL7 v2 over MLLP	Channels, acknowledgements and error handling
Field access	Named attributes and path expressions
Config tables	The translate method used above