Coverage for collectra/evaluator/base.py: 92.31%
221 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-16 09:56 +0000
1__all__ = ["Evaluator"]
3import csv
4from dataclasses import dataclass
5from pathlib import Path
7import yaml
8from rich.table import Table
10from ..logger import get_logger
11from ..types import ArtefactNode
12from ..utils import change_dir
14logger = get_logger(__name__)
17@dataclass
18class FileEvaluationResult:
19 """Results for a single file pair evaluation."""
21 filename: str
22 label_metrics: dict[str, dict]
25@dataclass
26class EvaluationReport:
27 """Complete evaluation report across all files."""
29 file_results: list[FileEvaluationResult]
30 aggregate: dict
31 threshold: float
33 def __post_init__(self):
34 self.aggregate_tables = []
35 self.tables = []
36 for data_type, metrics in self.aggregate.items():
37 """Turn the aggregate dict into a rich Table for display."""
38 table = Table(title=f"Aggregate Evaluation Metrics for {data_type}")
39 table.add_column("Metric", style="cyan", no_wrap=True)
40 table.add_column("Value", justify="right", style="green")
42 # Add overall metrics
43 overall = metrics.get("overall", {})
44 table.add_row("Precision (Overall)", f"{overall.get('precision', 0):.3f}")
45 table.add_row("Recall (Overall)", f"{overall.get('recall', 0):.3f}")
46 table.add_row("F1 Micro", f"{overall.get('f1_micro', 0):.3f}")
47 table.add_row(
48 "F1 File-Averaged", f"{overall.get('f1_file_averaged', 0):.3f}"
49 )
50 table.add_row("Total Files", str(overall.get("total_files", 0)))
51 self.aggregate_tables.append(table)
52 # Create per-label table
53 per_label = metrics.get("per_label", {})
54 if not per_label:
55 continue
56 table = Table(title=f"Per-Label Metrics for Artefact Type: {data_type}")
57 table.add_column("Label", style="magenta")
58 table.add_column("Precision", justify="right", style="green")
59 table.add_column("Recall", justify="right", style="green")
60 table.add_column("F1", justify="right", style="green")
61 table.add_column("Mean Score", justify="right", style="green")
62 table.add_column("Support", justify="right", style="blue")
63 for label_name, metrics in per_label.items():
64 table.add_row(
65 label_name,
66 f"{metrics.get('precision', 0):.3f}",
67 f"{metrics.get('recall', 0):.3f}",
68 f"{metrics.get('f1', 0):.3f}",
69 f"{metrics.get('mean_score', 0):.3f}",
70 str(metrics.get("support", 0)),
71 )
72 self.tables.append(table)
74 def save_csv(self, output: Path) -> None:
75 """Save file- and label-level evaluation metrics as a tidy CSV."""
76 output = Path(output).expanduser()
77 output.parent.mkdir(parents=True, exist_ok=True)
78 fieldnames = [
79 "filename",
80 "label",
81 "artefact_type",
82 "precision",
83 "recall",
84 "f1",
85 "mean_score",
86 "num_predicted",
87 "num_gold",
88 "true_positives",
89 "false_positives",
90 "false_negatives",
91 ]
92 with output.open("w", newline="") as stream:
93 writer = csv.DictWriter(stream, fieldnames=fieldnames)
94 writer.writeheader()
95 for result in self.file_results:
96 for label, metrics in result.label_metrics.items():
97 types = metrics.get("types", []) or [""]
98 for artefact_type in types:
99 writer.writerow(
100 {
101 "filename": result.filename,
102 "label": label,
103 "artefact_type": artefact_type,
104 "precision": metrics.get("precision", 0),
105 "recall": metrics.get("recall", 0),
106 "f1": metrics.get("f1", 0),
107 "mean_score": metrics.get("mean_score", 0),
108 "num_predicted": metrics.get("num_predicted", 0),
109 "num_gold": metrics.get("num_gold", 0),
110 "true_positives": metrics.get("num_matched", 0),
111 "false_positives": metrics.get(
112 "num_false_positives", 0
113 ),
114 "false_negatives": metrics.get(
115 "num_false_negatives", 0
116 ),
117 }
118 )
121class EvaluationError(Exception):
122 """Raised when evaluation cannot proceed (e.g., missing gold files)."""
124 pass
127class Evaluator:
128 """
129 End-to-end evaluation orchestrator for Collectra workflows.
131 Responsibilities:
132 1. Match input files to gold standard files by filename
133 2. Load and parse .grapto (or other ext) folder structures
134 3. Delegate per-label evaluation to ArtefactNode.evaluate()
135 4. Aggregate metrics across files and labels
136 5. Export results in tabular format
137 """
139 def __init__(
140 self,
141 predicted_folder: Path,
142 gold_folder: Path,
143 ext: str,
144 labels: set[str] | None = None,
145 match_by_order: bool = False,
146 ) -> None:
147 self.ext = ext.lower().replace(".", "")
148 self.labels = labels
149 predicted_folder = Path(predicted_folder).expanduser().resolve()
150 gold_folder = Path(gold_folder).expanduser().resolve()
151 direct_pair = self._is_collectra_folder(
152 predicted_folder
153 ) and self._is_collectra_folder(gold_folder)
154 if match_by_order or direct_pair:
155 self.entries = self._pair_folders_by_order(predicted_folder, gold_folder)
156 else:
157 entries = self._discover_collectra_folders(predicted_folder)
158 self.entries = self._discover_collectra_folders(
159 gold_folder, gold=True, entries=entries
160 )
161 self.results: list[FileEvaluationResult] = []
162 self.aggregate_metrics: dict = {}
164 def _is_collectra_folder(self, path: Path) -> bool:
165 return path.is_dir() and path.suffix == f".{self.ext}"
167 def _pair_folders_by_order(
168 self, predicted_folder: Path, gold_folder: Path
169 ) -> dict[str, list[Path | None]]:
170 """Pair independently sorted prediction and gold result folders."""
171 predicted_entries = self._discover_collectra_folders(predicted_folder)
172 gold_entries = self._discover_collectra_folders(gold_folder, gold=True)
173 predictions = sorted(
174 (paths[1] for paths in predicted_entries.values() if paths[1]),
175 key=lambda path: path.name,
176 )
177 golds = sorted(
178 (paths[0] for paths in gold_entries.values() if paths[0]),
179 key=lambda path: path.name,
180 )
181 return {
182 f"pair-{index:06d}": [
183 golds[index] if index < len(golds) else None,
184 predictions[index] if index < len(predictions) else None,
185 ]
186 for index in range(max(len(predictions), len(golds)))
187 }
189 def __call__(self) -> None:
190 self.evaluate()
192 def _discover_collectra_folders(
193 self,
194 folder: Path,
195 entries: dict[str, list[Path | None]] | None = None,
196 gold: bool = False,
197 ) -> dict[str, list[Path | None]]:
198 """
199 Find .{ext} directories (e.g., .grapto folders).
201 ``folder`` may be either one result directory or a parent containing
202 multiple result directories.
204 Args:
205 folder: Path to search for .{ext} directories
206 entries: dictionary of tuples base_name: (gold_path, predicted_path) to extend if empty, or to have entries updated.
207 gold: Whether to look for gold files (as opposed to predicted files)
209 Returns:
210 dictionary mapping base_name to tuples (gold_path, predicted_path) for each discovered .{ext} directory
211 """
212 if entries is None:
213 entries = {}
214 folder = Path(folder).expanduser().resolve()
215 if not folder.is_dir():
216 raise FileNotFoundError(f"Evaluation path is not a directory: {folder}")
218 items = [folder] if folder.suffix == f".{self.ext}" else list(folder.iterdir())
219 for item in items:
220 if not (item.is_dir() and item.suffix == f".{self.ext}"):
221 continue
222 base_name = item.stem
223 if base_name not in entries:
224 if gold:
225 entries[base_name] = [item, None]
226 else:
227 entries[base_name] = [None, item]
228 else:
229 existing_gold, existing_predicted = entries[base_name]
230 if gold:
231 entries[base_name] = [item, existing_predicted]
232 else:
233 entries[base_name] = [existing_gold, item]
234 return entries
236 def _validate_file_pairs(
237 self,
238 ) -> tuple[list[tuple[Path, Path]], list[Path], list[Path]]:
239 """
240 Match input files to gold files and identify mismatches.
242 Returns:
243 - matched_pairs: list of (input_path, gold_path, base_name)
244 - missing_gold: input files without corresponding gold
245 - extra_gold: gold files without corresponding input
246 """
248 matched_pairs: list[tuple[Path, Path]] = []
249 missing_golds: list[Path] = []
250 extra_golds: list[Path] = []
252 for key, (gold_path, predicted_path) in self.entries.items():
253 missing_gold = gold_path is None
254 missing_predicted = predicted_path is None
255 if not missing_gold and not missing_predicted:
256 matched_pairs.append((predicted_path, gold_path))
257 if missing_gold and missing_predicted:
258 raise ValueError(
259 f"Entry '{key}' has neither gold nor predicted file. This should not happen."
260 )
261 if missing_gold and predicted_path is not None:
262 missing_golds.append(predicted_path)
263 if missing_predicted and gold_path is not None:
264 extra_golds.append(gold_path)
266 return matched_pairs, list(missing_golds), list(extra_golds)
268 def _load_collectra_file(self, path: Path) -> dict[str, ArtefactNode]:
269 """
270 Parse a collectra file data into ArtefactNodes organized by label.
272 Each label (e.g., "registration_number", "local_text") becomes
273 a separate ArtefactNode containing its items (ImageCrops, Texts).
275 Returns:
276 dict mapping label_name -> ArtefactNode
277 """
278 labels = dict()
279 with change_dir(path):
280 with open("results.yaml", "r") as f:
281 data = yaml.safe_load(f)
282 data.pop("collectra_results_metadata", "")
283 node_names = list(data.keys())
284 for name in node_names:
285 node = ArtefactNode(name=name)
286 node.process(key=node.name, value=path, skip_type_check=True)
287 labels[name] = node
288 return labels
290 def _create_missing_prediction_metrics(self, gold_node: ArtefactNode) -> dict:
291 """
292 Generate metrics when predictions are entirely missing for a gold label.
294 Scenario: Gold has items for this label, but predictions have none.
295 Result:
296 - Precision = 0.0 (undefined 0/0, defaults to 0.0 for consistency)
297 - Recall = 0.0 (missed everything)
298 - F1 = 0.0
299 - All gold items are false negatives
301 This mirrors the edge case handling in ArtefactNode.evaluate()
302 when self.items is empty but gold_items is not.
303 """
304 gold_ids = list(gold_node.items.keys())
306 return {
307 "types": list(
308 set([type(item).__name__ for item in gold_node.items.values()])
309 ),
310 "precision": 0.0, # No predictions = undefined (0/0), defaults to 0.0
311 "recall": 0.0, # Missed all gold items
312 "f1": 0.0,
313 "mean_score": 0.0,
314 "num_predicted": 0,
315 "num_gold": len(gold_ids),
316 "num_matched": 0,
317 "num_false_positives": 0,
318 "num_false_negatives": len(gold_ids),
319 "matches": [],
320 "unmatched_predicted": [],
321 "unmatched_gold": gold_ids,
322 }
324 def _create_extra_prediction_metrics(self, input_node: ArtefactNode) -> dict:
325 """
326 Generate metrics when predictions exist for a label not in gold.
328 Scenario: Predictions have items for this label, but gold has none.
329 Result:
330 - Precision = 0.0 (all predictions are false positives)
331 - Recall = 1.0 (nothing to miss, vacuously true)
332 - F1 = 0.0
333 - All predicted items are false positives
335 This mirrors the edge case handling in ArtefactNode.evaluate()
336 when gold_items.items is empty but self.items is not.
338 Note: This typically indicates either:
339 1. Model hallucinated a label class
340 2. Mismatch in label naming conventions
341 3. Gold annotation is incomplete
342 Consider logging a warning when this occurs.
343 """
344 predicted_ids = list(input_node.items.keys())
346 return {
347 "types": list(
348 set([type(item).__name__ for item in input_node.items.values()])
349 ),
350 "precision": 0.0, # All predictions are wrong
351 "recall": 1.0, # Nothing to find = found everything
352 "f1": 0.0,
353 "mean_score": 0.0,
354 "num_predicted": len(predicted_ids),
355 "num_gold": 0,
356 "num_matched": 0,
357 "num_false_positives": len(predicted_ids),
358 "num_false_negatives": 0,
359 "matches": [],
360 "unmatched_predicted": predicted_ids,
361 "unmatched_gold": [],
362 }
364 def _evaluate_file_pair(
365 self, input_path: Path, gold_path: Path, threshold: float = 0.5
366 ) -> FileEvaluationResult:
367 """
368 Evaluate a single input file against its gold standard.
370 Process:
371 1. Load both files into label -> ArtefactNode mappings
372 2. For each label present in gold, evaluate predictions
373 3. Handle labels present in only one file (missing predictions or extra labels)
375 Returns:
376 FileEvaluationResult containing per-label metrics
377 """
378 input_labels = self._load_collectra_file(input_path)
379 gold_labels = self._load_collectra_file(gold_path)
381 file_result = FileEvaluationResult(filename=input_path.stem, label_metrics={})
383 all_labels = {
384 l
385 for l in (set(input_labels.keys()) | set(gold_labels.keys()))
386 if not l.endswith("_draft") and (self.labels is None or l in self.labels)
387 }
389 for label_name in all_labels:
390 input_node = input_labels.get(label_name)
391 gold_node = gold_labels.get(label_name)
393 if input_node is None and gold_node is None:
394 # This should not happen since label is in all_labels
395 continue
396 elif input_node is None:
397 # Missing predictions for this label
398 metrics = self._create_missing_prediction_metrics(gold_node)
399 elif gold_node is None:
400 # Extra predictions for non-existent gold label
401 metrics = self._create_extra_prediction_metrics(input_node)
402 else:
403 # Both exist - use ArtefactNode.evaluate()
404 metrics = input_node.evaluate(gold_node, threshold=threshold)
406 file_result.label_metrics[label_name] = metrics
408 return file_result
410 def _aggregate_results(self) -> dict:
411 """
412 Compute aggregate metrics across all files.
414 Aggregation strategy:
415 - Macro average: Average of per-file metrics (treats each file equally)
416 - Micro average: Pool all TP/FP/FN then compute (treats each item equally)
417 - Per-label aggregates: Metrics grouped by label class
418 """
419 from collections import defaultdict
421 metrics = ["precision", "recall", "f1", "mean_score"]
422 data_types_set = set()
423 for result in self.results:
424 for label_name, label_metrics in result.label_metrics.items():
425 data_types = label_metrics.get("types", [])
426 for data_type in data_types:
427 data_types_set.add(data_type)
429 data_types = list(data_types_set)
431 aggregated_results = dict()
433 for data_type in data_types:
434 # Initialize accumulators
435 label_pools = defaultdict(
436 lambda: {
437 "total_tp": 0,
438 "total_fp": 0,
439 "total_fn": 0,
440 "total_score": 0.0,
441 "total_matches": 0,
442 }
443 )
445 file_f1_scores = []
447 for file_result in self.results:
448 file_tp = file_fp = file_fn = 0
449 for label_name, metrics in file_result.label_metrics.items():
450 if data_type not in metrics.get("types", []):
451 continue
452 pool = label_pools[label_name]
453 pool["total_tp"] += metrics["num_matched"]
454 pool["total_fp"] += metrics["num_false_positives"]
455 pool["total_fn"] += metrics["num_false_negatives"]
456 pool["total_score"] += (
457 metrics["mean_score"] * metrics["num_matched"]
458 )
459 pool["total_matches"] += metrics["num_matched"]
460 pool["types"] = metrics.get("types", [])
462 file_tp += metrics["num_matched"]
463 file_fp += metrics["num_false_positives"]
464 file_fn += metrics["num_false_negatives"]
466 # Per-file F1
467 file_precision = (
468 file_tp / (file_tp + file_fp) if (file_tp + file_fp) > 0 else 0
469 )
470 file_recall = (
471 file_tp / (file_tp + file_fn) if (file_tp + file_fn) > 0 else 0
472 )
473 file_f1 = (
474 2 * file_precision * file_recall / (file_precision + file_recall)
475 if (file_precision + file_recall) > 0
476 else 0
477 )
478 # Issues 2 & 3: skip files with no items of this type (vacuous truth)
479 if (file_tp + file_fp + file_fn) == 0:
480 logger.warning(
481 "Skipping vacuous file-level metrics for '%s' (no items of "
482 "type '%s'). Excluded from file-averaged F1.",
483 file_result.filename,
484 data_type,
485 )
486 # Issue 5: skip files with predictions but no gold items for this type
487 elif (file_tp + file_fn) == 0:
488 logger.warning(
489 "File '%s' has predictions but no gold items for type '%s'. "
490 "Excluding from file-averaged F1 to avoid inflating recall.",
491 file_result.filename,
492 data_type,
493 )
494 else:
495 file_f1_scores.append(file_f1)
497 # Compute per-label micro metrics
498 per_label_metrics = {}
499 for label_name, pool in label_pools.items():
500 tp, fp, fn = pool["total_tp"], pool["total_fp"], pool["total_fn"]
501 precision = tp / (tp + fp) if (tp + fp) > 0 else 0
502 recall = tp / (tp + fn) if (tp + fn) > 0 else 0
503 f1 = (
504 2 * precision * recall / (precision + recall)
505 if (precision + recall) > 0
506 else 0
507 )
508 mean_score = (
509 pool["total_score"] / pool["total_matches"]
510 if pool["total_matches"] > 0
511 else 0
512 )
514 per_label_metrics[label_name] = {
515 "precision": precision,
516 "recall": recall,
517 "f1": f1,
518 "mean_score": mean_score,
519 "support": pool["total_tp"]
520 + pool["total_fn"], # Number of gold items
521 }
523 # Overall metrics
524 total_tp = sum(p["total_tp"] for p in label_pools.values())
525 total_fp = sum(p["total_fp"] for p in label_pools.values())
526 total_fn = sum(p["total_fn"] for p in label_pools.values())
528 overall_precision = (
529 total_tp / (total_tp + total_fp) if (total_tp + total_fp) > 0 else 0
530 )
531 overall_recall = (
532 total_tp / (total_tp + total_fn) if (total_tp + total_fn) > 0 else 0
533 )
534 overall_f1_micro = (
535 2
536 * overall_precision
537 * overall_recall
538 / (overall_precision + overall_recall)
539 if (overall_precision + overall_recall) > 0
540 else 0
541 )
542 overall_f1_file_averaged = (
543 sum(file_f1_scores) / len(file_f1_scores) if file_f1_scores else 0
544 )
545 aggregated_results[data_type] = {
546 "overall": {
547 "precision": overall_precision,
548 "recall": overall_recall,
549 "f1_micro": overall_f1_micro,
550 "f1_file_averaged": overall_f1_file_averaged,
551 "total_files": len(self.results),
552 },
553 "per_label": per_label_metrics,
554 }
556 return aggregated_results
558 def evaluate(self, threshold: float = 0.5) -> EvaluationReport:
559 """
560 Execute full evaluation across all matched file pairs.
562 Raises:
563 EvaluationError: If any input file lacks a corresponding gold file
564 """
565 self.results = []
566 matched_pairs, missing_gold, extra_gold = self._validate_file_pairs()
568 # Fail fast if gold files are missing
569 if missing_gold:
570 raise EvaluationError(f"Missing gold standard files for: {missing_gold}")
572 # Warn about extra gold files (not an error)
573 if extra_gold:
574 logger.warning(f"Gold files without predictions: {extra_gold}")
576 # Evaluate each pair
577 for input_path, gold_path in matched_pairs:
578 file_result = self._evaluate_file_pair(input_path, gold_path, threshold)
579 self.results.append(file_result)
581 # Aggregate across all files
582 self.aggregate_metrics = self._aggregate_results()
584 return EvaluationReport(
585 file_results=self.results,
586 aggregate=self.aggregate_metrics,
587 threshold=threshold,
588 )