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:

Python

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

Python

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
    """Defines in which column we should match the desired alarm ids in the database"""

    self._validate_dict_of_lists(alarms, "alarms", key_type=str, item_type=int)

    self._alarms = alarms
    """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, ...], ...}"""

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.

fetched property

Attribute that defines if get_data() has been called on this requirement.

True even when the fetch returned no data (e.g. an optional requirement that found nothing). Use this to distinguish "never fetched" from "fetched but empty/None".

Returns:

• bool –

  True if get_data() has been called at least once.

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

Check that the requirement is met.

This concrete implementation handles two concerns automatically so that subclasses only need to implement _do_check():

1. Already-checked guard — returns True immediately if check() has already succeeded for this instance, avoiding redundant DB round-trips when _fetch_requirements() iterates requirements on every _compute() call.
2. Per-thread caching — when _check_cache_key() returns a non-None key, the result produced by _do_check() is stored in a thread-local cache and reused by subsequent instances in the same thread with the same key. Because the cache is never shared across threads, no locking is needed and concurrent Polars operations inside _do_check cannot deadlock.

The optional guard is intentionally delegated to _do_check() because different subclasses have different optional semantics (see _do_check docs).

Returns:

• bool –

  True if the requirement is met; raises on unmet non-optional requirements.

Source code in echo_energycalc/calculation_requirements_core.py

Python

def check(self) -> bool:
    """
    Check that the requirement is met.

    This concrete implementation handles two concerns automatically so that
    subclasses only need to implement ``_do_check()``:

    1. **Already-checked guard** — returns ``True`` immediately if ``check()`` has
       already succeeded for this instance, avoiding redundant DB round-trips when
       ``_fetch_requirements()`` iterates requirements on every ``_compute()`` call.
    2. **Per-thread caching** — when ``_check_cache_key()`` returns a non-None key,
       the result produced by ``_do_check()`` is stored in a thread-local cache and
       reused by subsequent instances in the same thread with the same key. Because
       the cache is never shared across threads, no locking is needed and concurrent
       Polars operations inside ``_do_check`` cannot deadlock.

    The **optional guard** is intentionally delegated to ``_do_check()`` because
    different subclasses have different optional semantics (see ``_do_check`` docs).

    Returns
    -------
    bool
        True if the requirement is met; raises on unmet non-optional requirements.
    """
    if self._checked:
        return True

    cache_key = self._check_cache_key()

    if cache_key is not None:
        _tl = type(self)._cache_local  # noqa: SLF001
        if not hasattr(_tl, "cache"):
            _tl.cache = {}
        cached = _tl.cache.get(cache_key)
        if cached is None:
            self._do_check()
            _tl.cache[cache_key] = self._get_cache_value()
            cached = _tl.cache[cache_key]
        else:
            logger.debug("Cache hit for %s (key=%s)", type(self).__name__, cache_key)
        self._set_from_cache(cached)
    else:
        self._do_check()

    self._checked = True
    return True

get_data(period, output_type='pl.DataFrame', **kwargs)

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

All internal data handling is done with Polars for performance; self._data always stores a Polars DataFrame. The result is converted to pandas when output_type="DataFrame".

Parameters:

• period

  (DateTimeRange) –

  Desired period for the features.

• output_type

  (Literal['DataFrame', 'pl.DataFrame'], default: 'pl.DataFrame' ) –

  Output format. By default "pl.DataFrame".

Returns:

• DataFrame | DataFrame –

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

Source code in echo_energycalc/calculation_requirement_alarms.py

Python

@validate_call
def get_data(
    self,
    period: DateTimeRange,
    output_type: Literal["DataFrame", "pl.DataFrame"] = "pl.DataFrame",
    **kwargs,  # noqa: ARG002
) -> pl.DataFrame | DataFrame:
    """
    Method used to get the alarms history for the required alarms and period.

    All internal data handling is done with Polars for performance; ``self._data`` always
    stores a Polars DataFrame. The result is converted to pandas when ``output_type="DataFrame"``.

    Parameters
    ----------
    period : DateTimeRange
        Desired period for the features.
    output_type : Literal["DataFrame", "pl.DataFrame"], optional
        Output format. By default ``"pl.DataFrame"``.

    Returns
    -------
    DataFrame | pl.DataFrame
        DataFrame with the alarms history for the required alarms and period.
    """
    # check if requirement has been checked
    if not self._checked:
        self.check()

    try:
        alarm_dfs: list[pl.DataFrame] = []

        for object_name, alarms in self.alarms.items():
            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="pl.DataFrame",
            )
            alarm_dfs.append(alarms_history)

        result_pl = pl.concat(alarm_dfs, how="diagonal") if alarm_dfs else pl.DataFrame()
        self._data = result_pl

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

    finally:
        self._fetched = True

    if self.data is None:
        return None
    if output_type == "pl.DataFrame":
        return self.data
    return self.data.to_pandas(use_pyarrow_extension_array=True) if not self.data.is_empty() else DataFrame()