Weekly counts

Report the number of results per test and outcome to a DHIS2 data set each week.

This page concludes the Lab results series. Individual results are written to tracker cases on the preceding pages. Aggregate reporting in DHIS2 is based on data sets, so a weekly job counts the results of the previous week per test and outcome and sends the counts to a data set.

The data set

The data set, its data elements and the organisation unit are identified by code rather than UID, which keeps the service independent of the DHIS2 instance it runs against. The data set and organisation unit codes come from the weekly section of dhis2.ini. Each data element code is the test and result option codes joined with an underscore and prefixed with LAB_, for instance LAB_COVID19_PCR_POSITIVE.

The reporting week

The job runs on Mondays and reports on the week that has just ended. The service computes the Monday and Sunday of that week and its DHIS2 period identifier:

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

# stdlib
from collections import Counter
from datetime import date, timedelta

# Zato
from zato.server.service import Service

class SendWeeklyLabCounts(Service):
    name = 'dhis2.lab.send-weekly-counts'

    def previous_week(self) -> 'tuple[date, date, str]':

        # Any day of the previous week, then its Monday and Sunday
        last_week = date.today() - timedelta(days=7)
        start = last_week - timedelta(days=last_week.weekday())
        end = start + timedelta(days=6)

        year, week, _ = start.isocalendar()
        period = f'{year}W{week}'

        return start, end, period

period is the DHIS2 weekly period identifier, such as 2026W40 for the week starting on 28 September 2026.

Read the events

The lab result events of the week are read from the tracker, with paging turned off so that one request returns all of them:

    def read_events(self, start:'date', end:'date') -> 'list':

        program = self.config.dhis2.program
        occurred_after = start.isoformat()
        occurred_before = end.isoformat()

        params = {
            'program': program.id,
            'programStage': program.stage_lab_result,
            'occurredAfter': occurred_after,
            'occurredBefore': occurred_before,
            'orgUnitMode': 'ACCESSIBLE',
            'fields': 'dataValues',
            'paging': 'false',
        }

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

        return response.data['events']

Count the events

Each event is counted under the data element code of its test and result:

    def count_events(self, events:'list') -> 'Counter':

        element = self.config.dhis2.element
        counts = Counter()

        for event in events:

            # Index the event's data values by data element ..
            values = {}
            for item in event['dataValues']:
                data_element = item['dataElement']
                values[data_element] = item['value']

            # .. and count the event under its test and result.
            test = values[element.test]
            result = values[element.result]
            counts[f'LAB_{test}_{result}'] += 1

        return counts

Send the counts

The counts are posted as one data value set for the period:

    def handle(self) -> 'None':

        weekly = self.config.dhis2.weekly
        start, end, period = self.previous_week()

        events = self.read_events(start, end)
        counts = self.count_events(events)

        # DHIS2 takes every value as a string, counts included.
        data_values = []
        for code, count in counts.items():
            value = str(count)
            data_values.append({'dataElement': code, 'value': value})

        payload = {
            'dataSet': weekly.data_set,
            'period': period,
            'orgUnit': weekly.org_unit,
            'dataValues': data_values,
        }

        params = {
            'dataSetIdScheme': 'code',
            'dataElementIdScheme': 'code',
            'orgUnitIdScheme': 'code',
        }

        conn = self.rest['DHIS2 Data Values']
        conn.post(payload, params)

The IdScheme parameters tell DHIS2 that the data set, data elements and organisation unit are given by code. Posting the same period again replaces the values of that period, so the job can be re-run after a correction.

The import summary

DHIS2 returns the counts of imported, updated and ignored values under response.importCount. A value DHIS2 cannot accept is ignored and listed under response.conflicts with the reason, while the other values are imported. A data element code that is not part of the data set makes DHIS2 reject the whole set with HTTP 409:

import_summary = response.data['response']
for conflict in import_summary['conflicts']:
    self.logger.warning('DHIS2 ignored a value: %s', conflict['value'])