Calculation Handler¶
The CalculationHandler class is responsible for handling the calculation of features. It do all the mapping from the features in the database and its respective FeatureCalculator subclass, calling the calculate method from the FeatureCalculator subclass and saving the results in the database.
The most clear usage of this now is the Airflow DAG feature-calculator that is responsible for periodically calculating the features from the previous hours.
Class Definition¶
CalculationHandler(features, object_filters=None, regex_feature_names=False, ignored_features=None)
¶
Class used to handle the calculation of features.
This will essentially be a wrapper for the FeatureCalculator classes, iterating object by object and feature by feature, performing the calculations and saving the results in performance_db.
An instance of this class should be created in most scripts that import data and them do some calculations to store in the database. Avoid using the FeatureCalculator classes directly.
Parameters:
-
(features¶str | list[str]) –List of features to be calculated. Can be a single feature as a string as well.
If a list is passed, the features will be calculated in the order they are passed.
-
(object_filters¶dict[str, list[str]], default:None) –A dict containing the keyword arguments that will be passed to the function perfdb.objects.instances.get(). These will be used to filter the objects that will be calculated.
Most common arguments are
object_names,object_types,object_models,parent_objectsandspe_names.By default None, which means that all objects will be calculated.
-
(regex_feature_names¶bool, default:False) –If True, the features will be treated as regular expressions. This is useful when you want to calculate all features that match a certain pattern.
For example, if features=['^active.', '^reactive.'], it will return all features that start with 'active' or 'reactive'.
By default False.
-
(ignored_features¶list[str] | None, default:None) –List of features that should be ignored. These will not be calculated. Keep in mind that they are not regex patterns even if regex_feature_names is True.
By default None.
Source code in echo_energycalc/calculation_handler.py
def __init__(
self,
features: str | list[str],
object_filters: dict[str, list[str]] | None = None,
regex_feature_names: bool = False,
ignored_features: list[str] | None = None,
) -> None:
"""
Class used to handle the calculation of features.
This will essentially be a wrapper for the FeatureCalculator classes, iterating object by object and feature by feature, performing the calculations and saving the results in performance_db.
An instance of this class should be created in most scripts that import data and them do some calculations to store in the database. Avoid using the FeatureCalculator classes directly.
Parameters
----------
features : str | list[str]
List of features to be calculated. Can be a single feature as a string as well.
If a list is passed, the features will be calculated in the order they are passed.
object_filters : dict[str, list[str]], optional
A dict containing the keyword arguments that will be passed to the function perfdb.objects.instances.get(). These will be used to filter the objects that will be calculated.
Most common arguments are `object_names`, `object_types`, `object_models`, `parent_objects` and `spe_names`.
By default None, which means that all objects will be calculated.
regex_feature_names : bool, optional
If True, the features will be treated as regular expressions. This is useful when you want to calculate all features that match a certain pattern.
For example, if features=['^active.*', '^reactive.*'], it will return all features that start with 'active' or 'reactive'.
By default False.
ignored_features : list[str] | None, optional
List of features that should be ignored. These will not be calculated. Keep in mind that they are not regex patterns even if regex_feature_names is True.
By default None.
"""
# object to store all the errors that occur during the calculations
if object_filters is None:
object_filters = {}
self._errors = ErrorDataSource("server_calc")
# creating structure that will be used to connect to performance_db
self._perfdb = PerfDB(application_name="CalculationHandler")
# getting objects from performance_db
# this assumes that all keyword arguments in object_filters are valid arguments for the method get_objects from echo_postgres like 'object_names', 'object_models'
self._objects = self._perfdb.objects.instances.get(
**object_filters,
output_type="DataFrame",
get_attributes=True,
attribute_names=["calculation_disabled"],
)
# checking if features is a valid input
all_features = None
if isinstance(features, str):
all_features = [features]
elif isinstance(features, list):
if len(features) == 0:
warnings.warn("The list of features is empty. No features will be calculated.", stacklevel=2)
logger.warning("The list of features is empty. No features will be calculated.")
self._features = None
return
if all(isinstance(feature, str) for feature in features):
all_features = features
if all_features is None:
raise TypeError(f"The features argument must be a string or a list of strings. Cannot be of type {type(features)}.")
# checking if all the features exist as a server_calc in performance_db
feature_df_list = []
for obj_model in self._objects["object_model_name"].unique():
# getting all the features for the current object model
obj_model_features = self._perfdb.features.definitions.get(
object_models=[obj_model],
feature_names=all_features,
data_source_types=["server_calc"],
get_attributes=True,
attribute_names=["server_calc_type"],
output_type="DataFrame",
regex_feature_names=regex_feature_names,
).reset_index(drop=False)
# saving number of features found for this object model
n_features = len(obj_model_features)
# adding column "server_calc_type" if necessary
if "server_calc_type" not in obj_model_features.columns:
obj_model_features["server_calc_type"] = None
# filtering only wanted columns
obj_model_features = obj_model_features[["object_model_name", "name", "data_source_type_name", "server_calc_type"]]
# getting all objects of this model
model_objects = self._objects[self._objects["object_model_name"] == obj_model].index.to_list()
# adding a column with the object names to the obj_model_features DataFrame and replicate obj_model_features for each object of the model
obj_model_features = pd.concat([obj_model_features] * len(model_objects), ignore_index=True)
# adding a column with the object names to the obj_model_features DataFrame
obj_model_features["object_name"] = list(itertools.chain.from_iterable(itertools.repeat(x, n_features) for x in model_objects))
feature_df_list.append(obj_model_features)
self._features = pd.concat(feature_df_list)
# checking if all the features were found
if not regex_feature_names:
missing_features = set(all_features) - set(self._features["name"])
if len(missing_features) > 0:
logger.debug(
f"""The following features are not available in performance_db for any of the models and will be skipped ({list(self._objects["object_model_name"].unique())}): {missing_features}.""",
)
# creating list of features to sort the _features DataFrame
feature_order = []
if not regex_feature_names:
feature_order = features
else:
for feature_regex in features:
matching_features = self._features["name"].str.contains(
feature_regex,
regex=True,
case=False,
) # important to ignore case to make the same type o regex matching done in the database
feature_order.extend(self._features[matching_features]["name"].unique().tolist())
# make column "name" a pd.Categorical column with the categories being the features in the order they where provided in "features" argument
self._features["name"] = pd.Categorical(self._features["name"], categories=feature_order, ordered=True)
# removing
# sorting features by feature and them by object_model_name
self._features = self._features.sort_values(by=["name", "object_model_name"])
# resetting index for later use
self._features = self._features.reset_index(drop=True)
# adding auxiliary columns
self._features["finished_calc"] = False
self._features["cached"] = False
# removing ignored features
if ignored_features is not None:
self._features = self._features[~self._features["name"].isin(ignored_features)].copy()
# dict to store results of the calculations
# First level of keys are the object names and the second level of keys are the feature names
self._results: dict[str, dict[str, FeatureCalculator]] = {}
# creating a DataFrame to save the cached results
self._cached_data = pd.DataFrame(columns=pd.MultiIndex.from_arrays([[], []], names=("object", "feature")))
errors
property
¶
Errors that occurred during calculations.
Returns:
-
ErrorDataSource–ErrorDataSource object with all the errors that occurred during calculations.
features
property
¶
objects
property
¶
List of objects for which the features will be calculated.
Returns:
-
list[str]–List of objects for which the features will be calculated.
results
property
¶
Results of the calculations.
Returns:
-
dict[str, dict[str, FeatureCalculator]]–Dict containing the results of the calculations for each object and feature.
First level of keys are the object names and the second level of keys are the feature names.
The values are FeatureCalculator objects with the results of the calculations.
calculate(period, save_into='all', save_method='end', free_up_memory=True, **kwargs)
¶
Method to calculate all the features for all the objects in the object list.
This will perform the following steps:
- Iterate over each feature in the exact order they were added to the calculator.
- Iterate over each object model.
- Iterate over each object for the object model.
- Calculate the feature for the object.
- If it fails, add the error message to self.errors for later use.
- If it succeeds and save_method is "end", add the result to self._results for later saving all features of each object at the same time.
- If it succeeds and save_method is "each", save the result to the performance_db.
Parameters:
-
(period¶DateTimeRange) –Desired period for the calculation.
-
(save_into¶Literal['all', 'performance_db'] | None, default:'all') –Argument that will be passed to the method "save". The options are: - "all": The feature will be saved in performance_db and bazefield. - "performance_db": the feature will be saved only in performance_db. - None: The feature will not be saved.
By default "all".
-
(save_method¶Literal['end', 'each'], default:'end') –If "end" will calculate everything and save the results to postgres only at the end of the all calculations for this object. This takes more time to get "live" results in the database but is more efficient in terms of storage and database performance.
If "each" will save the results to postgres after each calculation. This takes less time to get "live" results in the database but is less efficient in terms of storage and database performance.
By default "end".
-
(free_up_memory¶bool, default:True) –If True, the results of the object will be removed from self._results after saving, by default True.
If you want to access the results later, set this to False.
-
(**kwargs¶dict, default:{}) –Kept for compatibility with other methods. Not used.
Returns:
-
ErrorDataSource–ErrorDataSource object with all the errors that occurred during calculations.
Source code in echo_energycalc/calculation_handler.py
def calculate(
self,
period: DateTimeRange,
save_into: Literal["all", "performance_db"] | None = "all",
save_method: Literal["end", "each"] = "end",
free_up_memory: bool = True,
**kwargs,
) -> ErrorDataSource:
"""
Method to calculate all the features for all the objects in the object list.
This will perform the following steps:
1. Iterate over each feature in the exact order they were added to the calculator.
2. Iterate over each object model.
3. Iterate over each object for the object model.
4. Calculate the feature for the object.
5. If it fails, add the error message to self.errors for later use.
6. If it succeeds and save_method is "end", add the result to self._results for later saving all features of each object at the same time.
7. If it succeeds and save_method is "each", save the result to the performance_db.
Parameters
----------
period : DateTimeRange
Desired period for the calculation.
save_into : Literal["all", "performance_db"] | None, optional
Argument that will be passed to the method "save". The options are:
- "all": The feature will be saved in performance_db and bazefield.
- "performance_db": the feature will be saved only in performance_db.
- None: The feature will not be saved.
By default "all".
save_method : Literal["end", "each"], optional
If "end" will calculate everything and save the results to postgres only at the end of the all calculations for this object. This takes more time to get "live" results in the database but is more efficient in terms of storage and database performance.
If "each" will save the results to postgres after each calculation. This takes less time to get "live" results in the database but is less efficient in terms of storage and database performance.
By default "end".
free_up_memory : bool, optional
If True, the results of the object will be removed from self._results after saving, by default True.
If you want to access the results later, set this to False.
**kwargs : dict, optional
Kept for compatibility with other methods. Not used.
Returns
-------
ErrorDataSource
ErrorDataSource object with all the errors that occurred during calculations.
"""
if self._features is None:
return self._errors
# resetting calculation history
self._reset_calc_history()
# iterating over calculated features
features_list = list(self._features["name"].unique())
for feature in features_list:
# iterating over object models
obj_models = list(self._features[self._features["name"] == feature]["object_model_name"].unique())
for obj_model in obj_models:
# iterating over objects of that model
selected_objects = list(self._objects[self._objects["object_model_name"] == obj_model].index)
for obj in selected_objects:
if "calculation_disabled" in self._objects.columns:
ignored_features = self._objects.loc[obj, "calculation_disabled"]
if not isinstance(ignored_features, list):
logger.warning(
f"Object '{obj}' has 'calculation_disabled' attribute that is not a list. Expected a list, got {type(ignored_features)}. Skipping this object.",
)
elif "all" in [x.lower() for x in ignored_features] or feature in ignored_features:
logger.info(
f"Skipping feature '{feature}' for object '{obj}' because it is disabled in calculation_disabled object attribute.",
)
continue
logger.info(
f"Calculating feature '{feature}' ({features_list.index(feature) + 1} of {len(features_list)}) for object '{obj}' ({selected_objects.index(obj) + 1} of {len(selected_objects)}) of model '{obj_model}' ({obj_models.index(obj_model) + 1} of {len(obj_models)})",
)
t0 = perf_counter()
# creating error object for this object and feature
# these will be used later to save errors in case they happen
self._errors.add_child(ErrorObject(name=obj))
self._errors.children[obj].add_child(ErrorFeature(name=feature))
# execute all within a try except block to catch any errors that may occur
feature_calc = None
try:
try:
# checking if class for that feature exists
server_calc_type = self._features[
(self._features["name"] == feature) & (self._features["object_model_name"] == obj_model)
]["server_calc_type"].iloc[0]
if server_calc_type not in FEATURE_CALC_CLASS_MAPPING:
raise ValueError(
f"Feature '{feature}' has a server_calc_type that is not configured yet: '{server_calc_type}'.",
)
if FEATURE_CALC_CLASS_MAPPING[server_calc_type] is None:
logger.info(f"Feature '{feature}' is set to be ignored. Skipping calculation.")
continue
# creating feature calculator object
feature_calc = FEATURE_CALC_CLASS_MAPPING[server_calc_type](
object_name=obj,
feature=feature,
)
# getting cached data if save_method is "end"
t1 = perf_counter()
cached_data = None
if save_method == "end":
cached_data = self._get_cached_data()
# defining save_into for function
calc_save_into = save_into if save_method == "each" else None
# performing calculation
t2 = perf_counter()
feature_calc.calculate(
period=period,
save_into=calc_save_into,
cached_data=cached_data,
**kwargs,
)
# adding result to results dict
self._add_result(object_name=obj, feature_name=feature, result=feature_calc)
t3 = perf_counter()
except Exception as e:
message = f"'{obj}' - '{feature}': Error while calculating feature."
logger.exception(message)
cprint(message, "red")
self._errors.children[obj].children[feature].add_exception(e)
# checking if there are any additional features that need to be added to self._features
# this will happen if a calculation returns a DataFrame with multiple columns instead of only a Series corresponding to the desired feature
# if this is not done getting cached data might result in many errors
if feature_calc is not None and feature_calc.result is not None and not feature_calc.result.empty:
if isinstance(feature_calc.result, pd.DataFrame):
calculated_features = feature_calc.result.columns.get_level_values("feature")
elif isinstance(feature_calc.result, pd.Series):
calculated_features = [feature_calc.result.name]
exisiting_features = self._features[
(self._features["object_name"] == obj) & (self._features["name"].isin(calculated_features))
]["name"]
missing_features = set(calculated_features) - set(exisiting_features)
if len(missing_features) > 0:
append_df = pd.DataFrame.from_dict(
{
"object_name": [obj] * len(missing_features),
"name": list(missing_features),
"finished_calc": [True] * len(missing_features),
"cached": [False] * len(missing_features),
},
)
self._features = pd.concat([self._features, append_df], ignore_index=True, axis=0, join="outer")
self._features = self._features.reset_index(drop=True)
# marking feature as finished in self._features
# this is done even if there was an error so that the logic to save at end of calculation works
idx = self._features[(self._features["name"] == feature) & (self._features["object_name"] == obj)].index[0]
self._features.loc[idx, "finished_calc"] = True
# if save_method is "end" then we need to save the results to performance_db
if save_method == "end":
try:
# checking if there memory usage is too high
# this is extremely important to avoid the complete calculation crashing due to high memory usage when cached data becomes too big
if psutil.virtual_memory().percent / 100 > MAX_MEMORY_USAGE:
self._save_all_obj_results(save_into=save_into, free_up_memory=free_up_memory, **kwargs)
self._remove_cached_data()
# checking if all features for this object are finished and saving
elif self._features[(self._features["object_name"] == obj) & (~self._features["finished_calc"])].empty:
# saving results to performance_db
self._save_obj_results(object_name=obj, save_into=save_into, free_up_memory=free_up_memory, **kwargs)
except Exception as e:
message = f"'{obj}' Error while saving results to database."
logger.exception(message)
self._errors.children[obj].add_exception(e)
if feature_calc is not None and feature_calc.result is not None:
message = f"{feature_calc} - {period=} - Preprocess in {t1 - t0:.2f}s, Cached Data in {t2 - t1:.2f}s, Calculation in {t3 - t2:.2f}s and Afterprocess {perf_counter() - t3:.2f}s."
logger.info(message)
print(message) # noqa: T201
except Exception as e:
message = f"'{obj}' Error while calculating feature {feature}."
logger.exception(message)
self._errors.children[obj].children[feature].add_exception(e)
return self._errors
from_calc_types(calc_types, object_filters=None, one_per_object=False)
classmethod
¶
Class method that returns a CalculationHandler for all features of the desired server_calc_types.
Parameters:
-
(calc_types¶list[str]) –Desired calculation types, as defined in
server_calc_typefeature attribute. -
(object_filters¶dict[str, list[str]] | None, default:None) –A dict containing the keyword arguments that will be passed to the function PerfDB.objects.instances.get(). These will be used to filter the objects that will be calculated.
Most common arguments are
object_names,object_types, "object_models,parent_objectsandspe_names`.By default None, which means that all objects will be calculated.
-
(one_per_object¶bool, default:False) –If True, one CalculationHandler per object will be returned. This is useful when you want to calculate only one feature per object.
Returns:
-
CalculationHandler | list[CalculationHandler]–CalculationHandler object with the features for the desired calculation types.
If one_per_object is True, a list of CalculationHandler objects will be returned, one for each object found.
Source code in echo_energycalc/calculation_handler.py
@classmethod
def from_calc_types(
cls,
calc_types: list[str],
object_filters: dict[str, list[str]] | None = None,
one_per_object: bool = False,
) -> Self | list[Self]:
"""
Class method that returns a CalculationHandler for all features of the desired `server_calc_types`.
Parameters
----------
calc_types : list[str]
Desired calculation types, as defined in `server_calc_type` feature attribute.
object_filters : dict[str, list[str]] | None, optional
A dict containing the keyword arguments that will be passed to the function PerfDB.objects.instances.get(). These will be used to filter the objects that will be calculated.
Most common arguments are `object_names`, `object_types`, "object_models`, `parent_objects` and `spe_names`.
By default None, which means that all objects will be calculated.
one_per_object : bool, optional
If True, one CalculationHandler per object will be returned. This is useful when you want to calculate only one feature per object.
Returns
-------
CalculationHandler | list[CalculationHandler]
CalculationHandler object with the features for the desired calculation types.
If one_per_object is True, a list of CalculationHandler objects will be returned, one for each object found.
"""
# connecting to postgres
if object_filters is None:
object_filters = {}
perfdb = PerfDB(application_name="CalculationHandler")
objs = perfdb.objects.instances.get(**object_filters, output_type="DataFrame")
# getting object models
models = objs["object_model_name"].unique().tolist()
# getting all features of the desired calc_types
model_features = pd.DataFrame()
for calc_type in calc_types:
this_model_features = perfdb.features.definitions.get(
object_models=models,
attributes={"server_calc_type": calc_type},
output_type="DataFrame",
)
model_features = pd.concat([model_features, this_model_features], ignore_index=False, axis=0, join="outer")
model_features = model_features.reset_index(drop=False)
if not one_per_object:
return CalculationHandler(
features=model_features["name"].to_list(),
object_filters=object_filters,
)
handlers = []
for obj in objs.index:
obj_model = objs.loc[obj, "object_model_name"]
obj_features = model_features[model_features["object_model_name"] == obj_model]["name"].to_list()
handlers.append(
CalculationHandler(
features=obj_features,
object_filters={"object_names": [obj]},
),
)
return handlers
from_type_defaults(object_type='infer', object_filters=None, ignored_features=None, one_per_object=False)
classmethod
¶
Class method that returns a CalculationHandler object with the features that should be calculated for each object type, in the correct order.
The type defaults are defined in the default_feature_order setting in performance database
Parameters:
-
(object_type¶str, default:'infer') –Object type that will be used to get the features that should be calculated.
If "infer" is used, the object type will be inferred from the object_filters. This will also cause the method to return a list of CalculationHandler objects, one for each object type found.
Default is "infer".
-
(object_filters¶dict[str, list[str]] | None, default:None) –A dict containing the keyword arguments that will be passed to the function PerfDB.objects.instances.get(). These will be used to filter the objects that will be calculated.
Most common arguments are
object_names,object_types, "object_models,parent_objectsandspe_names`.By default None, which means that all objects will be calculated.
-
(ignored_features¶list[str] | None, default:None) –List of features that should be ignored. These will not be calculated. By default None.
-
(one_per_object¶bool, default:False) –If True, one CalculationHandler per object will be returned. This is useful when you want to calculate only one feature per object.
Returns:
-
CalculationHandler | list[CalculationHandler]–CalculationHandler object with the features that should be calculated for the desired object type, in the correct order.
If object_type is "infer" or one_per_object is True, a list of CalculationHandler objects will be returned, one for each object type or object found.
Source code in echo_energycalc/calculation_handler.py
@classmethod
def from_type_defaults(
cls,
object_type: str = "infer",
object_filters: dict[str, list[str]] | None = None,
ignored_features: list[str] | None = None,
one_per_object: bool = False,
) -> Self | list[Self]:
"""
Class method that returns a CalculationHandler object with the features that should be calculated for each object type, in the correct order.
The type defaults are defined in the default_feature_order setting in performance database
Parameters
----------
object_type : str, optional
Object type that will be used to get the features that should be calculated.
If "infer" is used, the object type will be inferred from the object_filters. This will also cause the method to return a list of CalculationHandler objects, one for each object type found.
Default is "infer".
object_filters : dict[str, list[str]] | None, optional
A dict containing the keyword arguments that will be passed to the function PerfDB.objects.instances.get(). These will be used to filter the objects that will be calculated.
Most common arguments are `object_names`, `object_types`, "object_models`, `parent_objects` and `spe_names`.
By default None, which means that all objects will be calculated.
ignored_features : list[str] | None, optional
List of features that should be ignored. These will not be calculated.
By default None.
one_per_object : bool, optional
If True, one CalculationHandler per object will be returned. This is useful when you want to calculate only one feature per object.
Returns
-------
CalculationHandler | list[CalculationHandler]
CalculationHandler object with the features that should be calculated for the desired object type, in the correct order.
If object_type is "infer" or one_per_object is True, a list of CalculationHandler objects will be returned, one for each object type or object found.
"""
perfdb = PerfDB(application_name="CalculationHandler")
if ignored_features is None:
ignored_features = []
if object_filters is None:
object_filters = {}
if object_type == "infer" or one_per_object:
objects = perfdb.objects.instances.get(**object_filters, output_type="DataFrame")
obects_def = objects.copy()
obj_list = objects.index.to_list()
# creating a dict grouping objects by "object_type_name" columns and values will be a list of objects for that name (object name is the index)
objects = objects.groupby("object_type_name").apply(lambda x: x.index.tolist(), include_groups=False).to_dict()
object_types = list(objects.keys())
logger.info(
f"Object type inferred as the following: \n'{jsbeautifier.beautify(json.dumps(objects, sort_keys=True), js_options)}'.",
)
else:
object_types = [object_type]
# getting feature order from database
feature_order = perfdb.settings.get()["feature_calculation"]["default_feature_order"]["value"]
# checking if object_type is defined in feature_order
missing_object_types = set(object_types) - set(feature_order.keys())
if len(missing_object_types) > 0:
raise ValueError(f"Object types '{missing_object_types}' are not defined in default_feature_order in database.")
if object_type != "infer":
# if no keys in object_filters have values, define key "object_types" with the desired object_type
if all(len(value) == 0 for value in object_filters.values()):
object_filters["object_types"] = [object_type]
if not one_per_object:
handler = CalculationHandler(
features=feature_order[object_type],
object_filters=object_filters,
regex_feature_names=True,
ignored_features=ignored_features,
)
else:
handler = []
for obj_name in obj_list:
obj_type = obects_def.loc[obj_name, "object_type_name"]
handler.append(
CalculationHandler(
features=feature_order[obj_type],
object_filters={"object_names": [obj_name]},
regex_feature_names=True,
ignored_features=ignored_features,
),
)
else:
handler = []
if not one_per_object:
for obj_type in object_types:
handler.append(
CalculationHandler(
features=feature_order[obj_type],
object_filters={"object_names": objects[obj_type]},
regex_feature_names=True,
ignored_features=ignored_features,
),
)
else:
for obj_name in obj_list:
obj_type = obects_def.loc[obj_name, "object_type_name"]
handler.append(
CalculationHandler(
features=feature_order[obj_type],
object_filters={"object_names": [obj_name]},
regex_feature_names=True,
ignored_features=ignored_features,
),
)
return handler
str()
¶
Method that returns a string representation of the object.
Currently the same as repr.
Source code in echo_energycalc/calculation_handler.py
def str(self) -> str:
"""
Method that returns a string representation of the object.
Currently the same as repr.
"""
return repr(self)