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:
```
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:

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

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 and "default_truncate_seconds" not in settings["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: 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:

DataFrame | None: –

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

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

Abstract method that should be implemented in child classes. Should calculate the alarm for the given object and period.

The method also should call the method "save" to save the calculated alarm in performance_db.

At the end of the method, the attribute "_result" should be filled with the result of the calculation.

Parameters:

period ¶
(DateTimeRange) –

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

Dict with DataFrame containing the cached data for each object. This is useful to avoid querying the database multiple times for the same data. 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 –

Pandas DataFrame with the calculated alarm.

Source code in echo_energycalc/alarm_calc_core.py

@abstractmethod
def calculate(
    self,
    period: DateTimeRange,
    cached_data: dict[str, DataFrame] | None = None,
    save: bool = True,
) -> DataFrame:
    """
    Abstract method that should be implemented in child classes. Should calculate the alarm for the given object and period.

    The method also should call the method "save" to save the calculated alarm in performance_db.

    At the end of the method, the attribute "_result" should be filled with the result of the calculation.

    Parameters
    ----------
    period : DateTimeRange
        Period for which the alarm will be calculated.
    cached_data : dict[str, DataFrame] | None, optional
        Dict with DataFrame containing the cached data for each object. This is useful to avoid querying the database multiple times for the same data. 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
    -------
    DataFrame
        Pandas DataFrame with the calculated alarm.
    """
    raise NotImplementedError("The method 'calculate' must be implemented in the child class.")

`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

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.")

    # getting result
    result = self._result

    # checking if there are rows where start and end are equal
    if not result.empty:
        equal_count = result[result["start"] == result["end"]].shape[0]
        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[result["start"] != result["end"]]
        end_before_start_count = result[result["end"] < result["start"]].shape[0]
        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[result["end"] >= result["start"]]

    # renaming columns
    result = result.rename(columns={"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.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)` `abstractmethod` ¶

`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) abstractmethod ¶

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)` `abstractmethod` ¶

`period` ¶

`cached_data` ¶

`save` ¶

`save()` ¶