Results from FHIR

Poll a FHIR server for new Observations and write each one to DHIS2.

This page follows Results over HL7 v2 in the Lab results series. Laboratories that expose a FHIR server publish each result as an Observation resource. Zato reads new Observations at an interval, resolves the Specimen each one refers to, and invokes dhis2.lab.write-result for each result.

The connection

The laboratory's FHIR server is a FHIR outgoing connection named Lab FHIR. The service reaches it as self.fhir['Lab FHIR'].

The marker

Each run reads the Observations updated since the previous run. The time of the last Observation processed is kept in the cache under a key from dhis2.ini:

[fhir]
marker_key = dhis2.lab.fhir.since

The key is created once, before the first run, with the date from which results are to be read.

Build the request

The test code is the first coding of the Observation's code, the result code is the first coding of its valueCodeableConcept, and the specimen ID is the accessionIdentifier of the Specimen the Observation refers to. Nested values are read with path access:

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

# Zato
from zato.server.service import Service

class ReadLabResultsFHIR(Service):
    name = 'dhis2.lab.from-fhir'

    def build_request(self, observation:'dict') -> 'dict':

        loinc_code = observation.get_by_path('code.coding.0.code')
        snomed_code = observation.get_by_path('valueCodeableConcept.coding.0.code')

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

        # The specimen ID is the accession identifier of the referenced Specimen
        specimen_reference = observation.get_by_path('specimen.reference')
        client = self.fhir['Lab FHIR']
        specimen = client.get(specimen_reference)
        specimen_id = specimen.get_by_path('accessionIdentifier.value')

        request = {
            'specimen_id': specimen_id,
            'test': test,
            'result': result,
            'result_date': observation['effectiveDateTime'],
        }

        return request

Read the results

The service searches for final Observations updated after the marker, writes each one to its case and advances the marker after each result:

    def handle(self) -> 'None':

        client = self.fhir['Lab FHIR']
        marker_key = self.config.dhis2.fhir.marker_key

        # Read the Observations finalised since the previous run ..
        since = self.cache.get(marker_key)
        since = since.decode()

        observations = client.resources('Observation')
        observations = observations.search(_lastUpdated__gt=since, status='final')
        observations = observations.fetch_all()

        # .. write each one to its case and move the marker past it.
        for observation in observations:

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

            last_updated = observation.get_by_path('meta.lastUpdated')
            self.cache.set(marker_key, last_updated)

fetch_all follows the next links of the search Bundle, so a run that finds more results than one page holds processes all of them. An Observation whose codes have no mapping in the config tables reaches DHIS2 with empty values and is rejected by the tracker importer, which the import summary reports as described in Write a result.

The schedule

The service runs from a scheduler job, created in the Dashboard under Scheduler with dhis2.lab.from-fhir as the service and an interval of five minutes.

Held results covers results that arrive before the case they belong to has been registered in DHIS2.

Feature	What it does
FHIR connections	Outgoing connections to FHIR servers
Path access	Reading nested values with get_by_path
Scheduler	Interval-based jobs
Cache	Keeping state between runs