feat: calculate service provision for unsaved planning variants

app/common/modules/effects_api_gateway.py

@@ -1,4 +1,8 @@
 import asyncio
+import hashlib
+import json
+import os
+from pathlib import Path
 
 import geopandas as gpd
 import pandas as pd
@@ -64,6 +68,39 @@ async def get_service_normative(
                 headers={USER_ID_HEADER: token} if token else None,
             )
             request_ter_id = territory_id
+        # Explicit local assessment fixture: real Urban values always take precedence.
+        fixture_path = os.getenv("TEST_SERVICE_NORMATIVES_FILE")
+        if fixture_path and not any(
+            (row.get("service_type") or {}).get("id") == service_type_id
+            for row in response or []
+        ):
+            raw = Path(fixture_path).read_bytes()
+            fixture = json.loads(raw)
+            if fixture.get("source_kind") != "test_mock":
+                raise ValueError("Normative fixture must declare source_kind=test_mock")
+            if request_ter_id in fixture["territory_ids"]:
+                rows = [
+                    row
+                    for row in fixture["normatives"]
+                    if row["service_type"]["id"] == service_type_id
+                ]
+                response = list(response or []) + [
+                    dict(
+                        row,
+                        source={
+                            "kind": "test_mock",
+                            "fixture_id": fixture["id"],
+                            "sha256": hashlib.sha256(raw).hexdigest(),
+                            "description": fixture["description"],
+                            "legal_compliance_claim": False,
+                        },
+                    )
+                    for row in rows
+                ]
+        if not response:
+            raise self._missing_service_normative(
+                territory_id, context_ids, request_ter_id, service_type_id, []
+            )
         response_df = pd.DataFrame.from_records(response)
         response_df["service_type_id"] = response_df["service_type"].apply(
             lambda x: x["id"] if x else None
@@ -72,17 +109,12 @@ async def get_service_normative(
             response_df["service_type_id"] == service_type_id
         ].copy()
         if len(service_type) < 1:
-            raise http_exception(
-                400,
-                msg="Service type id not found in urban_db for provided territory/context ids. ",
-                _input={
-                    "territory_id": territory_id,
-                    "context_ids": context_ids,
-                    "service_type_id": service_type_id,
-                },
-                _detail={
-                    "Available service ids": response_df["service_type_id"].to_list()
-                },
+            raise self._missing_service_normative(
+                territory_id,
+                context_ids,
+                request_ter_id,
+                service_type_id,
+                response_df["service_type_id"].dropna().to_list(),
             )
 
         service_type = (
@@ -140,6 +172,30 @@ async def get_service_normative(
             _detail={"Available service ids": response_df["service_type_id"].to_list()},
         )
 
+    @staticmethod
+    def _missing_service_normative(
+        territory_id, context_ids, request_ter_id, service_type_id, available_ids
+    ):
+        return http_exception(
+            400,
+            msg="Норматив обеспеченности для выбранного вида услуг не задан в Urban API.",
+            _input={
+                "territory_id": territory_id,
+                "context_ids": context_ids,
+                "request_ter_id": request_ter_id,
+                "service_type_id": service_type_id,
+            },
+            _detail={
+                "code": "missing_service_normative",
+                "Available service ids": available_ids,
+                "required_action": (
+                    f"Укажите или добавьте в Urban API норматив для территории {request_ter_id} "
+                    f"и вида услуг {service_type_id}: радиус или время доступности и норму "
+                    "обеспеченности. Либо выберите территорию с заданным нормативом."
+                ),
+            },
+        )
+
     async def get_project_data(
         self, project_id: int, token: str
     ) -> dict[str, int | dict]:

app/effects/effects_service.py

@@ -375,7 +375,7 @@ async def calculate_effects(
                 before_prove_data["services"],
                 after_prove_data["services"],
             )
-        return EffectsSchema(**result)
+        return EffectsSchema(**result, normative=normative_data)
 
     @staticmethod
     async def form_llm_context(

app/effects/shemas/effects_base_schema.py

@@ -23,6 +23,7 @@ class PivotSchema(BaseModel):
 
 
 class EffectsSchema(BaseModel):
+    normative: dict | None = None
 
     before_prove_data: ProvisionSchema
     after_prove_data: ProvisionSchema

app/provision/provision_mcp.py

@@ -9,11 +9,54 @@
 from app.schemas.provision_base_schema import (
     MultiProvisionRequestSchema,
     ServiceInfoSchema,
+    VariantProvisionRequestSchema,
 )
 
 provision_mcp = FastMCP("Object Provision MCP server", auth=service_token_verifier)
 
 
+@provision_mcp.tool(
+    name="CalculateVariantServicesProvision",
+    title="Evaluate an unsaved planning variant",
+    description="Calculate provision using new GenBuilder residential buildings and/or proposed services, "
+    "preserving existing scenario buildings and services. No Urban API writes. "
+    "generated_buildings is a WGS84 FeatureCollection with properties.zone=residential and floors_count; "
+    "is_excluded features are ignored because existing buildings and attributes are retained from Urban. "
+    "additional_services maps service type IDs to new WGS84 layers with explicit positive capacity and service_type_id. "
+    "target_population is the TOTAL population of the whole scenario including existing residents. "
+    "Demand is distributed using the existing floor-area restoration and gravity accessibility model. "
+    "Call CalculateServicesProvision separately for the baseline; compare summaries and full layers.",
+)
+async def calc_variant_services_provision(
+    scenario_id: int,
+    services: dict[int, ServiceInfoSchema],
+    target_population: int,
+    generated_buildings: dict | None = None,
+    additional_services: dict[int, dict] | None = None,
+):
+    if generated_buildings is None and not additional_services:
+        raise ValueError(
+            "A variant requires generated buildings or additional services"
+        )
+    params = VariantProvisionRequestSchema(
+        scenario_id=scenario_id,
+        services=services,
+        target_population=target_population,
+        generated_buildings=generated_buildings,
+        additional_services=additional_services or {},
+    )
+    result = await provision_mcp_service.calculate_multi_provision(
+        params, get_mcp_user_id()
+    )
+    return {
+        **result.model_dump(),
+        "scenario_id": scenario_id,
+        "target_population": target_population,
+        "methodology": "Existing buildings and services preserved; population distributed by restored floor area; gravity accessibility model",
+        "variant": True,
+    }
+
+
 @provision_mcp.tool(
     name="CalculateServiceProvision",
     title="Get service provision for scenario",
@@ -52,9 +95,7 @@ async def calc_service_provision(
             service_type_id=service_type_id,
             target_population=target_population,
         )
-        result = await provision_mcp_service.calculate_provision(
-            provision_dto, user_id
-        )
+        result = await provision_mcp_service.calculate_provision(provision_dto, user_id)
         return result.model_dump()
     except Exception as e:
         tb = traceback.format_exc()

app/provision/provision_service.py

@@ -14,6 +14,7 @@
     objectnat_calculator,
 )
 from app.dto.provision_dto import ProvisionDTO
+from app.provision.variant import add_generated_buildings, additional_services
 from app.schemas.provision_base_schema import (
     MultiProvisionRequestSchema,
     MultiProvisionSchema,
@@ -105,6 +106,7 @@ async def _calculate_for_service(
             service_type_id=service_type_id,
             token=token,
         )
+        shared_data.setdefault("normatives", {})[service_type_id] = normative_data
         context_buildings = await asyncio.to_thread(
             data_restorator.restore_demands,
             buildings=shared_data["context_buildings"].copy(),
@@ -146,6 +148,23 @@ async def _calculate_for_service(
             services=target_scenario_services,
             service_default_capacity=service_default_capacity,
         )
+        extra = shared_data.get("additional_services", {}).get(service_type_id)
+        if extra is not None:
+            first_id = (
+                min(
+                    [
+                        0,
+                        *context_services.get("service_id", []),
+                        *target_scenario_services.get("service_id", []),
+                    ]
+                )
+                - 1
+            )
+            proposed = additional_services(extra, service_type_id, first_id)
+            target_scenario_services = gpd.GeoDataFrame(
+                pd.concat([target_scenario_services, proposed], ignore_index=True),
+                crs=4326,
+            )
         before_buildings = await asyncio.to_thread(
             pd.concat,
             objs=[context_buildings, target_scenario_buildings],
@@ -260,7 +279,12 @@ async def calculate_provision(
         logger.info(
             f"Calculated PROVISION for {provision_params.scenario_id} and {provision_params.service_type_id}"
         )
-        return ProvisionSchema(**result)
+        return ProvisionSchema(
+            **result,
+            normative=shared_data.get("normatives", {}).get(
+                provision_params.service_type_id
+            ),
+        )
 
     async def calculate_multi_provision(
         self, multi_params: MultiProvisionRequestSchema, token: str
@@ -288,6 +312,17 @@ async def calculate_multi_provision(
         )
         if multi_params.target_population:
             shared_data["target_scenario_population"] = multi_params.target_population
+        generated = getattr(multi_params, "generated_buildings", None)
+        additions = getattr(multi_params, "additional_services", {})
+        if generated is not None:
+            shared_data["target_scenario_buildings"] = add_generated_buildings(
+                shared_data["target_scenario_buildings"], generated
+            )
+        if not set(additions) <= set(multi_params.services):
+            raise ValueError(
+                "Additional services must belong to requested service types"
+            )
+        shared_data["additional_services"] = additions
         results = {}
         for service_type_id, service_info in multi_params.services.items():
             try:
@@ -321,6 +356,7 @@ async def calculate_multi_provision(
                     services=before_prove_data["services"],
                 ),
                 layers=layers,
+                normative=shared_data.get("normatives", {}).get(service_type_id),
             )
         logger.info(f"Calculated MULTI PROVISION for {multi_params.scenario_id}")
         return MultiProvisionSchema(services=results)

app/provision/variant.py

@@ -0,0 +1,91 @@
+"""Transient additions to a scenario; never write a design back to Urban API."""
+
+from math import isfinite
+
+import geopandas as gpd
+import pandas as pd
+from shapely.geometry import shape
+
+
+def validated_features(layer):
+    if not isinstance(layer, dict) or layer.get("type") != "FeatureCollection":
+        raise ValueError("A WGS84 FeatureCollection is required")
+    features = layer.get("features")
+    if not isinstance(features, list):
+        raise ValueError("FeatureCollection.features must be a list")
+    for feature in features:
+        geometry = shape(feature["geometry"])
+        if not geometry.is_valid or geometry.is_empty:
+            raise ValueError("Variant geometry must be valid and nonempty")
+        xmin, ymin, xmax, ymax = geometry.bounds
+        if not (-180 <= xmin <= xmax <= 180 and -90 <= ymin <= ymax <= 90):
+            raise ValueError("Variant geometry must use WGS84 longitude/latitude")
+        yield geometry, feature.get("properties") or {}
+
+
+def add_generated_buildings(baseline, layer):
+    """Keep baseline attributes; GenBuilder excluded features carry zeroed values."""
+    rows = []
+    first_id = min([0, *baseline.get("building_id", [])]) - 1
+    for geometry, props in validated_features(layer):
+        if props.get("is_excluded"):
+            continue  # Baseline buildings are retained, not replaced by these stubs.
+        if props.get("zone") != "residential":
+            continue
+        floors = props.get("floors_count")
+        if (
+            isinstance(floors, bool)
+            or not isinstance(floors, (int, float))
+            or not isfinite(floors)
+            or floors <= 0
+        ):
+            raise ValueError(
+                "Generated residential buildings require positive floors_count"
+            )
+        if geometry.geom_type not in {"Polygon", "MultiPolygon"}:
+            raise ValueError("Generated buildings must be polygons")
+        rows.append(
+            {
+                "geometry": geometry,
+                "building_id": first_id - len(rows),
+                "storeys_count": floors,
+            }
+        )
+    if not rows:
+        raise ValueError("No generated residential buildings in supplied variant")
+    added = gpd.GeoDataFrame(rows, geometry="geometry", crs=4326)
+    if baseline.empty:
+        return added
+    return gpd.GeoDataFrame(
+        pd.concat([baseline.to_crs(4326), added], ignore_index=True), crs=4326
+    )
+
+
+def additional_services(layer, service_type_id, first_id=-1):
+    rows = []
+    for geometry, props in validated_features(layer):
+        capacity = props.get("capacity")
+        if (
+            isinstance(capacity, bool)
+            or not isinstance(capacity, (int, float))
+            or not isfinite(capacity)
+            or capacity <= 0
+        ):
+            raise ValueError("Proposed services require an explicit positive capacity")
+        if props.get("service_type_id") != service_type_id:
+            raise ValueError(
+                "Proposed service_type_id must match its requested service group"
+            )
+        rows.append(
+            {
+                "geometry": geometry,
+                "service_id": first_id - len(rows),
+                "capacity": capacity,
+            }
+        )
+    return gpd.GeoDataFrame(
+        rows,
+        columns=["geometry", "service_id", "capacity"],
+        geometry="geometry",
+        crs=4326,
+    )

app/schemas/provision_base_schema.py

@@ -32,6 +32,8 @@ class FeatureCollectionSchema(BaseModel):
 
 class ProvisionSchema(BaseModel):
 
+    normative: dict | None = None
+
     buildings: FeatureCollectionSchema
     services: FeatureCollectionSchema
     links: FeatureCollectionSchema
@@ -85,8 +87,25 @@ class ProvisionSummarySchema(BaseModel):
     median_provision_value: float | None
 
 
+class VariantProvisionRequestSchema(MultiProvisionRequestSchema):
+    target_population: int = Field(
+        gt=0,
+        description="Total population of the whole target scenario, including preserved buildings; not only new residents",
+    )
+    generated_buildings: dict | None = Field(
+        default=None,
+        description="GenBuilder WGS84 FeatureCollection; new residential buildings are added to existing scenario buildings",
+    )
+    additional_services: dict[int, dict] = Field(
+        default_factory=dict,
+        description="New service layers keyed by service type ID; properties.capacity and properties.service_type_id required",
+    )
+
+
 class ServiceProvisionResultSchema(BaseModel):
 
+    normative: dict | None = None
+
     name: str
     summary: ProvisionSummarySchema | None = None
     layers: ProvisionSchema | None = None