Buckets:
| import os | |
| import unittest | |
| from pathlib import Path | |
| from unittest.mock import patch | |
| import numpy as np | |
| class ModelFactoryTests(unittest.TestCase): | |
| def test_model_loaders_select_cpu_and_expected_tasks(self): | |
| from training import model_factory as factory | |
| class local_model: | |
| def __init__(self, path): | |
| self.task = "obb" if "-obb" in str(path) else "semantic" | |
| self.names = {0: "plane"} if self.task == "obb" else {0: "building"} | |
| def to(self, device): | |
| self.device = device | |
| return self | |
| with patch.object(factory, "_checkpoint", side_effect=lambda cache, name: cache / name), patch.object( | |
| factory, "_yolo_class", return_value=local_model | |
| ): | |
| detector = factory.load_pretrained_detector(Path("training/cache")) | |
| segmenter = factory.load_pretrained_segmenter(Path("training/cache")) | |
| self.assertEqual((detector.task, detector.device), ("obb", "cpu")) | |
| self.assertEqual((segmenter.task, segmenter.device), ("semantic", "cpu")) | |
| self.assertTrue(Path(os.environ["YOLO_CONFIG_DIR"]).is_relative_to(factory.ROOT)) | |
| with patch("torch.cuda.is_available", return_value=True): | |
| detector = factory.load_pretrained_detector(Path("training/cache"), "cuda:0") | |
| self.assertEqual(str(detector.device), "cuda:0") | |
| def test_baseline_class_mapping_exposes_only_supported_classes(self): | |
| from training.model_factory import public_mapping | |
| mapped = public_mapping({0: "plane", 1: "helicopter", 2: "ship", 3: "small-vehicle", | |
| 4: "large-vehicle", 5: "bridge", 6: "storage-tank", 7: "tennis-court"}, "detection") | |
| self.assertEqual(mapped, {0: 6, 1: 6, 2: 7, 3: 4, 4: 5, 5: 3, 6: 8}) | |
| self.assertEqual(public_mapping({0: "building", 1: "road", 2: "water", 3: "tree", | |
| 4: "field", 5: "earth", 6: "land"}, "segmentation"), | |
| {0: 1, 1: 2, 2: 12}) | |
| def test_detector_checkpoint_classes_are_mapped_by_name(self): | |
| from training.model_factory import detector_transfer_plan, transfer_head_rows | |
| names = {0: "ship", 1: "plane", 2: "bridge"} | |
| plan = detector_transfer_plan(names) | |
| self.assertEqual(len(plan["target_names"]), 16) | |
| self.assertEqual(plan["source_indices"]["plane"], 1) | |
| self.assertEqual(plan["source_indices"]["ship"], 0) | |
| self.assertIn("container-crane", plan["missing_classes"]) | |
| source = np.array([[7, 8], [1, 2], [3, 4]]) | |
| initialized = np.full((16, 2), -1) | |
| adapted = transfer_head_rows(source, initialized, names) | |
| np.testing.assert_array_equal(adapted[0], [1, 2]) | |
| np.testing.assert_array_equal(adapted[6], [7, 8]) | |
| np.testing.assert_array_equal(adapted[15], [-1, -1]) | |
| np.testing.assert_array_equal(initialized, np.full((16, 2), -1)) | |
| def test_official_space_separated_dota_names_are_exact_name_matches(self): | |
| from training.model_factory import public_mapping, detector_transfer_plan | |
| names = {0: "small vehicle", 1: "large vehicle", 2: "storage tank", | |
| 3: "baseball diamond", 4: "container crane"} | |
| self.assertEqual(public_mapping(names, "detection"), {0: 4, 1: 5, 2: 8}) | |
| plan = detector_transfer_plan(names) | |
| self.assertEqual(plan["source_indices"]["baseball-diamond"], 3) | |
| self.assertNotIn("container-crane", plan["missing_classes"]) | |
| class QuickEvalTests(unittest.TestCase): | |
| def test_semantic_result_unwraps_dense_tensor_before_numpy_save(self): | |
| from evaluation import quick_eval | |
| from ultralytics.engine.results import Results | |
| import torch | |
| result = Results(np.zeros((2, 2, 3), dtype=np.uint8), path="tile", | |
| names={0: "building"}, semantic_mask=torch.tensor([[0, 1], [2, 3]])) | |
| raw = quick_eval.semantic_array(result) | |
| self.assertIsInstance(raw, np.ndarray) | |
| np.testing.assert_array_equal(raw, [[0, 1], [2, 3]]) | |
| def test_fast_calibration_matches_reference_and_rejects_quick_ids(self): | |
| import json | |
| from evaluation.calibrate_thresholds import calibrate_thresholds | |
| from evaluation.quick_eval import selection_thresholds | |
| manifest = json.loads(Path("evaluation/quick_eval_manifest.json").read_text()) | |
| sample_id = next(row["sample_id"] for row in manifest["selection_validation"] | |
| if row["task"] == "detection") | |
| polygon = [0, 0, 10, 0, 10, 10, 0, 10] | |
| target = {"sample_id": sample_id, "split": "val", "provenance": "selection_validation", | |
| "annotations": [{"category_id": 3, "obb_polygon": polygon}]} | |
| predicted = {**target, "annotations": [{"category_id": 3, "confidence": .3, "obb_polygon": polygon}]} | |
| self.assertEqual(selection_thresholds([predicted], [target]), calibrate_thresholds([predicted], [target])) | |
| predicted["sample_id"] = manifest["quick_eval"]["detection"][0]["sample_id"] | |
| with self.assertRaises(ValueError): | |
| selection_thresholds([predicted], [target]) | |
| def test_detection_target_translates_only_source_frame(self): | |
| from evaluation.quick_eval import detection_target | |
| sample = {"metadata": {"id": "tile", "tile_coordinates": [100, 200, 110, 210]}, | |
| "annotations": [{"category_id": 3, "obb_polygon": [100, 200, 110, 200, 110, 210, 100, 210]}], | |
| "ignored_annotations": [{"category_id": None, "clipped_polygon": [0, 0, 2, 0, 2, 2, 0, 2]}]} | |
| row = detection_target(sample, "quick_eval") | |
| self.assertEqual(row["annotations"][0]["obb_polygon"], [0, 0, 10, 0, 10, 10, 0, 10]) | |
| self.assertEqual(row["ignored_annotations"], sample["ignored_annotations"]) | |
| self.assertEqual(sample["annotations"][0]["obb_polygon"][0], 100) | |
| def test_ap_pool_is_not_truncated_by_operating_thresholds(self): | |
| from evaluation.quick_eval import detection_metrics | |
| polygon = [0, 0, 10, 0, 10, 10, 0, 10] | |
| predicted = [{"sample_id": "tile", "annotations": [ | |
| {"category_id": 3, "confidence": .2, "obb_polygon": polygon}]}] | |
| targets = [{"sample_id": "tile", "annotations": [{"category_id": 3, "obb_polygon": polygon}]}] | |
| metrics = detection_metrics(predicted, targets, {"3": .8}) | |
| self.assertEqual(metrics["ap"]["ap50"], 1.0) | |
| self.assertEqual(metrics["operating"]["recall"], 0.0) | |
| def test_repeated_iou_thresholds_reuse_polygon_geometry(self): | |
| from evaluation import metrics | |
| polygon = [0, 0, 10, 0, 10, 10, 0, 10] | |
| predicted = [{"sample_id": "tile", "annotations": [ | |
| {"category_id": 3, "confidence": .8, "obb_polygon": polygon}, | |
| {"category_id": 3, "confidence": .2, "obb_polygon": polygon}]}] | |
| targets = [{"sample_id": "tile", "annotations": [{"category_id": 3, "obb_polygon": polygon}]}] | |
| with patch.object(metrics, "_polygon", wraps=metrics._polygon) as build: | |
| result = metrics.evaluate_detection(predicted, targets) | |
| self.assertEqual(result["ap50"], 1) | |
| self.assertLessEqual(build.call_count, 2, "geometry must be reused across AP thresholds") | |
| def test_semantic_unmapped_pixels_are_errors_not_ignored(self): | |
| from evaluation.quick_eval import segmentation_counts | |
| counts = segmentation_counts(np.array([[0, 3], [1, 3]]), np.array([[1, 1], [2, 255]]), | |
| {0: 1, 1: 2}, [1, 2, 12, 13, 14, 15, 16]) | |
| self.assertEqual(counts["1"], {"intersection": 1, "union": 2, "targets": 2}) | |
| self.assertEqual(counts["13"]["intersection"], 0) | |
| def test_timeout_and_altered_manifest_fail_closed(self): | |
| from evaluation.quick_eval import run_quick_eval | |
| with self.assertRaises(ValueError): | |
| run_quick_eval("BASELINE_PRETRAINED", None, None, Path("missing.json"), | |
| Path("evaluation/invalid_manifest_output"), timeout_seconds=301) | |
| with self.assertRaises(ValueError): | |
| run_quick_eval("BASELINE_PRETRAINED", None, None, Path("missing.json"), | |
| Path("evaluation/invalid_manifest_output"), timeout_seconds=300) | |
| if __name__ == "__main__": | |
| unittest.main() | |
Xet Storage Details
- Size:
- 8.5 kB
- Xet hash:
- ebc140af227444a74bf871132392c283ec988e07490fdd16528a14a98ab9a4d7
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.