feat: add source-grounded evolution and Ollama assistant

backend/app/services/temporal_analysis_service.py

@@ -10,13 +10,15 @@ from shapely.geometry import mapping
 from sqlalchemy.orm import Session
 
 from app.core.errors import AppError
-from app.models import Dataset, VectorFeature
+from app.models import Area, Dataset, VectorFeature
 from app.schemas.temporal import (
     TemporalComparisonRequest,
     TemporalComparisonResponse,
     TemporalDatasetRef,
     TemporalMetricComparison,
     TemporalObjectChanges,
+    TemporalObservation,
+    TemporalObservationMetric,
     TemporalSeriesDataset,
     TemporalSeriesRead,
 )
@@ -104,22 +106,38 @@ class TemporalAnalysisService:
             )
 
         bbox = payload.bbox.model_dump()
-        earlier_summary = VectorFeatureService.summarize_features_by_bbox(db, dataset=earlier, bbox=bbox)
-        later_summary = VectorFeatureService.summarize_features_by_bbox(db, dataset=later, bbox=bbox)
-        if (
-            earlier_summary["aggregation_method"] != later_summary["aggregation_method"]
-            or earlier_summary["metric_unit"] != later_summary["metric_unit"]
-        ):
+        selection_area = TemporalAnalysisService._get_selection_area(db, project_id, payload.area_id)
+        summaries: dict[UUID, dict[str, Any]] = {}
+
+        def summarize(dataset: Dataset) -> dict[str, Any]:
+            cached = summaries.get(dataset.id)
+            if cached is not None:
+                return cached
+            kwargs: dict[str, Any] = {"dataset": dataset, "bbox": bbox}
+            if selection_area is not None:
+                kwargs["selection_geometry"] = selection_area.geometry
+                kwargs["full_dataset_area"] = VectorFeatureService.can_use_full_area_fast_path(
+                    dataset,
+                    selection_area.id,
+                )
+            summary = VectorFeatureService.summarize_features_by_bbox(db, **kwargs)
+            summaries[dataset.id] = summary
+            return summary
+
+        earlier_summary = summarize(earlier)
+        later_summary = summarize(later)
+        metric_comparisons = TemporalAnalysisService._compare_summary_metrics(earlier_summary, later_summary)
+        if not metric_comparisons:
             raise AppError(
                 code="INCOMPATIBLE_TEMPORAL_AGGREGATION",
                 message="Dataset snapshots use incompatible aggregation semantics",
                 status_code=400,
             )
-
-        earlier_value = float(earlier_summary["metric_value"])
-        later_value = float(later_summary["metric_value"])
-        absolute_change = later_value - earlier_value
-        percent_change = (absolute_change / earlier_value * 100.0) if earlier_value else None
+        primary_key = str(later_summary.get("primary_metric_key") or metric_comparisons[0].metric_key)
+        primary_metric = next(
+            (metric for metric in metric_comparisons if metric.metric_key == primary_key),
+            metric_comparisons[0],
+        )
         warnings = [
             warning
             for warning in {earlier_summary.get("warning"), later_summary.get("warning")}
@@ -132,8 +150,26 @@ class TemporalAnalysisService:
             later=later,
             bbox=bbox,
             preview_limit=payload.preview_limit,
+            selection_geometry=selection_area.geometry if selection_area is not None else None,
+            earlier_full_dataset_area=(
+                VectorFeatureService.can_use_full_area_fast_path(earlier, selection_area.id)
+                if selection_area is not None
+                else False
+            ),
+            later_full_dataset_area=(
+                VectorFeatureService.can_use_full_area_fast_path(later, selection_area.id)
+                if selection_area is not None
+                else False
+            ),
         )
         warnings.extend(identity_warnings)
+        timeline = TemporalAnalysisService._build_timeline(
+            db,
+            project_id=project_id,
+            series_key=earlier.temporal_series_key,
+            fallback_datasets=[earlier, later],
+            summarize=summarize,
+        )
 
         return TemporalComparisonResponse(
             temporal_series_key=earlier.temporal_series_key,
@@ -150,22 +186,132 @@ class TemporalAnalysisService:
                 source_version=later.source_version,
             ),
             selection_bbox=payload.bbox,
-            metric=TemporalMetricComparison(
-                label=str(later_summary["metric_label"]),
-                unit=str(later_summary["metric_unit"]),
-                aggregation_method=str(later_summary["aggregation_method"]),
-                earlier_value=earlier_value,
-                later_value=later_value,
-                absolute_change=absolute_change,
-                percent_change=percent_change,
-                is_estimate=bool(earlier_summary["is_estimate"] or later_summary["is_estimate"]),
-            ),
+            selection_area_id=selection_area.id if selection_area is not None else None,
+            metric=primary_metric,
+            metrics=metric_comparisons,
+            timeline=timeline,
             object_changes=object_changes,
             geojson=geojson,
             warnings=warnings,
             generated_at=datetime.now(timezone.utc),
         )
 
+    @staticmethod
+    def _get_selection_area(db: Session, project_id: UUID, area_id: UUID | None) -> Area | None:
+        if area_id is None:
+            return None
+        area = db.get(Area, area_id)
+        if area is None or area.project_id != project_id:
+            raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404)
+        return area
+
+    @staticmethod
+    def _summary_metrics(summary: dict[str, Any]) -> list[dict[str, Any]]:
+        configured = summary.get("metrics")
+        if isinstance(configured, list) and configured:
+            return [item for item in configured if isinstance(item, dict)]
+        return [
+            {
+                "metric_key": summary.get("primary_metric_key") or "primary",
+                "metric_label": summary["metric_label"],
+                "metric_value": summary["metric_value"],
+                "metric_unit": summary["metric_unit"],
+                "aggregation_method": summary["aggregation_method"],
+                "is_estimate": summary.get("is_estimate", False),
+                "warning": summary.get("warning"),
+            }
+        ]
+
+    @staticmethod
+    def _compare_summary_metrics(
+        earlier_summary: dict[str, Any],
+        later_summary: dict[str, Any],
+    ) -> list[TemporalMetricComparison]:
+        earlier_metrics = {
+            str(item.get("metric_key") or item.get("aggregation_method") or "primary"): item
+            for item in TemporalAnalysisService._summary_metrics(earlier_summary)
+        }
+        comparisons: list[TemporalMetricComparison] = []
+        for later_metric in TemporalAnalysisService._summary_metrics(later_summary):
+            key = str(later_metric.get("metric_key") or later_metric.get("aggregation_method") or "primary")
+            earlier_metric = earlier_metrics.get(key)
+            if earlier_metric is None:
+                continue
+            if (
+                earlier_metric.get("aggregation_method") != later_metric.get("aggregation_method")
+                or earlier_metric.get("metric_unit") != later_metric.get("metric_unit")
+            ):
+                continue
+            earlier_value = float(earlier_metric.get("metric_value") or 0.0)
+            later_value = float(later_metric.get("metric_value") or 0.0)
+            absolute_change = later_value - earlier_value
+            warning = later_metric.get("warning") or earlier_metric.get("warning")
+            comparisons.append(
+                TemporalMetricComparison(
+                    metric_key=key,
+                    label=str(later_metric.get("metric_label") or key),
+                    unit=str(later_metric.get("metric_unit") or ""),
+                    aggregation_method=str(later_metric.get("aggregation_method") or "feature_count"),
+                    earlier_value=earlier_value,
+                    later_value=later_value,
+                    absolute_change=absolute_change,
+                    percent_change=(absolute_change / earlier_value * 100.0) if earlier_value else None,
+                    is_estimate=bool(earlier_metric.get("is_estimate") or later_metric.get("is_estimate")),
+                    warning=str(warning) if warning else None,
+                )
+            )
+        return comparisons
+
+    @staticmethod
+    def _build_timeline(
+        db: Session,
+        *,
+        project_id: UUID,
+        series_key: str,
+        fallback_datasets: list[Dataset],
+        summarize,
+    ) -> list[TemporalObservation]:
+        if hasattr(db, "query"):
+            datasets = (
+                db.query(Dataset)
+                .filter(Dataset.project_id == project_id)
+                .filter(Dataset.temporal_series_key == series_key)
+                .filter(Dataset.observed_at.isnot(None))
+                .order_by(Dataset.observed_at.asc())
+                .all()
+            )
+        else:
+            datasets = fallback_datasets
+        unique = {dataset.id: dataset for dataset in datasets}
+        ordered = sorted(unique.values(), key=lambda item: item.observed_at or datetime.min.replace(tzinfo=timezone.utc))
+        observations: list[TemporalObservation] = []
+        for dataset in ordered:
+            if dataset.observed_at is None:
+                continue
+            metrics = [
+                TemporalObservationMetric(
+                    metric_key=str(item.get("metric_key") or item.get("aggregation_method") or "primary"),
+                    label=str(item.get("metric_label") or "Meting"),
+                    value=float(item.get("metric_value") or 0.0),
+                    unit=str(item.get("metric_unit") or ""),
+                    aggregation_method=str(item.get("aggregation_method") or "feature_count"),
+                    is_estimate=bool(item.get("is_estimate")),
+                )
+                for item in TemporalAnalysisService._summary_metrics(summarize(dataset))
+            ]
+            observations.append(
+                TemporalObservation(
+                    dataset=TemporalDatasetRef(
+                        id=dataset.id,
+                        name=dataset.name,
+                        observed_at=dataset.observed_at,
+                        source_version=dataset.source_version,
+                    ),
+                    metrics=metrics,
+                )
+            )
+        return observations
+
     @staticmethod
     def _get_temporal_dataset(db: Session, project_id: UUID, dataset_id: UUID, label: str) -> Dataset:
         dataset = db.get(Dataset, dataset_id)
@@ -193,6 +339,9 @@ class TemporalAnalysisService:
         later: Dataset,
         bbox: dict[str, Any],
         preview_limit: int,
+        selection_geometry: Any | None = None,
+        earlier_full_dataset_area: bool = False,
+        later_full_dataset_area: bool = False,
     ) -> tuple[TemporalObjectChanges, dict[str, Any], list[str]]:
         earlier_config = earlier.source_metadata if isinstance(earlier.source_metadata, dict) else {}
         later_config = later.source_metadata if isinstance(later.source_metadata, dict) else {}
@@ -204,27 +353,29 @@ class TemporalAnalysisService:
             )
 
         normalized_bbox = VectorFeatureService._normalize_selection_bbox(bbox)
-        envelope = ST_MakeEnvelope(
-            normalized_bbox["min_x"],
-            normalized_bbox["min_y"],
-            normalized_bbox["max_x"],
-            normalized_bbox["max_y"],
-            4326,
-        )
+        selection_shape = selection_geometry
+        if selection_shape is None:
+            selection_shape = ST_MakeEnvelope(
+                normalized_bbox["min_x"],
+                normalized_bbox["min_y"],
+                normalized_bbox["max_x"],
+                normalized_bbox["max_y"],
+                4326,
+            )
 
-        def load(dataset_id: UUID) -> list[VectorFeature]:
+        def load(dataset_id: UUID, full_dataset_area: bool) -> list[VectorFeature]:
+            query = db.query(VectorFeature).filter(VectorFeature.dataset_id == dataset_id)
+            if not full_dataset_area:
+                query = query.filter(ST_Intersects(VectorFeature.geometry, selection_shape))
             return (
-                db.query(VectorFeature)
-                .filter(VectorFeature.dataset_id == dataset_id)
-                .filter(ST_Intersects(VectorFeature.geometry, envelope))
-                .filter(VectorFeature.source_feature_id.isnot(None))
+                query.filter(VectorFeature.source_feature_id.isnot(None))
                 .order_by(VectorFeature.source_feature_id.asc())
                 .limit(TemporalAnalysisService.IDENTITY_COMPARISON_LIMIT + 1)
                 .all()
             )
 
-        earlier_rows = load(earlier.id)
-        later_rows = load(later.id)
+        earlier_rows = load(earlier.id, earlier_full_dataset_area)
+        later_rows = load(later.id, later_full_dataset_area)
         if (
             len(earlier_rows) > TemporalAnalysisService.IDENTITY_COMPARISON_LIMIT
             or len(later_rows) > TemporalAnalysisService.IDENTITY_COMPARISON_LIMIT