from __future__ import annotations import importlib.util import math from pathlib import Path import sys import pytest 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("operator_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 test_operator_sample_registry_includes_kempen_reference_and_background_candidates() -> None: module = load_sample_preparer() expected_reference_slugs = {"geel", "mol", "turnhout", "herentals", "balen", "retie", "westerlo"} expected_background_slugs = { "postel_bos", "lommel_heide", "kasterlee_bos", "dessel_heide", "ravels_bos", "meerhout_bos", "geel_bel", "arendonk_heide", } assert expected_reference_slugs.issubset(module.SAMPLES) assert expected_background_slugs.issubset(module.SAMPLES) assert all(not module.SAMPLES[slug].allow_empty_reference for slug in expected_reference_slugs) assert all(module.SAMPLES[slug].allow_empty_reference for slug in expected_background_slugs) assert all(module.SAMPLES[slug].sample_role == "background_candidate" for slug in expected_background_slugs) def test_operator_training_expansion_preserves_geographically_separate_holdouts() -> None: module = load_sample_preparer() expected_expansion = {"olen_center", "lille_center", "oud_turnhout_center", "kasterlee_center"} expected_holdouts = {"turnhout", "retie", "westerlo", "arendonk_heide"} assert module.TRAINING_EXPANSION_SAMPLE_SLUGS == frozenset(expected_expansion) assert expected_holdouts.issubset(module.DEFAULT_VALIDATION_SAMPLE_SLUGS) assert all(module.SAMPLES[slug].sample_role == "reference" for slug in expected_expansion) assert all(not module.SAMPLES[slug].allow_empty_reference for slug in expected_expansion) assert all(module.recommended_split_for_sample(module.SAMPLES[slug]) == "train" for slug in expected_expansion) assert all(module.recommended_split_for_sample(module.SAMPLES[slug]) == "val" for slug in expected_holdouts) 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)) reference_holdouts = expected_holdouts - {"arendonk_heide"} for expansion_slug in expected_expansion: expansion = module.SAMPLES[expansion_slug] assert min( distance_m(expansion, module.SAMPLES[holdout_slug]) for holdout_slug in reference_holdouts ) >= 2_000 def test_small_building_expansion_has_separate_training_and_validation_centers() -> None: module = load_sample_preparer() expected_training = { "beerse_center", "rijkevorsel_center", "hoogstraten_center", "vorselaar_center", } expected_validation = {"vosselaar_center", "grobbendonk_center"} assert module.SMALL_BUILDING_TRAINING_SAMPLE_SLUGS == frozenset(expected_training) assert module.SMALL_BUILDING_VALIDATION_SAMPLE_SLUGS == frozenset(expected_validation) assert expected_validation.issubset(module.DEFAULT_VALIDATION_SAMPLE_SLUGS) assert all(module.SAMPLES[slug].sample_role == "reference" for slug in expected_training | expected_validation) assert all( module.recommended_split_for_sample(module.SAMPLES[slug]) == "train" for slug in expected_training ) assert all( module.recommended_split_for_sample(module.SAMPLES[slug]) == "val" for slug in expected_validation ) 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)) protected_holdouts = expected_validation | {"turnhout", "retie", "westerlo"} for training_slug in expected_training: training_sample = module.SAMPLES[training_slug] assert min( distance_m(training_sample, module.SAMPLES[holdout_slug]) for holdout_slug in protected_holdouts ) >= 2_000 def test_operator_background_candidates_are_unique_enough_for_hard_negative_training() -> None: module = load_sample_preparer() background_samples = [ sample for sample in module.SAMPLES.values() if sample.sample_role == "background_candidate" ] centers = {(round(sample.center_lon, 4), round(sample.center_lat, 4)) for sample in background_samples} half_sizes = {sample.half_size_m for sample in background_samples} assert len(background_samples) >= 8 assert len(centers) == len(background_samples) assert min(sample.center_lon for sample in background_samples) < 4.85 assert max(sample.center_lon for sample in background_samples) > 5.25 assert min(sample.center_lat for sample in background_samples) < 51.18 assert max(sample.center_lat for sample in background_samples) > 51.33 assert half_sizes == {260.0} def test_operator_sample_can_be_scaled_for_larger_training_aoi(tmp_path: Path) -> None: module = load_sample_preparer() sample = module.OperatorSample( slug="geel", display_name="Geel", center_lon=5.0, center_lat=51.0, half_size_m=250.0, ) configured = module.apply_sample_overrides(sample, width=1024, height=1024, half_size_scale=2.0) ortho_path, reference_path = module.sample_artifact_paths(configured, tmp_path) assert configured.width == 1024 assert configured.height == 1024 assert configured.half_size_m == 500.0 assert ortho_path.name == "geel_orthophoto_wms_1024.tif" assert reference_path.name == "geel_grb_gbg_buildings.geojson" def test_background_candidate_can_write_empty_reference_geojson(tmp_path: Path, monkeypatch) -> None: module = load_sample_preparer() class EmptyFeatureResponse: headers = {"content-type": "application/geo+json"} def raise_for_status(self) -> None: return None def json(self) -> dict: return {"type": "FeatureCollection", "features": []} class FakeRequests: @staticmethod def get(*args, **kwargs): return EmptyFeatureResponse() monkeypatch.setattr(module, "requests", FakeRequests) monkeypatch.setattr(module, "prepared_url", lambda url, params: f"{url}?prepared=true") sample = module.OperatorSample( slug="background", display_name="Background", center_lon=5.0, center_lat=51.0, allow_empty_reference=True, sample_role="background_candidate", ) reference_path = tmp_path / "background.geojson" source_url, feature_count = module.fetch_reference(sample, reference_path, [4.9, 50.9, 5.1, 51.1]) assert source_url.endswith("?prepared=true") assert feature_count == 0 payload = reference_path.read_text(encoding="utf-8") assert '"features": []' in payload assert '"sample_role": "background_candidate"' in payload def test_fetch_reference_follows_grb_next_links_until_complete(tmp_path: Path, monkeypatch) -> None: module = load_sample_preparer() requested: list[tuple[str, dict | None]] = [] def feature(feature_id: str) -> dict: return { "type": "Feature", "id": feature_id, "geometry": {"type": "Polygon", "coordinates": []}, "properties": {}, } class FeatureResponse: def __init__(self, payload: dict) -> None: self.payload = payload def raise_for_status(self) -> None: return None def json(self) -> dict: return self.payload class FakeRequests: Request = module.requests.Request if module.requests else object @staticmethod def get(url, params=None, timeout=120): requested.append((url, params)) if len(requested) == 1: return FeatureResponse( { "type": "FeatureCollection", "features": [feature("GBG.1")], "numberReturned": 1, "links": [ { "rel": "next", "type": "application/geo+json", "href": "https://example.test/grb?page=2", } ], } ) return FeatureResponse( { "type": "FeatureCollection", "features": [feature("GBG.2")], "numberReturned": 1, "links": [], } ) monkeypatch.setattr(module, "requests", FakeRequests) monkeypatch.setattr(module, "prepared_url", lambda url, params: f"{url}?prepared=true") sample = module.OperatorSample( slug="urban", display_name="Urban", center_lon=5.0, center_lat=51.0, ) source_url, feature_count = module.fetch_reference(sample, tmp_path / "urban.geojson", [4.9, 50.9, 5.1, 51.1]) payload = (tmp_path / "urban.geojson").read_text(encoding="utf-8") assert source_url.endswith("?prepared=true") assert feature_count == 2 assert requested == [ (module.GRB_GBG_URL, {"f": "application/geo+json", "limit": "1000", "bbox": "4.90000000,50.90000000,5.10000000,51.10000000"}), ("https://example.test/grb?page=2", None), ] assert '"id": "GBG.1"' in payload assert '"id": "GBG.2"' in payload def test_reference_sample_still_rejects_empty_grb_response(tmp_path: Path, monkeypatch) -> None: module = load_sample_preparer() class EmptyFeatureResponse: def raise_for_status(self) -> None: return None def json(self) -> dict: return {"type": "FeatureCollection", "features": []} class FakeRequests: @staticmethod def get(*args, **kwargs): return EmptyFeatureResponse() monkeypatch.setattr(module, "requests", FakeRequests) monkeypatch.setattr(module, "prepared_url", lambda url, params: f"{url}?prepared=true") sample = module.OperatorSample( slug="urban", display_name="Urban", center_lon=5.0, center_lat=51.0, ) with pytest.raises(SystemExit, match="returned no building features"): module.fetch_reference(sample, tmp_path / "urban.geojson", [4.9, 50.9, 5.1, 51.1])