Temperature readings

Turn the readings in PQS E006 reports into the morning and afternoon minimum, current and maximum of the Temperature Monitoring Tool.

This page follows Setup in the Supply chain series. The Temperature Monitoring Tool records what a storekeeper reads off a thermometer twice a day - the minimum, current and maximum temperature in the morning and in the afternoon. A monitoring device records a reading every few minutes instead, and the service below turns the readings into those six values.

The report

A PQS E006 report describes one appliance and holds its readings under records. Each record has the time of the reading in ABST, in UTC, the temperature of the vaccine compartment in TVC and any alarm in ALRM:

{
  "ASER": "RF-0001-7734",
  "LSER": "LOG-55120",
  "records": [
    {"ABST": "20261005T061500Z", "TVC": 4.1, "ALRM": null},
    {"ABST": "20261005T063000Z", "TVC": 4.6, "ALRM": null},
    {"ABST": "20261005T064500Z", "TVC": null, "ALRM": null}
  ]
}

The service reads the time of each record from ABST, so sending it with every record is part of what is agreed with each vendor. A temperature the logger knows to be wrong is sent as null.

Morning and afternoon

The half of the day a reading belongs to is decided by the local time of the facility:

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

# stdlib
from datetime import datetime, timedelta, timezone
from json import dumps, loads

# Zato
from zato.server.service import Service

class ReceiveReport(Service):
    name = 'dhis2.cold-chain.receive-report'

    def to_local_time(self, abst:'str') -> 'datetime':

        offset_hours = self.config.cold_chain.tmt.utc_offset_hours
        offset = timedelta(hours=offset_hours)
        local_zone = timezone(offset)

        # ABST is always in UTC, as YYYYMMDDThhmmssZ.
        utc_time = datetime.strptime(abst, '%Y%m%dT%H%M%SZ')
        utc_time = utc_time.replace(tzinfo=timezone.utc)

        out = utc_time.astimezone(local_zone)
        return out

Each valid reading goes under its day and its half of the day:

    def read_points(self, records:'list') -> 'dict':

        afternoon_from_hour = self.config.cold_chain.tmt.afternoon_from_hour
        out = {}

        for record in records:

            # A record without a valid temperature has nothing to add ..
            temperature = record.get('TVC')
            if temperature is None:
                continue

            # .. and every other one belongs to a half of its local day.
            abst = record['ABST']
            local_time = self.to_local_time(abst)
            day = local_time.strftime('%Y%m%d')
            if local_time.hour < afternoon_from_hour:
                half = 'morning'
            else:
                half = 'afternoon'

            key = day, half
            if key not in out:
                out[key] = []
            time = local_time.isoformat()
            point = [time, temperature]
            half_day = out[key]
            half_day.append(point)

        return out

Readings across reports

A vendor sends a report every few minutes or once a day, so the readings of one morning can arrive in several reports. The service keeps the readings of each half-day in the cache and computes the three values from all of them, which makes each report correct the values the earlier ones sent:

    def merge(self, serial:'str', day:'str', half:'str', points:'list') -> 'dict':

        cache_seconds = self.config.cold_chain.tmt.cache_seconds
        key = f'cold-chain.{serial}.{day}.{half}'

        # The readings of the same half-day from earlier reports ..
        cached = self.cache.get(key)
        if cached is not None:
            earlier = loads(cached)
            points.extend(earlier)

        # .. kept together with these ones, in time order.
        points.sort()
        serialized = dumps(points)
        self.cache.set(key, serialized, ex=cache_seconds)

        temperatures = []
        for _, temperature in points:
            temperatures.append(temperature)

        _, current = points[-1]
        minimum = min(temperatures)
        maximum = max(temperatures)

        out = {'minimum': minimum, 'current': current, 'maximum': maximum}
        return out

The current temperature is the latest reading of the half-day. A report sent again after a failure adds the same readings a second time, which changes none of the three values.

The data values

Each of the three values is one data value of the appliance, with the category option combination of its half of the day and its measurement:

    def build_values(self, serial:'str', day:'str', half:'str', summary:'dict') -> 'list':

        appliance = self.config.cold_chain.appliances[serial]
        combos = self.config.cold_chain.combos
        out = []

        for measurement, temperature in summary.items():
            combo_name = f'{half}_{measurement}'
            combo = combos[combo_name]
            value = str(temperature)

            data_value = {
                'dataElement': appliance.data_element,
                'period': day,
                'orgUnit': appliance.org_unit,
                'categoryOptionCombo': combo,
                'value': value,
            }
            out.append(data_value)

        return out

The report as a whole

A report from an appliance that is not in cold_chain.ini is refused with an error, so the vendor sends it again later, by which time the appliance may have been added:

    def handle(self) -> 'None':

        report = self.request.payload
        serial = report['ASER']
        appliances = self.config.cold_chain.appliances

        if serial not in appliances:
            raise Exception(f'Unknown appliance {serial}')

        records = report['records']
        points_by_half = self.read_points(records)

        data_values = []
        for (day, half), points in points_by_half.items():
            summary = self.merge(serial, day, half, points)
            values = self.build_values(serial, day, half, summary)
            data_values.extend(values)

        payload = {'dataValues': data_values}
        conn = self.rest['DHIS2 Data Values']
        conn.post(payload)

        # Alarms in the same report are checked separately.
        self.invoke('dhis2.cold-chain.check-alarms', report)

The period of a daily data set is the date as YYYYMMDD, which read_points already gives each half-day. The data set's expiry days have to leave room for vendors that send readings up to a day after they were taken.

Alarms notifies the district of the alarms in a report.