Thresholds

Check the week just ended against an alert rule per disease and raise an alert for each district that crosses it.

This page follows Weekly data in the Outbreak alerts series. Each disease in alerts.ini names the rule it is checked with. A rule is a Python function that takes the twelve weekly counts of one district and the disease's configuration, and says whether the week just ended is an alert.

The rules

A disease for which a single case is an alert, such as cholera, is checked with the simplest rule:

# stdlib
from datetime import date
from statistics import mean, stdev

# Zato
from zato.common.typing_ import any_

def is_single_case(counts:'list', disease:'any_') -> 'bool':

    # One case in the week just ended is enough.
    this_week = counts[-1]

    out = this_week > 0
    return out

A disease that is always present, such as malaria, alerts when the week is well above the weeks before it. The threshold is the mean of the earlier weeks plus a number of standard deviations, and a minimum count keeps a district with very few cases from alerting on small changes:

def is_above_baseline(counts:'list', disease:'any_') -> 'bool':

    # The weeks before the one just ended are the baseline ..
    this_week = counts[-1]
    start = -1 - disease.baseline_weeks
    baseline = counts[start:-1]

    # .. the threshold is their mean plus a number of standard deviations ..
    baseline_mean = mean(baseline)
    baseline_deviation = stdev(baseline)
    spread = disease.deviations * baseline_deviation
    threshold = baseline_mean + spread

    # .. and the week is an alert when it is above both the threshold and the minimum.
    if this_week < disease.minimum:
        out = False
    else:
        out = this_week > threshold

    return out

_rules = {
    'single_case': is_single_case,
    'baseline': is_above_baseline,
}

baseline_weeks can be up to eleven, which is how many weeks the request returns before the one just ended. A new rule is a function of the same shape, an entry in _rules, and its name in the section of each disease that uses it.

Check every district

Each series is keyed by its data element, and the diseases in alerts.ini say which disease each data element reports:

    def diseases_by_element(self) -> 'dict':

        diseases = self.config.alerts.diseases
        out = {}

        for name, disease in diseases.items():
            data_element = disease.data_element
            out[data_element] = name

        return out

An alert names the disease, the district and the week, which is the last of the twelve, with the number of cases and the date on which the alert was detected:

    def new_alert(self, name:'str', district:'str', counts:'list', week:'str') -> 'dict':

        district_config = self.config.districts[district]
        this_week = counts[-1]
        cases = int(this_week)

        today = date.today()
        detected = today.isoformat()

        alert = {
            'disease': name,
            'district': district,
            'district_name': district_config.name,
            'week': week,
            'cases': cases,
            'detected': detected,
        }

        return alert

Each series is checked with the rule of its disease:

    def handle(self) -> 'None':

        diseases = self.config.alerts.diseases
        by_element = self.diseases_by_element()
        weeks, series = self.read_series()
        week = weeks[-1]

        for (data_element, district_id), counts in series.items():

            # Check the series with the rule of its disease ..
            name = by_element[data_element]
            disease = diseases[name]
            rule = _rules[disease.rule]

            # .. and raise an alert for its district if the rule says so.
            if rule(counts, disease):
                alert = self.new_alert(name, district_id, counts, week)
                self.raise_alert(alert)

Raise the alert

Each alert goes to three services. The district is notified first, because the date of the notification is one of the dates the 7-1-7 register keeps. The service that records the alert is the one alerts.ini names, which is how the same code serves DHIS2 and an EMS:

    def raise_alert(self, alert:'dict') -> 'None':

        # Notify the district ..
        response = self.invoke('dhis2.alerts.notify-district', alert)
        alert['notified'] = response['notified']

        # .. record the alert where the configuration says ..
        record_service = self.config.alerts.record.service
        self.invoke(record_service, alert)

        # .. and keep its dates in the 7-1-7 register.
        self.invoke('dhis2.alerts.update-register', alert)