Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
291 lines
12 KiB
Python
291 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def load(name: str):
|
|
path = ROOT / "scripts" / f"{name}.py"
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
miner = load("build_building_proposal_classifier_dataset")
|
|
trainer = load("train_building_proposal_classifier")
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def _write_json(path: Path, payload: dict) -> None:
|
|
path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
|
|
|
|
|
|
def _governed_proposal_crop_fixture(tmp_path: Path) -> tuple[Path, Path]:
|
|
"""Create a complete tiny crop release without importing Torch or YOLO."""
|
|
|
|
corpus_manifest = tmp_path / "corpus-manifest.json"
|
|
_write_json(corpus_manifest, {"samples": []})
|
|
corpus_sha256 = _sha256(corpus_manifest)
|
|
corpus_freeze = tmp_path / "corpus-freeze.json"
|
|
_write_json(corpus_freeze, {"immutable": True})
|
|
summary = tmp_path / "source-summary.json"
|
|
_write_json(summary, {"source_manifest_sha256": corpus_sha256, "tiles": []})
|
|
|
|
dataset_dir = tmp_path / "proposal-crops"
|
|
entries: list[dict] = []
|
|
for split in ("train", "val"):
|
|
for label, colour in (("negative", (0, 0, 0)), ("positive", (255, 255, 255))):
|
|
crop_path = dataset_dir / split / label / f"{split}-{label}.jpg"
|
|
crop_path.parent.mkdir(parents=True, exist_ok=True)
|
|
Image.new("RGB", (2, 2), colour).save(crop_path)
|
|
relative_path = crop_path.relative_to(dataset_dir).as_posix()
|
|
entries.append(
|
|
{
|
|
"relative_path": relative_path,
|
|
"sha256": _sha256(crop_path),
|
|
"size_bytes": crop_path.stat().st_size,
|
|
"split": split,
|
|
"label": label,
|
|
"sample_slug": f"{split}-{label}",
|
|
"proposal_index": 0,
|
|
"proposal_score": 0.8,
|
|
"source_box_xyxy": [0.0, 0.0, 1.0, 1.0],
|
|
"source_image_path": str(tmp_path / "source-image.tif"),
|
|
"source_image_sha256": "a" * 64,
|
|
"source_label_path": str(tmp_path / "source-label.txt"),
|
|
"source_label_sha256": "b" * 64,
|
|
}
|
|
)
|
|
entries.sort(key=lambda item: item["relative_path"])
|
|
counts = {f"{split}/{label}": 1 for split in ("train", "val") for label in ("negative", "positive")}
|
|
payload: dict = {
|
|
"schema_version": 1,
|
|
"status": "ok",
|
|
"immutable": True,
|
|
"dataset_kind": "building_proposal_classifier_crops",
|
|
"fixture_mode": False,
|
|
"governed_corpus_live_recheck": True,
|
|
"source": {
|
|
"corpus_manifest": {"path": str(corpus_manifest), "sha256": corpus_sha256},
|
|
"corpus_freeze": {"path": str(corpus_freeze), "sha256": _sha256(corpus_freeze)},
|
|
"summary": {
|
|
"path": str(summary),
|
|
"sha256": _sha256(summary),
|
|
"source_manifest_sha256": corpus_sha256,
|
|
},
|
|
"training_release": {
|
|
"dataset_yaml_path": "fixture-dataset.yaml",
|
|
"dataset_yaml_sha256": "d" * 64,
|
|
"corpus_manifest_sha256": corpus_sha256,
|
|
},
|
|
"proposal_model": {"path": str(tmp_path / "proposal.pt"), "sha256": "c" * 64},
|
|
},
|
|
"parameters": {"crop_scale": 1.4},
|
|
"counts": counts,
|
|
"sample_counts": {item["sample_slug"]: 1 for item in entries},
|
|
"tile_count": 2,
|
|
"crop_count": len(entries),
|
|
"crops_sha256": hashlib.sha256(trainer._canonical_json_bytes({"crops": entries})).hexdigest(),
|
|
"crops": entries,
|
|
}
|
|
payload["manifest_sha256"] = trainer._payload_sha256(payload)
|
|
_write_json(dataset_dir / trainer.PROPOSAL_DATASET_PROVENANCE_NAME, payload)
|
|
return dataset_dir, corpus_manifest
|
|
|
|
|
|
def test_classify_proposals_consumes_reference_once() -> None:
|
|
reference = [(0.0, 0.0, 10.0, 10.0)]
|
|
proposals = [((0.0, 0.0, 10.0, 10.0), 0.9), ((0.0, 0.0, 10.0, 10.0), 0.8)]
|
|
assert [item[0] for item in miner.classify_proposals(proposals, reference, 0.25)] == ["positive", "negative"]
|
|
|
|
|
|
@pytest.mark.parametrize("protected_split", ["calibration", "test", "background-test", "challenge"])
|
|
def test_eligible_tiles_rejects_protected_manifest_split(protected_split: str) -> None:
|
|
manifest = {"samples": [{"sample_slug": "x", "region": "flanders", "split": protected_split}]}
|
|
summary = {"tiles": [{"sample_slug": "x", "split": "train", "image_path": "x.png"}]}
|
|
with pytest.raises(ValueError, match="protected"):
|
|
miner.eligible_tiles(summary, manifest, "flanders")
|
|
|
|
|
|
def test_binary_metrics() -> None:
|
|
result = trainer.binary_metrics([0.9, 0.8, 0.2, 0.1], [1, 0, 1, 0])
|
|
assert result == {"tp": 1, "fp": 1, "fn": 1, "precision": 0.5, "recall": 0.5, "f1": 0.5}
|
|
|
|
|
|
def test_builder_rechecks_frozen_corpus_against_live_governed_state(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
observed: dict[str, object] = {}
|
|
|
|
def fake_assert(path: Path, **kwargs: object) -> dict:
|
|
observed["path"] = path
|
|
observed.update(kwargs)
|
|
return {"samples": []}
|
|
|
|
monkeypatch.setattr(miner, "assert_frozen_manifest_training_eligible", fake_assert)
|
|
manifest_path = tmp_path / "corpus.json"
|
|
manifest_path.write_text("{}", encoding="utf-8")
|
|
assert miner.load_governed_corpus_manifest(manifest_path, fixture_mode=False) == {"samples": []}
|
|
assert observed == {"path": manifest_path, "fixture_mode": False, "verify_live": True}
|
|
|
|
|
|
def test_builder_rejects_summary_not_bound_to_exact_corpus_manifest(tmp_path: Path) -> None:
|
|
manifest_path = tmp_path / "corpus.json"
|
|
manifest_path.write_text('{"samples": []}', encoding="utf-8")
|
|
with pytest.raises(ValueError, match="not bound"):
|
|
miner.assert_summary_source_manifest_binding({"source_manifest_sha256": "0" * 64}, manifest_path)
|
|
|
|
|
|
def test_builder_emits_immutable_checksum_bound_crop_provenance(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
class FakeTensor:
|
|
def __init__(self, values: list[list[float]] | list[float]) -> None:
|
|
self._values = values
|
|
|
|
def cpu(self) -> "FakeTensor":
|
|
return self
|
|
|
|
def tolist(self) -> list[list[float]] | list[float]:
|
|
return self._values
|
|
|
|
class FakeBoxes:
|
|
xyxy = FakeTensor([[40.0, 40.0, 60.0, 60.0], [0.0, 0.0, 10.0, 10.0]])
|
|
conf = FakeTensor([0.9, 0.8])
|
|
|
|
class FakeResult:
|
|
boxes = FakeBoxes()
|
|
|
|
class FakeYOLO:
|
|
def __init__(self, model: str) -> None:
|
|
self.model = model
|
|
|
|
def predict(self, sources: list[str], **_kwargs: object) -> list[FakeResult]:
|
|
return [FakeResult() for _source in sources]
|
|
|
|
corpus_manifest = tmp_path / "corpus-manifest.json"
|
|
manifest = {
|
|
"samples": [
|
|
{"sample_slug": "train-a", "split": "train", "region": "flanders"},
|
|
{"sample_slug": "val-b", "split": "val", "region": "flanders"},
|
|
]
|
|
}
|
|
_write_json(corpus_manifest, manifest)
|
|
_write_json(tmp_path / "corpus-freeze.json", {"immutable": True})
|
|
tiles: list[dict[str, object]] = []
|
|
for sample_slug, split in (("train-a", "train"), ("val-b", "val")):
|
|
image_path = tmp_path / f"{sample_slug}.png"
|
|
Image.new("RGB", (100, 100), (50, 100, 150)).save(image_path)
|
|
label_path = tmp_path / f"{sample_slug}.txt"
|
|
label_path.write_text("0 0.5 0.5 0.2 0.2\n", encoding="utf-8")
|
|
tiles.append(
|
|
{
|
|
"sample_slug": sample_slug,
|
|
"split": split,
|
|
"kept": True,
|
|
"image_path": str(image_path),
|
|
"label_path": str(label_path),
|
|
}
|
|
)
|
|
summary_path = tmp_path / "source-summary.json"
|
|
_write_json(
|
|
summary_path,
|
|
{
|
|
"source_manifest_sha256": _sha256(corpus_manifest),
|
|
"training_release_manifest": "fixture-training-release.json",
|
|
"training_release_manifest_sha256": "a" * 64,
|
|
"training_asset_manifest": "fixture-training-assets.json",
|
|
"tiles": tiles,
|
|
},
|
|
)
|
|
model_path = tmp_path / "proposal-model.pt"
|
|
model_path.write_bytes(b"proposal-model")
|
|
output_dir = tmp_path / "proposal-crops"
|
|
|
|
monkeypatch.setattr(miner, "load_governed_corpus_manifest", lambda *_args, **_kwargs: manifest)
|
|
monkeypatch.setattr(
|
|
miner,
|
|
"assert_yolo_summary_bound_to_embedded_training_release",
|
|
lambda **_kwargs: {
|
|
"dataset_yaml": {"path": "fixture-dataset.yaml", "sha256": "d" * 64},
|
|
"corpus": {"manifest_sha256": _sha256(corpus_manifest)},
|
|
},
|
|
)
|
|
monkeypatch.setitem(sys.modules, "ultralytics", types.SimpleNamespace(YOLO=FakeYOLO))
|
|
monkeypatch.setattr(
|
|
miner.sys,
|
|
"argv",
|
|
[
|
|
"build_building_proposal_classifier_dataset.py",
|
|
"--model",
|
|
str(model_path),
|
|
"--summary",
|
|
str(summary_path),
|
|
"--corpus-manifest",
|
|
str(corpus_manifest),
|
|
"--output-dir",
|
|
str(output_dir),
|
|
],
|
|
)
|
|
|
|
assert miner.main() == 0
|
|
provenance_path = output_dir / miner.PROPOSAL_DATASET_PROVENANCE_NAME
|
|
provenance = json.loads(provenance_path.read_text(encoding="utf-8"))
|
|
assert provenance["immutable"] is True
|
|
assert provenance["crop_count"] == 4
|
|
assert provenance["manifest_sha256"] == miner._immutable_payload_sha256(provenance)
|
|
assert all(_sha256(output_dir / item["relative_path"]) == item["sha256"] for item in provenance["crops"])
|
|
with pytest.raises(RuntimeError, match="immutable provenance"):
|
|
miner._write_immutable_json(provenance_path, {"different": True})
|
|
|
|
|
|
def test_trainer_validates_every_crop_and_source_binding_before_torch(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
dataset_dir, corpus_manifest = _governed_proposal_crop_fixture(tmp_path)
|
|
monkeypatch.setattr(
|
|
trainer,
|
|
"assert_yolo_summary_bound_to_embedded_training_release",
|
|
lambda **_kwargs: {
|
|
"dataset_yaml": {"path": "fixture-dataset.yaml", "sha256": "d" * 64},
|
|
"corpus": {"manifest_sha256": _sha256(corpus_manifest)},
|
|
},
|
|
)
|
|
payload = trainer.validate_proposal_dataset_provenance(dataset_dir, corpus_manifest, fixture_mode=False)
|
|
assert payload["crop_count"] == 4
|
|
|
|
crop = dataset_dir / "train" / "positive" / "train-positive.jpg"
|
|
crop.write_bytes(b"tampered")
|
|
with pytest.raises(trainer.ProposalDatasetProvenanceError, match="checksum"):
|
|
trainer.validate_proposal_dataset_provenance(dataset_dir, corpus_manifest, fixture_mode=False)
|
|
|
|
|
|
def test_trainer_rechecks_live_governed_corpus_before_pytorch(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
observed: dict[str, object] = {}
|
|
|
|
def fake_assert(path: Path, **kwargs: object) -> dict:
|
|
observed["path"] = path
|
|
observed.update(kwargs)
|
|
return {"samples": []}
|
|
|
|
monkeypatch.setattr(trainer, "assert_frozen_manifest_training_eligible", fake_assert)
|
|
manifest_path = tmp_path / "corpus.json"
|
|
manifest_path.write_text("{}", encoding="utf-8")
|
|
assert trainer.load_governed_corpus_manifest(manifest_path, fixture_mode=False) == {"samples": []}
|
|
assert observed == {"path": manifest_path, "fixture_mode": False, "verify_live": True}
|