cpdbench.control.CPDDatasetResult

  1from enum import Enum
  2
  3from cpdbench.exception.ResultSetInconsistentException import ResultSetInconsistentException
  4import traceback
  5
  6from cpdbench.task.Task import Task
  7
  8
  9class ErrorType(str, Enum):
 10    """Enum for all error types which can occur during the CPDBench execution"""
 11    DATASET_ERROR = "DATASET_ERROR"
 12    ALGORITHM_ERROR = "ALGORITHM_ERROR"
 13    METRIC_ERROR = "METRIC_ERROR"
 14
 15
 16class CPDDatasetResult:
 17    """Container for all results of one single dataset including algorithm and metric results"""
 18
 19    def __init__(self, dataset: Task, algorithms: list[Task], metrics: list[Task]):
 20        """Constructs a dataset result with the basic attributes
 21        :param dataset: task which created the dataset
 22        :param algorithms: list of all algorithm tasks which were used with this dataset
 23        :param metrics: list of all metric tasks which were used with this dataset
 24        """
 25
 26        self._dataset = dataset.get_task_name()
 27        self._algorithms = [a.get_task_name() for a in algorithms]
 28        self._metrics = [m.get_task_name() for m in metrics]
 29
 30        self._indexes = {}
 31        self._scores = {}
 32        self._metric_scores = {}
 33
 34        self._errors = []
 35        self._parameters = ({self._dataset: dataset.get_param_dict()}
 36                            | {task.get_task_name(): task.get_param_dict() for task in algorithms}
 37                            | {task.get_task_name(): task.get_param_dict() for task in metrics})
 38
 39        self._dataset_runtime = -1
 40        self._algorithm_runtimes = {}
 41        for a in self._algorithms:
 42            self._metric_scores[a] = {}
 43
 44    def add_dataset_runtime(self, runtime: float) -> None:
 45        """Adds the runtime of the dataset task to the result object.
 46        Once a runtime was added, the value is immutable.
 47        :param runtime: the runtime of the task in seconds
 48        """
 49        if self._dataset_runtime == -1:
 50            self._dataset_runtime = runtime
 51
 52    def add_algorithm_result(self, indexes: list[int], scores: list[float], algorithm: str, runtime: float) -> None:
 53        """Adds an algorithm result with indexes and confidence scores to the result container.
 54        :param indexes: list of calculated changepoint indexes
 55        :param scores: list of calculated confidence scores respective to the indexes list
 56        :param algorithm: name of the calculated algorithm
 57        :param runtime: runtime of the algorithm execution in seconds
 58        """
 59
 60        if algorithm not in self._algorithms:
 61            raise ResultSetInconsistentException(f"Algorithm {algorithm} does not exist")
 62        self._indexes[algorithm] = indexes
 63        self._scores[algorithm] = scores
 64        self._algorithm_runtimes[algorithm] = {}
 65        self._algorithm_runtimes[algorithm]["runtime"] = runtime
 66
 67    def add_metric_score(self, metric_score: float, algorithm: str, metric: str, runtime: float) -> None:
 68        """Adds a metric result of an algorithm/dataset to the result container.
 69        :param metric_score: calculated metric score as float
 70        :param algorithm: name of the calculated algorithm
 71        :param metric: name of the used metric
 72        :param runtime: runtime of the metric execution in seconds
 73        """
 74
 75        if (algorithm not in self._algorithms
 76                or metric not in self._metrics):
 77            raise ResultSetInconsistentException()
 78        if self._indexes.get(algorithm) is None:
 79            raise ResultSetInconsistentException()
 80        self._metric_scores[algorithm][metric] = metric_score
 81        self._algorithm_runtimes[algorithm][metric] = runtime
 82
 83    def add_error(self, exception: Exception, error_type: ErrorType, algorithm: str = None, metric: str = None) -> None:
 84        """Adds a thrown error to the result container.
 85        :param exception: the thrown exception object
 86        :param error_type: the error type of the thrown exception
 87        :param algorithm: name of the algorithm where the exception occurred if applicable
 88        :param metric: name of the metric where the exception occurred if applicable
 89        """
 90
 91        self._errors.append((type(exception).__name__, ''.join(traceback.format_exception(None, exception,
 92                                                                                          exception.__traceback__)),
 93                             error_type, algorithm, metric))
 94
 95    def get_result_as_dict(self) -> dict:
 96        """Returns the result container formatted as dictionary.
 97        :returns: the complete results with indexes, scores and metric scores of one dataset as dict
 98        """
 99
100        return {
101            self._dataset: {
102                "indexes": self._indexes,
103                "scores": self._scores,
104                "metric_scores": self._metric_scores
105            }
106        }
107
108    def get_errors_as_list(self) -> list:
109        """Returns the list of errors occurred around the dataset.
110        :returns: all errors of the dataset as python list
111        """
112
113        return [
114            {
115                "dataset": self._dataset,
116                "error_type": error[2],
117                "algorithm": error[3],
118                "metric": error[4],
119                "exception_type": error[0],
120                "trace_back": error[1]
121            }
122            for error in self._errors
123        ]
124
125    def get_parameters(self) -> dict:
126        """Returns the parameters of all included tasks as dict.
127        :returns: the parameters as python dict
128        """
129        return self._parameters
130
131    def get_runtimes(self) -> dict:
132        """Returns the runtimes of all included tasks as dict.
133        :returns: the runtimes as python dict
134        """
135        if self._dataset_runtime == -1:
136            result_dict = {self._dataset: self._algorithm_runtimes}
137        else:
138            result_dict = {
139                self._dataset: self._algorithm_runtimes | {
140                    "runtime": self._dataset_runtime,
141                }
142            }
143        return result_dict