Alarms

Notify the district of heat, freeze, door, power and disconnection alarms, once per alarm rather than once per record.

This page follows Temperature readings in the Supply chain series. A record of a PQS E006 report carries an alarm code in ALRM while the appliance is in an alarm condition:

Code	Alarm
`HEAT`	The temperature has been too high for too long
`FRZE`	The temperature has been below freezing for too long
`DOOR`	The door or lid has been open for too long
`POWR`	The appliance has had no power for more than a day
`DCNT`	The monitoring device has been disconnected from the appliance

An alarm is recorded in every record for as long as it lasts, and once someone at the facility acknowledges it, for example by muting it, the code gains an ACK suffix, as in HEATACK. The service below notifies the district of a new alarm, and of an alarm that is still on after a set time, rather than of every record.

The configuration

The names of the alarms in messages, the time after which an alarm still on is notified again and the sender of the emails go to cold_chain.ini:

[alarms]
repeat_after_seconds = 21600
sender = cold-chain@example.com

[[names]]
HEAT = Heat
FRZE = Freeze
DOOR = Door open
POWR = Power outage
DCNT = Monitoring device disconnected

The notification

The message names the alarm and the appliance, and goes to the district of the appliance by email and SMS:

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

# Zato
from zato.common import SMTPMessage
from zato.server.service import Service

class CheckAlarms(Service):
    name = 'dhis2.cold-chain.check-alarms'

    def notify(self, serial:'str', code:'str') -> 'None':

        cold_chain = self.config.cold_chain
        appliance = cold_chain.appliances[serial]
        district_id = appliance.district
        district = self.config.districts[district_id]
        name = cold_chain.alarms.names[code]
        text = f'{name} alarm: {appliance.label}'

        message = SMTPMessage()
        message.subject = text
        message.to = district.email
        message.from_ = cold_chain.alarms.sender
        message.body = text

        smtp = self.email.smtp['District Alerts'].conn
        smtp.send(message)

        sms = {'to': district.phone, 'text': text}
        conn = self.rest['SMS Gateway']
        conn.post(sms)

A heat alarm of the first refrigerator in the setup reads Heat alarm: Health Centre 12 - Vaccine refrigerator 1.

Once per alarm

The cache key of an alarm is made of the appliance and the code, and it expires after repeat_after_seconds. Setting it with nx=True succeeds only when the key is not there, which is the case for a new alarm and for one still on after the key expired:

    def handle(self) -> 'None':

        report = self.request.payload
        serial = report['ASER']
        records = report['records']
        repeat_after = self.config.cold_chain.alarms.repeat_after_seconds

        for record in records:

            # Most records carry no alarm ..
            code = record.get('ALRM')
            if code is None:
                continue

            # .. one that someone at the facility acknowledged needs no notification ..
            if code.endswith('ACK'):
                continue

            # .. and an alarm still on is notified once per repeat_after_seconds.
            key = f'cold-chain.alarm.{serial}.{code}'
            is_new = self.cache.set(key, '1', ex=repeat_after, nx=True)
            if is_new:
                self.notify(serial, code)