Buckets:
| import unittest | |
| from pathlib import Path | |
| import numpy as np | |
| def annotation(name="plane", *, difficult=False, visible=1.0): | |
| return {"original_class": name, "source_annotation_id": "scene:1", | |
| "obb_polygon": [100, 200, 112, 200, 112, 208, 100, 208], | |
| "clipped_polygon": [0, 0, 12, 0, 12, 8, 0, 8], | |
| "bbox": [0, 0, 12, 8], "source_difficult": difficult, | |
| "visible_fraction": visible, "category_id": None} | |
| def sample(annotations=(), ignored=()): | |
| return {"image": np.full((32, 32, 3), 100, np.uint8), | |
| "metadata": {"id": "scene", "task": "detection", "split": "train", | |
| "tile_coordinates": [100, 200, 132, 232], | |
| "valid_width": 30, "valid_height": 30}, | |
| "annotations": list(annotations), "ignored_annotations": list(ignored)} | |
| class DetectionDataTests(unittest.TestCase): | |
| def test_unmapped_dota_categories_are_auxiliary_targets(self): | |
| from training.detection_data import SOURCE_CLASSES, convert_detection_sample | |
| from training.model_factory import DOTA_NAMES, public_mapping | |
| self.assertEqual(tuple(SOURCE_CLASSES), DOTA_NAMES) | |
| example = convert_detection_sample(sample(ignored=[annotation(n) for n in DOTA_NAMES])) | |
| self.assertEqual(example.class_ids.tolist(), list(range(16))) | |
| self.assertEqual(len(example.ignore_polygons), 0) | |
| self.assertTrue(example.valid_mask[2, 2]) | |
| self.assertEqual(set(public_mapping(dict(enumerate(DOTA_NAMES)), "detection").values()), | |
| {3, 4, 5, 6, 7, 8}) | |
| def test_original_coordinates_and_clipped_nonquadrilateral(self): | |
| from training.detection_data import convert_detection_sample | |
| a = annotation() | |
| example = convert_detection_sample(sample([a])) | |
| np.testing.assert_allclose(example.polygons[0], np.array(a["obb_polygon"]).reshape(4, 2) - [100, 200]) | |
| a["clipped_polygon"] = [0, 0, 10, 0, 12, 2, 12, 8, 0, 8] | |
| a["visible_fraction"] = .9 | |
| example = convert_detection_sample(sample([a])) | |
| self.assertEqual(example.polygons.shape, (1, 4, 2)) | |
| self.assertGreater(example.boxes[0, 2], 0) | |
| self.assertGreater(example.boxes[0, 3], 0) | |
| def test_rotated_geometry_round_trip_and_real_clip_extent(self): | |
| import cv2 | |
| from shapely.geometry import Polygon | |
| from training.detection_data import convert_detection_sample | |
| a = annotation() | |
| local = cv2.boxPoints(((15., 15.), (14., 6.), 27.)) | |
| a.update(obb_polygon=(local + [100, 200]).reshape(-1).tolist(), | |
| clipped_polygon=local.reshape(-1).tolist(), bbox=[7, 7, 16, 16]) | |
| e = convert_detection_sample(sample([a])) | |
| x, y, w, h, angle = e.boxes[0] | |
| reconstructed = cv2.boxPoints(((float(x), float(y)), (float(w), float(h)), float(np.rad2deg(angle)))) | |
| self.assertGreater(Polygon(local).intersection(Polygon(reconstructed)).area / Polygon(local).union(Polygon(reconstructed)).area, .99999) | |
| a.update(visible_fraction=.9, clipped_polygon=[0, 0, 10, 0, 12, 2, 12, 8, 0, 8]) | |
| e = convert_detection_sample(sample([a])) | |
| np.testing.assert_allclose(e.boxes[0, :2], [6, 4], atol=1e-5) | |
| def test_difficult_fragments_padding_and_black_are_ignored(self): | |
| from training.detection_data import convert_detection_sample | |
| s = sample(ignored=[annotation(difficult=True), annotation(visible=.5)]) | |
| s["image"][20, 20] = 0 | |
| e = convert_detection_sample(s) | |
| self.assertEqual(len(e.class_ids), 0) | |
| self.assertEqual(len(e.ignore_polygons), 2) | |
| self.assertFalse(e.valid_mask[2, 2]) | |
| self.assertFalse(e.valid_mask[31, 31]) | |
| self.assertFalse(e.valid_mask[20, 20]) | |
| def test_transform_keeps_geometry_and_mask_aligned(self): | |
| from training.detection_data import convert_detection_sample, transform_example | |
| e = convert_detection_sample(sample([annotation()])) | |
| transformed = transform_example(e, horizontal=True, rotations=1) | |
| restored = transform_example(transform_example(transformed, rotations=3), horizontal=True) | |
| np.testing.assert_array_equal(restored.image, e.image) | |
| np.testing.assert_array_equal(restored.valid_mask, e.valid_mask) | |
| np.testing.assert_allclose(restored.polygons, e.polygons, atol=1e-5) | |
| def test_dior_partial_coverage_preserves_positives_and_masks_unknown_classes(self): | |
| from training.detection_data import convert_detection_sample,detection_batch | |
| s=sample([annotation()],ignored=[{**annotation('unknown'),'training_ignore':True}]) | |
| s['metadata'].update(source_dataset='dior-r',annotated_source_classes=[0]) | |
| e=convert_detection_sample(s) | |
| self.assertTrue(e.valid_mask[2,2]) | |
| self.assertFalse(e.valid_mask[31,31]) | |
| batch=detection_batch([e]) | |
| self.assertEqual(batch['class_valid_mask'].sum().item(),1) | |
| self.assertTrue(batch['class_valid_mask'][0,0]) | |
| s['metadata']['annotated_source_classes']=[2] | |
| with self.assertRaises(ValueError):detection_batch([convert_detection_sample(s)]) | |
| def test_unknown_classes_fail_closed(self): | |
| from training.detection_data import convert_detection_sample | |
| with self.assertRaises(ValueError): | |
| convert_detection_sample(sample([annotation("unknown")])) | |
| class DetectionLossTests(unittest.TestCase): | |
| def test_ignored_polygon_anchors_have_zero_classification_weight(self): | |
| import torch | |
| from training.detection_data import convert_detection_sample | |
| from training.detection_loss import anchor_classification_weights | |
| e = convert_detection_sample(sample(ignored=[annotation(difficult=True)])) | |
| points = torch.tensor([[2., 2.], [20., 20.], [31., 31.], [-1., 0.]]) | |
| weights = anchor_classification_weights(points, torch.from_numpy(e.valid_mask)[None]) | |
| self.assertEqual(weights.tolist(), [[0., 1., 0., 0.]]) | |
| def test_actual_head_transfer_and_masked_end_to_end_loss(self): | |
| import torch | |
| from training.detection_loss import build_adapted_detector, SourceAwareOBBLoss | |
| from training.model_factory import load_pretrained_detector, DOTA_NAMES, detector_transfer_plan | |
| torch.set_num_threads(4) | |
| source = load_pretrained_detector(Path("training/cache/checkpoints")) | |
| source_head = source.model.model[-1] | |
| before = source_head.cv3[0][-1].weight.detach().clone() | |
| model, report = build_adapted_detector(source, seed=42) | |
| self.assertEqual(model.model[-1].nc, 16) | |
| self.assertEqual(model.names, dict(enumerate(DOTA_NAMES))) | |
| self.assertEqual(report["missing_classes"], ["container-crane"]) | |
| for branch in ("cv3", "one2one_cv3"): | |
| for old, new in zip(getattr(source_head, branch), getattr(model.model[-1], branch)): | |
| for target, name in enumerate(DOTA_NAMES[:-1]): | |
| idx = detector_transfer_plan(source.names)["source_indices"][name] | |
| torch.testing.assert_close(new[-1].weight[target], old[-1].weight[idx]) | |
| torch.testing.assert_close(source_head.cv3[0][-1].weight, before) | |
| model.train() | |
| batch = {"img": torch.zeros(1, 3, 64, 64), "batch_idx": torch.empty(0), | |
| "cls": torch.empty(0, 1), "bboxes": torch.empty(0, 5), | |
| "valid_mask": torch.zeros(1, 64, 64, dtype=torch.bool)} | |
| predictions = model(batch["img"]) | |
| criterion = SourceAwareOBBLoss(model) | |
| loss, _ = criterion(predictions, batch) | |
| self.assertEqual(loss.sum().detach().item(), 0.) | |
| loss.sum().backward() | |
| gradients = [p.grad for p in model.parameters() if p.grad is not None] | |
| self.assertTrue(gradients) | |
| self.assertTrue(all(torch.isfinite(g).all() and not g.any() for g in gradients)) | |
| # With every anchor valid, the adapter must preserve the stock criterion. | |
| from ultralytics.utils.loss import E2ELoss, v8OBBLoss | |
| batch["valid_mask"].fill_(True) | |
| batch["batch_idx"] = torch.tensor([0.]) | |
| batch["cls"] = torch.tensor([[0.]]) | |
| batch["bboxes"] = torch.tensor([[.5, .5, .4, .2, .3]]) | |
| reference, _ = E2ELoss(model, v8OBBLoss)(predictions, batch) | |
| actual, _ = criterion(predictions, batch) | |
| torch.testing.assert_close(actual, reference) | |
| self.assertGreater(actual.sum().detach().item(), 0.) | |
| # Ignore masking applies to both branches and both target/background scores. | |
| batch["valid_mask"][:, :, :32] = False | |
| changed = {k: {name: value.detach().clone() if torch.is_tensor(value) else value | |
| for name, value in branch.items()} for k, branch in predictions.items()} | |
| from ultralytics.utils.tal import make_anchors | |
| for branch in changed.values(): | |
| points, strides = make_anchors(branch["feats"], model.stride, .5) | |
| invalid = (points * strides)[:, 0] < 32 | |
| branch["scores"][:, :, invalid] += 100 | |
| original_loss, _ = criterion(predictions, batch) | |
| changed_loss, _ = criterion(changed, batch) | |
| torch.testing.assert_close(original_loss, changed_loss) | |
| # Unannotated source classes must have no classification effect. | |
| batch['class_valid_mask'] = torch.zeros(1,16,dtype=torch.bool) | |
| batch['class_valid_mask'][:,0] = True | |
| unknown_changed = {k: {name: value.detach().clone() if torch.is_tensor(value) else value | |
| for name,value in branch.items()} for k,branch in predictions.items()} | |
| for branch in unknown_changed.values(): | |
| branch['scores'][:,1:,:] += 100 | |
| original_loss,_ = criterion(predictions,batch) | |
| unknown_loss,_ = criterion(unknown_changed,batch) | |
| torch.testing.assert_close(original_loss,unknown_loss) | |
| batch.pop('class_valid_mask') | |
| from copy import deepcopy | |
| predictor = deepcopy(source) | |
| predictor.model, predictor.predictor = deepcopy(model).eval(), None | |
| with torch.inference_mode(): | |
| result = predictor.predict(np.zeros((64, 64, 3), np.uint8), imgsz=64, conf=.001, | |
| device="cpu", save=False, verbose=False)[0] | |
| self.assertEqual(result.names, dict(enumerate(DOTA_NAMES))) | |
| self.assertIsNotNone(model.model[-1].cv3) | |
| # Use an empty target pool to isolate background classification invariance. | |
| batch.update(batch_idx=torch.empty(0), cls=torch.empty(0, 1), bboxes=torch.empty(0, 5)) | |
| original_loss, _ = criterion(predictions, batch) | |
| changed_loss, _ = criterion(changed, batch) | |
| torch.testing.assert_close(original_loss, changed_loss) | |
| class TrainerContractTests(unittest.TestCase): | |
| def test_synthetic_native_fit_saves_reloadable_checkpoint(self): | |
| import json | |
| import tempfile | |
| from copy import deepcopy | |
| from types import SimpleNamespace | |
| from unittest.mock import patch | |
| import torch | |
| from training.model_factory import ROOT | |
| from training.detection_trainer import DetectorTrainConfig, make_detection_trainer | |
| from ultralytics import YOLO | |
| manifest = json.loads((ROOT / "evaluation/quick_eval_manifest.json").read_text()) | |
| selected = next(r["sample_id"] for r in manifest["selection_validation"] if r["task"] == "detection") | |
| a = annotation() | |
| a.update(obb_polygon=[400, 400, 624, 400, 624, 560, 400, 560], | |
| clipped_polygon=[400, 400, 624, 400, 624, 560, 400, 560], | |
| bbox=[400, 400, 224, 160], category_id=6) | |
| generated = {"image": np.full((1024, 1024, 3), 100, np.uint8), | |
| "metadata": {"id": "synthetic", "task": "detection", "split": "train", | |
| "tile_coordinates": [0, 0, 1024, 1024], | |
| "valid_width": 1024, "valid_height": 1024}, | |
| "annotations": [a], "ignored_annotations": []} | |
| def load(root, row, hashes): | |
| result = deepcopy(generated) | |
| result["metadata"].update(id=row["id"], split=row["split"]) | |
| return result | |
| with tempfile.TemporaryDirectory(dir=ROOT / "training/cache") as directory: | |
| output = Path(directory) / "fit" | |
| config = DetectorTrainConfig(output=output, epochs=2, record_initial_metrics=True) | |
| from training.detection_trainer import torch as trainer_torch | |
| actual_adam = trainer_torch.optim.AdamW | |
| steps = [] | |
| def optimizer_factory(*args, **kwargs): | |
| optimizer = actual_adam(*args, **kwargs) | |
| step = optimizer.step | |
| def checked_step(*step_args, **step_kwargs): | |
| self.assertTrue((output / "initial_evaluation.json").is_file()) | |
| progress = json.loads((output / "progress.json").read_text()) | |
| self.assertEqual(progress["history"][0]["epoch"], 0) | |
| if steps: | |
| self.assertTrue((output / "best.pt").is_file()) | |
| self.assertTrue((output / "progress.json").is_file()) | |
| steps.append(1) | |
| return step(*step_args, **step_kwargs) | |
| optimizer.step = checked_step | |
| return optimizer | |
| with patch("training.detection_trainer.dataset", side_effect=[ | |
| SimpleNamespace(samples=[{"id": "synthetic", "split": "train"}], expected_hashes={}), | |
| SimpleNamespace(samples=[{"id": selected, "split": "val"}], expected_hashes={})]), \ | |
| patch("training.detection_trainer.load_sample", side_effect=load), \ | |
| patch("training.detection_trainer.torch.optim.AdamW", side_effect=optimizer_factory): | |
| result = make_detection_trainer(config).fit(["synthetic"], [selected], max_seconds=120) | |
| self.assertEqual(result["status"], "COMPLETED") | |
| self.assertEqual(len(result["history"]), 3) | |
| self.assertEqual(result["history"][0]["evaluation_ids"], [selected]) | |
| self.assertEqual(len(steps), 2) | |
| self.assertTrue((output / "initial_evaluation.json").is_file()) | |
| checkpoint = Path(result["best_checkpoint"]) | |
| self.assertTrue(checkpoint.is_file()) | |
| model = YOLO(str(checkpoint)) | |
| with torch.inference_mode(): | |
| predictions = model.predict(generated["image"][:, :, ::-1].copy(), imgsz=1024, | |
| conf=.001, device="cpu", save=False, verbose=False) | |
| self.assertEqual(len(predictions), 1) | |
| self.assertEqual(model.task, "obb") | |
| self.assertIsNotNone(predictions[0].obb) | |
| self.assertEqual(len(model.names), 16) | |
| self.assertEqual(model.model.args["imgsz"], 1024) | |
| self.assertEqual(model.ckpt["train_args"]["lr0"], config.learning_rate) | |
| self.assertEqual(model.ckpt["train_args"]["optimizer"], "AdamW") | |
| self.assertEqual(model.ckpt["augmentation"]["uniform_quarter_turns"], [0, 1, 2, 3]) | |
| def test_invalid_budget_and_config_fail_before_model_loading(self): | |
| from training.detection_trainer import DetectorTrainConfig, make_detection_trainer | |
| trainer = make_detection_trainer(DetectorTrainConfig()) | |
| for seconds in [0, 3601, True]: | |
| with self.assertRaises(ValueError): | |
| trainer.fit([], [], max_seconds=seconds) | |
| for kwargs in [{"epochs": 11}, {"patience": 4}, {"learning_rate": float("nan")}, | |
| {"output": Path("/tmp/not-approved")}]: | |
| with self.assertRaises(ValueError): | |
| DetectorTrainConfig(**kwargs) | |
| def test_fit_rejects_quick_and_test_ids_without_writes(self): | |
| from types import SimpleNamespace | |
| from unittest.mock import patch | |
| from training.detection_trainer import DetectorTrainConfig, make_detection_trainer | |
| fake = SimpleNamespace(samples=[{"id": "forbidden"}]) | |
| trainer = make_detection_trainer(DetectorTrainConfig()) | |
| with patch("training.detection_trainer.dataset", return_value=fake): | |
| with self.assertRaises(ValueError): | |
| trainer.fit(["forbidden"], ["forbidden"], max_seconds=1) | |
| def test_expired_fit_preserves_result_contract_without_loading_weights(self): | |
| import json | |
| import tempfile | |
| from types import SimpleNamespace | |
| from unittest.mock import patch | |
| from training.detection_trainer import DetectorTrainConfig, make_detection_trainer | |
| from training.model_factory import ROOT | |
| manifest = json.loads((ROOT / "evaluation/quick_eval_manifest.json").read_text()) | |
| selected = next(r["sample_id"] for r in manifest["selection_validation"] if r["task"] == "detection") | |
| with tempfile.TemporaryDirectory(dir=ROOT / "training/cache") as temporary: | |
| config = DetectorTrainConfig.from_json(ROOT / "training/configs/detector.json", output=Path(temporary) / "run") | |
| with patch("training.detection_trainer.dataset", side_effect=[SimpleNamespace(samples=[{"id": "train"}]), | |
| SimpleNamespace(samples=[{"id": selected}])]), \ | |
| patch("training.detection_trainer.time.monotonic", side_effect=[0, 2, 3]), \ | |
| patch("training.detection_trainer.load_pretrained_detector") as load: | |
| result = make_detection_trainer(config).fit(["train"], [selected], max_seconds=1) | |
| load.assert_not_called() | |
| self.assertEqual(set(result), {"status", "best_checkpoint", "history", "best_metrics", "elapsed_seconds"}) | |
| self.assertEqual(result["status"], "TIME_LIMIT") | |
| self.assertIsNone(result["best_checkpoint"]) | |
| self.assertEqual(json.loads((config.output / "fit.json").read_text()), result) | |
| if __name__ == "__main__": | |
| unittest.main() | |
Xet Storage Details
- Size:
- 18.1 kB
- Xet hash:
- 5d5a9e18585a1716caa9125fcb5b5ff610ee289553764c0f467f9a24361225f3
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.