"""
File: qumphy/uq/deepensembles.py
Project: 22HLT01 QUMPHY
Contact: oskar.pfeffer@ptb.de
Gitlab: https://gitlab.com/qumphy
Description: DeepEnsemble Trainer.
"""
import qumphy
import torch
import numpy as np
[docs]
def append_nested_dicts(dicts_list):
"""
Take a list of nested dictionaries and create a single dictionary
with the values replaced by arrays of the values of the individual dictionaries.
The dictionaries must have the same structure.
"""
dictionary = {}
for key, value in dicts_list[0].items():
if isinstance(value, dict):
dictionary[key] = append_nested_dicts([d[key] for d in dicts_list])
else:
dictionary[key] = np.array([d[key] for d in dicts_list])
return dictionary
[docs]
def reduce_nested_dict_list(dicts_list: list[dict], reduction_function: callable):
"""
Take a list of nested dictionaries and reduce them to a single dictionary
using the given reduction function.
The dictionaries must have the same structure.
Parameters
----------
dicts_list : list
List of nested dictionaries.
reduction_function : function, optional
Function to use for reduction.
Returns
-------
dict
Reduced dictionary.
"""
dictionary = append_nested_dicts(dicts_list)
def reduce(dictionary, reduction_function=max):
for key, value in dictionary.items():
if isinstance(value, dict):
dictionary[key] = reduce(value, reduction_function)
else:
dictionary[key] = reduction_function(value)
return dictionary
return reduce(dictionary, reduction_function)
[docs]
class DeepBeatEvaluation:
"""Evaluation utilities for DeepBeat ensemble predictions."""
[docs]
def evaluation_function(self, target, predictions, ensemble_predictions):
"""Evaluate individual DeepBeat models and ensemble predictions.
Parameters
----------
target : np.ndarray
Ground truth target values.
predictions : np.ndarray
Predictions from individual ensemble members.
ensemble_predictions : np.ndarray
Aggregated ensemble predictions.
Returns
-------
None
The function prints the calculated metrics.
"""
model_metrics = []
ensemble_predictions = np.squeeze(ensemble_predictions)
for prediction in predictions:
prediction = np.squeeze(prediction)
model_metrics.append(qumphy.metrics.all_binary_metrics(target, prediction))
mean_metrics = reduce_nested_dict_list(model_metrics, np.mean)
max_metrics = reduce_nested_dict_list(model_metrics, max)
min_metrics = reduce_nested_dict_list(model_metrics, min)
ensemble_metrics = qumphy.metrics.all_binary_metrics(
target, ensemble_predictions
)
self.print_metrics(mean_metrics, max_metrics, min_metrics, ensemble_metrics)
[docs]
def print_metrics(self, mean_metrics, max_metrics, min_metrics, ensemble_metrics):
for key, value in mean_metrics.items():
print(f"Mean {key}: {value}")
for key, value in max_metrics.items():
print(f"Max {key}: {value}")
for key, value in min_metrics.items():
print(f"Min {key}: {value}")
for key, value in ensemble_metrics.items():
print(f"Ensemble {key}: {value}")
[docs]
def reduce(self, predictions):
"""mean of predictions"""
ensemble_predictions = np.mean(predictions, axis=0)
return ensemble_predictions
[docs]
class PulseDBEvaluation:
"""
A class for evaluating PulseDB models and ensembles.
Contains methods for evaluating individual models and ensembles,
as well as methods for printing and saving the results.
"""
[docs]
def denormalize(self, dataset, predictions):
"""Denormalize PulseDB predictions.
Parameters
----------
dataset : object
Dataset object containing BP_mean and BP_std attributes.
predictions : np.ndarray
Normalized predictions containing mean and standard deviation values.
Returns
-------
np.ndarray
Denormalized predictions.
"""
prediction_mean = predictions[:, :2]
prediction_std = predictions[:, 2:]
prediction_mean = (
prediction_mean * dataset.BP_std.numpy() + dataset.BP_mean.numpy()
)
prediction_std = prediction_std * dataset.BP_std.numpy()
return np.concatenate([prediction_mean, prediction_std], axis=-1)
[docs]
def evaluation_function(self, target, predictions, ensemble_predictions):
"""Evaluate individual PulseDB models and ensemble predictions.
Parameters
----------
target : np.ndarray
Ground truth blood pressure targets.
predictions : np.ndarray
Predictions from individual ensemble members.
ensemble_predictions : np.ndarray
Aggregated ensemble predictions.
Returns
-------
None
The function prints the calculated metrics.
"""
model_metrics = []
SBP = target[:, 0]
DBP = target[:, 1]
for prediction in predictions:
SBP_hat = prediction[:, 0]
DBP_hat = prediction[:, 1]
model_metrics.append(
{
"SBP": qumphy.metrics.all_regression_metrics(SBP, SBP_hat),
"DBP": qumphy.metrics.all_regression_metrics(DBP, DBP_hat),
}
)
mean_metrics = reduce_nested_dict_list(model_metrics, np.mean)
max_metrics = reduce_nested_dict_list(model_metrics, max)
min_metrics = reduce_nested_dict_list(model_metrics, min)
SBP_hat = ensemble_predictions[:, 0]
DBP_hat = ensemble_predictions[:, 1]
ensemble_metrics = {
"SBP": qumphy.metrics.all_regression_metrics(SBP, SBP_hat),
"DBP": qumphy.metrics.all_regression_metrics(DBP, DBP_hat),
}
self.print_metrics(mean_metrics, max_metrics, min_metrics, ensemble_metrics)
[docs]
def print_metrics(self, mean_metrics, max_metrics, min_metrics, ensemble_metrics):
for key, value in mean_metrics.items():
print(f"Mean {key}: {value}")
for key, value in max_metrics.items():
print(f"Max {key}: {value}")
for key, value in min_metrics.items():
print(f"Min {key}: {value}")
for key, value in ensemble_metrics.items():
print(f"Ensemble {key}: {value}")
[docs]
def reduce(self, predictions):
"""GAUSSIAN MIXTURE AS IN LAKSMINARAYANAN PAPER
The predictions are given as mean and std and returned the same way.
Parameters
----------
predictions : np.ndarray
Predictions from individual ensemble members.
Returns
-------
np.ndarray
Aggregated ensemble prediction.
"""
num_models = predictions.shape[0]
BP_mean = predictions[:, :, :2]
BP_std = predictions[:, :, 2:]
ensemble_BP_mean = np.mean(BP_mean, axis=0)
# \sigma^2 = 1/N * \sum_{i=1}^N (\sigma_i^2 + \mu_i^2) - \mu^2
ensemble_BP_std = np.sqrt(
(np.sum(BP_std**2 + BP_mean**2, axis=0)) / num_models - ensemble_BP_mean**2
)
return np.concatenate([ensemble_BP_mean, ensemble_BP_std], axis=-1)
[docs]
class DeepEnsembleEvaluate:
"""Evaluation pipeline for deep ensemble predictions."""
def __init__(self, config):
"""Initialize the deep ensemble evaluation pipeline.
Parameters
----------
config : dict
Configuration dictionary containing the evaluation class, dataset,
prediction paths, and post-processing options.
"""
self.config = config
self.load_evaluation_class()
self.load_dataset()
self.load_predictions()
[docs]
def load_evaluation_class(self):
"""Load the evaluation class from the configuration.
Returns
-------
None
The function stores the evaluation class as an attribute.
"""
self.evaluation_class = qumphy.misc.misc.instantiate_class(
self.config["evaluation_class"]
)
[docs]
def load_dataset(self):
"""Load the dataset from the configuration.
Returns
-------
None
The function stores the dataset and target values as attributes.
"""
self.dataset = qumphy.misc.misc.instantiate_class(self.config["dataset"])
self.target = self.dataset.target.numpy()
[docs]
def load_predictions(self):
"""Load and process prediction files.
Returns
-------
None
The function loads predictions, optionally applies extra processing
and denormalization, and stores individual and ensemble predictions.
"""
predictions = []
for path in self.config["prediction_paths"]:
prediction = torch.load(path).numpy()
if self.config["extra_function"]:
prediction = self.evaluation_class.extra_function(prediction)
if self.config["denormalize_predictions"]:
prediction = self.evaluation_class.denormalize(self.dataset, prediction)
predictions.append(prediction)
self.predictions = np.stack(predictions, axis=0)
self.ensemble_predictions = self.evaluation_class.reduce(self.predictions)
[docs]
def calculate_metrics(self):
self.evaluation_class.evaluation_function(
self.target, self.predictions, self.ensemble_predictions
)