Diff of
/src/utils/metrics.py
[000000]
..
[66326d]
Switch to side-by-side view
--- a +++ b/src/utils/metrics.py @@ -0,0 +1,2240 @@ +""" +Custom binary prediction metrics using Avalanche +https://github.com/ContinualAI/avalanche/blob/master/notebooks/from-zero-to-hero-tutorial/05_evaluation.ipynb +""" + +from typing import List, Union, Dict +from collections import defaultdict + +import torch +import numpy as np +from torch import Tensor, arange +from avalanche.evaluation import Metric, PluginMetric, GenericPluginMetric +from avalanche.evaluation.metrics.mean import Mean +from avalanche.evaluation.metric_utils import phase_and_task + +from sklearn.metrics import average_precision_score, roc_auc_score + + +def confusion(prediction, truth): + """Returns the confusion matrix for the values in the `prediction` and `truth` + tensors, i.e. the amount of positions where the values of `prediction` + and `truth` are + - 1 and 1 (True Positive) + - 1 and 0 (False Positive) + - 0 and 0 (True Negative) + - 0 and 1 (False Negative) + + Source: https://gist.github.com/the-bass/cae9f3976866776dea17a5049013258d + """ + + confusion_vector = prediction / truth + # Element-wise division of the 2 tensors returns a new tensor which holds a + # unique value for each case: + # 1 where prediction and truth are 1 (True Positive) + # inf where prediction is 1 and truth is 0 (False Positive) + # nan where prediction and truth are 0 (True Negative) + # 0 where prediction is 0 and truth is 1 (False Negative) + + true_positives = torch.sum(confusion_vector == 1).item() + false_positives = torch.sum(confusion_vector == float("inf")).item() + true_negatives = torch.sum(torch.isnan(confusion_vector)).item() + false_negatives = torch.sum(confusion_vector == 0).item() + + return true_positives, false_positives, true_negatives, false_negatives + + +# https://github.com/ContinualAI/avalanche/blob/master/avalanche/evaluation/metrics/mean_scores.py +# Use above for AUPRC etc templates. + + +class BalancedAccuracy(Metric[float]): + """ + The BalancedAccuracy metric. This is a standalone metric. + + The metric keeps a dictionary of <task_label, balancedaccuracy value> pairs. + and update the values through a running average over multiple + <prediction, target> pairs of Tensors, provided incrementally. + The "prediction" and "target" tensors may contain plain labels or + one-hot/logit vectors. + + Each time `result` is called, this metric emits the average balancedaccuracy + across all predictions made since the last `reset`. + + The reset method will bring the metric to its initial state. By default + this metric in its initial state will return an balancedaccuracy value of 0. + """ + + def __init__(self): + """ + Creates an instance of the standalone BalancedAccuracy metric. + + By default this metric in its initial state will return an balancedaccuracy + value of 0. The metric can be updated by using the `update` method + while the running balancedaccuracy can be retrieved using the `result` method. + """ + super().__init__() + self._mean_balancedaccuracy = defaultdict(Mean) + """ + The mean utility that will be used to store the running balancedaccuracy + for each task label. + """ + + @torch.no_grad() + def update( + self, predicted_y: Tensor, true_y: Tensor, task_labels: Union[float, Tensor] + ) -> None: + """ + Update the running balancedaccuracy given the true and predicted labels. + Parameter `task_labels` is used to decide how to update the inner + dictionary: if Float, only the dictionary value related to that task + is updated. If Tensor, all the dictionary elements belonging to the + task labels will be updated. + + :param predicted_y: The model prediction. Both labels and logit vectors + are supported. + :param true_y: The ground truth. Both labels and one-hot vectors + are supported. + :param task_labels: the int task label associated to the current + experience or the task labels vector showing the task label + for each pattern. + + :return: None. + """ + if len(true_y) != len(predicted_y): + raise ValueError("Size mismatch for true_y and predicted_y tensors") + + if isinstance(task_labels, Tensor) and len(task_labels) != len(true_y): + raise ValueError("Size mismatch for true_y and task_labels tensors") + + true_y = torch.as_tensor(true_y) + predicted_y = torch.as_tensor(predicted_y) + + # Check if logits or labels + if len(predicted_y.shape) > 1: + # Logits -> transform to labels + predicted_y = torch.max(predicted_y, 1)[1] + + if len(true_y.shape) > 1: + # Logits -> transform to labels + true_y = torch.max(true_y, 1)[1] + + if isinstance(task_labels, int): + ( + true_positives, + false_positives, + true_negatives, + false_negatives, + ) = confusion(predicted_y, true_y) + + try: + tpr = true_positives / (true_positives + false_negatives) + except ZeroDivisionError: + tpr = 1 + + try: + tnr = true_negatives / (true_negatives + false_positives) + except ZeroDivisionError: + tnr = 1 + + self._mean_balancedaccuracy[task_labels].update( + (tpr + tnr) / 2, len(predicted_y) + ) + elif isinstance(task_labels, Tensor): + raise NotImplementedError + else: + raise ValueError( + f"Task label type: {type(task_labels)}, expected int/float or Tensor" + ) + + def result(self, task_label=None) -> Dict[int, float]: + """ + Retrieves the running balancedaccuracy. + + Calling this method will not change the internal state of the metric. + + :param task_label: if None, return the entire dictionary of balanced accuracies + for each task. Otherwise return the dictionary + `{task_label: balancedaccuracy}`. + :return: A dict of running balanced accuracies for each task label, + where each value is a float value between 0 and 1. + """ + assert task_label is None or isinstance(task_label, int) + if task_label is None: + return {k: v.result() for k, v in self._mean_balancedaccuracy.items()} + else: + return {task_label: self._mean_balancedaccuracy[task_label].result()} + + def reset(self, task_label=None) -> None: + """ + Resets the metric. + :param task_label: if None, reset the entire dictionary. + Otherwise, reset the value associated to `task_label`. + + :return: None. + """ + assert task_label is None or isinstance(task_label, int) + if task_label is None: + self._mean_balancedaccuracy = defaultdict(Mean) + else: + self._mean_balancedaccuracy[task_label].reset() + + +class BalancedAccuracyPluginMetric(GenericPluginMetric[float]): + """ + Base class for all balanced accuracies plugin metrics + """ + + def __init__(self, reset_at, emit_at, mode): + self._balancedaccuracy = BalancedAccuracy() + super(BalancedAccuracyPluginMetric, self).__init__( + self._balancedaccuracy, reset_at=reset_at, emit_at=emit_at, mode=mode + ) + + def reset(self, strategy=None) -> None: + if self._reset_at == "stream" or strategy is None: + self._metric.reset() + else: + self._metric.reset(phase_and_task(strategy)[1]) + + def result(self, strategy=None) -> float: + if self._emit_at == "stream" or strategy is None: + return self._metric.result() + else: + return self._metric.result(phase_and_task(strategy)[1]) + + def update(self, strategy): + # task labels defined for each experience + task_labels = strategy.experience.task_labels + if len(task_labels) > 1: + # task labels defined for each pattern + task_labels = strategy.mb_task_id + else: + task_labels = task_labels[0] + self._balancedaccuracy.update(strategy.mb_output, strategy.mb_y, task_labels) + + +class MinibatchBalancedAccuracy(BalancedAccuracyPluginMetric): + """ + The minibatch plugin balancedaccuracy metric. + This metric only works at training time. + + This metric computes the average balancedaccuracy over patterns + from a single minibatch. + It reports the result after each iteration. + + If a more coarse-grained logging is needed, consider using + :class:`EpochBalancedAccuracy` instead. + """ + + def __init__(self): + """ + Creates an instance of the MinibatchBalancedAccuracy metric. + """ + super(MinibatchBalancedAccuracy, self).__init__( + reset_at="iteration", emit_at="iteration", mode="train" + ) + + def __str__(self): + return "BalAcc_MB" + + +class EpochBalancedAccuracy(BalancedAccuracyPluginMetric): + """ + The average balancedaccuracy over a single training epoch. + This plugin metric only works at training time. + + The balancedaccuracy will be logged after each training epoch by computing + the number of correctly predicted patterns during the epoch divided by + the overall number of patterns encountered in that epoch. + """ + + def __init__(self): + """ + Creates an instance of the EpochBalancedAccuracy metric. + """ + + super(EpochBalancedAccuracy, self).__init__( + reset_at="epoch", emit_at="epoch", mode="train" + ) + + def __str__(self): + return "BalAcc_Epoch" + + +class RunningEpochBalancedAccuracy(BalancedAccuracyPluginMetric): + """ + The average balancedaccuracy across all minibatches up to the current + epoch iteration. + This plugin metric only works at training time. + + At each iteration, this metric logs the balancedaccuracy averaged over all patterns + seen so far in the current epoch. + The metric resets its state after each training epoch. + """ + + def __init__(self): + """ + Creates an instance of the RunningEpochBalancedAccuracy metric. + """ + + super(RunningEpochBalancedAccuracy, self).__init__( + reset_at="epoch", emit_at="iteration", mode="train" + ) + + def __str__(self): + return "RunningBalAcc_Epoch" + + +class ExperienceBalancedAccuracy(BalancedAccuracyPluginMetric): + """ + At the end of each experience, this plugin metric reports + the average balancedaccuracy over all patterns seen in that experience. + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of ExperienceBalancedAccuracy metric + """ + super(ExperienceBalancedAccuracy, self).__init__( + reset_at="experience", emit_at="experience", mode="eval" + ) + + def __str__(self): + return "BalAcc_Exp" + + +class StreamBalancedAccuracy(BalancedAccuracyPluginMetric): + """ + At the end of the entire stream of experiences, this plugin metric + reports the average balancedaccuracy over all patterns seen in all experiences. + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of StreamBalancedAccuracy metric + """ + super(StreamBalancedAccuracy, self).__init__( + reset_at="stream", emit_at="stream", mode="eval" + ) + + def __str__(self): + return "BalAcc_Stream" + + +class TrainedExperienceBalancedAccuracy(BalancedAccuracyPluginMetric): + """ + At the end of each experience, this plugin metric reports the average + balancedaccuracy for only the experiences that the model has been trained on so far. + + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of TrainedExperienceBalancedAccuracy metric by first + constructing BalancedAccuracyPluginMetric + """ + super(TrainedExperienceBalancedAccuracy, self).__init__( + reset_at="stream", emit_at="stream", mode="eval" + ) + self._current_experience = 0 + + def after_training_exp(self, strategy) -> None: + self._current_experience = strategy.experience.current_experience + # Reset average after learning from a new experience + BalancedAccuracyPluginMetric.reset(self, strategy) + return BalancedAccuracyPluginMetric.after_training_exp(self, strategy) + + def update(self, strategy): + """ + Only update the balancedaccuracy with results from experiences that have been + trained on + """ + if strategy.experience.current_experience <= self._current_experience: + BalancedAccuracyPluginMetric.update(self, strategy) + + def __str__(self): + return "BalancedAccuracy_On_Trained_Experiences" + + +def balancedaccuracy_metrics( + *, + minibatch=False, + epoch=False, + epoch_running=False, + experience=False, + stream=False, + trained_experience=False, +) -> List[PluginMetric]: + """ + Helper method that can be used to obtain the desired set of + plugin metrics. + + :param minibatch: If True, will return a metric able to log + the minibatch balancedaccuracy at training time. + :param epoch: If True, will return a metric able to log + the epoch balancedaccuracy at training time. + :param epoch_running: If True, will return a metric able to log + the running epoch balancedaccuracy at training time. + :param experience: If True, will return a metric able to log + the balancedaccuracy on each evaluation experience. + :param stream: If True, will return a metric able to log + the balancedaccuracy averaged over the entire evaluation stream of experiences. + :param trained_experience: If True, will return a metric able to log + the average evaluation balancedaccuracy only for experiences that the + model has been trained on + + :return: A list of plugin metrics. + """ + + metrics = [] + if minibatch: + metrics.append(MinibatchBalancedAccuracy()) + + if epoch: + metrics.append(EpochBalancedAccuracy()) + + if epoch_running: + metrics.append(RunningEpochBalancedAccuracy()) + + if experience: + metrics.append(ExperienceBalancedAccuracy()) + + if stream: + metrics.append(StreamBalancedAccuracy()) + + if trained_experience: + metrics.append(TrainedExperienceBalancedAccuracy()) + + return metrics + + +class Sensitivity(Metric[float]): + """ + The Sensitivity metric. This is a standalone metric. + + The metric keeps a dictionary of <task_label, Sensitivity value> pairs. + and update the values through a running average over multiple + <prediction, target> pairs of Tensors, provided incrementally. + The "prediction" and "target" tensors may contain plain labels or + one-hot/logit vectors. + + Each time `result` is called, this metric emits the average Sensitivity + across all predictions made since the last `reset`. + + The reset method will bring the metric to its initial state. By default + this metric in its initial state will return an Sensitivity value of 0. + """ + + def __init__(self): + """ + Creates an instance of the standalone Sensitivity metric. + + By default this metric in its initial state will return an Sensitivity + value of 0. The metric can be updated by using the `update` method + while the running Sensitivity can be retrieved using the `result` method. + """ + super().__init__() + self._mean_Sensitivity = defaultdict(Mean) + """ + The mean utility that will be used to store the running Sensitivity + for each task label. + """ + + @torch.no_grad() + def update( + self, predicted_y: Tensor, true_y: Tensor, task_labels: Union[float, Tensor] + ) -> None: + """ + Update the running Sensitivity given the true and predicted labels. + Parameter `task_labels` is used to decide how to update the inner + dictionary: if Float, only the dictionary value related to that task + is updated. If Tensor, all the dictionary elements belonging to the + task labels will be updated. + + :param predicted_y: The model prediction. Both labels and logit vectors + are supported. + :param true_y: The ground truth. Both labels and one-hot vectors + are supported. + :param task_labels: the int task label associated to the current + experience or the task labels vector showing the task label + for each pattern. + + :return: None. + """ + if len(true_y) != len(predicted_y): + raise ValueError("Size mismatch for true_y and predicted_y tensors") + + if isinstance(task_labels, Tensor) and len(task_labels) != len(true_y): + raise ValueError("Size mismatch for true_y and task_labels tensors") + + true_y = torch.as_tensor(true_y) + predicted_y = torch.as_tensor(predicted_y) + + # Check if logits or labels + if len(predicted_y.shape) > 1: + # Logits -> transform to labels + predicted_y = torch.max(predicted_y, 1)[1] + + if len(true_y.shape) > 1: + # Logits -> transform to labels + true_y = torch.max(true_y, 1)[1] + + if isinstance(task_labels, int): + ( + true_positives, + false_positives, + true_negatives, + false_negatives, + ) = confusion(predicted_y, true_y) + + try: + tpr = true_positives / (true_positives + false_negatives) + except ZeroDivisionError: + tpr = 1 + + self._mean_Sensitivity[task_labels].update(tpr, len(predicted_y)) + elif isinstance(task_labels, Tensor): + raise NotImplementedError + else: + raise ValueError( + f"Task label type: {type(task_labels)}, expected int/float or Tensor" + ) + + def result(self, task_label=None) -> Dict[int, float]: + """ + Retrieves the running Sensitivity. + + Calling this method will not change the internal state of the metric. + + :param task_label: if None, return the entire dictionary of sensitivities + for each task. Otherwise return the dictionary + `{task_label: Sensitivity}`. + :return: A dict of running sensitivities for each task label, + where each value is a float value between 0 and 1. + """ + assert task_label is None or isinstance(task_label, int) + if task_label is None: + return {k: v.result() for k, v in self._mean_Sensitivity.items()} + else: + return {task_label: self._mean_Sensitivity[task_label].result()} + + def reset(self, task_label=None) -> None: + """ + Resets the metric. + :param task_label: if None, reset the entire dictionary. + Otherwise, reset the value associated to `task_label`. + + :return: None. + """ + assert task_label is None or isinstance(task_label, int) + if task_label is None: + self._mean_Sensitivity = defaultdict(Mean) + else: + self._mean_Sensitivity[task_label].reset() + + +class SensitivityPluginMetric(GenericPluginMetric[float]): + """ + Base class for all sensitivities plugin metrics + """ + + def __init__(self, reset_at, emit_at, mode): + self._Sensitivity = Sensitivity() + super(SensitivityPluginMetric, self).__init__( + self._Sensitivity, reset_at=reset_at, emit_at=emit_at, mode=mode + ) + + def reset(self, strategy=None) -> None: + if self._reset_at == "stream" or strategy is None: + self._metric.reset() + else: + self._metric.reset(phase_and_task(strategy)[1]) + + def result(self, strategy=None) -> float: + if self._emit_at == "stream" or strategy is None: + return self._metric.result() + else: + return self._metric.result(phase_and_task(strategy)[1]) + + def update(self, strategy): + # task labels defined for each experience + task_labels = strategy.experience.task_labels + if len(task_labels) > 1: + # task labels defined for each pattern + task_labels = strategy.mb_task_id + else: + task_labels = task_labels[0] + self._Sensitivity.update(strategy.mb_output, strategy.mb_y, task_labels) + + +class MinibatchSensitivity(SensitivityPluginMetric): + """ + The minibatch plugin Sensitivity metric. + This metric only works at training time. + + This metric computes the average Sensitivity over patterns + from a single minibatch. + It reports the result after each iteration. + + If a more coarse-grained logging is needed, consider using + :class:`EpochSensitivity` instead. + """ + + def __init__(self): + """ + Creates an instance of the MinibatchSensitivity metric. + """ + super(MinibatchSensitivity, self).__init__( + reset_at="iteration", emit_at="iteration", mode="train" + ) + + def __str__(self): + return "Sens_MB" + + +class EpochSensitivity(SensitivityPluginMetric): + """ + The average Sensitivity over a single training epoch. + This plugin metric only works at training time. + + The Sensitivity will be logged after each training epoch by computing + the number of correctly predicted patterns during the epoch divided by + the overall number of patterns encountered in that epoch. + """ + + def __init__(self): + """ + Creates an instance of the EpochSensitivity metric. + """ + + super(EpochSensitivity, self).__init__( + reset_at="epoch", emit_at="epoch", mode="train" + ) + + def __str__(self): + return "Sens_Epoch" + + +class RunningEpochSensitivity(SensitivityPluginMetric): + """ + The average Sensitivity across all minibatches up to the current + epoch iteration. + This plugin metric only works at training time. + + At each iteration, this metric logs the Sensitivity averaged over all patterns + seen so far in the current epoch. + The metric resets its state after each training epoch. + """ + + def __init__(self): + """ + Creates an instance of the RunningEpochSensitivity metric. + """ + + super(RunningEpochSensitivity, self).__init__( + reset_at="epoch", emit_at="iteration", mode="train" + ) + + def __str__(self): + return "RunningSens_Epoch" + + +class ExperienceSensitivity(SensitivityPluginMetric): + """ + At the end of each experience, this plugin metric reports + the average Sensitivity over all patterns seen in that experience. + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of ExperienceSensitivity metric + """ + super(ExperienceSensitivity, self).__init__( + reset_at="experience", emit_at="experience", mode="eval" + ) + + def __str__(self): + return "Sens_Exp" + + +class StreamSensitivity(SensitivityPluginMetric): + """ + At the end of the entire stream of experiences, this plugin metric + reports the average Sensitivity over all patterns seen in all experiences. + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of StreamSensitivity metric + """ + super(StreamSensitivity, self).__init__( + reset_at="stream", emit_at="stream", mode="eval" + ) + + def __str__(self): + return "Sens_Stream" + + +class TrainedExperienceSensitivity(SensitivityPluginMetric): + """ + At the end of each experience, this plugin metric reports the average + Sensitivity for only the experiences that the model has been trained on so far. + + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of TrainedExperienceSensitivity metric by first + constructing SensitivityPluginMetric + """ + super(TrainedExperienceSensitivity, self).__init__( + reset_at="stream", emit_at="stream", mode="eval" + ) + self._current_experience = 0 + + def after_training_exp(self, strategy) -> None: + self._current_experience = strategy.experience.current_experience + # Reset average after learning from a new experience + SensitivityPluginMetric.reset(self, strategy) + return SensitivityPluginMetric.after_training_exp(self, strategy) + + def update(self, strategy): + """ + Only update the Sensitivity with results from experiences that have been + trained on + """ + if strategy.experience.current_experience <= self._current_experience: + SensitivityPluginMetric.update(self, strategy) + + def __str__(self): + return "Sensitivity_On_Trained_Experiences" + + +def sensitivity_metrics( + *, + minibatch=False, + epoch=False, + epoch_running=False, + experience=False, + stream=False, + trained_experience=False, +) -> List[PluginMetric]: + """ + Helper method that can be used to obtain the desired set of + plugin metrics. + + :param minibatch: If True, will return a metric able to log + the minibatch Sensitivity at training time. + :param epoch: If True, will return a metric able to log + the epoch Sensitivity at training time. + :param epoch_running: If True, will return a metric able to log + the running epoch Sensitivity at training time. + :param experience: If True, will return a metric able to log + the Sensitivity on each evaluation experience. + :param stream: If True, will return a metric able to log + the Sensitivity averaged over the entire evaluation stream of experiences. + :param trained_experience: If True, will return a metric able to log + the average evaluation Sensitivity only for experiences that the + model has been trained on + + :return: A list of plugin metrics. + """ + + metrics = [] + if minibatch: + metrics.append(MinibatchSensitivity()) + + if epoch: + metrics.append(EpochSensitivity()) + + if epoch_running: + metrics.append(RunningEpochSensitivity()) + + if experience: + metrics.append(ExperienceSensitivity()) + + if stream: + metrics.append(StreamSensitivity()) + + if trained_experience: + metrics.append(TrainedExperienceSensitivity()) + + return metrics + + +class Specificity(Metric[float]): + """ + The Specificity metric. This is a standalone metric. + + The metric keeps a dictionary of <task_label, Specificity value> pairs. + and update the values through a running average over multiple + <prediction, target> pairs of Tensors, provided incrementally. + The "prediction" and "target" tensors may contain plain labels or + one-hot/logit vectors. + + Each time `result` is called, this metric emits the average Specificity + across all predictions made since the last `reset`. + + The reset method will bring the metric to its initial state. By default + this metric in its initial state will return an Specificity value of 0. + """ + + def __init__(self): + """ + Creates an instance of the standalone Specificity metric. + + By default this metric in its initial state will return an Specificity + value of 0. The metric can be updated by using the `update` method + while the running Specificity can be retrieved using the `result` method. + """ + super().__init__() + self._mean_Specificity = defaultdict(Mean) + """ + The mean utility that will be used to store the running Specificity + for each task label. + """ + + @torch.no_grad() + def update( + self, predicted_y: Tensor, true_y: Tensor, task_labels: Union[float, Tensor] + ) -> None: + """ + Update the running Specificity given the true and predicted labels. + Parameter `task_labels` is used to decide how to update the inner + dictionary: if Float, only the dictionary value related to that task + is updated. If Tensor, all the dictionary elements belonging to the + task labels will be updated. + + :param predicted_y: The model prediction. Both labels and logit vectors + are supported. + :param true_y: The ground truth. Both labels and one-hot vectors + are supported. + :param task_labels: the int task label associated to the current + experience or the task labels vector showing the task label + for each pattern. + + :return: None. + """ + if len(true_y) != len(predicted_y): + raise ValueError("Size mismatch for true_y and predicted_y tensors") + + if isinstance(task_labels, Tensor) and len(task_labels) != len(true_y): + raise ValueError("Size mismatch for true_y and task_labels tensors") + + true_y = torch.as_tensor(true_y) + predicted_y = torch.as_tensor(predicted_y) + + # Check if logits or labels + if len(predicted_y.shape) > 1: + # Logits -> transform to labels + predicted_y = torch.max(predicted_y, 1)[1] + + if len(true_y.shape) > 1: + # Logits -> transform to labels + true_y = torch.max(true_y, 1)[1] + + if isinstance(task_labels, int): + ( + true_positives, + false_positives, + true_negatives, + false_negatives, + ) = confusion(predicted_y, true_y) + + try: + tnr = true_negatives / (true_negatives + false_positives) + except ZeroDivisionError: + tnr = 1 + + self._mean_Specificity[task_labels].update(tnr, len(predicted_y)) + elif isinstance(task_labels, Tensor): + raise NotImplementedError + else: + raise ValueError( + f"Task label type: {type(task_labels)}, expected int/float or Tensor" + ) + + def result(self, task_label=None) -> Dict[int, float]: + """ + Retrieves the running Specificity. + + Calling this method will not change the internal state of the metric. + + :param task_label: if None, return the entire dictionary of specificities + for each task. Otherwise return the dictionary + `{task_label: Specificity}`. + :return: A dict of running specificities for each task label, + where each value is a float value between 0 and 1. + """ + assert task_label is None or isinstance(task_label, int) + if task_label is None: + return {k: v.result() for k, v in self._mean_Specificity.items()} + else: + return {task_label: self._mean_Specificity[task_label].result()} + + def reset(self, task_label=None) -> None: + """ + Resets the metric. + :param task_label: if None, reset the entire dictionary. + Otherwise, reset the value associated to `task_label`. + + :return: None. + """ + assert task_label is None or isinstance(task_label, int) + if task_label is None: + self._mean_Specificity = defaultdict(Mean) + else: + self._mean_Specificity[task_label].reset() + + +class SpecificityPluginMetric(GenericPluginMetric[float]): + """ + Base class for all specificities plugin metrics + """ + + def __init__(self, reset_at, emit_at, mode): + self._Specificity = Specificity() + super(SpecificityPluginMetric, self).__init__( + self._Specificity, reset_at=reset_at, emit_at=emit_at, mode=mode + ) + + def reset(self, strategy=None) -> None: + if self._reset_at == "stream" or strategy is None: + self._metric.reset() + else: + self._metric.reset(phase_and_task(strategy)[1]) + + def result(self, strategy=None) -> float: + if self._emit_at == "stream" or strategy is None: + return self._metric.result() + else: + return self._metric.result(phase_and_task(strategy)[1]) + + def update(self, strategy): + # task labels defined for each experience + task_labels = strategy.experience.task_labels + if len(task_labels) > 1: + # task labels defined for each pattern + task_labels = strategy.mb_task_id + else: + task_labels = task_labels[0] + self._Specificity.update(strategy.mb_output, strategy.mb_y, task_labels) + + +class MinibatchSpecificity(SpecificityPluginMetric): + """ + The minibatch plugin Specificity metric. + This metric only works at training time. + + This metric computes the average Specificity over patterns + from a single minibatch. + It reports the result after each iteration. + + If a more coarse-grained logging is needed, consider using + :class:`EpochSpecificity` instead. + """ + + def __init__(self): + """ + Creates an instance of the MinibatchSpecificity metric. + """ + super(MinibatchSpecificity, self).__init__( + reset_at="iteration", emit_at="iteration", mode="train" + ) + + def __str__(self): + return "Spec_MB" + + +class EpochSpecificity(SpecificityPluginMetric): + """ + The average Specificity over a single training epoch. + This plugin metric only works at training time. + + The Specificity will be logged after each training epoch by computing + the number of correctly predicted patterns during the epoch divided by + the overall number of patterns encountered in that epoch. + """ + + def __init__(self): + """ + Creates an instance of the EpochSpecificity metric. + """ + + super(EpochSpecificity, self).__init__( + reset_at="epoch", emit_at="epoch", mode="train" + ) + + def __str__(self): + return "Spec_Epoch" + + +class RunningEpochSpecificity(SpecificityPluginMetric): + """ + The average Specificity across all minibatches up to the current + epoch iteration. + This plugin metric only works at training time. + + At each iteration, this metric logs the Specificity averaged over all patterns + seen so far in the current epoch. + The metric resets its state after each training epoch. + """ + + def __init__(self): + """ + Creates an instance of the RunningEpochSpecificity metric. + """ + + super(RunningEpochSpecificity, self).__init__( + reset_at="epoch", emit_at="iteration", mode="train" + ) + + def __str__(self): + return "RunningSpec_Epoch" + + +class ExperienceSpecificity(SpecificityPluginMetric): + """ + At the end of each experience, this plugin metric reports + the average Specificity over all patterns seen in that experience. + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of ExperienceSpecificity metric + """ + super(ExperienceSpecificity, self).__init__( + reset_at="experience", emit_at="experience", mode="eval" + ) + + def __str__(self): + return "Spec_Exp" + + +class StreamSpecificity(SpecificityPluginMetric): + """ + At the end of the entire stream of experiences, this plugin metric + reports the average Specificity over all patterns seen in all experiences. + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of StreamSpecificity metric + """ + super(StreamSpecificity, self).__init__( + reset_at="stream", emit_at="stream", mode="eval" + ) + + def __str__(self): + return "Spec_Stream" + + +class TrainedExperienceSpecificity(SpecificityPluginMetric): + """ + At the end of each experience, this plugin metric reports the average + Specificity for only the experiences that the model has been trained on so far. + + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of TrainedExperienceSpecificity metric by first + constructing SpecificityPluginMetric + """ + super(TrainedExperienceSpecificity, self).__init__( + reset_at="stream", emit_at="stream", mode="eval" + ) + self._current_experience = 0 + + def after_training_exp(self, strategy) -> None: + self._current_experience = strategy.experience.current_experience + # Reset average after learning from a new experience + SpecificityPluginMetric.reset(self, strategy) + return SpecificityPluginMetric.after_training_exp(self, strategy) + + def update(self, strategy): + """ + Only update the Specificity with results from experiences that have been + trained on + """ + if strategy.experience.current_experience <= self._current_experience: + SpecificityPluginMetric.update(self, strategy) + + def __str__(self): + return "Specificity_On_Trained_Experiences" + + +def specificity_metrics( + *, + minibatch=False, + epoch=False, + epoch_running=False, + experience=False, + stream=False, + trained_experience=False, +) -> List[PluginMetric]: + """ + Helper method that can be used to obtain the desired set of + plugin metrics. + + :param minibatch: If True, will return a metric able to log + the minibatch Specificity at training time. + :param epoch: If True, will return a metric able to log + the epoch Specificity at training time. + :param epoch_running: If True, will return a metric able to log + the running epoch Specificity at training time. + :param experience: If True, will return a metric able to log + the Specificity on each evaluation experience. + :param stream: If True, will return a metric able to log + the Specificity averaged over the entire evaluation stream of experiences. + :param trained_experience: If True, will return a metric able to log + the average evaluation Specificity only for experiences that the + model has been trained on + + :return: A list of plugin metrics. + """ + + metrics = [] + if minibatch: + metrics.append(MinibatchSpecificity()) + + if epoch: + metrics.append(EpochSpecificity()) + + if epoch_running: + metrics.append(RunningEpochSpecificity()) + + if experience: + metrics.append(ExperienceSpecificity()) + + if stream: + metrics.append(StreamSpecificity()) + + if trained_experience: + metrics.append(TrainedExperienceSpecificity()) + + return metrics + + +class Precision(Metric[float]): + """ + The Precision metric. This is a standalone metric. + + The metric keeps a dictionary of <task_label, Precision value> pairs. + and update the values through a running average over multiple + <prediction, target> pairs of Tensors, provided incrementally. + The "prediction" and "target" tensors may contain plain labels or + one-hot/logit vectors. + + Each time `result` is called, this metric emits the average Precision + across all predictions made since the last `reset`. + + The reset method will bring the metric to its initial state. By default + this metric in its initial state will return an Precision value of 0. + """ + + def __init__(self): + """ + Creates an instance of the standalone Precision metric. + + By default this metric in its initial state will return a Precision + value of 0. The metric can be updated by using the `update` method + while the running Precision can be retrieved using the `result` method. + """ + super().__init__() + self._mean_Precision = defaultdict(Mean) + """ + The mean utility that will be used to store the running Precision + for each task label. + """ + + @torch.no_grad() + def update( + self, predicted_y: Tensor, true_y: Tensor, task_labels: Union[float, Tensor] + ) -> None: + """ + Update the running Precision given the true and predicted labels. + Parameter `task_labels` is used to decide how to update the inner + dictionary: if Float, only the dictionary value related to that task + is updated. If Tensor, all the dictionary elements belonging to the + task labels will be updated. + + :param predicted_y: The model prediction. Both labels and logit vectors + are supported. + :param true_y: The ground truth. Both labels and one-hot vectors + are supported. + :param task_labels: the int task label associated to the current + experience or the task labels vector showing the task label + for each pattern. + + :return: None. + """ + if len(true_y) != len(predicted_y): + raise ValueError("Size mismatch for true_y and predicted_y tensors") + + if isinstance(task_labels, Tensor) and len(task_labels) != len(true_y): + raise ValueError("Size mismatch for true_y and task_labels tensors") + + true_y = torch.as_tensor(true_y) + predicted_y = torch.as_tensor(predicted_y) + + # Check if logits or labels + if len(predicted_y.shape) > 1: + # Logits -> transform to labels + predicted_y = torch.max(predicted_y, 1)[1] + + if len(true_y.shape) > 1: + # Logits -> transform to labels + true_y = torch.max(true_y, 1)[1] + + if isinstance(task_labels, int): + ( + true_positives, + false_positives, + true_negatives, + false_negatives, + ) = confusion(predicted_y, true_y) + + try: + ppv = true_positives / (true_positives + false_positives) + except ZeroDivisionError: + ppv = 1 + + self._mean_Precision[task_labels].update(ppv, len(predicted_y)) + elif isinstance(task_labels, Tensor): + raise NotImplementedError + else: + raise ValueError( + f"Task label type: {type(task_labels)}, expected int/float or Tensor" + ) + + def result(self, task_label=None) -> Dict[int, float]: + """ + Retrieves the running Precision. + + Calling this method will not change the internal state of the metric. + + :param task_label: if None, return the entire dictionary of precisions + for each task. Otherwise return the dictionary + `{task_label: Precision}`. + :return: A dict of running precisions for each task label, + where each value is a float value between 0 and 1. + """ + assert task_label is None or isinstance(task_label, int) + if task_label is None: + return {k: v.result() for k, v in self._mean_Precision.items()} + else: + return {task_label: self._mean_Precision[task_label].result()} + + def reset(self, task_label=None) -> None: + """ + Resets the metric. + :param task_label: if None, reset the entire dictionary. + Otherwise, reset the value associated to `task_label`. + + :return: None. + """ + assert task_label is None or isinstance(task_label, int) + if task_label is None: + self._mean_Precision = defaultdict(Mean) + else: + self._mean_Precision[task_label].reset() + + +class PrecisionPluginMetric(GenericPluginMetric[float]): + """ + Base class for all precisions plugin metrics + """ + + def __init__(self, reset_at, emit_at, mode): + self._Precision = Precision() + super(PrecisionPluginMetric, self).__init__( + self._Precision, reset_at=reset_at, emit_at=emit_at, mode=mode + ) + + def reset(self, strategy=None) -> None: + if self._reset_at == "stream" or strategy is None: + self._metric.reset() + else: + self._metric.reset(phase_and_task(strategy)[1]) + + def result(self, strategy=None) -> float: + if self._emit_at == "stream" or strategy is None: + return self._metric.result() + else: + return self._metric.result(phase_and_task(strategy)[1]) + + def update(self, strategy): + # task labels defined for each experience + task_labels = strategy.experience.task_labels + if len(task_labels) > 1: + # task labels defined for each pattern + task_labels = strategy.mb_task_id + else: + task_labels = task_labels[0] + self._Precision.update(strategy.mb_output, strategy.mb_y, task_labels) + + +class MinibatchPrecision(PrecisionPluginMetric): + """ + The minibatch plugin Precision metric. + This metric only works at training time. + + This metric computes the average Precision over patterns + from a single minibatch. + It reports the result after each iteration. + + If a more coarse-grained logging is needed, consider using + :class:`EpochPrecision` instead. + """ + + def __init__(self): + """ + Creates an instance of the MinibatchPrecision metric. + """ + super(MinibatchPrecision, self).__init__( + reset_at="iteration", emit_at="iteration", mode="train" + ) + + def __str__(self): + return "Prec_MB" + + +class EpochPrecision(PrecisionPluginMetric): + """ + The average Precision over a single training epoch. + This plugin metric only works at training time. + + The Precision will be logged after each training epoch by computing + the number of correctly predicted patterns during the epoch divided by + the overall number of patterns encountered in that epoch. + """ + + def __init__(self): + """ + Creates an instance of the EpochPrecision metric. + """ + + super(EpochPrecision, self).__init__( + reset_at="epoch", emit_at="epoch", mode="train" + ) + + def __str__(self): + return "Prec_Epoch" + + +class RunningEpochPrecision(PrecisionPluginMetric): + """ + The average Precision across all minibatches up to the current + epoch iteration. + This plugin metric only works at training time. + + At each iteration, this metric logs the Precision averaged over all patterns + seen so far in the current epoch. + The metric resets its state after each training epoch. + """ + + def __init__(self): + """ + Creates an instance of the RunningEpochPrecision metric. + """ + + super(RunningEpochPrecision, self).__init__( + reset_at="epoch", emit_at="iteration", mode="train" + ) + + def __str__(self): + return "RunningPrec_Epoch" + + +class ExperiencePrecision(PrecisionPluginMetric): + """ + At the end of each experience, this plugin metric reports + the average Precision over all patterns seen in that experience. + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of ExperiencePrecision metric + """ + super(ExperiencePrecision, self).__init__( + reset_at="experience", emit_at="experience", mode="eval" + ) + + def __str__(self): + return "Prec_Exp" + + +class StreamPrecision(PrecisionPluginMetric): + """ + At the end of the entire stream of experiences, this plugin metric + reports the average Precision over all patterns seen in all experiences. + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of StreamPrecision metric + """ + super(StreamPrecision, self).__init__( + reset_at="stream", emit_at="stream", mode="eval" + ) + + def __str__(self): + return "Prec_Stream" + + +class TrainedExperiencePrecision(PrecisionPluginMetric): + """ + At the end of each experience, this plugin metric reports the average + Precision for only the experiences that the model has been trained on so far. + + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of TrainedExperiencePrecision metric by first + constructing PrecisionPluginMetric + """ + super(TrainedExperiencePrecision, self).__init__( + reset_at="stream", emit_at="stream", mode="eval" + ) + self._current_experience = 0 + + def after_training_exp(self, strategy) -> None: + self._current_experience = strategy.experience.current_experience + # Reset average after learning from a new experience + PrecisionPluginMetric.reset(self, strategy) + return PrecisionPluginMetric.after_training_exp(self, strategy) + + def update(self, strategy): + """ + Only update the Precision with results from experiences that have been + trained on + """ + if strategy.experience.current_experience <= self._current_experience: + PrecisionPluginMetric.update(self, strategy) + + def __str__(self): + return "Precision_On_Trained_Experiences" + + +def precision_metrics( + *, + minibatch=False, + epoch=False, + epoch_running=False, + experience=False, + stream=False, + trained_experience=False, +) -> List[PluginMetric]: + """ + Helper method that can be used to obtain the desired set of + plugin metrics. + + :param minibatch: If True, will return a metric able to log + the minibatch Precision at training time. + :param epoch: If True, will return a metric able to log + the epoch Precision at training time. + :param epoch_running: If True, will return a metric able to log + the running epoch Precision at training time. + :param experience: If True, will return a metric able to log + the Precision on each evaluation experience. + :param stream: If True, will return a metric able to log + the Precision averaged over the entire evaluation stream of experiences. + :param trained_experience: If True, will return a metric able to log + the average evaluation Precision only for experiences that the + model has been trained on + + :return: A list of plugin metrics. + """ + + metrics = [] + if minibatch: + metrics.append(MinibatchPrecision()) + + if epoch: + metrics.append(EpochPrecision()) + + if epoch_running: + metrics.append(RunningEpochPrecision()) + + if experience: + metrics.append(ExperiencePrecision()) + + if stream: + metrics.append(StreamPrecision()) + + if trained_experience: + metrics.append(TrainedExperiencePrecision()) + + return metrics + + +class AUPRC(Metric[float]): + """ + The AUPRC metric. This is a standalone metric. + + The metric keeps a dictionary of <task_label, AUPRC value> pairs. + and update the values through a running average over multiple + <prediction, target> pairs of Tensors, provided incrementally. + The "prediction" and "target" tensors may contain plain labels or + one-hot/logit vectors. + + Each time `result` is called, this metric emits the average AUPRC + across all predictions made since the last `reset`. + + The reset method will bring the metric to its initial state. By default + this metric in its initial state will return an AUPRC value of 0. + """ + + def __init__(self): + """ + Creates an instance of the standalone AUPRC metric. + + By default this metric in its initial state will return a AUPRC + value of 0. The metric can be updated by using the `update` method + while the running AUPRC can be retrieved using the `result` method. + """ + super().__init__() + self._mean_AUPRC = defaultdict(Mean) + """ + The mean utility that will be used to store the running AUPRC + for each task label. + """ + + @torch.no_grad() + def update( + self, predicted_y: Tensor, true_y: Tensor, task_labels: Union[float, Tensor] + ) -> None: + """ + Update the running AUPRC given the true and predicted labels. + Parameter `task_labels` is used to decide how to update the inner + dictionary: if Float, only the dictionary value related to that task + is updated. If Tensor, all the dictionary elements belonging to the + task labels will be updated. + + :param predicted_y: The model prediction. Both labels and logit vectors + are supported. + :param true_y: The ground truth. Both labels and one-hot vectors + are supported. + :param task_labels: the int task label associated to the current + experience or the task labels vector showing the task label + for each pattern. + + :return: None. + """ + if len(true_y) != len(predicted_y): + raise ValueError("Size mismatch for true_y and predicted_y tensors") + + if isinstance(task_labels, Tensor) and len(task_labels) != len(true_y): + raise ValueError("Size mismatch for true_y and task_labels tensors") + + true_y = torch.as_tensor(true_y) + predicted_y = torch.as_tensor(predicted_y) + + assert len(predicted_y.size()) == 2, ( + "Predictions need to be logits or scores, not labels" + ) + + if len(true_y.shape) > 1: + # Logits -> transform to labels + true_y = torch.max(true_y, 1)[1] + + scores = predicted_y[arange(len(true_y)), true_y] + + with np.errstate(divide="ignore", invalid="ignore"): + average_precision_score_val = average_precision_score( + true_y.cpu(), scores.cpu() + ) + + if np.isnan(average_precision_score_val): + average_precision_score_val = 0 + + if isinstance(task_labels, int): + self._mean_AUPRC[task_labels].update( + average_precision_score_val, len(predicted_y) + ) + elif isinstance(task_labels, Tensor): + raise NotImplementedError + else: + raise ValueError( + f"Task label type: {type(task_labels)}, expected int/float or Tensor" + ) + + def result(self, task_label=None) -> Dict[int, float]: + """ + Retrieves the running AUPRC. + + Calling this method will not change the internal state of the metric. + + :param task_label: if None, return the entire dictionary of AUPRCs + for each task. Otherwise return the dictionary + `{task_label: AUPRC}`. + :return: A dict of running AUPRCs for each task label, + where each value is a float value between 0 and 1. + """ + assert task_label is None or isinstance(task_label, int) + if task_label is None: + return {k: v.result() for k, v in self._mean_AUPRC.items()} + else: + return {task_label: self._mean_AUPRC[task_label].result()} + + def reset(self, task_label=None) -> None: + """ + Resets the metric. + :param task_label: if None, reset the entire dictionary. + Otherwise, reset the value associated to `task_label`. + + :return: None. + """ + assert task_label is None or isinstance(task_label, int) + if task_label is None: + self._mean_AUPRC = defaultdict(Mean) + else: + self._mean_AUPRC[task_label].reset() + + +class AUPRCPluginMetric(GenericPluginMetric[float]): + """ + Base class for all AUPRCs plugin metrics + """ + + def __init__(self, reset_at, emit_at, mode): + self._AUPRC = AUPRC() + super(AUPRCPluginMetric, self).__init__( + self._AUPRC, reset_at=reset_at, emit_at=emit_at, mode=mode + ) + + def reset(self, strategy=None) -> None: + if self._reset_at == "stream" or strategy is None: + self._metric.reset() + else: + self._metric.reset(phase_and_task(strategy)[1]) + + def result(self, strategy=None) -> float: + if self._emit_at == "stream" or strategy is None: + return self._metric.result() + else: + return self._metric.result(phase_and_task(strategy)[1]) + + def update(self, strategy): + # task labels defined for each experience + task_labels = strategy.experience.task_labels + if len(task_labels) > 1: + # task labels defined for each pattern + task_labels = strategy.mb_task_id + else: + task_labels = task_labels[0] + self._AUPRC.update(strategy.mb_output, strategy.mb_y, task_labels) + + +class MinibatchAUPRC(AUPRCPluginMetric): + """ + The minibatch plugin AUPRC metric. + This metric only works at training time. + + This metric computes the average AUPRC over patterns + from a single minibatch. + It reports the result after each iteration. + + If a more coarse-grained logging is needed, consider using + :class:`EpochAUPRC` instead. + """ + + def __init__(self): + """ + Creates an instance of the MinibatchAUPRC metric. + """ + super(MinibatchAUPRC, self).__init__( + reset_at="iteration", emit_at="iteration", mode="train" + ) + + def __str__(self): + return "AUPRC_MB" + + +class EpochAUPRC(AUPRCPluginMetric): + """ + The average AUPRC over a single training epoch. + This plugin metric only works at training time. + + The AUPRC will be logged after each training epoch by computing + the number of correctly predicted patterns during the epoch divided by + the overall number of patterns encountered in that epoch. + """ + + def __init__(self): + """ + Creates an instance of the EpochAUPRC metric. + """ + + super(EpochAUPRC, self).__init__( + reset_at="epoch", emit_at="epoch", mode="train" + ) + + def __str__(self): + return "AUPRC_Epoch" + + +class RunningEpochAUPRC(AUPRCPluginMetric): + """ + The average AUPRC across all minibatches up to the current + epoch iteration. + This plugin metric only works at training time. + + At each iteration, this metric logs the AUPRC averaged over all patterns + seen so far in the current epoch. + The metric resets its state after each training epoch. + """ + + def __init__(self): + """ + Creates an instance of the RunningEpochAUPRC metric. + """ + + super(RunningEpochAUPRC, self).__init__( + reset_at="epoch", emit_at="iteration", mode="train" + ) + + def __str__(self): + return "RunningAUPRC_Epoch" + + +class ExperienceAUPRC(AUPRCPluginMetric): + """ + At the end of each experience, this plugin metric reports + the average AUPRC over all patterns seen in that experience. + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of ExperienceAUPRC metric + """ + super(ExperienceAUPRC, self).__init__( + reset_at="experience", emit_at="experience", mode="eval" + ) + + def __str__(self): + return "AUPRC_Exp" + + +class StreamAUPRC(AUPRCPluginMetric): + """ + At the end of the entire stream of experiences, this plugin metric + reports the average AUPRC over all patterns seen in all experiences. + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of StreamAUPRC metric + """ + super(StreamAUPRC, self).__init__( + reset_at="stream", emit_at="stream", mode="eval" + ) + + def __str__(self): + return "AUPRC_Stream" + + +class TrainedExperienceAUPRC(AUPRCPluginMetric): + """ + At the end of each experience, this plugin metric reports the average + AUPRC for only the experiences that the model has been trained on so far. + + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of TrainedExperienceAUPRC metric by first + constructing AUPRCPluginMetric + """ + super(TrainedExperienceAUPRC, self).__init__( + reset_at="stream", emit_at="stream", mode="eval" + ) + self._current_experience = 0 + + def after_training_exp(self, strategy) -> None: + self._current_experience = strategy.experience.current_experience + # Reset average after learning from a new experience + AUPRCPluginMetric.reset(self, strategy) + return AUPRCPluginMetric.after_training_exp(self, strategy) + + def update(self, strategy): + """ + Only update the AUPRC with results from experiences that have been + trained on + """ + if strategy.experience.current_experience <= self._current_experience: + AUPRCPluginMetric.update(self, strategy) + + def __str__(self): + return "AUPRC_On_Trained_Experiences" + + +def auprc_metrics( + *, + minibatch=False, + epoch=False, + epoch_running=False, + experience=False, + stream=False, + trained_experience=False, +) -> List[PluginMetric]: + """ + Helper method that can be used to obtain the desired set of + plugin metrics. + + :param minibatch: If True, will return a metric able to log + the minibatch AUPRC at training time. + :param epoch: If True, will return a metric able to log + the epoch AUPRC at training time. + :param epoch_running: If True, will return a metric able to log + the running epoch AUPRC at training time. + :param experience: If True, will return a metric able to log + the AUPRC on each evaluation experience. + :param stream: If True, will return a metric able to logAUPRCperiences. + :param trained_experience: If True, will return a metric able to log + the average evaluation AUPRC only for experiences that the + model has been trained on + + :return: A list of plugin metrics. + """ + + metrics = [] + if minibatch: + metrics.append(MinibatchAUPRC()) + + if epoch: + metrics.append(EpochAUPRC()) + + if epoch_running: + metrics.append(RunningEpochAUPRC()) + + if experience: + metrics.append(ExperienceAUPRC()) + + if stream: + metrics.append(StreamAUPRC()) + + if trained_experience: + metrics.append(TrainedExperienceAUPRC()) + + return metrics + + +class ROCAUC(Metric[float]): + """ + The ROCAUC metric. This is a standalone metric. + + The metric keeps a dictionary of <task_label, ROCAUC value> pairs. + and update the values through a running average over multiple + <prediction, target> pairs of Tensors, provided incrementally. + The "prediction" and "target" tensors may contain plain labels or + one-hot/logit vectors. + + Each time `result` is called, this metric emits the average ROCAUC + across all predictions made since the last `reset`. + + The reset method will bring the metric to its initial state. By default + this metric in its initial state will return an ROCAUC value of 0. + """ + + def __init__(self): + """ + Creates an instance of the standalone ROCAUC metric. + + By default this metric in its initial state will return a ROCAUC + value of 0. The metric can be updated by using the `update` method + while the running ROCAUC can be retrieved using the `result` method. + """ + super().__init__() + self._mean_ROCAUC = defaultdict(Mean) + """ + The mean utility that will be used to store the running ROCAUC + for each task label. + """ + + @torch.no_grad() + def update( + self, predicted_y: Tensor, true_y: Tensor, task_labels: Union[float, Tensor] + ) -> None: + """ + Update the running ROCAUC given the true and predicted labels. + Parameter `task_labels` is used to decide how to update the inner + dictionary: if Float, only the dictionary value related to that task + is updated. If Tensor, all the dictionary elements belonging to the + task labels will be updated. + + :param predicted_y: The model prediction. Both labels and logit vectors + are supported. + :param true_y: The ground truth. Both labels and one-hot vectors + are supported. + :param task_labels: the int task label associated to the current + experience or the task labels vector showing the task label + for each pattern. + + :return: None. + """ + if len(true_y) != len(predicted_y): + raise ValueError("Size mismatch for true_y and predicted_y tensors") + + if isinstance(task_labels, Tensor) and len(task_labels) != len(true_y): + raise ValueError("Size mismatch for true_y and task_labels tensors") + + true_y = torch.as_tensor(true_y) + predicted_y = torch.as_tensor(predicted_y) + + assert len(predicted_y.size()) == 2, ( + "Predictions need to be logits or scores, not labels" + ) + + if len(true_y.shape) > 1: + # Logits -> transform to labels + true_y = torch.max(true_y, 1)[1] + + scores = predicted_y[arange(len(true_y)), true_y] + + try: + roc_auc_score_val = roc_auc_score(true_y.cpu(), scores.cpu()) + except ValueError: + roc_auc_score_val = 1 + + if isinstance(task_labels, int): + self._mean_ROCAUC[task_labels].update(roc_auc_score_val, len(predicted_y)) + elif isinstance(task_labels, Tensor): + raise NotImplementedError + else: + raise ValueError( + f"Task label type: {type(task_labels)}, expected int/float or Tensor" + ) + + def result(self, task_label=None) -> Dict[int, float]: + """ + Retrieves the running ROCAUC. + + Calling this method will not change the internal state of the metric. + + :param task_label: if None, return the entire dictionary of ROCAUCs + for each task. Otherwise return the dictionary + `{task_label: ROCAUC}`. + :return: A dict of running ROCAUCs for each task label, + where each value is a float value between 0 and 1. + """ + assert task_label is None or isinstance(task_label, int) + if task_label is None: + return {k: v.result() for k, v in self._mean_ROCAUC.items()} + else: + return {task_label: self._mean_ROCAUC[task_label].result()} + + def reset(self, task_label=None) -> None: + """ + Resets the metric. + :param task_label: if None, reset the entire dictionary. + Otherwise, reset the value associated to `task_label`. + + :return: None. + """ + assert task_label is None or isinstance(task_label, int) + if task_label is None: + self._mean_ROCAUC = defaultdict(Mean) + else: + self._mean_ROCAUC[task_label].reset() + + +class ROCAUCPluginMetric(GenericPluginMetric[float]): + """ + Base class for all ROCAUCs plugin metrics + """ + + def __init__(self, reset_at, emit_at, mode): + self._ROCAUC = ROCAUC() + super(ROCAUCPluginMetric, self).__init__( + self._ROCAUC, reset_at=reset_at, emit_at=emit_at, mode=mode + ) + + def reset(self, strategy=None) -> None: + if self._reset_at == "stream" or strategy is None: + self._metric.reset() + else: + self._metric.reset(phase_and_task(strategy)[1]) + + def result(self, strategy=None) -> float: + if self._emit_at == "stream" or strategy is None: + return self._metric.result() + else: + return self._metric.result(phase_and_task(strategy)[1]) + + def update(self, strategy): + # task labels defined for each experience + task_labels = strategy.experience.task_labels + if len(task_labels) > 1: + # task labels defined for each pattern + task_labels = strategy.mb_task_id + else: + task_labels = task_labels[0] + self._ROCAUC.update(strategy.mb_output, strategy.mb_y, task_labels) + + +class MinibatchROCAUC(ROCAUCPluginMetric): + """ + The minibatch plugin ROCAUC metric. + This metric only works at training time. + + This metric computes the average ROCAUC over patterns + from a single minibatch. + It reports the result after each iteration. + + If a more coarse-grained logging is needed, consider using + :class:`EpochROCAUC` instead. + """ + + def __init__(self): + """ + Creates an instance of the MinibatchROCAUC metric. + """ + super(MinibatchROCAUC, self).__init__( + reset_at="iteration", emit_at="iteration", mode="train" + ) + + def __str__(self): + return "ROCAUC_MB" + + +class EpochROCAUC(ROCAUCPluginMetric): + """ + The average ROCAUC over a single training epoch. + This plugin metric only works at training time. + + The ROCAUC will be logged after each training epoch by computing + the number of correctly predicted patterns during the epoch divided by + the overall number of patterns encountered in that epoch. + """ + + def __init__(self): + """ + Creates an instance of the EpochROCAUC metric. + """ + + super(EpochROCAUC, self).__init__( + reset_at="epoch", emit_at="epoch", mode="train" + ) + + def __str__(self): + return "ROCAUC_Epoch" + + +class RunningEpochROCAUC(ROCAUCPluginMetric): + """ + The average ROCAUC across all minibatches up to the current + epoch iteration. + This plugin metric only works at training time. + + At each iteration, this metric logs the ROCAUC averaged over all patterns + seen so far in the current epoch. + The metric resets its state after each training epoch. + """ + + def __init__(self): + """ + Creates an instance of the RunningEpochROCAUC metric. + """ + + super(RunningEpochROCAUC, self).__init__( + reset_at="epoch", emit_at="iteration", mode="train" + ) + + def __str__(self): + return "RunningROCAUC_Epoch" + + +class ExperienceROCAUC(ROCAUCPluginMetric): + """ + At the end of each experience, this plugin metric reports + the average ROCAUC over all patterns seen in that experience. + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of ExperienceROCAUC metric + """ + super(ExperienceROCAUC, self).__init__( + reset_at="experience", emit_at="experience", mode="eval" + ) + + def __str__(self): + return "ROCAUC_Exp" + + +class StreamROCAUC(ROCAUCPluginMetric): + """ + At the end of the entire stream of experiences, this plugin metric + reports the average ROCAUC over all patterns seen in all experiences. + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of StreamROCAUC metric + """ + super(StreamROCAUC, self).__init__( + reset_at="stream", emit_at="stream", mode="eval" + ) + + def __str__(self): + return "ROCAUC_Stream" + + +class TrainedExperienceROCAUC(ROCAUCPluginMetric): + """ + At the end of each experience, this plugin metric reports the average + ROCAUC for only the experiences that the model has been trained on so far. + + This metric only works at eval time. + """ + + def __init__(self): + """ + Creates an instance of TrainedExperienceROCAUC metric by first + constructing ROCAUCPluginMetric + """ + super(TrainedExperienceROCAUC, self).__init__( + reset_at="stream", emit_at="stream", mode="eval" + ) + self._current_experience = 0 + + def after_training_exp(self, strategy) -> None: + self._current_experience = strategy.experience.current_experience + # Reset average after learning from a new experience + ROCAUCPluginMetric.reset(self, strategy) + return ROCAUCPluginMetric.after_training_exp(self, strategy) + + def update(self, strategy): + """ + Only update the ROCAUC with results from experiences that have been + trained on + """ + if strategy.experience.current_experience <= self._current_experience: + ROCAUCPluginMetric.update(self, strategy) + + def __str__(self): + return "ROCAUC_On_Trained_Experiences" + + +def rocauc_metrics( + *, + minibatch=False, + epoch=False, + epoch_running=False, + experience=False, + stream=False, + trained_experience=False, +) -> List[PluginMetric]: + """ + Helper method that can be used to obtain the desired set of + plugin metrics. + + :param minibatch: If True, will return a metric able to log + the minibatch ROCAUC at training time. + :param epoch: If True, will return a metric able to log + the epoch ROCAUC at training time. + :param epoch_running: If True, will return a metric able to log + the running epoch ROCAUC at training time. + :param experience: If True, will return a metric able to log + the ROCAUC on each evaluation experience. + :param stream: If True, will return a metric able to logROCAUCperiences. + :param trained_experience: If True, will return a metric able to log + the average evaluation ROCAUC only for experiences that the + model has been trained on + + :return: A list of plugin metrics. + """ + + metrics = [] + if minibatch: + metrics.append(MinibatchROCAUC()) + + if epoch: + metrics.append(EpochROCAUC()) + + if epoch_running: + metrics.append(RunningEpochROCAUC()) + + if experience: + metrics.append(ExperienceROCAUC()) + + if stream: + metrics.append(StreamROCAUC()) + + if trained_experience: + metrics.append(TrainedExperienceROCAUC()) + + return metrics + + +__all__ = [ + "BalancedAccuracy", + "MinibatchBalancedAccuracy", + "EpochBalancedAccuracy", + "RunningEpochBalancedAccuracy", + "ExperienceBalancedAccuracy", + "StreamBalancedAccuracy", + "TrainedExperienceBalancedAccuracy", + "balancedaccuracy_metrics", + "Sensitivity", + "MinibatchSensitivity", + "EpochSensitivity", + "RunningEpochSensitivity", + "ExperienceSensitivity", + "StreamSensitivity", + "TrainedExperienceSensitivity", + "sensitivity_metrics", + "Specificity", + "MinibatchSpecificity", + "EpochSpecificity", + "RunningEpochSpecificity", + "ExperienceSpecificity", + "StreamSpecificity", + "TrainedExperienceSpecificity", + "specificity_metrics", + "Precision", + "MinibatchPrecision", + "EpochPrecision", + "RunningEpochPrecision", + "ExperiencePrecision", + "StreamPrecision", + "TrainedExperiencePrecision", + "precision_metrics", + "AUPRC", + "MinibatchAUPRC", + "EpochAUPRC", + "RunningEpochAUPRC", + "ExperienceAUPRC", + "StreamAUPRC", + "TrainedExperienceAUPRC", + "auprc_metrics", + "ROCAUC", + "MinibatchROCAUC", + "EpochROCAUC", + "RunningEpochROCAUC", + "ExperienceROCAUC", + "StreamROCAUC", + "TrainedExperienceROCAUC", + "rocauc_metrics", +]