Skip to content

Alarm Manager API

The AlarmManager is a singleton service responsible for: 1. Creating and storing alarm instances. 2. Validating alarm configurations (e.g., detecting conflicting thresholds). 3. Acting as an Observer for Tag changes to trigger alarm logic. 4. Persisting alarm states to the database.

class automation.managers.alarms.AlarmManager(*args, **kwargs)

Singleton class that manages all alarms in the system.

It handles the creation, update, deletion, and retrieval of alarms. It also validates trigger conditions and manages communication with the frontend via SocketIO.

get_queue(self)

Retrieves the internal tag queue used for observer notifications.

Returns:

  • queue.Queue: The queue instance.
append_alarm(self, name, tag, type='BOOL', trigger_value=True, description='', identifier=None, state='Normal', timestamp=None, ack_timestamp=None, user=None, reload=False, sio=None)

Creates and registers a new alarm in the manager.

Parameters:

  • name (str): Alarm name.
  • tag (str): Associated Tag name.
  • type (str): Alarm type (BOOL, HH, H, L, LL).
  • trigger_value (bool|float): Value that triggers the alarm.
  • description (str, optional): Alarm description.
  • identifier (str, optional): Unique ID.
  • state (str, optional): Initial state.
  • timestamp (str, optional): Last trigger timestamp.
  • ack_timestamp (str, optional): Last acknowledgment timestamp.
  • user (User, optional): User creating the alarm.
  • reload (bool, optional): If reloading from DB.
  • sio (SocketIO, optional): SocketIO instance for real-time updates.

Returns:

  • tuple[Alarm, str]: The created Alarm object and a status message.
put(self, id, name=None, tag=None, description=None, alarm_type=None, trigger_value=None, user=None)

Updates an existing alarm configuration.

Parameters:

  • id (str): Alarm identifier.
  • name (str, optional): New name.
  • tag (str, optional): New tag name.
  • description (str, optional): New description.
  • alarm_type (str, optional): New alarm type.
  • trigger_value (float, optional): New trigger value.
  • user (User, optional): User performing the update.

Returns:

  • tuple[Alarm, str]: The updated Alarm object and a status message.
delete_alarm(self, id, user=None)

Removes an alarm from the manager and takes it out of service.

Parameters:

  • id (str): Alarm ID.
  • user (User, optional): User performing the deletion.
get_alarm(self, id)

Retrieves an alarm by its ID.

Parameters:

  • id (str): Alarm ID.

Returns:

  • Alarm: The alarm object if found.
get_alarm_by_name(self, name)

Retrieves an alarm by its name.

Parameters:

  • name (str): Alarm name.

Returns:

  • Alarm: The alarm object if found.
get_alarm_by_tag(self, tag)

Retrieves a list of alarms associated with a specific tag.

Parameters:

  • tag (str): Tag name.

Returns:

  • list[Alarm]: List of Alarm objects.
get_alarms(self)

Retrieves all registered alarms.

Returns:

  • dict: Dictionary of all Alarm objects.
get_lasts_active_alarms(self, lasts=None)

Retrieves the most recent active alarms.

Parameters:

  • lasts (int, optional): Number of alarms to retrieve.

Returns:

  • list: List of serialized active alarms sorted by timestamp.
serialize(self)

Serializes all alarms managed by this instance.

Returns:

  • list: List of serialized alarm dictionaries.
get_tag_alarms(self)

Retrieves a list of Tags that have alarms associated with them.

Returns:

  • list: List of Tag objects.
tags(self)

Retrieves a unique list of Tag names bound to alarms.

Returns:

  • list: List of Tag names.
filter_by(self, **fields)

Filters historical alarms via the database model.

Parameters:

  • fields: Filtering criteria (name, state, timestamp, etc.).

Returns:

  • tuple: (Result data, HTTP status code 200).
get_lasts(self, lasts=10)

Retrieves the last N alarm summary records.

Parameters:

  • lasts (int): Number of records.

Returns:

  • tuple: (List of records, HTTP status code 200).
summary(self)

Generates a summary of the current alarm manager state.

Returns:

  • dict: Summary including total alarms, alarm names, and associated tags.
attach(self, alarm_name)

Attaches a tag observer to a specific alarm's tag.

Parameters:

  • alarm_name (str): Name of the alarm.
execute(self, tag_name)

Evaluates alarm conditions for a given tag based on its current value.

Also handles auto-unshelving of alarms if their shelved duration has expired.

Parameters:

  • tag_name (str): Name of the tag to evaluate.