Zoonotic notifications

Carry notifications of zoonoses from the animal health DHIS2 instance to public health, under public health's org units and codes.

This page follows Setup in the One Health series. A notification of a zoonosis in the animal health instance, such as a rabid dog that has bitten someone, has to reach public health quickly, so that the person bitten is followed up for post-exposure prophylaxis. The service below runs every few minutes, reads the notifications updated since its previous run and creates them in the public health notification program.

Read what changed

DHIS2 lists the events of a program updated after a given time, and ordering them by that time makes the last one the newest:

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

# Zato
from zato.server.service import Service

class CarryNotifications(Service):
    name = 'dhis2.one-health.carry-notifications'

    def read_events(self, updated_after:'str') -> 'list':

        program = self.config.one_health.animal_program
        fields = 'event,orgUnit,occurredAt,updatedAt,dataValues[dataElement,value]'

        params = {
            'program': program.id,
            'updatedAfter': updated_after,
            'order': 'updatedAt:asc',
            'paging': 'false',
            'fields': fields,
        }

        conn = self.rest['Animal Health Events']
        response = conn.get(params=params)

        out = response.data['events']
        return out

Translate the codes

The org unit and the disease of a notification are translated with the config tables from the setup:

    def read_disease(self, event:'dict') -> 'str':

        element = self.config.one_health.animal_program.element_disease

        # The disease is compulsory in the animal health program,
        # so every notification has it.
        out = ''
        for data_value in event['dataValues']:
            if data_value['dataElement'] == element:
                out = data_value['value']

        return out

    def translate_codes(self, event:'dict') -> 'tuple':

        org_units = self.config.org_units
        diseases = self.config.diseases

        uid = event['orgUnit']
        animal_disease = self.read_disease(event)

        org_unit = org_units.translate(
            source='ANIMAL_HEALTH', code=uid, target='PUBLIC_HEALTH')
        disease = diseases.translate(
            source='ANIMAL_HEALTH', code=animal_disease, target='PUBLIC_HEALTH')

        out = org_unit, disease
        return out

Create them in public health

The event in public health takes the UID of the event in animal health. A notification updated after it was carried is therefore sent again under the same UID and DHIS2 updates the event it already has rather than creating a second one:

    def build_event(self, event:'dict', org_unit:'str', disease:'str') -> 'dict':

        program = self.config.one_health.public_program
        data_value = {'dataElement': program.element_disease, 'value': disease}

        out = {
            'event': event['event'],
            'program': program.id,
            'programStage': program.stage,
            'orgUnit': org_unit,
            'occurredAt': event['occurredAt'],
            'status': 'COMPLETED',
            'dataValues': [data_value],
        }

        return out

All events of a run go to DHIS2 in one request:

    def carry(self, events:'list') -> 'None':

        carried = []

        for event in events:
            org_unit, disease = self.translate_codes(event)

            # A code missing from either file is logged with the event's UID ..
            if org_unit is None or disease is None:
                uid = event['event']
                self.logger.warning('No mapping for event %s', uid)
                continue

            # .. and everything else is carried.
            public_event = self.build_event(event, org_unit, disease)
            carried.append(public_event)

        payload = {'events': carried}
        params = {'async': 'false'}

        conn = self.rest['Public Health Tracker']
        response = conn.post(payload, params)

        status = response.data['status']
        if status == 'ERROR':
            raise Exception(f'Notifications rejected: {response.text}')

An event logged as having no mapping is carried by the first run after its org unit or disease is added to the file and the event itself is updated in animal health.

Keep the place

The service keeps the update time of the newest event in the cache, and the first run, which finds nothing there, starts from the time in one_health.ini:

    def handle(self) -> 'None':

        notifications = self.config.one_health.notifications
        marker = self.cache.get(notifications.marker_key)

        # Start where the previous run ended ..
        if marker is None:
            updated_after = notifications.start
        else:
            updated_after = marker.decode('utf8')

        events = self.read_events(updated_after)

        if not events:
            return

        # .. carry what changed since then ..
        self.carry(events)

        # .. and move the place to the newest event.
        newest = events[-1]
        updated_at = newest['updatedAt']
        self.cache.set(notifications.marker_key, updated_at)

When DHIS2 rejects the request, carry raises before the place is moved, so the next run reads the same events again. DHIS2 includes events updated exactly at updatedAfter, which means the newest event of each run is read once more by the next one and sent under the same UID, which updates it to the same values.