184 lines
6.0 KiB
Python
184 lines
6.0 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPT = Path(__file__).parents[2] / "scripts" / "run_belgium_building_training_loop.py"
|
|
SPEC = importlib.util.spec_from_file_location("training_loop", SCRIPT)
|
|
assert SPEC and SPEC.loader
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
|
|
def test_training_command_is_cuda_deterministic_and_bound_to_frozen_inputs(tmp_path: Path) -> None:
|
|
command = MODULE.training_command(
|
|
"yolo",
|
|
model=tmp_path / "base.pt",
|
|
data=tmp_path / "dataset.yaml",
|
|
project=tmp_path / "runs",
|
|
name="iteration-001",
|
|
epochs=160,
|
|
seed=42,
|
|
batch=2,
|
|
workers=4,
|
|
)
|
|
assert command[:2] == ["yolo", "train"]
|
|
assert "device=0" in command
|
|
assert "deterministic=True" in command
|
|
assert "seed=42" in command
|
|
assert "epochs=160" in command
|
|
assert "max_det=1000" in command
|
|
assert "imgsz=640" in command
|
|
assert "optimizer=auto" in command
|
|
assert "mosaic=1.0" in command
|
|
|
|
|
|
def test_training_command_supports_conservative_aerial_finetuning(tmp_path: Path) -> None:
|
|
command = MODULE.training_command(
|
|
"yolo",
|
|
model=tmp_path / "base.pt",
|
|
data=tmp_path / "dataset.yaml",
|
|
project=tmp_path / "runs",
|
|
name="aerial",
|
|
epochs=50,
|
|
seed=42,
|
|
batch=2,
|
|
workers=0,
|
|
optimizer="AdamW",
|
|
lr0=0.0001,
|
|
mosaic=0.0,
|
|
scale=0.2,
|
|
translate=0.05,
|
|
)
|
|
assert "optimizer=AdamW" in command
|
|
assert "lr0=0.0001" in command
|
|
assert "mosaic=0.0" in command
|
|
assert "scale=0.2" in command
|
|
assert "translate=0.05" in command
|
|
assert f"data={tmp_path / 'dataset.yaml'}" in command
|
|
|
|
|
|
def test_failed_iteration_builds_train_only_sampling_for_next_checkpoint(tmp_path: Path) -> None:
|
|
command = MODULE.failure_sampling_command(
|
|
scripts_dir=tmp_path / "scripts",
|
|
train_summary=tmp_path / "train-summary.json",
|
|
corpus_manifest=tmp_path / "manifest.json",
|
|
assessment=tmp_path / "assessment.json",
|
|
output_dir=tmp_path / "iteration-001" / "failure-driven-training",
|
|
)
|
|
assert command[1].endswith("build_failure_driven_yolo_sampling.py")
|
|
assert command[command.index("--summary") + 1].endswith("train-summary.json")
|
|
assert command[command.index("--assessment") + 1].endswith("assessment.json")
|
|
assert command[command.index("--output-dir") + 1].endswith("failure-driven-training")
|
|
|
|
|
|
def test_loop_refuses_failed_dataset_audit(tmp_path: Path) -> None:
|
|
audit = tmp_path / "audit.json"
|
|
audit.write_text(json.dumps({"status": "needs_attention", "low_variance_positive_tile_count": 4}))
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(SCRIPT),
|
|
"--initial-model",
|
|
str(tmp_path / "base.pt"),
|
|
"--train-yaml",
|
|
str(tmp_path / "dataset.yaml"),
|
|
"--train-summary",
|
|
str(tmp_path / "train-summary.json"),
|
|
"--dataset-audit",
|
|
str(audit),
|
|
"--calibration-summary",
|
|
str(tmp_path / "cal.json"),
|
|
"--test-summary",
|
|
str(tmp_path / "test.json"),
|
|
"--background-summary",
|
|
str(tmp_path / "background.json"),
|
|
"--corpus-manifest",
|
|
str(tmp_path / "manifest.json"),
|
|
"--output-dir",
|
|
str(tmp_path / "output"),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
assert result.returncode != 0
|
|
assert "Dataset audit is not eligible for training" in result.stderr
|
|
|
|
|
|
def test_pending_human_review_does_not_block_objective_training() -> None:
|
|
audit = {
|
|
"status": "needs_human_review",
|
|
"failures": [],
|
|
"manifest_immutable": True,
|
|
"spatial_leakage_status": "ok",
|
|
"low_variance_positive_tile_count": 0,
|
|
"review_complete": False,
|
|
}
|
|
assert MODULE.dataset_audit_failures(audit) == []
|
|
|
|
|
|
def test_training_audit_still_fails_closed_on_automated_integrity_gates() -> None:
|
|
audit = {
|
|
"status": "needs_human_review",
|
|
"failures": ["wallonia/test below minimum"],
|
|
"manifest_immutable": False,
|
|
"spatial_leakage_status": "failed",
|
|
"low_variance_positive_tile_count": 2,
|
|
}
|
|
failures = MODULE.dataset_audit_failures(audit)
|
|
assert "wallonia/test below minimum" in failures
|
|
assert "corpus manifest is not immutable" in failures
|
|
assert "spatial leakage audit is not ok" in failures
|
|
assert "dataset contains blank/low-variance positive tiles" in failures
|
|
|
|
|
|
def test_calibration_failure_blocks_protected_evaluation() -> None:
|
|
chosen = {
|
|
"threshold": 0.1,
|
|
"aggregate": {"f1": 0.54},
|
|
"regions": {
|
|
"flanders": {"f1": 0.44, "precision": 0.49, "recall": 0.39},
|
|
"wallonia": {"f1": 0.6, "precision": 0.6, "recall": 0.6},
|
|
},
|
|
"pure_empty_false_positives": 0,
|
|
}
|
|
failures = MODULE.calibration_failures(
|
|
chosen,
|
|
min_aggregate_f1=0.55,
|
|
min_region_f1=0.45,
|
|
min_region_precision=0.5,
|
|
min_region_recall=0.4,
|
|
max_pure_empty_fp=0,
|
|
)
|
|
assert failures == [
|
|
"calibration_aggregate_f1_below_gate",
|
|
"calibration_flanders_f1_below_gate",
|
|
"calibration_flanders_precision_below_gate",
|
|
"calibration_flanders_recall_below_gate",
|
|
]
|
|
|
|
|
|
def test_threshold_selection_uses_worst_region_then_aggregate() -> None:
|
|
report = {
|
|
"sweeps": [
|
|
{
|
|
"threshold": 0.1,
|
|
"aggregate": {"f1": 0.8},
|
|
"regions": {"a": {"f1": 0.4}, "b": {"f1": 0.7}},
|
|
"pure_empty_false_positives": 0,
|
|
},
|
|
{
|
|
"threshold": 0.2,
|
|
"aggregate": {"f1": 0.6},
|
|
"regions": {"a": {"f1": 0.5}, "b": {"f1": 0.5}},
|
|
"pure_empty_false_positives": 0,
|
|
},
|
|
]
|
|
}
|
|
assert MODULE.select_calibration_threshold(report)["threshold"] == 0.2
|