perf: optimize trusted full-area analysis
GeoIntel CI / docs-smoke (push) Canceled after 0s
GeoIntel CI / contract-smoke (push) Canceled after 0s

This commit is contained in:
Codex
2026-07-14 22:57:09 +02:00
parent c4392c17fd
commit 0381fdd636
10 changed files with 164 additions and 15 deletions
@@ -7,11 +7,16 @@ import json
from pathlib import Path
import sys
import zipfile
from types import SimpleNamespace
from uuid import uuid4
import numpy as np
import rasterio
from rasterio.transform import from_origin
from app.models import Dataset
from app.services.vector_feature_service import VectorFeatureService
ROOT = Path(__file__).resolve().parents[2]
SCRIPTS = ROOT / "scripts"
@@ -184,3 +189,94 @@ def test_end_user_dataset_sources_are_human_readable() -> None:
assert "statbel: 'Statbel'" in display
assert "getDatasetSourceDisplayName(activeThemeDataset)" in workspace
assert "dataset ? getDatasetSourceDisplayName(dataset)" in workspace
def test_full_area_fast_path_requires_matching_area_and_clipped_operator_provenance() -> None:
project_id = uuid4()
area_id = uuid4()
trusted = Dataset(
id=uuid4(),
project_id=project_id,
area_id=area_id,
name="Regional forest",
dataset_type="vector",
source="operator_official_import",
provenance_metadata={"operator_tool": "provision_official_landuse_timeseries.py"},
)
explicit = Dataset(
id=uuid4(),
project_id=project_id,
area_id=area_id,
name="Clipped vector",
dataset_type="vector",
source="manual",
source_metadata={"geometry_clipped_to_area": True},
)
untrusted = Dataset(
id=uuid4(),
project_id=project_id,
area_id=area_id,
name="Assigned only",
dataset_type="vector",
source="manual",
)
assert VectorFeatureService.can_use_full_area_fast_path(trusted, area_id) is True
assert VectorFeatureService.can_use_full_area_fast_path(explicit, area_id) is True
assert VectorFeatureService.can_use_full_area_fast_path(untrusted, area_id) is False
assert VectorFeatureService.can_use_full_area_fast_path(trusted, uuid4()) is False
assert VectorFeatureService.can_use_full_area_fast_path(trusted, None) is False
def test_full_area_summary_uses_exact_stored_values_without_partial_intersection() -> None:
class ScalarQuery:
def __init__(self, value: float) -> None:
self.value = value
def filter(self, *_args):
return self
def scalar(self):
return self.value
class ScalarSession:
def __init__(self, value: float) -> None:
self.value = value
self.query_count = 0
def query(self, *_args):
self.query_count += 1
return ScalarQuery(self.value)
dataset = Dataset(
id=uuid4(),
project_id=uuid4(),
name="Population",
dataset_type="vector",
source="operator_official_import",
source_metadata={
"selection_aggregation": {
"method": "area_weighted_sum",
"property": "population_total",
"label": "Inwoners",
"unit": "inwoners",
"warning_only_when_estimate": True,
"warning": "Partial-sector estimate",
}
},
)
session = ScalarSession(506_473.0)
result = VectorFeatureService.summarize_features_by_bbox(
session,
dataset=dataset,
bbox={"min_x": 4.5, "min_y": 51.0, "max_x": 5.3, "max_y": 51.6, "crs": "EPSG:4326"},
total_feature_count=733,
selection_geometry=SimpleNamespace(),
full_dataset_area=True,
)
assert result["metric_value"] == 506_473.0
assert result["is_estimate"] is False
assert result["warning"] is None
assert session.query_count == 1