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
118 lines
5.4 KiB
Python
118 lines
5.4 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import math
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def load_sample_preparer():
|
|
script_path = ROOT / "scripts" / "prepare_operator_real_data_samples.py"
|
|
spec = importlib.util.spec_from_file_location("mol_operational_sample_preparer", script_path)
|
|
assert spec is not None
|
|
assert spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def distance_m(left, right) -> float:
|
|
radius_m = 6_371_008.8
|
|
left_lat = math.radians(left.center_lat)
|
|
right_lat = math.radians(right.center_lat)
|
|
delta_lat = right_lat - left_lat
|
|
delta_lon = math.radians(right.center_lon - left.center_lon)
|
|
haversine = (
|
|
math.sin(delta_lat / 2) ** 2
|
|
+ math.cos(left_lat) * math.cos(right_lat) * math.sin(delta_lon / 2) ** 2
|
|
)
|
|
return 2 * radius_m * math.asin(math.sqrt(haversine))
|
|
|
|
|
|
def test_mol_operational_registry_has_distinct_real_world_zones_and_holdouts() -> None:
|
|
module = load_sample_preparer()
|
|
|
|
expected = {
|
|
"mol": "center",
|
|
"mol_achterbos": "residential",
|
|
"mol_gompel": "mixed_settlement",
|
|
"mol_donk": "canal_industrial",
|
|
"mol_postel": "rural_village",
|
|
}
|
|
assert tuple(expected) == module.MOL_OPERATIONAL_SAMPLE_SLUGS
|
|
assert module.MOL_BACKGROUND_CONTROL_SAMPLE_SLUGS == ("postel_bos",)
|
|
assert module.MOL_OPERATIONAL_VALIDATION_SAMPLE_SLUGS == frozenset(expected) - {"mol"}
|
|
|
|
samples = [module.SAMPLES[slug] for slug in expected]
|
|
assert all(sample.municipality == "Mol" for sample in samples)
|
|
assert {sample.operational_zone for sample in samples} == set(expected.values())
|
|
assert all(5.09 < sample.center_lon < 5.20 for sample in samples)
|
|
assert all(51.18 < sample.center_lat < 51.30 for sample in samples)
|
|
assert all(
|
|
module.recommended_split_for_sample(module.SAMPLES[slug]) == "val"
|
|
for slug in module.MOL_OPERATIONAL_VALIDATION_SAMPLE_SLUGS
|
|
)
|
|
|
|
new_holdouts = [module.SAMPLES[slug] for slug in module.MOL_OPERATIONAL_VALIDATION_SAMPLE_SLUGS]
|
|
assert min(
|
|
distance_m(left, right)
|
|
for index, left in enumerate(new_holdouts)
|
|
for right in new_holdouts[index + 1 :]
|
|
) >= 1_500
|
|
|
|
|
|
def test_mol_sample_metadata_is_persisted_in_operator_manifest_records(tmp_path: Path, monkeypatch) -> None:
|
|
module = load_sample_preparer()
|
|
sample = module.SAMPLES["mol_achterbos"]
|
|
monkeypatch.setattr(module, "sample_bounds", lambda _sample: ((1.0, 2.0, 3.0, 4.0), [5.0, 51.0, 5.1, 51.1]))
|
|
monkeypatch.setattr(module, "sample_artifact_paths", lambda _sample, _output: (tmp_path / "ortho.tif", tmp_path / "reference.geojson"))
|
|
monkeypatch.setattr(module, "fetch_orthophoto", lambda *_args: "https://example.test/ortho")
|
|
monkeypatch.setattr(module, "fetch_reference", lambda *_args, **_kwargs: ("https://example.test/grb", 42))
|
|
monkeypatch.setattr(module, "raster_summary", lambda _path: {"crs": "EPSG:31370"})
|
|
|
|
prepared = module.prepare_sample(sample, tmp_path, force=True)
|
|
|
|
assert prepared["municipality"] == "Mol"
|
|
assert prepared["operational_zone"] == "residential"
|
|
assert prepared["recommended_split"] == "val"
|
|
assert prepared["wgs84_bbox"] == [5.0, 51.0, 5.1, 51.1]
|
|
|
|
|
|
def test_real_data_matrix_propagates_project_region_and_persisted_area() -> None:
|
|
workflow = (ROOT / "scripts" / "verify_real_data_detection_qa_workflow.sh").read_text(encoding="utf-8")
|
|
matrix = (ROOT / "scripts" / "run_detection_quality_matrix.sh").read_text(encoding="utf-8")
|
|
multi = (ROOT / "scripts" / "run_multi_sample_detection_quality_matrix.sh").read_text(encoding="utf-8")
|
|
|
|
assert 'REAL_PROJECT_REGION="${REAL_PROJECT_REGION:-Kempen}"' in workflow
|
|
assert 'REAL_AREA_BBOX="${REAL_AREA_BBOX:-}"' in workflow
|
|
assert '/api/v1/projects/${project_id}/areas' in workflow
|
|
assert 'echo "Area: ${area_id}"' in workflow
|
|
assert 'REAL_PROJECT_REGION="${REAL_PROJECT_REGION}"' in matrix
|
|
assert 'REAL_AREA_BBOX="${REAL_AREA_BBOX}"' in matrix
|
|
assert 'REAL_AREA_BBOX="${wgs84_bbox}"' in multi
|
|
assert 'REAL_AREA_NAME="${sample_slug} AOI"' in multi
|
|
assert 'project_region="Mol, Kempen"' in multi
|
|
assert 'REAL_PROJECT_REGION="${project_region}"' in multi
|
|
assert 'manifest_path = Path(sys.argv[3])' in multi
|
|
assert '"operator_sample_manifest_path": str(manifest_path)' in multi
|
|
|
|
|
|
def test_mol_operational_runner_uses_real_positive_qa_and_background_paths() -> None:
|
|
runner = (ROOT / "scripts" / "run_mol_operational_validation.sh").read_text(encoding="utf-8")
|
|
readiness = (ROOT / "scripts" / "run_readiness_check.sh").read_text(encoding="utf-8")
|
|
dockerfile = (ROOT / "deploy" / "unraid" / "Dockerfile.all-in-one").read_text(encoding="utf-8")
|
|
|
|
assert "mol_achterbos mol_gompel mol_donk mol_postel" in runner
|
|
assert "run_multi_sample_detection_quality_matrix.sh" in runner
|
|
assert "run_operator_hard_negative_detection_matrix.sh" in runner
|
|
assert "mol_operational_validation_summary.json" in runner
|
|
assert "fixture_mode" not in runner
|
|
assert "manual-fixture-detector" not in runner
|
|
assert "/app/storage/operator-evidence/mol-operational-validation" in runner
|
|
assert "bash -n scripts/run_mol_operational_validation.sh" in readiness
|
|
assert "COPY scripts/run_mol_operational_validation.sh" in dockerfile
|