from __future__ import annotations from abc import ABC, abstractmethod 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): """Resolves a Property's coordinates from hosted reference data by UPRN. A Repo, not a Fetcher (ADR-0011): it reads stored Ordnance Survey Open-UPRN data, with no live API call. Returns None when the UPRN is not covered. """ @abstractmethod def coordinates_for(self, uprn: int) -> Optional[Coordinates]: ... def coordinates_for_uprns( self, uprns: list[int] ) -> dict[int, Coordinates]: """Resolve many UPRNs at once, returning only those covered. The default is a per-UPRN loop; adapters whose storage is partitioned (e.g. the S3 Open-UPRN parquet) override this to read each partition once for all the UPRNs it covers — far fewer reads when the UPRNs are co-located, as closely-numbered UPRNs share a partition.""" resolved: dict[int, Coordinates] = {} for uprn in uprns: coordinates = self.coordinates_for(uprn) if coordinates is not None: resolved[uprn] = coordinates return resolved 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). Defaults to None (unknown → unrestricted) so reference sources that don't carry the flags need not implement it.""" return None