feat(geospatial): one-read spatial reference (coords + restrictions)

Slice 3c.1. Ingestion will persist a UPRN's coordinates and planning
protections together as a write-through cache, so resolve them in a single
partition read rather than two. `SpatialReference` bundles the coordinates
(which drive the Solar fetch) and the `PlanningRestrictions` (which gate wall
insulation per ADR-0019/ADR-0020); `GeospatialRepository.spatial_for(uprn)`
returns it, and `coordinates_for`/`planning_restrictions_for` now delegate to
the one lookup.

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

domain/geospatial/spatial_reference.py

@@ -0,0 +1,25 @@
+"""One UPRN's row of Ordnance Survey spatial reference data.
+
+Bundles the two things the geospatial partition co-locates against a UPRN — the
+coordinates (which drive the Solar fetch) and the planning protections (which
+gate wall insulation, ADR-0019/ADR-0020) — so Ingestion resolves them in a
+single reference lookup and persists them together as a write-through cache
+(`property_details_spatial`). Coordinates are Optional because the legacy
+nearest-UPRN proxy fallback yields the flags with the coordinates nulled.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional
+
+from domain.geospatial.coordinates import Coordinates
+from domain.geospatial.planning_restrictions import PlanningRestrictions
+
+
+@dataclass(frozen=True)
+class SpatialReference:
+    """A Property's resolved spatial reference data, keyed by UPRN."""
+
+    coordinates: Optional[Coordinates]
+    restrictions: PlanningRestrictions

repositories/geospatial/geospatial_repository.py

@@ -5,6 +5,7 @@ from typing import Optional
 
 from domain.geospatial.coordinates import Coordinates
 from domain.geospatial.planning_restrictions import PlanningRestrictions
+from domain.geospatial.spatial_reference import SpatialReference
 
 
 class GeospatialRepository(ABC):
@@ -17,6 +18,13 @@ class GeospatialRepository(ABC):
     @abstractmethod
     def coordinates_for(self, uprn: int) -> Optional[Coordinates]: ...
 
+    def spatial_for(self, uprn: int) -> Optional[SpatialReference]:
+        """The Property's coordinates and planning protections together, in one
+        reference lookup (ADR-0020) — Ingestion uses the coordinates to drive
+        the Solar fetch and persists the whole reference. Defaults to None so
+        reference sources that don't carry the flags need not implement it."""
+        return None
+
     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).

repositories/geospatial/geospatial_s3_repository.py

@@ -7,6 +7,7 @@ import pandas as pd
 
 from domain.geospatial.coordinates import Coordinates
 from domain.geospatial.planning_restrictions import PlanningRestrictions
+from domain.geospatial.spatial_reference import SpatialReference
 from repositories.geospatial.geospatial_repository import GeospatialRepository
 
 ParquetReader = Callable[[str], pd.DataFrame]
@@ -41,21 +42,26 @@ class GeospatialS3Repository(GeospatialRepository):
             return None
         return rows.iloc[0]
 
-    def coordinates_for(self, uprn: int) -> Optional[Coordinates]:
+    def spatial_for(self, uprn: int) -> Optional[SpatialReference]:
         row = self._row_for(uprn)
         if row is None:
             return None
-        return Coordinates(
-            longitude=float(row["LONGITUDE"]),
-            latitude=float(row["LATITUDE"]),
+        return SpatialReference(
+            coordinates=Coordinates(
+                longitude=float(row["LONGITUDE"]),
+                latitude=float(row["LATITUDE"]),
+            ),
+            restrictions=PlanningRestrictions(
+                in_conservation_area=bool(row["conservation_status"]),
+                is_listed=bool(row["is_listed_building"]),
+                is_heritage=bool(row["is_heritage_building"]),
+            ),
         )
 
+    def coordinates_for(self, uprn: int) -> Optional[Coordinates]:
+        reference: Optional[SpatialReference] = self.spatial_for(uprn)
+        return reference.coordinates if reference is not None else None
+
     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"]),
-        )
+        reference: Optional[SpatialReference] = self.spatial_for(uprn)
+        return reference.restrictions if reference is not None else None

tests/repositories/geospatial/test_geospatial_repository.py

@@ -13,8 +13,11 @@ from pathlib import Path
 
 import pandas as pd
 
+from typing import Optional
+
 from domain.geospatial.coordinates import Coordinates
 from domain.geospatial.planning_restrictions import PlanningRestrictions
+from domain.geospatial.spatial_reference import SpatialReference
 from repositories.geospatial.geospatial_s3_repository import GeospatialS3Repository
 
 
@@ -100,3 +103,31 @@ def test_planning_restrictions_for_returns_none_when_uprn_absent(
 
     # Act / Assert
     assert repo.planning_restrictions_for(99999) is None
+
+
+def test_spatial_for_returns_coordinates_and_restrictions_together(
+    tmp_path: Path,
+) -> None:
+    # Arrange — one partition row carries the coordinates and the planning flags.
+    _write_open_uprn(tmp_path)
+    repo = GeospatialS3Repository(_reader(tmp_path))
+
+    # Act — a single reference lookup yields both, so Ingestion reads the row once.
+    reference: Optional[SpatialReference] = repo.spatial_for(12346)
+
+    # Assert
+    assert reference == SpatialReference(
+        coordinates=Coordinates(longitude=-0.2000, latitude=51.6000),
+        restrictions=PlanningRestrictions(
+            in_conservation_area=False, is_listed=True, is_heritage=False
+        ),
+    )
+
+
+def test_spatial_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.spatial_for(99999) is None