Skip to content

Required Object Attributes

Overview

The RequiredObjectAttributes class is a subclass of CalculationRequirement that is used to check if a list of attributes are present for an object. This is useful when a calculation requires a specific set of attributes to be present in an object, and it is necessary to check if they exist before proceeding with the calculation.

Usage

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

Python
requirement = RequiredObjectAttributes(attributes={"SDM1-VRN1-01": ["nominal_power", "rotor_diameter"]})

After calling check and get_data methods, the data attribute of the requirement will be a dictionary with the object as the key and a dictionary with the attributes as the value. Below there is an example of how this data is stored:

Python
{
    "SDM1-VRN1-01": {
        "nominal_power": 4200.0,
        "rotor_diameter": 150.0
    }
}

Database Requirements

This requirements expects that the object_attributes table is set with the necessary attributes for each object. This can easily be done with the database function set_attribute. Below is a query example to set the attributes for the object SDM1-VRN1-01:

SQL
SELECT * FROM performance.fn_set_attribute('SDM1-VRN1-01', 'object', 'nominal_power', '{"attribute_value": 4200.0}');
SELECT * FROM performance.fn_set_attribute('SDM1-VRN1-01', 'object', 'rotor_diameter', '{"attribute_value": 150.0}');

To check if the attributes are set correctly, go to the v_object_attributes view in the database.

Class Definition

RequiredObjectAttributes(attributes, optional=False)

Subclass of CalculationRequirement that defines the object attributes that are required for the calculation.

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

Parameters:

  • attributes

    (dict[str, list[str]]) –

    Object attributes that are required for the calculation.

    Should be in the format {object_name: [attribute_name, ...], ...}.

  • optional

    (bool, default: False ) –

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

Source code in echo_energycalc/calculation_requirement_object_attributes.py
Python
def __init__(self, attributes: dict[str, list[str]], optional: bool = False) -> None:
    """
    Constructor of the RequiredObjectAttributes class.

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

    Parameters
    ----------
    attributes : dict[str, list[str]]
        Object attributes that are required for the calculation.

        Should be in the format {object_name: [attribute_name, ...], ...}.
    optional : bool, optional
        Set to True if this is an optional requirement. by default False
    """
    super().__init__(optional)

    self._validate_dict_of_lists(attributes, "attributes", key_type=str, item_type=str)

    self._attributes = attributes

attributes property

Object attributes that are required for the calculation.

Returns:

  • dict[str, list[str]]

    Object attributes that are required for the calculation in the format {object_name: [attribute_name, ...], ...}.

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

Object attributes that are required for the calculation.

Returns:

  • dict[str, dict[str, Any]]

    Object attributes that are required for the calculation in the format {object_name: {attribute_name: attribute_value, ...}, ...}.

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.

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

Method used to get the required object attributes from performance database.

This will not do anything other than call the check() method because the check() method already gets the data.

Returns:

  • dict[str, dict[str, Any]]

    Dict in the format {object_name: {attribute_name: attribute_value, ...}, ...}.

Source code in echo_energycalc/calculation_requirement_object_attributes.py
Python
def get_data(self, **kwargs) -> dict[str, dict[str, Any]]:  # noqa: ARG002
    """
    Method used to get the required object attributes from performance database.

    This will not do anything other than call the check() method because the check() method already gets the data.

    Returns
    -------
    dict[str, dict[str, Any]]
        Dict in the format {object_name: {attribute_name: attribute_value, ...}, ...}.
    """
    if not self._checked:
        self.check()

    self._fetched = True
    return self.data