Buckets:
| import unittest | |
| import json | |
| import hashlib | |
| from pathlib import Path | |
| from unittest.mock import patch | |
| import numpy as np | |
| from PIL import Image | |
| from evaluation.calibrate_thresholds import calibrate_thresholds | |
| from evaluation.manifest import _choose, select_quick_eval | |
| from evaluation.metrics import evaluate_detection, evaluate_segmentation | |
| from load_dataset import load_sample | |
| ROOT = Path(__file__).resolve().parents[2] | |
| CORPUS = ROOT / "datasets" / "corpus" | |
| DETECTION_CLASSES = {3, 4, 5, 6, 7, 8} | |
| SEGMENTATION_CLASSES = {1, 2, 12, 13, 14, 15, 16} | |
| class QuickEvaluationManifestTests(unittest.TestCase): | |
| def setUpClass(cls) -> None: | |
| cls.manifest = select_quick_eval(CORPUS) | |
| def test_manifest_has_fixed_validation_coverage_and_provenance(self) -> None: | |
| quick_eval = self.manifest["quick_eval"] | |
| detection = quick_eval["detection"] | |
| segmentation = quick_eval["segmentation"] | |
| self.assertEqual(len(detection), 32) | |
| self.assertEqual(len(segmentation), 20) | |
| self.assertTrue(all(row["split"] == "val" for row in detection + segmentation)) | |
| self.assertTrue( | |
| all(row["provenance"] == "quick_eval" for row in detection + segmentation) | |
| ) | |
| self.assertTrue(all(row.get("sample_id") and row.get("scene_id") for row in detection + segmentation)) | |
| self.assertTrue( | |
| all(not Path(row[key]).is_absolute() for row in detection for key in ("image", "labels")) | |
| ) | |
| self.assertTrue( | |
| all( | |
| not Path(row[key]).is_absolute() | |
| for row in segmentation | |
| for key in ("image", "mask") | |
| ) | |
| ) | |
| detected_classes = set() | |
| for row in detection: | |
| labels = json.loads((CORPUS / row["labels"]).read_text()) | |
| detected_classes.update(annotation["category_id"] for annotation in labels["annotations"]) | |
| self.assertEqual(detected_classes, DETECTION_CLASSES) | |
| segmented_classes = set() | |
| class_15_tiles = [] | |
| for row in segmentation: | |
| pixels = row["pixels"] | |
| with Image.open(CORPUS / row["mask"]) as image: | |
| ids, counts = np.unique(np.asarray(image), return_counts=True) | |
| self.assertEqual(pixels, {str(int(key)): int(count) for key, count in zip(ids, counts)}) | |
| segmented_classes.update(int(class_id) for class_id, count in pixels.items() if int(class_id) != 255 and count > 0) | |
| if int(pixels.get("15", 0)) > 0: | |
| class_15_tiles.append(row["sample_id"]) | |
| self.assertEqual(segmented_classes, SEGMENTATION_CLASSES) | |
| self.assertEqual(class_15_tiles, ["oem_kagera_37_0_0"]) | |
| self.assertEqual( | |
| self.manifest["coverage_summary"]["sparse_evidence"][ | |
| "validation_positive_tiles" | |
| ], | |
| 1, | |
| ) | |
| self.assertGreater( | |
| self.manifest["coverage_summary"]["segmentation_classes"]["15"][ | |
| "positive_pixels" | |
| ], | |
| 0, | |
| ) | |
| self.assertGreaterEqual( | |
| sum(row["negative_for_annotated_classes"] for row in detection), 2 | |
| ) | |
| self.assertGreaterEqual(sum(row["ignored_instances"] > 0 for row in detection), 2) | |
| def test_manifest_is_deterministic_and_calibration_rows_are_disjoint(self) -> None: | |
| self.assertEqual(self.manifest, select_quick_eval(CORPUS)) | |
| quick_ids = { | |
| row["sample_id"] | |
| for rows in self.manifest["quick_eval"].values() | |
| for row in rows | |
| } | |
| calibration_rows = self.manifest["selection_validation"] | |
| calibration_ids = {row["sample_id"] for row in calibration_rows} | |
| self.assertTrue(quick_ids.isdisjoint(calibration_ids)) | |
| self.assertTrue( | |
| all( | |
| row["split"] == "val" | |
| and row["provenance"] == "selection_validation" | |
| and row.get("sample_id") | |
| and row.get("scene_id") | |
| for row in calibration_rows | |
| ) | |
| ) | |
| self.assertTrue( | |
| all( | |
| not Path(row[key]).is_absolute() | |
| for row in calibration_rows | |
| for key in ("image", "labels", "mask") | |
| if key in row | |
| ) | |
| ) | |
| self.assertEqual(len(calibration_rows), 89) | |
| validation_ids = { | |
| row["id"] | |
| for row in map(json.loads, (CORPUS / "metadata" / "samples.jsonl").read_text().splitlines()) | |
| if row["split"] == "val" | |
| } | |
| self.assertEqual(quick_ids | calibration_ids, validation_ids) | |
| self.assertEqual(len(quick_ids), 52) | |
| saved = json.loads((ROOT / "evaluation" / "quick_eval_manifest.json").read_text()) | |
| self.assertEqual(self.manifest, saved) | |
| def test_selector_accepts_exactly_enough_negative_candidates(self) -> None: | |
| for hard in (False, True): | |
| samples = [ | |
| { | |
| "metadata": { | |
| "id": str(index), "group": str(index), "task": "detection", | |
| "negative_for_annotated_classes": not hard, | |
| "ignored_instances": int(hard), | |
| }, | |
| "annotations": [], "ignored_annotations": [], | |
| } | |
| for index in range(2) | |
| ] | |
| with self.subTest(hard=hard): | |
| chosen = _choose( | |
| samples, 2, 1, set(), | |
| empty_minimum=0 if hard else 2, | |
| hard_negative_minimum=2 if hard else 0, | |
| ) | |
| self.assertEqual(len(chosen), 2) | |
| def test_selector_rejects_counts_too_small_for_class_coverage(self) -> None: | |
| samples = [ | |
| { | |
| "metadata": { | |
| "id": "class-3", | |
| "group": "scene-3", | |
| "task": "detection", | |
| "negative_for_annotated_classes": False, | |
| "ignored_instances": 0, | |
| }, | |
| "annotations": [{"category_id": 3}], | |
| "ignored_annotations": [], | |
| }, | |
| { | |
| "metadata": { | |
| "id": "class-4", | |
| "group": "scene-4", | |
| "task": "detection", | |
| "negative_for_annotated_classes": False, | |
| "ignored_instances": 0, | |
| }, | |
| "annotations": [{"category_id": 4}], | |
| "ignored_annotations": [], | |
| }, | |
| ] | |
| with self.assertRaisesRegex(ValueError, "too small"): | |
| _choose(samples, 1, 20261024, {3, 4}) | |
| class DetectionMetricTests(unittest.TestCase): | |
| def setUp(self) -> None: | |
| self.polygon = [0, 0, 10, 0, 10, 10, 0, 10] | |
| def test_perfect_obb_match_has_perfect_metrics(self) -> None: | |
| target = { | |
| "sample_id": "tile", | |
| "annotations": [{"category_id": 3, "obb_polygon": self.polygon}], | |
| "ignored_annotations": [], | |
| } | |
| prediction = { | |
| "sample_id": "tile", | |
| "annotations": [ | |
| { | |
| "category_id": 3, | |
| "confidence": 0.99, | |
| "obb_polygon": self.polygon, | |
| } | |
| ], | |
| } | |
| metrics = evaluate_detection([prediction], [target]) | |
| for key in ("precision", "recall", "f1", "ap50", "map50_95"): | |
| self.assertEqual(metrics[key], 1.0, key) | |
| def test_false_positive_lowers_precision(self) -> None: | |
| target = { | |
| "sample_id": "tile", | |
| "annotations": [{"category_id": 3, "obb_polygon": self.polygon}], | |
| "ignored_annotations": [], | |
| } | |
| prediction = { | |
| "sample_id": "tile", | |
| "annotations": [ | |
| {"category_id": 3, "confidence": 0.99, "obb_polygon": self.polygon}, | |
| { | |
| "category_id": 3, | |
| "confidence": 0.9, | |
| "obb_polygon": [20, 20, 30, 20, 30, 30, 20, 30], | |
| }, | |
| ], | |
| } | |
| self.assertLess( | |
| evaluate_detection([prediction], [target])["precision"], 1.0 | |
| ) | |
| def test_ignored_region_prediction_is_not_a_false_positive(self) -> None: | |
| targets = [ | |
| { | |
| "sample_id": "positive", | |
| "annotations": [{"category_id": 3, "obb_polygon": self.polygon}], | |
| "ignored_annotations": [], | |
| }, | |
| { | |
| "sample_id": "empty", | |
| "annotations": [], | |
| "ignored_annotations": [ | |
| { | |
| "category_id": 3, | |
| "clipped_polygon": [20, 20, 30, 20, 30, 30, 20, 30], | |
| } | |
| ], | |
| }, | |
| ] | |
| predictions = [ | |
| { | |
| "sample_id": "positive", | |
| "annotations": [ | |
| {"category_id": 3, "confidence": 0.99, "obb_polygon": self.polygon} | |
| ], | |
| }, | |
| { | |
| "sample_id": "empty", | |
| "annotations": [ | |
| { | |
| "category_id": 3, | |
| "confidence": 0.9, | |
| "obb_polygon": [21, 21, 29, 21, 29, 29, 21, 29], | |
| } | |
| ], | |
| }, | |
| ] | |
| metrics = evaluate_detection(predictions, targets) | |
| self.assertEqual(metrics["false_positives"], 0) | |
| self.assertEqual(metrics["precision"], 1.0) | |
| def test_empty_tile_with_no_predictions_has_no_false_positives(self) -> None: | |
| metrics = evaluate_detection( | |
| [{"sample_id": "empty", "annotations": []}], | |
| [ | |
| { | |
| "sample_id": "empty", | |
| "annotations": [], | |
| "ignored_annotations": [], | |
| } | |
| ], | |
| ) | |
| self.assertEqual(metrics["false_positives"], 0) | |
| def test_ignored_unsupported_class_region_suppresses_false_positive(self) -> None: | |
| metrics = evaluate_detection( | |
| [{"sample_id": "tile", "annotations": [ | |
| {"category_id": 3, "confidence": 0.9, "obb_polygon": self.polygon} | |
| ]}], | |
| [{"sample_id": "tile", "annotations": [], "ignored_annotations": [ | |
| {"category_id": None, "clipped_polygon": self.polygon} | |
| ]}], | |
| ) | |
| self.assertEqual(metrics["false_positives"], 0) | |
| def test_oriented_iou_and_ap_are_not_axis_aligned_matches(self) -> None: | |
| diamond = [5, 0, 10, 5, 5, 10, 0, 5] | |
| metrics = evaluate_detection( | |
| [{"sample_id": "tile", "annotations": [ | |
| {"category_id": 3, "confidence": 0.9, "obb_polygon": self.polygon} | |
| ]}], | |
| [{"sample_id": "tile", "annotations": [ | |
| {"category_id": 3, "obb_polygon": diamond} | |
| ]}], | |
| ) | |
| self.assertEqual(metrics["ap50"], 1.0) | |
| self.assertEqual(metrics["map50_95"], 0.1) | |
| def test_high_confidence_false_positive_reduces_ap(self) -> None: | |
| metrics = evaluate_detection( | |
| [{"sample_id": "tile", "annotations": [ | |
| {"category_id": 3, "confidence": 0.99, | |
| "obb_polygon": [20, 20, 30, 20, 30, 30, 20, 30]}, | |
| {"category_id": 3, "confidence": 0.8, "obb_polygon": self.polygon}, | |
| ]}], | |
| [{"sample_id": "tile", "annotations": [ | |
| {"category_id": 3, "obb_polygon": self.polygon} | |
| ]}], | |
| ) | |
| self.assertEqual(metrics["precision"], 0.5) | |
| self.assertEqual(metrics["ap50"], 0.5) | |
| def test_duplicate_predictions_match_each_target_only_once(self) -> None: | |
| annotation = {"category_id": 3, "confidence": 0.9, "obb_polygon": self.polygon} | |
| metrics = evaluate_detection( | |
| [{"sample_id": "tile", "annotations": [annotation, annotation]}], | |
| [{"sample_id": "tile", "annotations": [annotation]}], | |
| ) | |
| self.assertEqual(metrics["true_positives"], 1) | |
| self.assertEqual(metrics["false_positives"], 1) | |
| self.assertEqual(metrics["recall"], 1.0) | |
| class SegmentationMetricTests(unittest.TestCase): | |
| def test_masks_require_two_dimensional_integer_class_ids(self) -> None: | |
| for mask in (np.ones((1, 2, 3)), np.array([[1.5, 2.0]])): | |
| with self.subTest(shape=mask.shape), self.assertRaises(ValueError): | |
| evaluate_segmentation(mask, mask) | |
| def test_ignore_prediction_on_labeled_pixel_is_rejected(self) -> None: | |
| with self.assertRaises(ValueError): | |
| evaluate_segmentation(np.array([[255]]), np.array([[1]])) | |
| def test_all_ignored_pixels_have_no_classes(self) -> None: | |
| result = evaluate_segmentation(np.array([[1]]), np.array([[255]])) | |
| self.assertEqual(result, {"per_class_iou": {}, "miou": 0.0, "valid_pixels": 0}) | |
| def test_iou_excludes_ignore_pixels(self) -> None: | |
| prediction = np.array([[1, 2, 2, 1]], dtype=np.uint8) | |
| target = np.array([[1, 1, 2, 255]], dtype=np.uint8) | |
| metrics = evaluate_segmentation(prediction, target) | |
| self.assertEqual(metrics["per_class_iou"], {"1": 0.5, "2": 0.5}) | |
| self.assertEqual(metrics["miou"], 0.5) | |
| self.assertEqual(metrics["valid_pixels"], 3) | |
| class ThresholdCalibrationTests(unittest.TestCase): | |
| def test_calibration_filters_low_confidence_false_positives(self) -> None: | |
| metadata = { | |
| "sample_id": "dota_P0007_466_819", "split": "val", | |
| "provenance": "selection_validation", | |
| } | |
| polygon = [0, 0, 10, 0, 10, 10, 0, 10] | |
| target = dict(metadata, annotations=[{"category_id": 3, "obb_polygon": polygon}]) | |
| prediction = dict(metadata, annotations=[ | |
| {"category_id": 3, "confidence": 0.8, "obb_polygon": polygon}, | |
| {"category_id": 3, "confidence": 0.2, | |
| "obb_polygon": [20, 20, 30, 20, 30, 30, 20, 30]}, | |
| ]) | |
| result = calibrate_thresholds([prediction], [target]) | |
| self.assertEqual(result["thresholds"], {"3": 0.8}) | |
| self.assertEqual(result["f1"], {"3": 1.0}) | |
| self.assertEqual(result["sparse_class_ids"], ["3"]) | |
| def test_calibration_rejects_unknown_and_segmentation_ids(self) -> None: | |
| manifest = json.loads((ROOT / "evaluation" / "quick_eval_manifest.json").read_text()) | |
| segmentation_id = next( | |
| row["sample_id"] for row in manifest["selection_validation"] | |
| if row["task"] == "segmentation" | |
| ) | |
| for sample_id in ("unknown", segmentation_id): | |
| row = {"sample_id": sample_id, "split": "val", | |
| "provenance": "selection_validation", "annotations": []} | |
| with self.subTest(sample_id=sample_id), self.assertRaises(ValueError): | |
| calibrate_thresholds([row], [row]) | |
| def test_calibration_rejects_forged_quick_eval_membership(self) -> None: | |
| manifest = json.loads((ROOT / "evaluation" / "quick_eval_manifest.json").read_text()) | |
| row = dict(manifest["quick_eval"]["detection"][0]) | |
| row.update(provenance="selection_validation", annotations=[]) | |
| with self.assertRaisesRegex(ValueError, "selection_validation"): | |
| calibrate_thresholds([row], [row]) | |
| def test_calibration_uses_selection_validation_records(self) -> None: | |
| polygon = [0, 0, 10, 0, 10, 10, 0, 10] | |
| prediction = { | |
| "sample_id": "dota_P0007_466_819", | |
| "split": "val", | |
| "provenance": "selection_validation", | |
| "annotations": [ | |
| {"category_id": 3, "confidence": 0.8, "obb_polygon": polygon} | |
| ], | |
| } | |
| target = { | |
| "sample_id": "dota_P0007_466_819", | |
| "split": "val", | |
| "provenance": "selection_validation", | |
| "annotations": [{"category_id": 3, "obb_polygon": polygon}], | |
| "ignored_annotations": [], | |
| } | |
| result = calibrate_thresholds([prediction], [target]) | |
| self.assertEqual(result["provenance"], "selection_validation") | |
| self.assertEqual(result["sample_ids"], ["dota_P0007_466_819"]) | |
| self.assertEqual(result["thresholds"], {"3": 0.8}) | |
| self.assertEqual(result["f1"], {"3": 1.0}) | |
| def test_calibration_rejects_records_outside_selection_validation(self) -> None: | |
| prediction = { | |
| "sample_id": "tile", | |
| "split": "val", | |
| "provenance": "quick_eval", | |
| "annotations": [], | |
| } | |
| target = { | |
| "sample_id": "tile", | |
| "split": "val", | |
| "provenance": "quick_eval", | |
| "annotations": [], | |
| "ignored_annotations": [], | |
| } | |
| with self.assertRaisesRegex(ValueError, "selection_validation"): | |
| calibrate_thresholds([prediction], [target]) | |
| def test_calibration_rejects_train_test_and_missing_provenance(self) -> None: | |
| for split, provenance in (("train", "selection_validation"), ("test", "selection_validation"), ("val", None)): | |
| row = {"sample_id": "dota_P0007_466_819", "split": split, | |
| "provenance": provenance, "annotations": []} | |
| with self.subTest(split=split), self.assertRaises(ValueError): | |
| calibrate_thresholds([row], [row]) | |
| class LoaderIntegrityTests(unittest.TestCase): | |
| def test_missing_approved_hash_is_rejected(self) -> None: | |
| with patch.object(Path, "read_bytes", return_value=b"binary\r\n"): | |
| with self.assertRaisesRegex(ValueError, "approved"): | |
| load_sample(CORPUS, {"image": "image.png"}, expected_hashes={}) | |
| def test_binary_hash_is_never_newline_normalized(self) -> None: | |
| with patch.object(Path, "read_bytes", return_value=b"binary\r\n"): | |
| with self.assertRaisesRegex(ValueError, "approved"): | |
| load_sample( | |
| CORPUS, {"image": "image.png"}, | |
| expected_hashes={"image.png": hashlib.sha256(b"binary\n").hexdigest()}, | |
| ) | |
| if __name__ == "__main__": | |
| unittest.main() | |
Xet Storage Details
- Size:
- 18.6 kB
- Xet hash:
- c03e6afefe7b9574dbd0a81c5ae7ebe3411b25645eb7272ce76407e5d8a65fae
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.