Jens/geointel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add semantic GIS selection metrics
GeoIntel CI / docs-smoke (push) Canceled after 0s
Details
GeoIntel CI / contract-smoke (push) Canceled after 0s
Details

Browse Source
This commit is contained in:
Codex
2026-07-15 06:07:34 +02:00
parent 3ff2d07f3c
commit 0baa9b069c
16 changed files with 520 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/vector_feature_service.py
+183 -7
View File
@@ -26,7 +26,110 @@ FULL_AREA_CLIPPED_OPERATOR_TOOLS = {
}
SEMANTIC_SELECTION_METRICS: dict[str, tuple[dict[str, Any], ...]] = {
"buildings": (
{
"metric_key": "footprint_area",
"method": "intersection_area",
"label": "Bebouwde grondoppervlakte",
"unit": "ha",
"geometry_dimension": 2,
"warning": "Dit is de grondoppervlakte van gebouwcontouren, niet de totale vloeroppervlakte of het gebouwvolume.",
},
),
"forest": (
{
"metric_key": "forest_area",
"method": "intersection_area",
"label": "Bosoppervlakte",
"unit": "ha",
"geometry_dimension": 2,
},
),
"water": (
{
"metric_key": "water_area",
"method": "intersection_area",
"label": "Wateroppervlakte",
"unit": "ha",
"geometry_dimension": 2,
"warning": "Watervolume is niet berekenbaar zonder betrouwbare diepte- of bathymetrische gegevens. De kaartbron levert alleen oppervlakte- en lijngeometrie.",
},
{
"metric_key": "watercourse_length",
"method": "intersection_length",
"label": "Lengte waterlopen",
"unit": "km",
"geometry_dimension": 1,
},
),
"roads": (
{
"metric_key": "road_length",
"method": "intersection_length",
"label": "Totale weglengte",
"unit": "km",
"geometry_dimension": 1,
"warning": "De lengte volgt de GRB-wegsegmenten en zegt niets over rijstroken, verkeersvolume of verhardingsoppervlakte.",
},
),
"parcels": (
{
"metric_key": "parcel_area",
"method": "intersection_area",
"label": "Perceeloppervlakte",
"unit": "ha",
"geometry_dimension": 2,
"warning": "GRB-percelen zijn een grafische referentie en vormen geen juridische grensopmeting.",
},
),
}
SEMANTIC_COUNT_LABELS = {
"buildings": "Gebouwen",
"population": "Statistische sectoren",
"forest": "Bosvlakken",
"water": "Waterobjecten",
"roads": "Wegsegmenten",
"parcels": "Percelen",
}
class VectorFeatureService:
@staticmethod
def _dataset_theme(dataset: Dataset) -> str | None:
source_metadata = dataset.source_metadata if isinstance(dataset.source_metadata, dict) else {}
candidates = (
source_metadata.get("theme"),
dataset.reference_layer_name,
source_metadata.get("layer_type"),
)
aliases = {
"building": "buildings",
"bebouwing": "buildings",
"population": "population",
"forest": "forest",
"forestry": "forest",
"waterways": "water",
"road": "roads",
"parcel": "parcels",
}
for candidate in candidates:
if not isinstance(candidate, str) or not candidate.strip():
continue
normalized = candidate.strip().lower()
if normalized.startswith("regional_"):
normalized = normalized.removeprefix("regional_")
normalized = aliases.get(normalized, normalized)
if normalized in {*SEMANTIC_SELECTION_METRICS, "population"}:
return normalized
return None
@staticmethod
def supports_selection_summary(dataset: Dataset) -> bool:
source_metadata = dataset.source_metadata if isinstance(dataset.source_metadata, dict) else {}
return isinstance(source_metadata.get("selection_aggregation"), dict) or VectorFeatureService._dataset_theme(dataset) is not None
@staticmethod
def can_use_full_area_fast_path(dataset: Dataset, selection_area_id: UUID | None) -> bool:
if selection_area_id is None or dataset.area_id != selection_area_id:
@@ -175,7 +278,7 @@ class VectorFeatureService:
selected_rows = rows[:safe_limit]
features = [VectorFeatureService._row_to_geojson_feature(row) for row in selected_rows]
summary = None
if dataset and isinstance(dataset.source_metadata, dict) and dataset.source_metadata.get("selection_aggregation"):
if dataset and VectorFeatureService.supports_selection_summary(dataset):
summary = VectorFeatureService.summarize_features_by_bbox(
db,
dataset=dataset,
@@ -232,13 +335,86 @@ class VectorFeatureService:
config = source_metadata.get("selection_aggregation")
if not isinstance(config, dict):
config = {}
theme = VectorFeatureService._dataset_theme(dataset)
configured_metric = {
"metric_key": str(config.get("metric_key") or config.get("method") or "feature_count"),
"method": str(config.get("method") or "feature_count"),
"label": str(config.get("label") or SEMANTIC_COUNT_LABELS.get(theme or "", "Objecten")),
"unit": str(config.get("unit") or "objecten"),
"warning": str(config["warning"]) if config.get("warning") else None,
"is_estimate": bool(config.get("is_estimate", False)),
**({"property": config.get("property")} if config.get("property") else {}),
}
semantic_metrics = [dict(metric) for metric in SEMANTIC_SELECTION_METRICS.get(theme or "", ())]
primary_config = configured_metric
if configured_metric["method"] == "feature_count" and semantic_metrics:
primary_config = semantic_metrics[0]
metric_configs = [primary_config]
for semantic_metric in semantic_metrics:
signature = (semantic_metric["method"], semantic_metric["unit"])
existing = {
(item["method"], item["unit"])
for item in metric_configs
}
if signature not in existing:
metric_configs.append(semantic_metric)
if not any(item["method"] == "feature_count" for item in metric_configs):
metric_configs.append(
{
"metric_key": "feature_count",
"method": "feature_count",
"label": SEMANTIC_COUNT_LABELS.get(theme or "", "Objecten"),
"unit": "objecten",
}
)
metrics = [
VectorFeatureService._calculate_selection_metric(
db,
dataset=dataset,
config=metric_config,
selection_filter=selection_filter,
selection_shape=selection_shape,
feature_count=feature_count,
full_dataset_area=full_dataset_area,
)
for metric_config in metric_configs
]
primary_metric = metrics[0]
return {
"metric_label": primary_metric["metric_label"],
"metric_value": primary_metric["metric_value"],
"metric_unit": primary_metric["metric_unit"],
"aggregation_method": primary_metric["aggregation_method"],
"primary_metric_key": primary_metric["metric_key"],
"feature_count": feature_count,
"is_estimate": primary_metric["is_estimate"],
"warning": primary_metric.get("warning"),
"metrics": metrics,
}
@staticmethod
def _calculate_selection_metric(
db,
*,
dataset: Dataset,
config: dict[str, Any],
selection_filter: tuple[Any, ...],
selection_shape: Any,
feature_count: int,
full_dataset_area: bool,
) -> dict[str, Any]:
method = str(config.get("method") or "feature_count")
label = str(config.get("label") or "Objecten")
unit = str(config.get("unit") or "objecten")
warning = str(config["warning"]) if config.get("warning") else None
is_estimate = bool(config.get("is_estimate", False))
metric_value = float(feature_count)
dimension = config.get("geometry_dimension")
metric_filter = selection_filter
if dimension in {1, 2}:
metric_filter += (func.ST_Dimension(VectorFeature.geometry) == int(dimension),)
if method == "intersection_area":
measured_geometry = (
VectorFeature.geometry
@@ -246,7 +422,7 @@ class VectorFeatureService:
else func.ST_Intersection(VectorFeature.geometry, selection_shape)
)
area_expression = func.ST_Area(func.ST_Transform(measured_geometry, 31370))
area_m2 = db.query(func.coalesce(func.sum(area_expression), 0.0)).filter(*selection_filter).scalar()
area_m2 = db.query(func.coalesce(func.sum(area_expression), 0.0)).filter(*metric_filter).scalar()
divisor = 10_000.0 if unit == "ha" else 1.0
metric_value = float(area_m2 or 0.0) / divisor
elif method == "intersection_length":
@@ -256,7 +432,7 @@ class VectorFeatureService:
else func.ST_Intersection(VectorFeature.geometry, selection_shape)
)
length_expression = func.ST_Length(func.ST_Transform(measured_geometry, 31370))
length_m = db.query(func.coalesce(func.sum(length_expression), 0.0)).filter(*selection_filter).scalar()
length_m = db.query(func.coalesce(func.sum(length_expression), 0.0)).filter(*metric_filter).scalar()
divisor = 1_000.0 if unit == "km" else 1.0
metric_value = float(length_m or 0.0) / divisor
elif method in {"sum", "area_weighted_sum"}:
@@ -307,11 +483,11 @@ class VectorFeatureService:
)
return {
"metric_label": label,
"metric_key": str(config.get("metric_key") or method),
"metric_label": str(config.get("label") or "Objecten"),
"metric_value": metric_value,
"metric_unit": unit,
"aggregation_method": method,
"feature_count": feature_count,
"is_estimate": is_estimate,
"warning": warning,
}