Alarm Utilities

The pydoover utilities module provides an alarm system for monitoring values against thresholds. The alarm system includes grace periods and minimum inter-alarm intervals to prevent rapid triggering and alert fatigue.

Import

from pydoover.utils import create_alarm

Overview

The alarm system allows you to:

Monitor function return values against custom thresholds
Configure grace periods before alarms trigger
Set minimum intervals between consecutive alarms
Execute callbacks when alarm conditions are met

This is useful for monitoring sensor values, system metrics, or any numeric data that needs threshold-based alerting.

create_alarm()

A function that wraps an async method to add alarm monitoring to its return value.

Signature

def create_alarm(
    func: Callable,
    threshold_met: Callable[[Any], bool],
    callback: Callable = None,
    grace_period: float = None,
    min_inter_alarm: float = None,
) -> Callable

Parameters

Parameter	Type	Default	Description
`func`	Callable	required	The async function to wrap
`threshold_met`	Callable	required	Function that returns True if threshold is exceeded
`callback`	Callable	`None`	Function to call when alarm triggers
`grace_period`	float	3600 (1 hour)	Seconds threshold must be met before alarm fires
`min_inter_alarm`	float	86400 (1 day)	Minimum seconds between consecutive alarms

Return Value

Returns an async wrapper function that monitors the original function's return value.

Basic Usage

Simple Threshold Alarm

from pydoover.utils import create_alarm

class TemperatureMonitor:
    def __init__(self):
        # Wrap the method with alarm monitoring
        self.get_temperature = create_alarm(
            self.get_temperature,
            lambda x: x > 80,  # Alarm if temperature > 80
            callback=self.on_high_temp,
            grace_period=60,   # Must be high for 60 seconds
            min_inter_alarm=300  # At most one alarm per 5 minutes
        )

    async def get_temperature(self):
        """Read temperature from sensor"""
        return await self._sensor.read()

    async def on_high_temp(self):
        """Called when temperature alarm triggers"""
        print("WARNING: High temperature detected!")
        await self.send_notification("Temperature exceeds 80 degrees")

Usage in Application Loop

import asyncio

async def main():
    monitor = TemperatureMonitor()

    while True:
        # Each call checks the alarm condition
        temp = await monitor.get_temperature()
        print(f"Current temperature: {temp}")
        await asyncio.sleep(1)

asyncio.run(main())

Threshold Functions

The threshold_met parameter accepts any callable that takes the function's return value and returns a boolean.

Greater Than

create_alarm(
    func,
    threshold_met=lambda x: x > 100,
    ...
)

Less Than

create_alarm(
    func,
    threshold_met=lambda x: x < 10,
    ...
)

Range Check

create_alarm(
    func,
    threshold_met=lambda x: x < 20 or x > 80,  # Outside safe range
    ...
)

Complex Conditions

def check_threshold(value):
    """Custom threshold logic"""
    if value is None:
        return False
    if isinstance(value, dict):
        return value.get('status') == 'critical'
    return value > 100

create_alarm(
    func,
    threshold_met=check_threshold,
    ...
)

Alarm Timing

Grace Period

The grace_period defines how long the threshold must be continuously met before the alarm fires. This prevents false alarms from brief spikes.

# Alarm triggers if temperature > 80 for at least 30 seconds
create_alarm(
    self.get_temperature,
    lambda x: x > 80,
    grace_period=30,  # 30 seconds
    ...
)

Behavior:

When threshold is first met, a timer starts
If threshold is met continuously for grace_period seconds, alarm fires
If threshold is no longer met, timer resets

Minimum Inter-Alarm

The min_inter_alarm defines the minimum time between consecutive alarm triggers. This prevents alert fatigue.

# At most one alarm every 10 minutes
create_alarm(
    self.get_temperature,
    lambda x: x > 80,
    min_inter_alarm=600,  # 600 seconds = 10 minutes
    ...
)

Behavior:

After an alarm fires, subsequent triggers are suppressed
Only after min_inter_alarm seconds can another alarm fire
The grace period must still be satisfied for each alarm

Combined Example

create_alarm(
    self.get_temperature,
    lambda x: x > 80,
    callback=self.alert_high_temp,
    grace_period=60,       # Must be high for 1 minute
    min_inter_alarm=3600   # Max 1 alarm per hour
)

Timeline example:

T=0: Temperature goes above 80
T=60: Grace period satisfied, alarm fires, callback called
T=120: Temperature still high, but min_inter_alarm not met, no alarm
T=3660: If still high and grace period satisfied again, alarm can fire

Callbacks

Sync Callback

def my_callback():
    print("Alarm triggered!")

create_alarm(func, threshold_met, callback=my_callback, ...)

Async Callback

async def my_async_callback():
    await send_notification("Alarm triggered!")
    await log_to_database("alarm_event")

create_alarm(func, threshold_met, callback=my_async_callback, ...)

The alarm system automatically detects if the callback is async and awaits it appropriately.

Accessing Alarm State

The wrapper provides access to the underlying Alarm object:

class Monitor:
    def __init__(self):
        self.get_value = create_alarm(
            self.get_value,
            lambda x: x > 100,
            callback=self.on_alarm,
            grace_period=60,
            min_inter_alarm=300
        )

    async def get_value(self):
        return await self._sensor.read()

    async def on_alarm(self):
        print("Alarm!")

    def reset_alarm(self):
        """Manually reset the alarm state"""
        self.get_value.alarm.reset_alarm()

    def get_alarm_info(self):
        """Get information about alarm state"""
        alarm = self.get_value.alarm
        return {
            "last_alarm_time": alarm.last_alarm_time,
            "initial_trigger_time": alarm.initial_trigger_time,
            "grace_period": alarm.grace_period,
            "min_inter_alarm": alarm.min_inter_alarm
        }

Resetting Alarms

# Reset alarm state (clears last alarm time and trigger time)
monitor.get_value.alarm.reset_alarm()

Integration with pydoover Applications

Full Application Example

from pydoover.docker import Application
from pydoover.utils import create_alarm

class ProcessMonitor(Application):
    def setup(self):
        self.platform = self.get_platform_interface()

        # Set up alarms for various sensors
        self.read_tank_level = create_alarm(
            self.read_tank_level,
            lambda x: x < 10,  # Low level alarm
            callback=self.on_low_tank_level,
            grace_period=120,    # 2 minutes grace
            min_inter_alarm=1800  # 30 min between alarms
        )

        self.read_pressure = create_alarm(
            self.read_pressure,
            lambda x: x > 150,  # High pressure alarm
            callback=self.on_high_pressure,
            grace_period=30,     # 30 second grace
            min_inter_alarm=600  # 10 min between alarms
        )

        self.read_temperature = create_alarm(
            self.read_temperature,
            lambda x: x > 85 or x < 5,  # Temperature out of range
            callback=self.on_temperature_alarm,
            grace_period=60,
            min_inter_alarm=900
        )

    async def read_tank_level(self):
        """Read tank level percentage"""
        raw = self.platform.get_analog_input(0)
        return (raw / 4095.0) * 100.0

    async def read_pressure(self):
        """Read pressure in PSI"""
        raw = self.platform.get_analog_input(1)
        return (raw / 4095.0) * 200.0

    async def read_temperature(self):
        """Read temperature in Celsius"""
        raw = self.platform.get_analog_input(2)
        return (raw / 4095.0) * 100.0

    async def on_low_tank_level(self):
        """Handle low tank level alarm"""
        self.log.warning("Low tank level alarm triggered")
        await self.send_alert("Tank level critically low")

    async def on_high_pressure(self):
        """Handle high pressure alarm"""
        self.log.warning("High pressure alarm triggered")
        await self.send_alert("Pressure exceeds safe limit")
        # Automatically reduce pressure
        self.platform.set_digital_output(0, True)  # Open relief valve

    async def on_temperature_alarm(self):
        """Handle temperature alarm"""
        self.log.warning("Temperature out of range")
        await self.send_alert("Temperature outside safe range")

    async def send_alert(self, message):
        """Send alert to cloud platform"""
        await self.device_agent.publish_message(
            "alerts",
            {"message": message, "timestamp": time.time()}
        )

    async def main_loop(self):
        while self.running:
            # Each read checks its alarm condition
            level = await self.read_tank_level()
            pressure = await self.read_pressure()
            temp = await self.read_temperature()

            # Update UI
            self.ui_manager.update_variable("tank_level", level)
            self.ui_manager.update_variable("pressure", pressure)
            self.ui_manager.update_variable("temperature", temp)

            await asyncio.sleep(1)

Multiple Alarms on Same Function

You can chain alarms or create separate monitored versions:

class MultiAlarmMonitor:
    def __init__(self):
        # Warning alarm at 70
        self.get_temp_warning = create_alarm(
            self._read_temperature,
            lambda x: x > 70,
            callback=self.on_temp_warning,
            grace_period=30,
            min_inter_alarm=300
        )

        # Critical alarm at 90
        self.get_temp_critical = create_alarm(
            self._read_temperature,
            lambda x: x > 90,
            callback=self.on_temp_critical,
            grace_period=10,  # Shorter grace for critical
            min_inter_alarm=60  # More frequent alerts allowed
        )

    async def _read_temperature(self):
        """Base temperature reading"""
        return await self._sensor.read()

    async def on_temp_warning(self):
        print("WARNING: Temperature approaching limit")

    async def on_temp_critical(self):
        print("CRITICAL: Temperature exceeded safe limit!")

Alarm Class Reference

The underlying Alarm class (accessible via .alarm) has these attributes:

Attribute	Description
`threshold_met`	The threshold checking function
`callback`	The callback function
`grace_period`	Grace period in seconds
`min_inter_alarm`	Minimum inter-alarm interval in seconds
`last_alarm_time`	Timestamp of last triggered alarm
`initial_trigger_time`	Timestamp when threshold was first met

Methods

Method	Description
`reset_alarm()`	Reset alarm state (clears times)
`check_value(value, threshold_met, grace_period, min_inter_alarm)`	Manually check a value

Best Practices

Set appropriate grace periods - Too short leads to false alarms; too long delays response
Use min_inter_alarm wisely - Balance between alert fatigue and missing important events
Keep callbacks lightweight - Offload heavy processing to avoid blocking the main loop
Log alarm events - Track when alarms trigger for debugging and analysis
Consider alarm acknowledgment - Use reset_alarm() when operators acknowledge alerts

Utilities Overview - Overview of all utility functions
Kalman Filter - Filter sensor data before alarm checking
Application Class - Building pydoover applications