diff --git a/domain/geospatial/planning_restrictions.py b/domain/geospatial/planning_restrictions.py new file mode 100644 index 00000000..941458b2 --- /dev/null +++ b/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 diff --git a/domain/modelling/generators/solid_wall_recommendation.py b/domain/modelling/generators/solid_wall_recommendation.py index f7af98fa..9da36bb1 100644 --- a/domain/modelling/generators/solid_wall_recommendation.py +++ b/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 diff --git a/repositories/geospatial/geospatial_repository.py b/repositories/geospatial/geospatial_repository.py index 558216bb..745fee9e 100644 --- a/repositories/geospatial/geospatial_repository.py +++ b/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 diff --git a/repositories/geospatial/geospatial_s3_repository.py b/repositories/geospatial/geospatial_s3_repository.py index c91a57e1..62586e77 100644 --- a/repositories/geospatial/geospatial_s3_repository.py +++ b/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"]), + ) diff --git a/tests/domain/modelling/test_elmhurst_cascade_pins.py b/tests/domain/modelling/test_elmhurst_cascade_pins.py index 95a12f69..731e352d 100644 --- a/tests/domain/modelling/test_elmhurst_cascade_pins.py +++ b/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 diff --git a/tests/repositories/geospatial/test_geospatial_repository.py b/tests/repositories/geospatial/test_geospatial_repository.py index 4b0834c9..245be627 100644 --- a/tests/repositories/geospatial/test_geospatial_repository.py +++ b/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