feat(geospatial): read planning restrictions co-located with coordinates

Slice 3a (ADR-0020). PlanningRestrictions relocated out of the solid-wall
generator into domain/geospatial/ as the shared, Property-level value object
(three distinct flags + measure-specific blocks_external/blocks_internal).
GeospatialRepository gains a non-abstract planning_restrictions_for defaulting
to None (sources without the flags need not implement it); GeospatialS3Repository
reads conservation_status/is_listed_building/is_heritage_building from the same
Open-UPRN partition as the coordinates (legacy column names — to confirm in the
S3 deep-dive). Shared _row_for helper dedups the partition lookup.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

domain/geospatial/planning_restrictions.py

@@ -0,0 +1,35 @@
+"""A Property's planning protections, resolved from geospatial reference data.
+
+Three distinct flags (never the legacy collapsed `restricted_measures` boolean
+— ADR-0020): a conservation area, a listed building, a heritage building. They
+gate retrofit measures differently — a conservation area blocks external work
+only, while listed/heritage protect the fabric itself — so the
+measure-specific interpretation (`blocks_external` / `blocks_internal`) lives
+here as derived queries. Sourced onto the Property from the geospatial layer
+(co-located with the coordinates); defaults to unrestricted.
+"""
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class PlanningRestrictions:
+    """The planning protections on a Property that gate wall insulation
+    (ADR-0019). Defaults to unrestricted."""
+
+    in_conservation_area: bool = False
+    is_listed: bool = False
+    is_heritage: bool = False
+
+    @property
+    def blocks_external(self) -> bool:
+        """External wall insulation is blocked by any protection (it alters the
+        external appearance / protected fabric)."""
+        return self.in_conservation_area or self.is_listed or self.is_heritage
+
+    @property
+    def blocks_internal(self) -> bool:
+        """Internal wall insulation is blocked only where the fabric itself is
+        protected — a listed or heritage building, not a plain conservation
+        area."""
+        return self.is_listed or self.is_heritage

domain/modelling/generators/solid_wall_recommendation.py

@@ -15,7 +15,6 @@ generator. Detection + pricing only; impact is produced later by scoring
 (ADR-0016).
 """
 
-from dataclasses import dataclass
 from typing import Final, Optional
 
 from datatypes.epc.domain.epc_property_data import (
@@ -24,6 +23,7 @@ from datatypes.epc.domain.epc_property_data import (
 )
 from datatypes.epc.domain.field_mappings import PROPERTY_TYPE_LOOKUP
 from domain.building_geometry import gross_heat_loss_wall_area
+from domain.geospatial.planning_restrictions import PlanningRestrictions
 from domain.modelling.recommendation import Cost, MeasureOption, Recommendation
 from domain.modelling.simulation import BuildingPartOverlay, EpcSimulation
 from repositories.product.product_repository import ProductRepository
@@ -31,27 +31,6 @@ from repositories.product.product_repository import ProductRepository
 _EXTERNAL_MEASURE_TYPE: Final[str] = "external_wall_insulation"
 _INTERNAL_MEASURE_TYPE: Final[str] = "internal_wall_insulation"
 
-
-@dataclass(frozen=True)
-class PlanningRestrictions:
-    """A Property's planning protections that gate wall insulation (ADR-0019).
-    A conservation area blocks EWI only (external appearance); a listed or
-    heritage building blocks both EWI and IWI (protected fabric). Sourced from
-    the geospatial layer onto the Property in a later slice (ADR-0020); defaults
-    to unrestricted."""
-
-    in_conservation_area: bool = False
-    is_listed: bool = False
-    is_heritage: bool = False
-
-    @property
-    def blocks_external(self) -> bool:
-        return self.in_conservation_area or self.is_listed or self.is_heritage
-
-    @property
-    def blocks_internal(self) -> bool:
-        return self.is_listed or self.is_heritage
-
 # RdSAP `wall_construction` codes (consistent across paths for 1-5).
 _WALL_SOLID_BRICK: Final[int] = 3
 _WALL_TIMBER_FRAME: Final[int] = 5

repositories/geospatial/geospatial_repository.py

@@ -4,6 +4,7 @@ from abc import ABC, abstractmethod
 from typing import Optional
 
 from domain.geospatial.coordinates import Coordinates
+from domain.geospatial.planning_restrictions import PlanningRestrictions
 
 
 class GeospatialRepository(ABC):
@@ -15,3 +16,10 @@ class GeospatialRepository(ABC):
 
     @abstractmethod
     def coordinates_for(self, uprn: int) -> Optional[Coordinates]: ...
+
+    def planning_restrictions_for(self, uprn: int) -> Optional[PlanningRestrictions]:
+        """The Property's planning protections (conservation/listed/heritage),
+        co-located with the coordinates in the reference data (ADR-0020).
+        Defaults to None (unknown → unrestricted) so reference sources that
+        don't carry the flags need not implement it."""
+        return None

repositories/geospatial/geospatial_s3_repository.py

@@ -1,11 +1,12 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from typing import Optional
+from typing import Any, Optional
 
 import pandas as pd
 
 from domain.geospatial.coordinates import Coordinates
+from domain.geospatial.planning_restrictions import PlanningRestrictions
 from repositories.geospatial.geospatial_repository import GeospatialRepository
 
 ParquetReader = Callable[[str], pd.DataFrame]
@@ -25,7 +26,9 @@ class GeospatialS3Repository(GeospatialRepository):
     def __init__(self, read_parquet: ParquetReader) -> None:
         self._read_parquet = read_parquet
 
-    def coordinates_for(self, uprn: int) -> Optional[Coordinates]:
+    def _row_for(self, uprn: int) -> Optional["pd.Series[Any]"]:
+        """The Open-UPRN partition row for ``uprn`` (coordinates + co-located
+        planning flags), or None when no partition covers it / it is absent."""
         meta = self._read_parquet(_META_KEY)
         covering = meta[(meta["lower"] <= uprn) & (meta["upper"] >= uprn)]
         if covering.empty:
@@ -36,8 +39,23 @@ class GeospatialS3Repository(GeospatialRepository):
         rows = partition[partition["UPRN"] == uprn]
         if rows.empty:
             return None
-        row = rows.iloc[0]
+        return rows.iloc[0]
+
+    def coordinates_for(self, uprn: int) -> Optional[Coordinates]:
+        row = self._row_for(uprn)
+        if row is None:
+            return None
         return Coordinates(
             longitude=float(row["LONGITUDE"]),
             latitude=float(row["LATITUDE"]),
         )
+
+    def planning_restrictions_for(self, uprn: int) -> Optional[PlanningRestrictions]:
+        row = self._row_for(uprn)
+        if row is None:
+            return None
+        return PlanningRestrictions(
+            in_conservation_area=bool(row["conservation_status"]),
+            is_listed=bool(row["is_listed_building"]),
+            is_heritage=bool(row["is_heritage_building"]),
+        )

tests/domain/modelling/test_elmhurst_cascade_pins.py

@@ -28,8 +28,8 @@ from domain.modelling.generators.floor_recommendation import recommend_floor_ins
 from domain.modelling.generators.roof_recommendation import recommend_loft_insulation
 from domain.modelling.simulation import BuildingPartOverlay, EpcSimulation
 from domain.modelling.generators.wall_recommendation import recommend_cavity_wall
+from domain.geospatial.planning_restrictions import PlanningRestrictions
 from domain.modelling.generators.solid_wall_recommendation import (
-    PlanningRestrictions,
     recommend_solid_wall,
 )
 from domain.modelling.recommendation import MeasureOption

tests/repositories/geospatial/test_geospatial_repository.py

@@ -14,6 +14,7 @@ from pathlib import Path
 import pandas as pd
 
 from domain.geospatial.coordinates import Coordinates
+from domain.geospatial.planning_restrictions import PlanningRestrictions
 from repositories.geospatial.geospatial_s3_repository import GeospatialS3Repository
 
 
@@ -35,6 +36,11 @@ def _write_open_uprn(base: Path) -> None:
             "UPRN": [12345, 12346],
             "LATITUDE": [51.5074, 51.6000],
             "LONGITUDE": [-0.1278, -0.2000],
+            # Planning flags co-located with the coordinates in the partition
+            # (legacy column names — confirm exact names in the S3 deep-dive).
+            "conservation_status": [True, False],
+            "is_listed_building": [False, True],
+            "is_heritage_building": [False, False],
         }
     ).to_parquet(spatial / "0_100000.parquet")
 
@@ -69,3 +75,28 @@ def test_coordinates_for_returns_none_when_no_partition_covers_uprn(
 
     # Act / Assert — uprn beyond every partition's range
     assert repo.coordinates_for(500000) is None
+
+
+def test_planning_restrictions_for_reads_the_co_located_flags(tmp_path: Path) -> None:
+    # Arrange — same partition, planning flags alongside the coordinates.
+    _write_open_uprn(tmp_path)
+    repo = GeospatialS3Repository(_reader(tmp_path))
+
+    # Act
+    restrictions = repo.planning_restrictions_for(12345)
+
+    # Assert — the three flags come back as the Property's PlanningRestrictions.
+    assert restrictions == PlanningRestrictions(
+        in_conservation_area=True, is_listed=False, is_heritage=False
+    )
+
+
+def test_planning_restrictions_for_returns_none_when_uprn_absent(
+    tmp_path: Path,
+) -> None:
+    # Arrange
+    _write_open_uprn(tmp_path)
+    repo = GeospatialS3Repository(_reader(tmp_path))
+
+    # Act / Assert
+    assert repo.planning_restrictions_for(99999) is None