from __future__ import annotations import importlib.util from pathlib import Path SCRIPT = Path(__file__).parents[2] / "scripts" / "build_failure_driven_yolo_sampling.py" SPEC = importlib.util.spec_from_file_location("failure_sampling", SCRIPT) assert SPEC and SPEC.loader MODULE = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(MODULE) def test_dataset_validation_source_preserves_manifest_path(tmp_path: Path): source = tmp_path / "dataset.yaml" source.write_text( "path: /data/source\ntrain: /data/source/train.txt\n" "val: /data/source/internal-val.txt\nnames:\n 0: building\n", encoding="utf-8", ) assert MODULE.dataset_validation_source(source) == "/data/source/internal-val.txt" def test_sampling_repeats_only_failed_region_train_tiles() -> None: manifest = { "samples": [ {"sample_slug": "train-fl", "split": "train", "region": "flanders"}, {"sample_slug": "train-wa", "split": "train", "region": "wallonia"}, {"sample_slug": "test-fl", "split": "test", "region": "flanders"}, ] } summary = { "tiles": [ {"sample_slug": "train-fl", "split": "train", "label_count": 2, "image_path": "/tmp/fl-pos.png"}, {"sample_slug": "train-fl", "split": "train", "label_count": 0, "image_path": "/tmp/fl-neg.png"}, {"sample_slug": "train-wa", "split": "train", "label_count": 1, "image_path": "/tmp/wa-pos.png"}, {"sample_slug": "test-fl", "split": "val", "label_count": 1, "image_path": "/tmp/protected.png"}, ] } assessment = { "status": "continue_training_loop", "gates": { "min_region_f1": 0.45, "min_region_precision": 0.5, "min_region_recall": 0.4, "max_pure_empty_false_positives": 0, }, "test": { "regions": { "flanders": {"f1": 0.2, "precision": 0.3, "recall": 0.2}, "wallonia": {"f1": 0.6, "precision": 0.6, "recall": 0.6}, } }, "background": {"pure_empty_false_positives": 2}, } paths, metadata = MODULE.build_sampling( summary=summary, manifest=manifest, assessment=assessment, max_region_share=1.0 ) assert paths.count(str(Path("/tmp/fl-pos.png").resolve())) == 3 assert paths.count(str(Path("/tmp/fl-neg.png").resolve())) == 4 assert paths.count(str(Path("/tmp/wa-pos.png").resolve())) == 1 assert not any("protected" in path for path in paths) assert metadata["protected_samples_in_training"] == [] assert metadata["weak_recall_regions"] == ["flanders"] def test_sampling_can_use_calibration_before_test_is_opened() -> None: manifest = {"samples": [{"sample_slug": "train-fl", "split": "train", "region": "flanders"}]} summary = { "tiles": [ {"sample_slug": "train-fl", "split": "train", "label_count": 1, "image_path": "/tmp/fl.png"} ] } assessment = { "status": "continue_training_loop", "gates": { "min_region_f1": 0.45, "min_region_precision": 0.5, "min_region_recall": 0.4, "max_pure_empty_false_positives": 0, }, "calibration": { "regions": {"flanders": {"f1": 0.4, "precision": 0.6, "recall": 0.35}} }, "test": None, "background": None, } paths, metadata = MODULE.build_sampling( summary=summary, manifest=manifest, assessment=assessment, max_region_share=1.0 ) assert len(paths) == 3 assert metadata["failure_evidence_source"] == "calibration" def test_precision_correction_can_balance_positive_and_negative_tiles() -> None: manifest = {"samples": [{"sample_slug": "train-fl", "split": "train", "region": "flanders"}]} summary = { "tiles": [ {"sample_slug": "train-fl", "split": "train", "label_count": 2, "image_path": "/tmp/fl-pos.png"}, {"sample_slug": "train-fl", "split": "train", "label_count": 0, "image_path": "/tmp/fl-neg.png"}, ] } assessment = { "status": "continue_training_loop", "gates": { "min_region_f1": 0.45, "min_region_precision": 0.5, "min_region_recall": 0.4, "max_pure_empty_false_positives": 0, }, "calibration": { "regions": {"flanders": {"f1": 0.46, "precision": 0.45, "recall": 0.46}} }, } paths, metadata = MODULE.build_sampling( summary=summary, manifest=manifest, assessment=assessment, precision_positive_repeat=2, negative_repeat=3, max_region_share=1.0, ) assert paths.count(str(Path("/tmp/fl-pos.png").resolve())) == 2 assert paths.count(str(Path("/tmp/fl-neg.png").resolve())) == 3 assert metadata["precision_positive_repeat"] == 2 def test_sampling_targets_failed_calibration_contexts_without_using_protected_tiles() -> None: manifest = { "samples": [ {"sample_slug": "train-industry", "split": "train", "region": "flanders", "context": "industrial"}, {"sample_slug": "train-suburban", "split": "train", "region": "flanders", "context": "suburban"}, {"sample_slug": "cal-industry", "split": "calibration", "region": "flanders", "context": "industrial"}, ] } summary = { "tiles": [ {"sample_slug": "train-industry", "split": "train", "label_count": 2, "image_path": "/tmp/industry-pos.png"}, {"sample_slug": "train-industry", "split": "train", "label_count": 0, "image_path": "/tmp/industry-neg.png"}, {"sample_slug": "train-suburban", "split": "train", "label_count": 2, "image_path": "/tmp/suburban-pos.png"}, {"sample_slug": "cal-industry", "split": "val", "label_count": 2, "image_path": "/tmp/protected.png"}, ] } assessment = { "status": "continue_training_loop", "gates": { "min_region_f1": 0.45, "min_region_precision": 0.5, "min_region_recall": 0.4, "max_pure_empty_false_positives": 0, }, "calibration": { "regions": {"flanders": {"f1": 0.3, "precision": 0.35, "recall": 0.27}}, "samples": {"cal-industry": {"f1": 0.2, "precision": 0.3, "recall": 0.15}}, }, } paths, metadata = MODULE.build_sampling( summary=summary, manifest=manifest, assessment=assessment, max_region_share=1.0 ) assert paths.count(str(Path("/tmp/industry-pos.png").resolve())) == 5 assert paths.count(str(Path("/tmp/industry-neg.png").resolve())) == 1 assert paths.count(str(Path("/tmp/suburban-pos.png").resolve())) == 3 assert not any("protected" in path for path in paths) assert metadata["weak_recall_contexts"] == ["flanders:industrial"] assert metadata["weak_precision_contexts"] == ["flanders:industrial"] assert metadata["recall_dominant_regions"] == ["flanders"] def test_recall_dominance_does_not_suppress_negatives_when_background_gate_failed() -> None: manifest = {"samples": [ {"sample_slug": "positive", "split": "train", "region": "flanders", "context": "industrial"}, {"sample_slug": "negative", "split": "train", "region": "flanders", "context": "industrial-hard-negative"}, ]} summary = {"tiles": [ {"sample_slug": "positive", "split": "train", "label_count": 1, "image_path": "/tmp/positive.png"}, {"sample_slug": "negative", "split": "train", "label_count": 0, "image_path": "/tmp/negative.png"}, ]} assessment = { "status": "continue_training_loop", "gates": {"min_region_f1": .45, "min_region_precision": .5, "min_region_recall": .4, "max_pure_empty_false_positives": 0}, "calibration": {"regions": { "flanders": {"f1": .25, "precision": .4, "recall": .2}, }}, "background": {"pure_empty_false_positives": 1}, } paths, metadata = MODULE.build_sampling( summary=summary, manifest=manifest, assessment=assessment, max_region_share=1.0, ) assert paths.count(str(Path("/tmp/negative.png").resolve())) == 4 assert metadata["recall_dominant_regions"] == [] def test_region_cap_drops_only_repeats_and_preserves_every_unique_tile() -> None: manifest = {"samples": [ {"sample_slug": "fl", "split": "train", "region": "flanders", "context": "industrial"}, {"sample_slug": "wa", "split": "train", "region": "wallonia", "context": "rural-town"}, {"sample_slug": "br", "split": "train", "region": "brussels", "context": "dense-urban"}, ]} summary = {"tiles": [ {"sample_slug": "fl", "split": "train", "label_count": 2, "image_path": f"/tmp/fl-{index}.png"} for index in range(4) ] + [ {"sample_slug": "wa", "split": "train", "label_count": 2, "image_path": f"/tmp/wa-{index}.png"} for index in range(2) ] + [ {"sample_slug": "br", "split": "train", "label_count": 2, "image_path": f"/tmp/br-{index}.png"} for index in range(2) ]} assessment = { "status": "continue_training_loop", "gates": {"min_region_f1": .45, "min_region_precision": .5, "min_region_recall": .4, "max_pure_empty_false_positives": 0}, "calibration": {"regions": { "flanders": {"f1": .2, "precision": .3, "recall": .2}, "wallonia": {"f1": .6, "precision": .6, "recall": .6}, "brussels": {"f1": .6, "precision": .6, "recall": .6}, }}, } paths, metadata = MODULE.build_sampling( summary=summary, manifest=manifest, assessment=assessment, positive_repeat=5, max_region_share=.65, ) assert all(str(Path(f"/tmp/fl-{index}.png").resolve()) in paths for index in range(4)) assert metadata["pre_cap_entries_by_region"]["flanders"] == 20 assert metadata["sampled_entries_by_region"]["flanders"] == 7 assert metadata["dropped_region_repeat_count"] == 13 assert metadata["sampled_entries_by_region"]["flanders"] / len(paths) <= .65 def test_region_cap_rotates_repeats_between_sampling_rounds() -> None: manifest = {"samples": [ {"sample_slug": "fl", "split": "train", "region": "flanders", "context": "industrial"}, {"sample_slug": "wa", "split": "train", "region": "wallonia", "context": "rural-town"}, {"sample_slug": "br", "split": "train", "region": "brussels", "context": "dense-urban"}, ]} summary = {"tiles": [ {"sample_slug": "fl", "split": "train", "label_count": 2, "image_path": f"/tmp/fl-{index}.png"} for index in range(4) ] + [ {"sample_slug": "wa", "split": "train", "label_count": 2, "image_path": f"/tmp/wa-{index}.png"} for index in range(2) ] + [ {"sample_slug": "br", "split": "train", "label_count": 2, "image_path": f"/tmp/br-{index}.png"} for index in range(2) ]} assessment = { "status": "continue_training_loop", "gates": {"min_region_f1": .45, "min_region_precision": .5, "min_region_recall": .4, "max_pure_empty_false_positives": 0}, "calibration": {"regions": { "flanders": {"f1": .2, "precision": .3, "recall": .2}, "wallonia": {"f1": .6, "precision": .6, "recall": .6}, "brussels": {"f1": .6, "precision": .6, "recall": .6}, }}, } first, first_metadata = MODULE.build_sampling( summary=summary, manifest=manifest, assessment=assessment, positive_repeat=5, max_region_share=.65, sampling_round=1, ) second, second_metadata = MODULE.build_sampling( summary=summary, manifest=manifest, assessment=assessment, positive_repeat=5, max_region_share=.65, sampling_round=2, ) assert first != second assert set(first) == set(second) assert first_metadata["sampling_round"] == 1 assert second_metadata["sampling_round"] == 2 def test_region_cap_preserves_failed_context_positive_before_hard_negative() -> None: manifest = {"samples": [ {"sample_slug": "target", "split": "train", "region": "flanders", "context": "industrial"}, {"sample_slug": "negative", "split": "train", "region": "flanders", "context": "industrial-hard-negative"}, {"sample_slug": "wa", "split": "train", "region": "wallonia", "context": "rural-town"}, {"sample_slug": "br", "split": "train", "region": "brussels", "context": "dense-urban"}, ]} summary = {"tiles": [ {"sample_slug": "target", "split": "train", "label_count": 2, "image_path": "/tmp/target.png"}, {"sample_slug": "negative", "split": "train", "label_count": 0, "image_path": "/tmp/negative.png"}, {"sample_slug": "wa", "split": "train", "label_count": 1, "image_path": "/tmp/wa.png"}, {"sample_slug": "br", "split": "train", "label_count": 1, "image_path": "/tmp/br.png"}, ]} assessment = { "status": "continue_training_loop", "gates": {"min_region_f1": .45, "min_region_precision": .5, "min_region_recall": .4, "max_pure_empty_false_positives": 0}, "calibration": { "regions": { "flanders": {"f1": .2, "precision": .2, "recall": .3}, "wallonia": {"f1": .6, "precision": .6, "recall": .6}, "brussels": {"f1": .6, "precision": .6, "recall": .6}, }, "samples": {"target": {"f1": .2, "precision": .2, "recall": .3}}, }, } paths, metadata = MODULE.build_sampling( summary=summary, manifest=manifest, assessment=assessment, max_region_share=.65, ) assert paths.count(str(Path("/tmp/target.png").resolve())) == 2 assert paths.count(str(Path("/tmp/negative.png").resolve())) == 1 assert metadata["priority_positive_repeat_count"] == 4 def test_precision_guard_band_keeps_near_gate_region_stabilized() -> None: manifest = {"samples": [ {"sample_slug": "wa-positive", "split": "train", "region": "wallonia", "context": "rural-town"}, {"sample_slug": "wa-negative", "split": "train", "region": "wallonia", "context": "farmland-hard-negative"}, ]} summary = {"tiles": [ {"sample_slug": "wa-positive", "split": "train", "label_count": 1, "image_path": "/tmp/wa-positive.png"}, {"sample_slug": "wa-negative", "split": "train", "label_count": 0, "image_path": "/tmp/wa-negative.png"}, ]} assessment = { "status": "continue_training_loop", "gates": {"min_region_f1": .45, "min_region_precision": .5, "min_region_recall": .4, "max_pure_empty_false_positives": 0}, "calibration": {"regions": { "wallonia": {"f1": .6, "precision": .52, "recall": .7}, }}, } paths, metadata = MODULE.build_sampling( summary=summary, manifest=manifest, assessment=assessment, max_region_share=1.0, ) assert "wallonia" in metadata["weak_precision_regions"] assert paths.count(str(Path("/tmp/wa-negative.png").resolve())) == 4 assert metadata["precision_guard_band"] == .03 def test_coastal_precision_failure_targets_port_and_dunes_negatives() -> None: manifest = {"samples": [ {"sample_slug": "coastal-train", "split": "train", "region": "flanders", "context": "coastal-urban"}, {"sample_slug": "port-negative", "split": "train", "region": "flanders", "context": "port-hard-negative"}, {"sample_slug": "dunes-negative", "split": "train", "region": "flanders", "context": "dunes-negative"}, {"sample_slug": "coastal-cal", "split": "calibration", "region": "flanders", "context": "coastal-urban"}, ]} summary = {"tiles": [ {"sample_slug": "coastal-train", "split": "train", "label_count": 2, "image_path": "/tmp/coastal.png"}, {"sample_slug": "port-negative", "split": "train", "label_count": 0, "image_path": "/tmp/port.png"}, {"sample_slug": "dunes-negative", "split": "train", "label_count": 0, "image_path": "/tmp/dunes.png"}, ]} assessment = { "status": "continue_training_loop", "gates": {"min_region_f1": .45, "min_region_precision": .5, "min_region_recall": .4, "max_pure_empty_false_positives": 0}, "calibration": { "regions": {"flanders": {"f1": .3, "precision": .2, "recall": .4}}, "samples": {"coastal-cal": {"f1": .1, "precision": .05, "recall": .2}}, }, } paths, metadata = MODULE.build_sampling( summary=summary, manifest=manifest, assessment=assessment, max_region_share=1.0 ) assert paths.count(str(Path("/tmp/port.png").resolve())) == 6 assert paths.count(str(Path("/tmp/dunes.png").resolve())) == 6 assert "flanders:port-hard-negative" in metadata["targeted_negative_contexts"] assert "flanders:dunes-negative" in metadata["targeted_negative_contexts"]