mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
95719dd587
Adds GeospatialRepository.coordinates_for_uprns(uprns) -> dict — a batch coordinate lookup returning only covered UPRNs. The S3 adapter overrides it to read the meta once, group UPRNs by their covering partition, and read each partition once for all the UPRNs it covers; co-located (closely-numbered) UPRNs share a partition, so an EPC Prediction cohort is typically one or two reads instead of one per neighbour. Default port impl is a per-UPRN loop. Feeds the EPC Prediction geo-proximity work: a cohort's UPRNs resolve to coordinates in a couple of reads (validated at corpus scale: 170 partition reads for 2683 UPRNs). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
48 lines
2.1 KiB
Python
48 lines
2.1 KiB
Python
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
|