Alarms

The Alarm class monitors values against a threshold and triggers a callback when the threshold has been met for a sustained period. It includes two mechanisms to prevent nuisance alarms: a grace period that requires the threshold to be met continuously before triggering, and a minimum inter-alarm interval that limits how frequently the alarm can fire.

1. Each time check_value() is called with a value, the threshold_met function evaluates whether the threshold condition is satisfied.
2. If the threshold is not met, the alarm resets its trigger timer and returns False.
3. If the threshold is met for the first time, the alarm records the time and starts the grace period countdown.
4. Once the threshold has been continuously met for the full grace_period, the alarm checks whether enough time has passed since the last alarm (min_inter_alarm).
5. If both conditions pass, the callback fires and the alarm time is recorded.

This two-gate approach prevents false alarms from transient sensor spikes (grace period) and prevents alarm fatigue from sustained fault conditions (minimum inter-alarm interval).

from pydoover.utils.alarm import Alarm

alarm = Alarm(
    threshold_met=lambda x: x > 85.0,  # Function that returns True when threshold is exceeded
    callback=my_alert_function,          # Called when alarm triggers
    grace_period=3600,                   # Seconds threshold must be met before triggering (default: 1 hour)
    min_inter_alarm=86400,               # Minimum seconds between alarms (default: 24 hours)
)

Call check_value() each time you have a new measurement. It is an async method:

await alarm.check_value(
    value,                    # The current measurement
    threshold_met,            # The threshold function (required)
    grace_period=None,        # Override grace period for this check
    min_inter_alarm=None,     # Override min inter-alarm for this check
)

If the grace period and inter-alarm conditions are both met, the callback is invoked. The callback can be either a regular function or an async coroutine -- the alarm detects this automatically.

Reset the alarm state to allow it to trigger again immediately:

alarm.reset_alarm()

This clears both the last alarm time and the initial trigger time. Use this when you want to re-arm the alarm after acknowledging a condition, or when the system transitions to a known-good state.

The create_alarm function wraps an existing async function so that the alarm is checked automatically every time the function is called. The original return value is passed through unchanged.

from pydoover.utils import create_alarm

class MyApp:
    async def setup(self):
        # Wrap get_temperature so every call also checks the alarm
        self.get_temperature = create_alarm(
            self.get_temperature,
            threshold_met=lambda x: x > 85.0,
            callback=self.send_temperature_alert,
            grace_period=300,       # 5 minutes
            min_inter_alarm=3600,   # 1 hour
        )

    async def get_temperature(self):
        return await self.read_sensor("temperature")

    async def send_temperature_alert(self):
        await self.send_notification("Temperature exceeds 85 degrees!")

After wrapping, every call to self.get_temperature() reads the sensor and checks the alarm. If the temperature stays above 85 for 5 continuous minutes, send_temperature_alert fires. It will not fire again for at least 1 hour.

The wrapped function exposes the underlying Alarm instance via the .alarm attribute:

# Access the alarm instance to reset it
self.get_temperature.alarm.reset_alarm()

This example monitors tank water level and sends a notification when it drops below a critical threshold:

from pydoover.utils.alarm import Alarm

class TankMonitor:
    async def setup(self):
        self.low_level_alarm = Alarm(
            threshold_met=lambda level: level < 20.0,
            callback=self.on_low_level,
            grace_period=600,       # Must be low for 10 minutes
            min_inter_alarm=7200,   # Alert at most every 2 hours
        )

    async def on_low_level(self):
        await self.send_notification(
            "Tank level is critically low. Check supply valve.",
            title="Low Level Alert",
        )

    async def main_loop(self):
        level = await self.read_level_sensor()

        await self.low_level_alarm.check_value(
            level,
            threshold_met=lambda x: x < 20.0,
        )

        await self.update_aggregate({"level": level})

The 10-minute grace period ensures that brief dips during filling or pumping cycles do not trigger false alarms. The 2-hour inter-alarm interval prevents repeated notifications during a sustained low-level condition.

• Kalman Filter -- smooth noisy sensor data before checking alarm thresholds
• PID Controller -- automated feedback control to prevent alarm conditions
• Utilities Overview -- all available utility modules