Antimicrobial resistance

Load the WHONET exports of human, animal and food laboratories into DHIS2 AMR data sets and the isolate program.

This page follows Zoonosis counts in the One Health series. Human, animal and food laboratories record their antimicrobial susceptibility tests in WHONET, which exports them for DHIS2 as files - a metadata file, a CSV of aggregate statistics in the DHIS2 data value format, and a CSV of isolates for an event program. The metadata file is imported into DHIS2 once, when the AMR data sets and program are created. The two CSV files arrive with every export, and Zato picks them up from each laboratory and loads them, so that no one imports them by hand, lab by lab.

Pick up the files

The laboratories upload their exports over SFTP, each into a directory of its own. The services reach the server through an SFTP connection named AMR Labs:

Each directory has two file transfer schedules, one per kind of file. For a laboratory uploading to /whonet/lab-01:

Name	Pattern	Service
`lab-01.data-set`	`*_dataset.csv`	`dhis2.amr.import-data-set`
`lab-01.isolates`	`*_isolates.csv`	`dhis2.amr.import-isolates`

Both schedules check that a file has stopped growing before they take it, so that no export is read before it has finished uploading, and move each file to processed once its service returns. A file the service raises an exception for stays where it is and is read again on the next run.

Aggregate statistics

The aggregate CSV is already in the format DHIS2 imports, so the service posts it as it is. WHONET writes the codes of org units and data elements rather than their UIDs, and amr.ini says which:

# config/user-conf/amr.ini

[data_set]
id_scheme = code

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

# Zato
from zato.server.service import Service

class ImportAMRDataSet(Service):
    name = 'dhis2.amr.import-data-set'

    def handle(self) -> 'None':

        request = self.request.input
        data = request.data
        text = data.decode('utf8')

        id_scheme = self.config.amr.data_set.id_scheme
        params = {'idScheme': id_scheme}
        headers = {'Content-Type': 'application/csv'}

        conn = self.rest['Public Health Data Values']
        response = conn.post(text, params, headers=headers)
        import_summary = response.data['response']
        status = import_summary['status']

        # A file DHIS2 rejects stays on the server for the next run ..
        if status == 'ERROR':
            raise Exception(f'{request.file_name} rejected: {response.text}')

        # .. while values it ignores are logged with the file they came from.
        for conflict in import_summary['conflicts']:
            self.logger.warning('%s: %s', request.file_name, conflict['value'])

Isolates

The isolate CSV has one row per value, and the rows of one isolate share its event identifier. The names of its columns are in amr.ini, together with the AMR program the events go to:

[isolates]
program = Bx7nR2kQm4T
stage = Cu3wL8pVz1H
id_scheme = CODE

[[columns]]
event = event
org_unit = orgUnit
occurred_at = eventDate
data_element = dataElement
value = value

The service reads every row under the names in the file, so a change to the export's columns is a change to columns alone:

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

# stdlib
from csv import DictReader
from io import StringIO

# Zato
from zato.server.service import Service

class ImportAMRIsolates(Service):
    name = 'dhis2.amr.import-isolates'

    def read_rows(self) -> 'list':

        columns = self.config.amr.isolates.columns
        data = self.request.input.data
        text = data.decode('utf8')
        stream = StringIO(text)

        reader = DictReader(stream)
        out = []

        for row in reader:
            isolate = {}
            for name, column in columns.items():
                isolate[name] = row[column]
            out.append(isolate)

        return out

Rows with the same event identifier become one event with all of their values:

    def group_events(self, rows:'list') -> 'dict':

        isolates = self.config.amr.isolates
        out = {}

        for row in rows:
            key = row['event']

            # The first row of an isolate opens its event ..
            if key not in out:
                out[key] = {
                    'program': isolates.program,
                    'programStage': isolates.stage,
                    'orgUnit': row['org_unit'],
                    'occurredAt': row['occurred_at'],
                    'status': 'COMPLETED',
                    'dataValues': [],
                }

            # .. and each row adds one value to it.
            event = out[key]
            data_value = {'dataElement': row['data_element'], 'value': row['value']}
            event['dataValues'].append(data_value)

        return out

The events of a file go to the tracker importer in one request, with org units and data elements read by code:

    def handle(self) -> 'None':

        rows = self.read_rows()
        events = self.group_events(rows)
        event_list = list(events.values())
        id_scheme = self.config.amr.isolates.id_scheme

        payload = {'events': event_list}
        params = {
            'async': 'false',
            'orgUnitIdScheme': id_scheme,
            'dataElementIdScheme': id_scheme,
        }

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

        # A file with an event DHIS2 rejects stays on the server for the next run.
        status = response.data['status']
        if status == 'ERROR':
            file_name = self.request.input.file_name
            raise Exception(f'{file_name} rejected: {response.text}')

DHIS2 validates every event before it imports any, and the rejection lists each problem under validationReport.errorReports with the message, the error code and the event it concerns. Once the data is in, DHIS2 charts resistance by pathogen and antibiotic from it.

Environmental data loads the daily readings of a national source.