Optimize assistant context by requested themes

backend/app/services/geo_assistant_service.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import re
 from datetime import datetime, timezone
 from typing import Any
 from urllib.error import HTTPError, URLError
@@ -57,12 +58,50 @@ class GeoAssistantService:
         "accessibility": "bereikbaarheidsscores",
         "services": "voorzieningenscores",
     }
+    THEME_QUERY_TERMS = {
+        "buildings": ("bebouwing", "gebouw", "gebouwen", "gebouwoppervlakte"),
+        "space_occupation": ("ruimtebeslag", "verharding"),
+        "open_space": ("open ruimte", "openruimte"),
+        "population": ("bevolking", "bevolkingsdichtheid", "inwoner", "inwoners"),
+        "forest": ("bos", "bossen", "bosoppervlakte", "groen"),
+        "nature_value": ("natuur", "natuurwaarde", "biodiversiteit", "habitat", "natura 2000"),
+        "agriculture": (
+            "landbouw",
+            "landbouwteelt",
+            "landbouwteelten",
+            "akker",
+            "akkers",
+            "teelt",
+            "teelten",
+            "gewas",
+            "gewassen",
+        ),
+        "soil": ("bodem", "bodemkaart", "bodemtype", "bodemtypes"),
+        "water": ("water", "waterloop", "waterlopen", "waterweg", "waterwegen", "rivier", "beek"),
+        "flood_hazard": ("overstroming", "overstromingen", "inundatie", "waterdiepte"),
+        "terrain": ("hoogte", "reliëf", "terrein", "dhmv"),
+        "accessibility": ("bereikbaarheid", "bereikbaar", "knooppuntwaarde", "collectief vervoer"),
+        "services": ("voorziening", "voorzieningen", "voorzieningenniveau"),
+        "roads": ("weg", "wegen", "wegennet", "rijbaan", "rijbanen", "straat", "straten"),
+        "parcels": ("perceel", "percelen", "kadastraal", "kadaster"),
+    }
 
     @classmethod
     def history_requested(cls, question: str) -> bool:
         normalized = question.casefold()
         return any(keyword in normalized for keyword in cls.HISTORY_KEYWORDS)
 
+    @classmethod
+    def requested_themes(cls, question: str) -> set[str] | None:
+        normalized = " ".join(re.sub(r"[^\w]+", " ", question.casefold()).split())
+        padded = f" {normalized} "
+        themes = {
+            theme
+            for theme, terms in cls.THEME_QUERY_TERMS.items()
+            if any(f" {term} " in padded for term in terms)
+        }
+        return themes or None
+
     @classmethod
     def ensure_estimate_disclosure(
         cls,
@@ -269,17 +308,33 @@ class GeoAssistantService:
             .all()
         )
         vector_datasets = [dataset for dataset in datasets if dataset.dataset_type in {"vector", "geojson"}]
+        requested_themes = self.requested_themes(payload.question)
+        relevant_vector_datasets = [
+            dataset
+            for dataset in vector_datasets
+            if requested_themes is None or VectorFeatureService._dataset_theme(dataset) in requested_themes
+        ]
         flood_hazard_datasets = [
             dataset
             for dataset in datasets
             if dataset.dataset_type == "raster" and dataset.source_name == FloodHazardAcquisitionService.PROVIDER
             and (area is None or dataset.area_id is None or dataset.area_id == area.id)
+            and (requested_themes is None or "flood_hazard" in requested_themes)
         ]
+        thematic_products = ThematicRasterAcquisitionService._products()
         thematic_candidates = [
             dataset
             for dataset in datasets
             if dataset.dataset_type == "raster" and dataset.source_name == ThematicRasterAcquisitionService.PROVIDER
             and (area is None or dataset.area_id is None or dataset.area_id == area.id)
+            and (
+                requested_themes is None
+                or (
+                    str((dataset.source_metadata or {}).get("product_key") or "") in thematic_products
+                    and thematic_products[str((dataset.source_metadata or {}).get("product_key") or "")].theme
+                    in requested_themes
+                )
+            )
         ]
         thematic_by_product: dict[str, Dataset] = {}
         for dataset in thematic_candidates:
@@ -294,7 +349,7 @@ class GeoAssistantService:
         current_context: list[dict[str, Any]] = []
 
         if bbox is not None:
-            for dataset in self._current_datasets(vector_datasets):
+            for dataset in self._current_datasets(relevant_vector_datasets):
                 kwargs: dict[str, Any] = {"dataset": dataset, "bbox": bbox}
                 if area is not None:
                     kwargs["selection_geometry"] = area.geometry
@@ -441,7 +496,7 @@ class GeoAssistantService:
         temporal_series: list[AssistantTemporalSeries] = []
         temporal_context: list[dict[str, Any]] = []
         include_history = self.history_requested(payload.question)
-        for key, observations in self._series(vector_datasets):
+        for key, observations in self._series(relevant_vector_datasets):
             first = observations[0]
             last = observations[-1]
             source_metadata = last.source_metadata if isinstance(last.source_metadata, dict) else {}
@@ -483,7 +538,12 @@ class GeoAssistantService:
 
         context = {
             "project": {"id": str(project.id), "name": project.name, "region": project.region},
-            "scope": {"label": scope_label, "bbox": bbox, "exact_area_geometry_used": area is not None},
+            "scope": {
+                "label": scope_label,
+                "bbox": bbox,
+                "exact_area_geometry_used": area is not None,
+                "requested_themes": sorted(requested_themes) if requested_themes is not None else None,
+            },
             "current_measurements": current_context,
             "available_temporal_series": temporal_context,
             "rules": {