Alarm Calculation Base Class¶

The AlarmCalc class is the base class for all alarm calculation classes. It determines the interface that all alarm calculation classes must implement. The AlarmCalc class is defined in the alarm_calc_core.py file and defines these top level methods:

A default __init__ method that receives the object name and alarm id to calculate. It will do some initial checks to make sure the alarm definition is correct and will create the necessary data structures to store the results.
The save method that will save the results of the calculation to the database. This method should be called after the calculation is done and the results are ready to be stored.
A series of properties that can be used to retrieve the main configurations related to the calculation, like the name of the alarm calculator, the object name and the alarm id, etc.

The subclasses of AlarmCalc must implement the following attributes and methods:

_name class attribute: A string with the name of the alarm calculator. This is used to identify the alarm calculator type in the database. For example, in case of alarms based on feature thresholds, the class definition starts like this:
Python
```
class AlarmCalcThreshold(AlarmCalc):
    _name = "threshold"
```
A __init__ method that that will call the parent __init__ method and will also check for the required settings of this alarm calculator.
A calculate method that will perform the actual calculation of the alarm. This method should receive the period of the calculation and return a DataFrame with the calculated alarms. At the end of the calculation, the resulting DataFrame must be saved in the _result attribute of the class.

Class Definition¶

`AlarmCalculator(object_name, alarm_id)` ¶

Base abstract class for alarm calculators. Already defines the methods and attributes that are common to all alarm calculators.

This should be overloaded in child classes of AlarmCalculator to define the name, requirements and logic of the alarm calculator.

It's extremely important to define the name of the alarm calculator in the child class. This name must be equal to the "trigger_name" value of the "trigger" column of the alarm in performance_db and will be used to check if the alarm calculator is the correct one for the alarm. Set the name of the alarm calculator in the child class as a class attribute "_name".

Also, the alarm must be defined in the database with the data source type "server_alarm".

Below there is a simple example on how to define a child class of AlarmCalculator:

Python

class AlarmCalculatorExample(AlarmCalculator):
    # name of the alarm calculator
    _name = "name"

    def __init__(self, object_name: str, alarm_id: int) -> None:
        # initialize parent class
        super().__init__(object_name, alarm_id)

Parameters:

object_name ¶
(str) –

Name of the object for which the alarm is calculated. It must exist in performance_db.
alarm_id ¶
(int) –

ID of the alarm (manufacturer id) for which the alarm calculator is being created. It must exist in performance_db for the model of the object.

Source code in echo_energycalc/alarm_calc_core.py

Python

def __init__(
    self,
    object_name: str,
    alarm_id: int,
) -> None:
    """
    Constructor of the AlarmCalculator class.

    This should be overloaded in child classes of AlarmCalculator to define the name, requirements and logic of the alarm calculator.

    It's extremely important to define the name of the alarm calculator in the child class. This name must be equal to the "trigger_name" value of the "trigger" column of the alarm in performance_db and will be used to check if the alarm calculator is the correct one for the alarm. Set the name of the alarm calculator in the child class as a class attribute "_name".

    Also, the alarm must be defined in the database with the data source type "server_alarm".

    Below there is a simple example on how to define a child class of AlarmCalculator:

    ```python
    class AlarmCalculatorExample(AlarmCalculator):
        # name of the alarm calculator
        _name = "name"

        def __init__(self, object_name: str, alarm_id: int) -> None:
            # initialize parent class
            super().__init__(object_name, alarm_id)
    ```

    Parameters
    ----------
    object_name : str
        Name of the object for which the alarm is calculated. It must exist in performance_db.
    alarm_id : int
        ID of the alarm (manufacturer id) for which the alarm calculator is being created. It must exist in performance_db for the model of the object.
    """
    # checking arguments
    if not isinstance(object_name, str):
        raise TypeError(f"object_name must be a string, not {type(object_name)}")
    if not isinstance(alarm_id, int):
        raise TypeError(f"alarm_id must be an integer, not {type(alarm_id)}")
    # check if self._name is defined in child class
    if not hasattr(self, "_name"):
        raise ValueError(f"AlarmCalculator name is not defined in {self.__class__.__name__}.")

    # creating structure that will be used to connect to performance_db
    self._perfdb = PerfDB(application_name=self.__class__.__name__)

    # check if object exists in performance_db
    obj_def = self._perfdb.objects.instances.get(object_names=[object_name], output_type="dict")
    if len(obj_def) == 0:
        raise ValueError(f"Object {object_name} does not exist in performance_db.")
    self._object = object_name
    # get object model
    self._object_model = obj_def[object_name]["object_model_name"]

    # check if alarm exists in performance_db
    alarm_def = self._perfdb.alarms.definitions.get(
        object_models=[self._object_model],
        data_source_types=["server_alarm"],
        alarm_ids=[alarm_id],
        match_alarm_id_on="manufacturer_id",
        output_type="dict",
    )
    if len(alarm_def) == 0:
        raise ValueError(
            f"Alarm {alarm_id} does not exist in performance_db for object {object_name} with data_source_type = 'server_alarm'.",
        )
    self._alarm_id = alarm_id
    # getting the type of the alarm
    self._alarm_type = next(iter(alarm_def[self._object_model][alarm_id].keys()))
    # getting id of the alarm in the database
    self._alarm_db_id = alarm_def[self._object_model][alarm_id][self._alarm_type]["id"]
    # getting other relevant information about the alarm
    self._alarm_settings = alarm_def[self._object_model][alarm_id][self._alarm_type]
    # popping the id from the settings
    self._alarm_settings.pop("id")

    # validating if trigger_type is equal to the name of the alarm calculator
    if "trigger_type" not in self.alarm_settings["trigger"]:
        raise ValueError(f"'trigger_type' not found in alarm {alarm_id} settings.")
    if self.alarm_settings["trigger"]["trigger_type"] != self.name:
        raise ValueError(f"'trigger_type' in alarm {alarm_id} settings is different from the alarm calculator name.")

    # getting maximum duration of alarm from database settings
    settings = self._perfdb.settings.get()
    if "alarms" not in settings or "default_truncate_seconds" not in settings.get("alarms", {}):
        logger.warning("Default truncate seconds not found in settings. Using 604800 seconds as default.")
        self._max_duration_seconds = 604800
    else:
        self._max_duration_seconds: int = settings["alarms"]["default_truncate_seconds"]["value"]["max_seconds"]

    # results of the calculation. It will be filled by the method "calculate".
    self._result: pl.DataFrame | None = None

    # period evaluated in the calculation
    self._evaluated_period: DateTimeRange | None = None

`alarm_db_id` `property` ¶

ID of the alarm in the database. This is used to get the alarm settings.

Returns:

int –

ID of the alarm in the database.

`alarm_id` `property` ¶

ID of the alarm that is calculated (manufacturer_id). This will be defined in the constructor and cannot be changed.

Returns:

int –

ID of the alarm that is calculated (manufacturer_id).

`alarm_settings` `property` ¶

Settings of the alarm. This is a dictionary with the settings of the alarm that is being calculated.

Returns:

dict[str, Any] –

Settings of the alarm.

`alarm_type` `property` ¶

Type of the alarm that is calculated. This will be defined in the constructor and cannot be changed.

Returns:

str –

Type of the alarm that is calculated.

`name` `property` ¶

Name of the alarm calculator. Is defined in child classes of AlarmCalculator.

This must be equal to the "server_calc_type" attribute of the alarm in performance_db.

Returns:

str –

Name of the alarm calculator.

`object` `property` ¶

Object for which the alarm is calculated. This will be defined in the constructor and cannot be changed.

Returns:

str –

Object name for which the alarm is calculated.

`result` `property` ¶

Result of the calculation. This is None until the method "calculate" is called.

Returns:

pl.DataFrame | None: –

Result of the calculation if the method "calculate" was called. None otherwise.

`calculate(period, cached_data=None, save=True)` ¶

Run the calculation for the given period and optionally save the result.

Calls :meth:_compute to get the result, stores it in :attr:result, then calls :meth:save. Subclasses should implement :meth:_compute instead of overriding this method.

Parameters:

period ¶
(DateTimeRange) –

Period for which the alarm will be calculated.
cached_data ¶
(dict[str, DataFrame] | None, default: None ) –

Dict with Polars DataFrames containing cached data, keyed by data type (e.g. "features"). Avoids re-querying the database across multiple alarm calculations in the same job. The default is None.
save ¶
(bool, default: True ) –

If True, the result of the alarms will be saved in the database. The default is True.

Returns:

DataFrame –

Polars DataFrame with the calculated alarm.

Source code in echo_energycalc/alarm_calc_core.py

Python

def calculate(
    self,
    period: DateTimeRange,
    cached_data: dict[str, pl.DataFrame | None] | None = None,
    save: bool = True,
) -> pl.DataFrame:
    """
    Run the calculation for the given period and optionally save the result.

    Calls :meth:`_compute` to get the result, stores it in :attr:`result`,
    then calls :meth:`save`. Subclasses should implement :meth:`_compute` instead
    of overriding this method.

    Parameters
    ----------
    period : DateTimeRange
        Period for which the alarm will be calculated.
    cached_data : dict[str, pl.DataFrame] | None, optional
        Dict with Polars DataFrames containing cached data, keyed by data type
        (e.g. ``"features"``). Avoids re-querying the database across multiple
        alarm calculations in the same job. The default is None.
    save : bool, optional
        If True, the result of the alarms will be saved in the database. The default is True.

    Returns
    -------
    pl.DataFrame
        Polars DataFrame with the calculated alarm.
    """
    result = self._compute(period, cached_data=cached_data)
    self._result = result
    if save:
        self.save()
    return result

`save()` ¶

Save the result of the calculation in the database.

If the method "calculate" was not called before, this method will raise an error.

Source code in echo_energycalc/alarm_calc_core.py

Python

def save(self) -> None:
    """
    Save the result of the calculation in the database.

    If the method "calculate" was not called before, this method will raise an error.
    """
    # checking if the result was calculated
    if self._result is None:
        raise ValueError("The method 'calculate' must be called before saving the result.")
    # checking if the period was evaluated
    if self._evaluated_period is None:
        raise ValueError("Evaluated period was not set during calculation.")

    result = self._result

    if not result.is_empty():
        equal_count = result.filter(pl.col("start") == pl.col("end")).height
        if equal_count > 0:
            logger.warning(f"Found {equal_count} rows with start equal to end in the result. These rows will be removed.")
            result = result.filter(pl.col("start") != pl.col("end"))
        end_before_start_count = result.filter(pl.col("end") < pl.col("start")).height
        if end_before_start_count > 0:
            logger.warning(f"Found {end_before_start_count} rows with end before start in the result. These rows will be removed.")
            result = result.filter(pl.col("end") >= pl.col("start"))

    result = result.rename({"alarm_id": "manufacturer_id"})

    # dropping existing alarms within the period
    logger.info(
        f"Deleting existing alarms for object {self.object} and alarm {self.alarm_id} within the period {self._evaluated_period}.",
    )
    self._perfdb.alarms.history.delete(
        object_names=[self.object],
        period=self._evaluated_period,
        alarm_ids=[self.alarm_db_id],
    )

    if not result.is_empty():
        # saving the result in the database
        logger.info(f"Saving alarms for object {self.object} and alarm {self.alarm_id} within the period {self._evaluated_period}.")
        self._perfdb.alarms.history.insert(
            df=result,
            on_conflict="update",
        )
    else:
        logger.info(f"No alarms to save for object {self.object} and alarm {self.alarm_id} within the period {self._evaluated_period}.")

Alarm Calculation Base Class¶

Class Definition¶

`AlarmCalculator(object_name, alarm_id)` ¶

`object_name` ¶

`alarm_id` ¶

`alarm_db_id` `property` ¶

`alarm_id` `property` ¶

`alarm_settings` `property` ¶

`alarm_type` `property` ¶

`name` `property` ¶

`object` `property` ¶

`result` `property` ¶

`calculate(period, cached_data=None, save=True)` ¶

`period` ¶

`cached_data` ¶

`save` ¶

`save()` ¶

Alarm Calculation Base Class¶

Class Definition¶

AlarmCalculator(object_name, alarm_id) ¶

object_name ¶

alarm_id ¶

alarm_db_id property ¶

alarm_id property ¶

alarm_settings property ¶

alarm_type property ¶

name property ¶

object property ¶

result property ¶

calculate(period, cached_data=None, save=True) ¶

period ¶

cached_data ¶

save ¶

save() ¶

`AlarmCalculator(object_name, alarm_id)` ¶

`object_name` ¶

`alarm_id` ¶

`alarm_db_id` `property` ¶

`alarm_id` `property` ¶

`alarm_settings` `property` ¶

`alarm_type` `property` ¶

`name` `property` ¶

`object` `property` ¶

`result` `property` ¶

`calculate(period, cached_data=None, save=True)` ¶

`period` ¶

`cached_data` ¶

`save` ¶

`save()` ¶