Required Alarms¶

Overview¶

The RequiredAlarms class is a subclass of CalculationRequirement that is used to check if a list of alarms are present for an object. This is useful for calculations based on alarms.

Usage¶

This requirement can be instantiated with a list of alarms that need to be present for each object. Below there is an example of how to use this requirement:

requirement = RequiredAlarms(alarms={"SDM1-VRN1-01": [1, 2]})

After calling check and get_data methods, the data attribute of the requirement will be a DataFrame with the required alarms for the desired period.

Database Requirements¶

This requirement expects that the alarms definition table is set with the necessary alarms for the model of the wanted object.

To check if the alarms are set correctly, go to the v_alarms_def view in the database.

Class Definition¶

`RequiredAlarms(alarms, match_id_on='manufacturer_id', optional=False)` ¶

Subclass of CalculationRequirement that defines the alarms that are required for the calculation.

This will check the performance database for the existence of the required alarms for the wanted objects.

Parameters:

alarms ¶
(dict[str, list[int]]) –

Dictionary with the alarms that are required for the calculation.

This must be in the following format: {object_name: [alarm_id_1, alarm_id_2, ...], ...}
match_id_on ¶
(Literal['manufacturer_id', 'id'], default: 'manufacturer_id' ) –

In which column should we match the desired ids. Can be one of ["manufacturer_id", "id"]. See view v_alarms_def in performance_db for more information.

By default "manufacturer_id"
optional ¶
(bool, default: False ) –

Set to True if this is an optional requirement. by default False

Source code in echo_energycalc/calculation_requirement_alarms.py

def __init__(
    self,
    alarms: dict[str, list[int]],
    match_id_on: Literal["manufacturer_id", "id"] = "manufacturer_id",
    optional: bool = False,
) -> None:
    """
    Constructor of the RequiredAlarms class.

    This will check the performance database for the existence of the required alarms for the wanted objects.

    Parameters
    ----------
    alarms : dict[str, list[int]]
        Dictionary with the alarms that are required for the calculation.

        This must be in the following format: {object_name: [alarm_id_1, alarm_id_2, ...], ...}
    match_id_on : Literal["manufacturer_id", "id"], optional
        In which column should we match the desired ids. Can be one of ["manufacturer_id", "id"]. See view `v_alarms_def` in performance_db for more information.

        By default "manufacturer_id"
    optional : bool, optional
        Set to True if this is an optional requirement. by default False
    """
    super().__init__(optional)

    # checking if "match_id_on" is a valid value
    if match_id_on not in ["manufacturer_id", "id"]:
        raise ValueError(f"match_id_on must be 'manufacturer_id' or 'id', not {match_id_on}")
    self._match_id_on = match_id_on

    # check if alarms is a dict with str keys and list of int values
    if not isinstance(alarms, dict):
        raise TypeError(f"alarms must be a dict, not {type(alarms)}")
    if not all(isinstance(key, str) for key in alarms):
        raise TypeError(f"all alarms keys must be str, not {[type(key) for key in alarms]}")
    if not all(isinstance(value, list) for value in alarms.values()):
        raise TypeError(f"all alarms values must be list, not {[type(value) for value in alarms.values()]}")
    if not all(all(isinstance(item, int) for item in value) for value in alarms.values()):
        raise TypeError(f"all alarms values must be list of int, not {[type(item) for value in alarms.values() for item in value]}")

    self._alarms = alarms

`alarms` `property` ¶

Alarms required.

Returns:

dict[str, list[int]] –

Alarms required.

`checked` `property` ¶

Attribute that defines if the requirement has been checked. It's value will start as False and will be set to True after the check method is called.

Returns:

bool –

True if the requirement has been checked.

`data` `property` ¶

Data required for the calculation.

Returns:

DataFrame –

DataFrame with the alarms history for the required alarms and period.

`match_id_on` `property` ¶

Type of id used to match the alarms.

Returns:

str –

Type of id used to match the alarms.

`optional` `property` ¶

Attribute that defines if the requirement is optional.

If optional is True, the requirement is only validated to check if it could exist, not if it is actually present. This is useful for requirements that are not necessary for all calculations, but are useful for some of them.

Returns:

bool –

True if the requirement is optional.

`check()` ¶

Method used to check if all required alarms are present in the database for each object.

This will raise an error if any of the required alarms are missing.

Returns:

bool –

Returns True if all required alarms are present in the database for each object.

Source code in echo_energycalc/calculation_requirement_alarms.py

def check(self) -> bool:
    """
    Method used to check if all required alarms are present in the database for each object.

    This will raise an error if any of the required alarms are missing.

    Returns
    -------
    bool
        Returns True if all required alarms are present in the database for each object.
    """
    if self.optional:
        return True

    # iterating each object and checking if all alarms are present
    for object_name, alarms in self.alarms.items():
        # getting object model
        object_model = self._perfdb.objects.instances.get(object_names=[object_name], output_type="dict")[object_name][
            "object_model_name"
        ]

        # getting existing alarms
        existing_alarms = self._perfdb.alarms.definitions.get_ids(
            object_models=[object_model],
            alarm_ids=alarms,
            match_alarm_id_on=self.match_id_on,
        )[object_model]

        # checking if all required alarms are present
        if self.match_id_on == "manufacturer_id":
            missing_alarms = set(alarms) - set(existing_alarms.keys())
        else:
            missing_alarms = set(alarms) - {
                alarm_id for alarm_type_def in existing_alarms.values() for alarm_id in alarm_type_def.values()
            }
        if len(missing_alarms) > 0:
            raise ValueError(f"Alarms ({self.match_id_on}) {missing_alarms} do not exist for object {object_name}")

    self._checked = True

    return True

`get_data(period, **kwargs)` ¶

Method used to get the alarms history for the required alarms and period.

Parameters:

period ¶
(DateTimeRange) –

Desired period for the features.

Returns:

DataFrame –

DataFrame with the alarms history for the required alarms and period.

Source code in echo_energycalc/calculation_requirement_alarms.py

def get_data(self, period: DateTimeRange, **kwargs) -> DataFrame:  # noqa: ARG002
    """
    Method used to get the alarms history for the required alarms and period.

    Parameters
    ----------
    period : DateTimeRange
        Desired period for the features.

    Returns
    -------
    DataFrame
        DataFrame with the alarms history for the required alarms and period.
    """
    # checking if necessary keyword arguments are present
    if not isinstance(period, DateTimeRange):
        raise TypeError(f"period must be a DateTimeRange object, not {type(period)}")

    # check if requirement has been checked
    if not self._checked:
        self.check()

    try:
        # creating empty DataFrame
        alarms_df = DataFrame()

        # iterating each object and getting alarms history
        for object_name, alarms in self.alarms.items():
            # getting alarms history
            alarms_history = self._perfdb.alarms.history.get(
                period=period,
                object_names=[object_name],
                match_alarm_id_on=self.match_id_on,
                alarm_ids=alarms,
                output_type="DataFrame",
            )
            # appending to data
            alarms_df = concat([alarms_df, alarms_history], axis=0, ignore_index=True)

        # resetting index
        alarms_df = alarms_df.reset_index(drop=True)

        # saving data
        self._data = alarms_df

    except Exception as e:
        if self.optional:
            self._data = None
        else:
            raise e

    return self.data

Required Alarms¶

Overview¶

Usage¶

Database Requirements¶

Class Definition¶

`RequiredAlarms(alarms, match_id_on='manufacturer_id', optional=False)` ¶

`alarms` ¶

`match_id_on` ¶

`optional` ¶

`alarms` `property` ¶

`checked` `property` ¶

`data` `property` ¶

`match_id_on` `property` ¶

`optional` `property` ¶

`check()` ¶

`get_data(period, **kwargs)` ¶

`period` ¶

Required Alarms¶

Overview¶

Usage¶

Database Requirements¶

Class Definition¶

RequiredAlarms(alarms, match_id_on='manufacturer_id', optional=False) ¶

alarms ¶

match_id_on ¶

optional ¶

alarms property ¶

checked property ¶

data property ¶

match_id_on property ¶

optional property ¶

check() ¶

get_data(period, **kwargs) ¶

period ¶

`RequiredAlarms(alarms, match_id_on='manufacturer_id', optional=False)` ¶

`alarms` ¶

`match_id_on` ¶

`optional` ¶

`alarms` `property` ¶

`checked` `property` ¶

`data` `property` ¶

`match_id_on` `property` ¶

`optional` `property` ¶

`check()` ¶

`get_data(period, **kwargs)` ¶

`period` ¶