metric.py (11455B)
1 """ 2 Module for handling socio-political event extraction evaluation. 3 4 All metrics are uniquely identified by a path with the following components: 5 split_dataset/field/metric 6 """ 7 8 from __future__ import annotations 9 10 import abc 11 import contextlib 12 import datetime 13 from typing import TYPE_CHECKING, Any, ClassVar 14 15 import torch 16 from torch.utils.tensorboard.writer import SummaryWriter 17 18 19 if TYPE_CHECKING: 20 import pathlib 21 22 import transformers 23 24 import t5.config 25 import t5.dataset 26 import t5.model 27 28 29 class Metrics: 30 """ 31 A collection of metrics to evaluate socio-political event extraction models. 32 33 This class act as a kind of Metric factory, where instantiating a Metrics object instantiate all registered Metric classes. 34 """ 35 36 metric_classes: ClassVar[dict[str, type[Metric]]] = {} 37 38 def __init__(self: Metrics, config: t5.config.Config, dataset: t5.dataset.Dataset, model: t5.model.Model, modeldir: pathlib.Path) -> None: 39 """Initialize socio-political event extraction metrics.""" 40 self.config: t5.config.Config = config 41 self.dataset: t5.dataset.Dataset = dataset 42 self.model: t5.model.Model = model 43 self.modeldir: pathlib.Path = modeldir # for tensorboard logs 44 self.metrics: list[Metric] = [] 45 for metric in Metrics.metric_classes.values(): 46 self.metrics.append(metric(self.config, self.dataset, self.model)) 47 48 @classmethod 49 def register(cls: type[Metrics], metric: type[Metric]) -> type[Metric]: 50 """Register a Metric class.""" 51 cls.metric_classes[metric.label()] = metric 52 return metric 53 54 def _zero_metrics(self: Metrics) -> None: 55 for metric in self.metrics: 56 metric.zero() 57 58 def _update_metrics(self: Metrics, evaluation_output: transformers.EvalPrediction) -> None: 59 batch_size: int = len(evaluation_output.inputs["id"]) 60 predictions: torch.Tensor = self.model.detect_prediction_generation(evaluation_output) 61 for batch_id in range(batch_size): 62 # evaluation_output.label_ids can be accessed through inputs["labels"] 63 inputs: dict[str, Any] = { key: value[batch_id] for key, value in evaluation_output.inputs.items() } 64 generated_text: torch.Tensor = predictions[batch_id] 65 outputs: dict[str, Any] = self.dataset.recover_data_from_output(inputs, generated_text) 66 for metric in self.metrics: 67 metric(inputs, outputs) 68 69 def _compute_metrics(self: Metrics) -> dict[str, float]: 70 values: dict[str, float] = {} 71 aggregate_accuracy: list[float] = [] 72 for metric in self.metrics: 73 for key, value in metric.compute().items(): 74 values[f"{self.dataset.config.name}/{key}"] = value 75 if key.endswith("/string accuracy"): 76 aggregate_accuracy.append(value) 77 values[f"{self.dataset.config.name}/aggregate/mean string accuracy"] = sum(aggregate_accuracy) / len(aggregate_accuracy) 78 return values 79 80 def log_graphs(self: Metrics, logdir: pathlib.Path) -> None: 81 """Log TensorBoard graphs for all metrics that produce them. Currently only histograms are available.""" 82 for metric in self.metrics: 83 if hasattr(metric, "log_histogram") and metric.log_histogram: # log all histograms 84 metric.log_histogram(logdir) 85 86 @torch.no_grad() 87 def __call__(self: Metrics, evaluation_output: transformers.EvalPrediction, *, compute_result: bool) -> None | dict[str, float]: 88 """Update and compute metrics for a given prediction.""" 89 self._update_metrics(evaluation_output) 90 if compute_result: 91 result: dict[str, float] = self._compute_metrics() 92 self.log_graphs(self.modeldir / "runs") 93 self._zero_metrics() 94 return result 95 return None 96 97 98 class Metric(abc.ABC): 99 """A single metric to evaluate (typically one) socio-political event field.""" 100 101 def __init__(self: Metric, config: t5.config.Config, dataset: t5.dataset.Dataset, model: t5.model.Model) -> None: 102 """Initialize the metric.""" 103 self.config: t5.config.Config = config 104 self.dataset: t5.dataset.Dataset = dataset 105 self.model: t5.model.Model = model 106 self.zero() 107 108 @abc.abstractmethod 109 def zero(self: Metric) -> None: 110 """Reinitialize the metric.""" 111 112 @classmethod 113 @abc.abstractmethod 114 def label(cls: type[Metric]) -> str: 115 """Get the name of the metric.""" 116 117 @abc.abstractmethod 118 def __call__(self: Metric, inputs: dict[str, Any], outputs: dict[str, Any]) -> None: 119 """Update the metric.""" 120 121 @abc.abstractmethod 122 def compute(self: Metric) -> dict[str, float]: 123 """Compute the metric.""" 124 125 126 for field in ["side_a_name", "side_b_name", "start_date", "end_date", "location_root_name", "location_adm1_name", "location_adm2_name", "location_where_name", "deaths_side_a", "deaths_side_b", "deaths_civilian", "deaths_unknown", "deaths_low", "deaths_high"]: 127 @Metrics.register 128 class StringAccuracy(Metric): 129 """Compute the accuracy.""" 130 131 field: str = field 132 133 def __init__(self: StringAccuracy, config: t5.config.Config, dataset: t5.dataset.Dataset, model: t5.model.Model) -> None: 134 """Initialize the accuracy counts.""" 135 super().__init__(config, dataset, model) 136 137 def zero(self: StringAccuracy) -> None: 138 """Reinitialize the metric.""" 139 self.correct: int = 0 140 self.total: int = 0 141 142 def __call__(self: StringAccuracy, inputs: dict[str, Any], outputs: dict[str, Any]) -> None: 143 """Update the metric.""" 144 self.total += 1 145 self.correct += (str(inputs[self.field]) == outputs.get(self.field, None)) 146 147 def compute(self: StringAccuracy) -> dict[str, float]: 148 """Compute the metric.""" 149 accuracy: float = self.correct / self.total if self.total != 0 else 0. 150 return { self.label(): accuracy } 151 152 @classmethod 153 def label(cls: type[StringAccuracy]) -> str: 154 """Get the name of the metric.""" 155 return f"{cls.field}/string accuracy" 156 157 158 for field in ["deaths_side_a", "deaths_side_b", "deaths_civilian", "deaths_unknown", "deaths_low", "deaths_high"]: 159 @Metrics.register 160 class RMSE(Metric): 161 """Compute the RMSE.""" 162 163 field: str = field 164 165 def __init__(self: RMSE, config: t5.config.Config, dataset: t5.dataset.Dataset, model: t5.model.Model) -> None: 166 """Initialize the error counts.""" 167 super().__init__(config, dataset, model) 168 169 def zero(self: RMSE) -> None: 170 """Reinitialize the metric.""" 171 self.squared_error: int = 0 172 self.total: int = 0 173 174 def __call__(self: RMSE, inputs: dict[str, Any], outputs: dict[str, Any]) -> None: 175 """Update the metric.""" 176 self.total += 1 177 value: int = 0 178 with contextlib.suppress(ValueError): # if it's not a number assume 0 179 value = int(outputs.get(self.field, "0")) 180 self.squared_error += (inputs[self.field] - value)**2 181 182 def compute(self: RMSE) -> dict[str, float]: 183 """Compute the metric.""" 184 rmse: float = (self.squared_error / self.total)**0.5 185 return { self.label(): rmse } 186 187 @classmethod 188 def label(cls: type[RMSE]) -> str: 189 """Get the name of the metric.""" 190 return f"{cls.field}/RMSE" 191 192 193 for field in ["start_date", "end_date"]: 194 @Metrics.register 195 class DateRMSE(Metric): 196 """Compute the RMSE (in days) for YYYY-MM-DD dates.""" 197 198 field: str = field 199 200 def __init__(self: DateRMSE, config: t5.config.Config, dataset: t5.dataset.Dataset, model: t5.model.Model) -> None: 201 """Initialize the error counts.""" 202 super().__init__(config, dataset, model) 203 204 def zero(self: DateRMSE) -> None: 205 """Reinitialize the metric.""" 206 self.squared_error: int = 0 207 self.total: int = 0 208 self.unparsable: int = 0 209 self.day_differences: list[int] = [] 210 211 def __call__(self: DateRMSE, inputs: dict[str, Any], outputs: dict[str, Any]) -> None: 212 """Update the metric.""" 213 # add to sample count 214 self.total += 1 215 216 # declare input publication date (input_source_date) for default values 217 input_source_date: Any = inputs.get("source_date") 218 input_start_date: Any = inputs.get(self.field) 219 if not isinstance(input_source_date, datetime.datetime) or not isinstance(input_start_date, datetime.date): # if faulty input, ignore 220 return 221 default_date: datetime.date = input_source_date.date() # remove hh:mm:ss info 222 output_start_date: Any = outputs.get(self.field) 223 224 # init dates 225 sample: datetime.date = input_start_date 226 value: datetime.date = default_date 227 228 # attempt parsing predicted start_date, if not possible, it will be the sample source_date 229 try: 230 value = datetime.datetime.strptime(output_start_date, "%Y-%m-%d").date() 231 except (ValueError, TypeError): 232 self.unparsable += 1 233 234 # calculate and store differences 235 day_difference = (sample - value).days 236 self.day_differences.append(day_difference) 237 self.squared_error += day_difference ** 2 238 239 def compute(self: DateRMSE) -> dict[str, float]: 240 """Compute the metric.""" 241 rmse: float = (self.squared_error / self.total)**0.5 242 return { self.label(): rmse, "debug/start_date/unparsable": self.unparsable / self.total} 243 244 def log_histogram(self: DateRMSE, logdir: pathlib.Path) -> None: 245 """Log the histogram of differences using TensorBoard.""" 246 writer: Any = SummaryWriter(logdir) 247 writer.add_histogram( 248 f"error_metrics/{self.dataset.config.name}/date_differences", 249 torch.tensor(self.day_differences) 250 ) 251 writer.close() 252 253 @classmethod 254 def label(cls: type[DateRMSE]) -> str: 255 """Get the name of the metric.""" 256 return f"{cls.field}/date RMSE in days" 257 258 259 @Metrics.register 260 class ParsableOutput(Metric): 261 """Verify whether the output of the model respect the template.""" 262 263 def __init__(self: ParsableOutput, config: t5.config.Config, dataset: t5.dataset.Dataset, model: t5.model.Model) -> None: 264 """Initialize the accuracy counts.""" 265 super().__init__(config, dataset, model) 266 267 def zero(self: ParsableOutput) -> None: 268 """Reinitialize the metric.""" 269 self.correct: int = 0 270 self.total: int = 0 271 272 def __call__(self: ParsableOutput, _: dict[str, Any], outputs: dict[str, Any]) -> None: 273 """Update the metric.""" 274 self.total += 1 275 self.correct += (len(outputs) > 1) # contains at least the "logits" field 276 277 def compute(self: ParsableOutput) -> dict[str, float]: 278 """Compute the metric.""" 279 accuracy: float = self.correct / self.total if self.total != 0 else 0. 280 return { self.label(): accuracy } 281 282 @classmethod 283 def label(cls: type[ParsableOutput]) -> str: 284 """Get the name of the metric.""" 285 return "debug/parsable output"