Alarm Calc - Threshold¶

The AlarmCalcThreshold class is a subclass of AlarmCalc and is used to calculate alarms based on feature thresholds. This means that an alarm will be created if the value of a feature is above or below a certain threshold.

This type of alarm calculation requires the following keys in the trigger column of the alarm definition in the database:

trigger_type: Must be threshold.
threshold_type: One of "high" or "low".
- "high": the alarm is triggered when the value is above the threshold.
- "low": the alarm is triggered when the value is below the threshold.
feature_name: Name of the feature in the performance_db that will be used to calculate the alarm.
threshold_value: Value of the threshold that will trigger the alarm.

Class Definition¶

`AlarmCalcThreshold(object_name, alarm_id)` ¶

Alarm calculator that calculates alarms based on thresholds.

It expects the following settings in the trigger columns of the alarm in performance_db:

trigger_type: "threshold"
threshold_type: One of "high" or "low":
- "high": the alarm is triggered when the value is above the threshold.
- "low": the alarm is triggered when the value is below the threshold.
feature_name: Name of the feature in the performance_db that will be used to calculate the alarm.
threshold_value: Value of the threshold that will trigger the alarm.

If non_overlapping_alarms is defined in the alarm settings, this alarm will not overlap with the alarms defined in the list, being the ones in this list considered as more important. ```

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_threshold.py

Python

def __init__(
    self,
    object_name: str,
    alarm_id: int,
) -> None:
    """Alarm calculator that calculates alarms based on thresholds.

    It expects the following settings in the trigger columns of the alarm in performance_db:

    - **trigger_type**: "threshold"
    - **threshold_type**: One of "high" or "low":
        - "high": the alarm is triggered when the value is above the threshold.
        - "low": the alarm is triggered when the value is below the threshold.
    - **feature_name**: Name of the feature in the performance_db that will be used to calculate the alarm.
    - **threshold_value**: Value of the threshold that will trigger the alarm.

    If `non_overlapping_alarms` is defined in the alarm settings, this alarm will not overlap with the alarms defined in the list, being the ones in this list considered as more important.
    ```

    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.

    """
    super().__init__(object_name, alarm_id)

    # validating if all the required columns are present in the alarm settings

    required_settings = {
        "threshold_type": {"type": str, "values": ["high", "low"]},
        "feature_name": {"type": str},
        "threshold_value": {"type": (int, float)},
    }
    for setting, setting_info in required_settings.items():
        if setting not in self.alarm_settings["trigger"]:
            raise ValueError(
                f"Setting '{setting}' is missing in the alarm settings of alarm {alarm_id} and object {object_name}.",
            )
        if not isinstance(
            self.alarm_settings["trigger"][setting],
            setting_info["type"],
        ):
            raise TypeError(
                f"Setting '{setting}' must be of type {setting_info['type']}, not {type(self.alarm_settings['trigger'][setting])}.",
            )
        if "values" in setting_info and self.alarm_settings["trigger"][setting] not in setting_info["values"]:
            raise ValueError(
                f"Setting '{setting}' must be one of {setting_info['values']}, not {self.alarm_settings['trigger'][setting]}.",
            )

`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 Calc - Threshold¶

Class Definition¶

`AlarmCalcThreshold(object_name, alarm_id)` ¶

Parameters¶

`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 Calc - Threshold¶

Class Definition¶

AlarmCalcThreshold(object_name, alarm_id) ¶

Parameters¶

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() ¶

`AlarmCalcThreshold(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()` ¶