mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
Merge branch 'main' into feature/abri-api-integration
This commit is contained in:
Daniel Roth
commit
4a1fdd87a8
18 changed files with 584 additions and 57 deletions
|
|
@ -205,13 +205,21 @@ def _flush_writes(engine: Engine, writes: list[_PropertyWrite]) -> None:
|
|||
uow.epc.save_batch(lodged_requests)
|
||||
if predicted_requests:
|
||||
uow.epc.save_batch(predicted_requests)
|
||||
# First plan wins: a property with no default Plan yet gets this Plan
|
||||
# as its default whatever the Scenario's flag says (the legacy
|
||||
# p.is_new rule — readers select one default Plan per property, so a
|
||||
# first run under a non-default Scenario must still produce one).
|
||||
# Also heals properties left default-less by runs before this rule.
|
||||
pids_with_default: set[int] = uow.plan.property_ids_with_default_plans(
|
||||
[w.property_id for w in writes]
|
||||
)
|
||||
plan_requests = [
|
||||
PlanSaveRequest(
|
||||
w.plan,
|
||||
property_id=w.property_id,
|
||||
scenario_id=w.scenario_id,
|
||||
portfolio_id=w.portfolio_id,
|
||||
is_default=w.is_default,
|
||||
is_default=w.is_default or w.property_id not in pids_with_default,
|
||||
)
|
||||
for w in writes
|
||||
]
|
||||
|
|
|
|||
|
|
@ -14,8 +14,13 @@ from utils.s3 import (
|
|||
)
|
||||
from datetime import datetime
|
||||
|
||||
from datatypes.address_match import UprnMatch
|
||||
from backend.utils.addressMatch import AddressMatch
|
||||
from backend.address2UPRN.scoring import all_uprns_match, rank_address_similarity
|
||||
from backend.address2UPRN.scoring import (
|
||||
all_uprns_match,
|
||||
rank_address_similarity,
|
||||
resolve_group_ambiguity,
|
||||
)
|
||||
from infrastructure.epc_client.epc_client_service import EpcClientService
|
||||
from repositories.historic_epc.historic_epc_resolver import HistoricEpcResolver
|
||||
from repositories.historic_epc.historic_epc_s3_repository import (
|
||||
|
|
@ -34,32 +39,39 @@ def get_epc_data_with_postcode(postcode: str) -> pd.DataFrame:
|
|||
service = EpcClientService(auth_token=token)
|
||||
results = service.search_by_postcode(postcode)
|
||||
return pd.DataFrame(
|
||||
[{"address": r.address_line_1, "uprn": r.uprn} for r in results]
|
||||
[
|
||||
{
|
||||
"address": r.address_line_1,
|
||||
"uprn": r.uprn,
|
||||
"certificate_number": r.certificate_number,
|
||||
}
|
||||
for r in results
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def get_uprn_from_historic_epc(
|
||||
user_inputed_address: str,
|
||||
postcode: str,
|
||||
) -> Optional[tuple[str, str, float]]:
|
||||
) -> Optional[UprnMatch]:
|
||||
"""Resolve a UPRN via historic EPC S3 data.
|
||||
|
||||
Returns (uprn, address, lexiscore) when the historic dataset agrees on a
|
||||
single rank-1 UPRN, None otherwise (no stored data, zero score, or
|
||||
ambiguous top rank). The score gate is `unambiguous_uprn`'s own (score > 0);
|
||||
the 0.7 heuristic used for the new-EPC source isn't applied here because
|
||||
historic addresses use a more verbose format that systematically depresses
|
||||
lexiscores.
|
||||
Returns (uprn, address, lexiscore, certificate_number) when the historic
|
||||
dataset agrees on a single rank-1 UPRN, None otherwise (no stored data,
|
||||
zero score, or ambiguous top rank). The score gate is `unambiguous_uprn`'s
|
||||
own (score > 0); the 0.7 heuristic used for the new-EPC source isn't
|
||||
applied here because historic addresses use a more verbose format that
|
||||
systematically depresses lexiscores.
|
||||
"""
|
||||
repo = HistoricEpcS3Repository.with_default_s3_client()
|
||||
return HistoricEpcResolver(repo).resolve_uprn(user_inputed_address, postcode)
|
||||
|
||||
|
||||
def get_uprn_with_epc_df(
|
||||
def get_uprn_from_epc_df(
|
||||
user_inputed_address: str,
|
||||
epc_df: pd.DataFrame,
|
||||
verbose: bool = False,
|
||||
) -> Optional[str | tuple[str, str, float]]:
|
||||
) -> Optional[str | UprnMatch]:
|
||||
"""
|
||||
Return uprn (str) using a pre-fetched EPC dataframe.
|
||||
This avoids calling the API multiple times for the same postcode.
|
||||
|
|
@ -88,6 +100,7 @@ def get_uprn_with_epc_df(
|
|||
|
||||
address = top_rank_df["address"].values[0]
|
||||
score = float(top_rank_df["lexiscore"].values[0])
|
||||
certificate_number = top_rank_df["certificate_number"].values[0]
|
||||
|
||||
logger.info(f"Address found to be: {address}, with lexiscore {score}")
|
||||
# Safe to return the agreed UPRN
|
||||
|
|
@ -98,7 +111,7 @@ def get_uprn_with_epc_df(
|
|||
return None
|
||||
|
||||
if verbose:
|
||||
return (found_uprn, address, score)
|
||||
return UprnMatch(found_uprn, address, score, certificate_number)
|
||||
else:
|
||||
return found_uprn
|
||||
|
||||
|
|
@ -116,11 +129,11 @@ def get_uprn(
|
|||
back to the historic EPC dataset on S3.
|
||||
|
||||
For processing multiple addresses in the same postcode, use
|
||||
get_uprn_with_epc_df instead.
|
||||
get_uprn_from_epc_df instead.
|
||||
"""
|
||||
df = get_epc_data_with_postcode(postcode=postcode)
|
||||
|
||||
result: Optional[tuple[str, str, float]] = get_uprn_with_epc_df(
|
||||
result: Optional[str | UprnMatch] = get_uprn_from_epc_df(
|
||||
user_inputed_address=user_inputed_address,
|
||||
epc_df=df,
|
||||
verbose=True,
|
||||
|
|
@ -393,6 +406,8 @@ def handler(event, context, local=False):
|
|||
"address2uprn_uprn": "invalid postcode",
|
||||
"address2uprn_address": "invalid postcode",
|
||||
"address2uprn_lexiscore": "invalid postcode",
|
||||
"address2uprn_certificate_number": "invalid postcode",
|
||||
"address2uprn_status": "invalid_postcode",
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
|
@ -409,7 +424,13 @@ def handler(event, context, local=False):
|
|||
)
|
||||
continue
|
||||
|
||||
# Process each address in this postcode with the same EPC data
|
||||
# Match each address in this postcode against the same EPC data,
|
||||
# collecting per-row results first. Cross-row ambiguity is resolved
|
||||
# for the whole group afterwards (ADR-0057) — a UPRN that is the
|
||||
# best match for two *different* addresses is withheld, so distinct
|
||||
# addresses are never coerced onto one UPRN (which the identity
|
||||
# insert would then silently merge).
|
||||
group_records: list[dict] = []
|
||||
for row in postcode_rows:
|
||||
try:
|
||||
# Concatenate Address columns directly
|
||||
|
|
@ -428,7 +449,7 @@ def handler(event, context, local=False):
|
|||
continue
|
||||
|
||||
# Get UPRN using the pre-fetched EPC data with all return options
|
||||
result: Optional[tuple[str, str, float]] = get_uprn_with_epc_df(
|
||||
result: Optional[UprnMatch] = get_uprn_from_epc_df(
|
||||
user_inputed_address=address2uprn_user_input,
|
||||
epc_df=epc_df,
|
||||
verbose=True,
|
||||
|
|
@ -451,31 +472,40 @@ def handler(event, context, local=False):
|
|||
f"Historic EPC matched {address2uprn_user_input} in {postcode}"
|
||||
)
|
||||
|
||||
norm_address = AddressMatch.normalise_address(
|
||||
address2uprn_user_input
|
||||
)
|
||||
|
||||
# Parse result tuple if successful
|
||||
if result:
|
||||
uprn, found_address, score = result
|
||||
uprn, found_address, score, certificate_number = result
|
||||
logger.info(
|
||||
f"Found UPRN for {address2uprn_user_input} in {postcode}: {uprn} (score: {score})"
|
||||
)
|
||||
|
||||
results_data.append(
|
||||
group_records.append(
|
||||
{
|
||||
**row, # Include all original data
|
||||
"address2uprn_uprn": uprn,
|
||||
"address2uprn_address": found_address,
|
||||
"address2uprn_lexiscore": score,
|
||||
"row": row,
|
||||
"uprn": uprn,
|
||||
"address": found_address,
|
||||
"lexiscore": score,
|
||||
"certificate_number": certificate_number,
|
||||
"norm_address": norm_address,
|
||||
"error": None,
|
||||
}
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
f"No UPRN found for {address2uprn_user_input} in {postcode}"
|
||||
)
|
||||
results_data.append(
|
||||
group_records.append(
|
||||
{
|
||||
**row, # Include all original data
|
||||
"address2uprn_uprn": None,
|
||||
"address2uprn_address": None,
|
||||
"address2uprn_lexiscore": None,
|
||||
"row": row,
|
||||
"uprn": None,
|
||||
"address": None,
|
||||
"lexiscore": None,
|
||||
"certificate_number": None,
|
||||
"norm_address": norm_address,
|
||||
"error": None,
|
||||
}
|
||||
)
|
||||
|
||||
|
|
@ -483,18 +513,43 @@ def handler(event, context, local=False):
|
|||
logger.error(
|
||||
f"Error processing address {row.get('address2uprn_user_input', 'unknown')}: {e}"
|
||||
)
|
||||
# Still add the row with error markers
|
||||
results_data.append(
|
||||
# Still record the row with error markers
|
||||
group_records.append(
|
||||
{
|
||||
**row,
|
||||
"address2uprn_uprn": None,
|
||||
"address2uprn_address": None,
|
||||
"address2uprn_lexiscore": None,
|
||||
"row": row,
|
||||
"uprn": None,
|
||||
"address": None,
|
||||
"lexiscore": None,
|
||||
"certificate_number": None,
|
||||
"norm_address": "",
|
||||
"error": str(e),
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
# Resolve cross-row ambiguity for the whole postcode group, then
|
||||
# emit. A withheld (ambiguous) UPRN drops to None but keeps its
|
||||
# lexiscore so the confirmation page can triage it (ADR-0057).
|
||||
decisions = resolve_group_ambiguity(
|
||||
[(rec["uprn"], rec["norm_address"]) for rec in group_records]
|
||||
)
|
||||
for rec, (final_uprn, status) in zip(group_records, decisions):
|
||||
emitted = {
|
||||
**rec["row"], # Include all original data
|
||||
"address2uprn_uprn": final_uprn,
|
||||
"address2uprn_address": rec["address"] if final_uprn else None,
|
||||
"address2uprn_lexiscore": rec["lexiscore"],
|
||||
"address2uprn_certificate_number": (
|
||||
rec["certificate_number"] if final_uprn else None
|
||||
),
|
||||
"address2uprn_status": (
|
||||
"error" if rec["error"] is not None else status
|
||||
),
|
||||
}
|
||||
if rec["error"] is not None:
|
||||
emitted["error"] = rec["error"]
|
||||
results_data.append(emitted)
|
||||
|
||||
# Create results DataFrame
|
||||
result_df = pd.DataFrame(results_data)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,8 +1,45 @@
|
|||
from collections import defaultdict
|
||||
from typing import NamedTuple, Optional
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from backend.utils.addressMatch import AddressMatch
|
||||
|
||||
|
||||
class GroupDecision(NamedTuple):
|
||||
"""One row's outcome after cross-row ambiguity resolution (ADR-0057)."""
|
||||
|
||||
uprn: Optional[str]
|
||||
status: str # "matched" | "ambiguous_duplicate" | "unmatched"
|
||||
|
||||
|
||||
def resolve_group_ambiguity(
|
||||
matches: list[tuple[Optional[str], str]],
|
||||
) -> list[GroupDecision]:
|
||||
"""Resolve cross-row UPRN ambiguity within one postcode group (ADR-0057).
|
||||
|
||||
``matches`` is ``(uprn, normalised_address)`` per row. A UPRN that is the
|
||||
best match for two rows with *different* normalised addresses is withheld
|
||||
on both (a coarse EPC record absorbing several real addresses, e.g. flats in
|
||||
a block); a UPRN shared only by identical addresses is a genuine re-listing
|
||||
and kept. Returns a ``GroupDecision`` per row, in input order.
|
||||
"""
|
||||
distinct_addresses: dict[str, set[str]] = defaultdict(set)
|
||||
for uprn, norm_address in matches:
|
||||
if uprn:
|
||||
distinct_addresses[uprn].add(norm_address)
|
||||
|
||||
resolved: list[GroupDecision] = []
|
||||
for uprn, _norm_address in matches:
|
||||
if not uprn:
|
||||
resolved.append(GroupDecision(None, "unmatched"))
|
||||
elif len(distinct_addresses[uprn]) > 1:
|
||||
resolved.append(GroupDecision(None, "ambiguous_duplicate"))
|
||||
else:
|
||||
resolved.append(GroupDecision(uprn, "matched"))
|
||||
return resolved
|
||||
|
||||
|
||||
def all_uprns_match(
|
||||
df: pd.DataFrame,
|
||||
target_uprn: str,
|
||||
|
|
|
|||
13
datatypes/address_match.py
Normal file
13
datatypes/address_match.py
Normal file
|
|
@ -0,0 +1,13 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from typing import NamedTuple, Optional
|
||||
|
||||
|
||||
class UprnMatch(NamedTuple):
|
||||
"""A confident address→UPRN match from either EPC source (new API or
|
||||
historic). Tuple-compatible, so existing unpacking keeps working."""
|
||||
|
||||
uprn: str
|
||||
address: str
|
||||
lexiscore: float
|
||||
certificate_number: Optional[str]
|
||||
80
docs/adr/0057-uprn-confirmation-precedes-finalise.md
Normal file
80
docs/adr/0057-uprn-confirmation-precedes-finalise.md
Normal file
|
|
@ -0,0 +1,80 @@
|
|||
# UPRN confirmation precedes finalise; the bulk-upload review is one two-tab page and the portfolio "unmatched" tab is retired
|
||||
|
||||
## Status
|
||||
|
||||
proposed
|
||||
|
||||
## Context
|
||||
|
||||
The bulk-upload pipeline is a chain of SQS-driven Lambda stages:
|
||||
`address2uprn → landlord-overrides → combiner → (review) → finaliser`.
|
||||
`address2uprn` matches each input row independently against per-postcode EPC
|
||||
candidates, keeping the best candidate above a `lexiscore >= 0.7` floor. There
|
||||
is **no cross-row uniqueness guard**: nothing stops one UPRN being the best
|
||||
match for two *different* addresses. The textbook case is a coarse EPC record
|
||||
`42 Moreton Road` (no flat token) that `Flat 1, 42 Moreton Road` and
|
||||
`Flat 2, 42 Moreton Road` both clear — the building-number guard passes and the
|
||||
flat guard never fires, so both distinct flats collapse onto one UPRN.
|
||||
|
||||
Downstream this is corrosive, not just noisy:
|
||||
|
||||
- The `property` identity insert is `on_conflict_do_nothing` on the partial
|
||||
unique index `(portfolio_id, uprn) WHERE uprn IS NOT NULL`. Two distinct
|
||||
addresses carrying the same UPRN therefore **merge into a single property** —
|
||||
one address silently loses its identity.
|
||||
- `_build_overrides` (the finaliser) keys `property_overrides` on
|
||||
`(property_id, override_component, building_part)`. Both merged rows emit the
|
||||
same tuple, so the single `INSERT ... ON CONFLICT DO UPDATE` raises
|
||||
`CardinalityViolation` ("cannot affect row a second time").
|
||||
|
||||
The classifier and finaliser were built assuming a UPRN is always intact and
|
||||
unique; it is not. Separately, a *post-onboarding* portfolio-level "Unmatched
|
||||
properties" tab let users fix no-UPRN properties after the fact — too late (the
|
||||
identity merge has already happened) and in a different place from the rest of
|
||||
review.
|
||||
|
||||
## Decision
|
||||
|
||||
1. **`address2uprn` emits a UPRN only when confident *and* unambiguous.** Within
|
||||
a postcode group, a UPRN that is the best match for two or more rows whose
|
||||
*normalised* input addresses differ is **withheld** (dropped to null) on
|
||||
every such row; a UPRN shared only by rows with the *same* normalised address
|
||||
is a genuine re-listing and is kept. `address2uprn` emits an
|
||||
`address2uprn_status` column (`matched | ambiguous_duplicate | unmatched |
|
||||
invalid_postcode | error`); withheld rows keep their `lexiscore` for triage.
|
||||
A `lexiscore` review band `[0.7, X)` that also withholds shaky matches is a
|
||||
follow-on once `X` is calibrated (see Consequences).
|
||||
2. **One pre-finalise review page, two tabs**, styled like the retired unmatched
|
||||
tab:
|
||||
- **Addresses** — the flagged rows (`unmatched` / `ambiguous_duplicate` /
|
||||
low-score); the user resolves each via OS Places (reusing the existing
|
||||
`MatchAddress` / `searchAddresses` / `useAssignUprn` flow + `postcode_search`
|
||||
cache), producing a valid residential UPRN, or explicitly marks *no-UPRN*.
|
||||
- **Classification** — the existing description→value verify gate, semantics
|
||||
unchanged (every `Unknown` must be mapped; the finaliser still fails loud on
|
||||
an unresolved value).
|
||||
Finalise is enabled only when **both** tabs are clear.
|
||||
3. The finaliser runs only after confirmation, so every row carries a confirmed
|
||||
UPRN **or** an explicit no-UPRN (a first-class terminal state: `property`
|
||||
identity with `uprn NULL`, no `property_overrides`, building-passport shows
|
||||
the "not matched yet" card).
|
||||
4. **Retire the portfolio-level "Unmatched properties" tab** — properties now
|
||||
arrive pre-matched, so post-onboarding UPRN cleanup is unnecessary.
|
||||
5. `PropertyOverridePostgresRepository.upsert_all` keeps a **defensive
|
||||
last-write-wins dedup** as a backstop for the single-statement invariant, but
|
||||
it is no longer the primary fix.
|
||||
|
||||
## Consequences
|
||||
|
||||
- Distinct addresses can no longer silently merge into one property; the
|
||||
`CardinalityViolation` can no longer arise from real data.
|
||||
- Onboarding reaching "complete" now implies confirmed identities, so portfolio
|
||||
pages need no unmatched tab.
|
||||
- The ambiguous/low-score subset adds bounded user work at review time; confident
|
||||
rows pass untouched.
|
||||
- The `lexiscore` review-band threshold `X` needs calibrating against a real
|
||||
portfolio's score distribution before it is switched on; until then only the
|
||||
ambiguous-duplicate rule (which needs no threshold) is active.
|
||||
- Migration is safe stage-by-stage: withheld UPRNs are caught by the *existing*
|
||||
unmatched tab until the two-tab page ships, and the tab is removed only after
|
||||
it does — so an ambiguous property always has somewhere to be resolved.
|
||||
|
|
@ -2,6 +2,7 @@ from __future__ import annotations
|
|||
|
||||
from typing import Optional
|
||||
|
||||
from datatypes.address_match import UprnMatch
|
||||
from datatypes.epc.domain.historic_epc import HistoricEpc
|
||||
from datatypes.epc.domain.historic_epc_matching import (
|
||||
HistoricEpcMatches,
|
||||
|
|
@ -47,11 +48,10 @@ class HistoricEpcResolver:
|
|||
return None
|
||||
return max(certs, key=lambda r: r.lodgement_date)
|
||||
|
||||
def resolve_uprn(
|
||||
self, user_address: str, postcode: str
|
||||
) -> Optional[tuple[str, str, float]]:
|
||||
"""``(uprn, matched_address, lexiscore)`` for an unambiguous rank-1
|
||||
match, else None (no data / ambiguous tie / zero score)."""
|
||||
def resolve_uprn(self, user_address: str, postcode: str) -> Optional[UprnMatch]:
|
||||
"""A ``UprnMatch`` for an unambiguous rank-1 match, else None (no data /
|
||||
ambiguous tie / zero score). ``certificate_number`` is the historic
|
||||
dataset's ``lmk_key``."""
|
||||
matches: HistoricEpcMatches = self.match(user_address, postcode)
|
||||
uprn: Optional[str] = matches.unambiguous_uprn()
|
||||
if not uprn or uprn == "nan":
|
||||
|
|
@ -59,4 +59,4 @@ class HistoricEpcResolver:
|
|||
top: Optional[ScoredHistoricEpc] = matches.top()
|
||||
if top is None:
|
||||
return None
|
||||
return uprn, top.record.address, top.lexiscore
|
||||
return UprnMatch(uprn, top.record.address, top.lexiscore, top.record.lmk_key)
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ from __future__ import annotations
|
|||
from typing import Any
|
||||
|
||||
from sqlalchemy import insert as _sa_insert
|
||||
from sqlmodel import Session, col, update
|
||||
from sqlmodel import Session, col, select, update
|
||||
|
||||
from domain.modelling.plan import Plan
|
||||
from infrastructure.postgres.modelling import PlanModel, RecommendationModel
|
||||
|
|
@ -26,8 +26,9 @@ class PlanPostgresRepository(PlanRepository):
|
|||
|
||||
A re-run INSERTs a fresh Plan rather than deleting the prior one (the cascade
|
||||
delete was slow); when the new Plan is the default it demotes any prior
|
||||
default Plan for the same (property_id, scenario_id) to ``is_default=False``,
|
||||
so readers can select the current Plan via ``is_default=True``."""
|
||||
default Plan for the same property — across all Scenarios — to
|
||||
``is_default=False``, so readers can select the property's one current Plan
|
||||
via ``is_default=True``."""
|
||||
|
||||
def __init__(self, session: Session) -> None:
|
||||
self._session = session
|
||||
|
|
@ -45,6 +46,19 @@ class PlanPostgresRepository(PlanRepository):
|
|||
[PlanSaveRequest(plan, property_id=property_id, scenario_id=scenario_id, portfolio_id=portfolio_id, is_default=is_default)]
|
||||
)[0]
|
||||
|
||||
def property_ids_with_default_plans(self, property_ids: list[int]) -> set[int]:
|
||||
if not property_ids:
|
||||
return set()
|
||||
rows = self._session.exec(
|
||||
select(col(PlanModel.property_id))
|
||||
.distinct()
|
||||
.where(
|
||||
col(PlanModel.property_id).in_(property_ids),
|
||||
col(PlanModel.is_default).is_(True),
|
||||
)
|
||||
).all()
|
||||
return {int(row) for row in rows}
|
||||
|
||||
def save_batch(self, requests: list[PlanSaveRequest]) -> list[int]:
|
||||
"""Persist all Plans in three statements regardless of batch size.
|
||||
|
||||
|
|
@ -56,17 +70,17 @@ class PlanPostgresRepository(PlanRepository):
|
|||
return []
|
||||
|
||||
# Demote prior default Plans for every property in the batch that is
|
||||
# receiving a new default Plan — one UPDATE for the whole batch.
|
||||
# receiving a new default Plan — one UPDATE for the whole batch,
|
||||
# across ALL scenarios: readers select a property's default by
|
||||
# portfolio + is_default alone, so the invariant is one default per
|
||||
# property. Scenario-scoped demotion left a promoted first Plan
|
||||
# (non-default scenario) defaulted forever alongside the default
|
||||
# scenario's later Plan (#1490).
|
||||
default_pids = [r.property_id for r in requests if r.is_default]
|
||||
if default_pids:
|
||||
# scenario_id is uniform per batch (one scenario per SQS message).
|
||||
scenario_id = requests[0].scenario_id
|
||||
self._session.exec( # type: ignore[call-overload]
|
||||
update(PlanModel)
|
||||
.where(
|
||||
col(PlanModel.property_id).in_(default_pids),
|
||||
col(PlanModel.scenario_id) == scenario_id,
|
||||
)
|
||||
.where(col(PlanModel.property_id).in_(default_pids))
|
||||
.values(is_default=False)
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -43,7 +43,8 @@ class PlanRepository(ABC):
|
|||
) -> int:
|
||||
"""Persist ``plan`` and return its Plan id. Keeps prior Plans for
|
||||
``(property_id, scenario_id)`` as history; when ``is_default`` is True,
|
||||
demotes those prior Plans to ``is_default=False``."""
|
||||
demotes the property's prior default Plan — whatever its Scenario — to
|
||||
``is_default=False`` (one default Plan per property)."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
|
|
@ -54,3 +55,11 @@ class PlanRepository(ABC):
|
|||
UPDATE only when at least one request has ``is_default=True``. Keeps
|
||||
prior Plans as history (ADR-0017)."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def property_ids_with_default_plans(self, property_ids: list[int]) -> set[int]:
|
||||
"""The subset of ``property_ids`` that already have a default Plan.
|
||||
A property outside this set is receiving its first Plan, which becomes
|
||||
its default whatever the Scenario says (the legacy ``p.is_new`` rule —
|
||||
readers need one default Plan per property)."""
|
||||
...
|
||||
|
|
|
|||
|
|
@ -31,6 +31,17 @@ class PropertyOverridePostgresRepository(PropertyOverrideRepository):
|
|||
return 0
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
# Defensive backstop (ADR-0057): a single INSERT ... ON CONFLICT DO
|
||||
# UPDATE may not touch the same target row twice — two batch rows sharing
|
||||
# the conflict key (property_id, override_component, building_part) make
|
||||
# Postgres raise CardinalityViolation ("cannot affect row a second
|
||||
# time"). ADR-0057 stops distinct addresses being coerced onto one UPRN
|
||||
# upstream, so this should no longer arise from real data; collapse to
|
||||
# the last occurrence (last-write-wins, the same outcome DO UPDATE gives
|
||||
# across separate statements) so the statement can never be invalid.
|
||||
deduped: dict[tuple[int, str, int], PropertyOverrideInsert] = {
|
||||
(r.property_id, r.override_component, r.building_part): r for r in rows
|
||||
}
|
||||
values = [
|
||||
{
|
||||
"property_id": r.property_id,
|
||||
|
|
@ -42,7 +53,7 @@ class PropertyOverridePostgresRepository(PropertyOverrideRepository):
|
|||
"created_at": now,
|
||||
"updated_at": now,
|
||||
}
|
||||
for r in rows
|
||||
for r in deduped.values()
|
||||
]
|
||||
|
||||
stmt = pg_insert(self._table).values(values)
|
||||
|
|
|
|||
|
|
@ -34,5 +34,9 @@ class PropertyOverrideRepository(ABC):
|
|||
"""Upsert each row on ``(property_id, override_component, building_part)``,
|
||||
refreshing ``override_value`` + ``original_spreadsheet_description`` +
|
||||
``updated_at`` on conflict (recalculate-on-rerun). Returns the number of
|
||||
rows affected; an empty list is a no-op returning 0."""
|
||||
rows affected; an empty list is a no-op returning 0.
|
||||
|
||||
Rows sharing a conflict key within one call are collapsed to the last
|
||||
occurrence (last-write-wins) before writing, so a single batch never
|
||||
upserts the same target row twice (ADR-0057 backstop)."""
|
||||
...
|
||||
|
|
|
|||
|
|
@ -47,7 +47,7 @@ sys.path.insert(0, str(_REPO_ROOT)) # worktree root first — avoid the import
|
|||
from backend.address2UPRN.main import ( # noqa: E402
|
||||
get_epc_data_with_postcode,
|
||||
get_uprn_from_historic_epc,
|
||||
get_uprn_with_epc_df,
|
||||
get_uprn_from_epc_df,
|
||||
)
|
||||
from backend.ordnanceSurvey.helpers import ( # noqa: E402
|
||||
lookup_os_places,
|
||||
|
|
@ -108,7 +108,7 @@ def resolve_epc(
|
|||
epc_df = get_epc_data_with_postcode(postcode=postcode_clean)
|
||||
epc_cache[postcode_clean] = epc_df
|
||||
|
||||
result = get_uprn_with_epc_df(
|
||||
result = get_uprn_from_epc_df(
|
||||
user_inputed_address=address, epc_df=epc_df, verbose=True
|
||||
)
|
||||
if isinstance(result, tuple):
|
||||
|
|
|
|||
|
|
@ -511,6 +511,70 @@ def test_attach_mode_partial_failure_persists_successes_then_records_failure() -
|
|||
]
|
||||
|
||||
|
||||
def test_first_plan_for_a_property_becomes_its_default() -> None:
|
||||
"""A property's first Plan is its default whatever the Scenario says (the
|
||||
legacy p.is_new rule — readers select one default Plan per property); a
|
||||
property that already has a default Plan keeps the Scenario's flag."""
|
||||
# Arrange — a non-default scenario; 111 has no default plan yet, 222 does
|
||||
pid_first, pid_has_default = 111, 222
|
||||
mock_engine = _engine_mock(
|
||||
[pid_first, pid_has_default], [1001, 1002], [POSTCODE, POSTCODE]
|
||||
)
|
||||
scenario = MagicMock()
|
||||
scenario.is_default = False
|
||||
|
||||
with ExitStack() as stack:
|
||||
stack.enter_context(patch("applications.modelling_e2e.handler.os.environ", _ENV))
|
||||
stack.enter_context(
|
||||
patch("applications.modelling_e2e.handler._get_engine", return_value=mock_engine)
|
||||
)
|
||||
stack.enter_context(
|
||||
patch("applications.modelling_e2e.handler.EpcClientService")
|
||||
).return_value.get_by_uprn.return_value = MagicMock()
|
||||
stack.enter_context(patch("applications.modelling_e2e.handler.GeospatialS3Repository"))
|
||||
stack.enter_context(patch("applications.modelling_e2e.handler.GoogleSolarApiClient"))
|
||||
stack.enter_context(
|
||||
patch("applications.modelling_e2e.handler._spatial_for", return_value=None)
|
||||
)
|
||||
stack.enter_context(
|
||||
patch("applications.modelling_e2e.handler._solar_insights_for", return_value=None)
|
||||
)
|
||||
stack.enter_context(
|
||||
patch("applications.modelling_e2e.handler.overlays_from", return_value=[])
|
||||
)
|
||||
stack.enter_context(
|
||||
patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader")
|
||||
).return_value.overrides_for_many.return_value = {}
|
||||
stack.enter_context(
|
||||
patch("applications.modelling_e2e.handler.ScenarioPostgresRepository")
|
||||
).return_value.get_many.return_value = [scenario]
|
||||
stack.enter_context(patch("applications.modelling_e2e.handler.catalogue_snapshot_with_off_catalogue_overrides"))
|
||||
stack.enter_context(patch("applications.modelling_e2e.handler.Session"))
|
||||
stack.enter_context(
|
||||
patch("applications.modelling_e2e.handler.run_modelling", return_value=_plan_mock())
|
||||
)
|
||||
MockUoW = stack.enter_context(patch("applications.modelling_e2e.handler.PostgresUnitOfWork"))
|
||||
mock_uow = MagicMock()
|
||||
mock_uow.plan.property_ids_with_default_plans.return_value = {pid_has_default}
|
||||
MockUoW.return_value.__enter__.return_value = mock_uow
|
||||
MockUoW.return_value.__exit__.return_value = False
|
||||
|
||||
# Act
|
||||
from applications.modelling_e2e.handler import handler
|
||||
handler.__wrapped__( # type: ignore[attr-defined]
|
||||
{"property_ids": [pid_first, pid_has_default],
|
||||
"portfolio_id": PORTFOLIO_ID, "scenario_id": SCENARIO_ID,
|
||||
"refetch_solar": False, "dry_run": False},
|
||||
None, _mock_orchestrator(), uuid4(),
|
||||
)
|
||||
|
||||
# Assert — the first-timer's plan is promoted; the other keeps the flag
|
||||
plan_requests = mock_uow.plan.save_batch.call_args.args[0]
|
||||
by_pid = {r.property_id: r for r in plan_requests}
|
||||
assert by_pid[pid_first].is_default is True
|
||||
assert by_pid[pid_has_default].is_default is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Lodged EPC path
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
0
tests/backend/address2UPRN/__init__.py
Normal file
0
tests/backend/address2UPRN/__init__.py
Normal file
70
tests/backend/address2UPRN/test_resolve_group_ambiguity.py
Normal file
70
tests/backend/address2UPRN/test_resolve_group_ambiguity.py
Normal file
|
|
@ -0,0 +1,70 @@
|
|||
"""Unit tests for ``resolve_group_ambiguity`` (ADR-0057).
|
||||
|
||||
Each address is matched independently, so one UPRN can be the best match for
|
||||
two *different* addresses (a coarse EPC record absorbing several real
|
||||
addresses). Withholding that UPRN is a pure function of a postcode group's
|
||||
``(uprn, normalised_address)`` pairs — no S3, EPC API, or scoring needed — so
|
||||
it is unit-tested directly here.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from backend.address2UPRN.scoring import resolve_group_ambiguity
|
||||
|
||||
|
||||
def test_distinct_addresses_sharing_one_uprn_are_withheld() -> None:
|
||||
# Flat 1 and Flat 2 both matched the same coarse EPC UPRN — distinct
|
||||
# addresses, so both are withheld rather than silently merged.
|
||||
result = resolve_group_ambiguity(
|
||||
[
|
||||
("100", "flat 1 42 moreton road"),
|
||||
("100", "flat 2 42 moreton road"),
|
||||
]
|
||||
)
|
||||
assert result == [(None, "ambiguous_duplicate"), (None, "ambiguous_duplicate")]
|
||||
|
||||
|
||||
def test_same_address_listed_twice_keeps_the_uprn() -> None:
|
||||
# A genuine re-listing of one property (identical normalised address) is a
|
||||
# real duplicate, not an ambiguous match — the UPRN is kept on both.
|
||||
result = resolve_group_ambiguity(
|
||||
[
|
||||
("100", "42 moreton road"),
|
||||
("100", "42 moreton road"),
|
||||
]
|
||||
)
|
||||
assert result == [("100", "matched"), ("100", "matched")]
|
||||
|
||||
|
||||
def test_distinct_uprns_are_each_matched() -> None:
|
||||
result = resolve_group_ambiguity(
|
||||
[
|
||||
("100", "42 moreton road"),
|
||||
("101", "44 moreton road"),
|
||||
]
|
||||
)
|
||||
assert result == [("100", "matched"), ("101", "matched")]
|
||||
|
||||
|
||||
def test_none_uprn_is_unmatched() -> None:
|
||||
result = resolve_group_ambiguity([(None, "99 nowhere lane")])
|
||||
assert result == [(None, "unmatched")]
|
||||
|
||||
|
||||
def test_order_is_preserved_across_mixed_group() -> None:
|
||||
# A withheld pair, an unmatched row, and a clean match — all in one group;
|
||||
# the output aligns positionally with the input.
|
||||
result = resolve_group_ambiguity(
|
||||
[
|
||||
("100", "flat 1 42 moreton road"), # ambiguous (with row 3)
|
||||
(None, "no epc candidate"), # unmatched
|
||||
("100", "flat 2 42 moreton road"), # ambiguous (with row 0)
|
||||
("200", "sole match road"), # matched
|
||||
]
|
||||
)
|
||||
assert result == [
|
||||
(None, "ambiguous_duplicate"),
|
||||
(None, "unmatched"),
|
||||
(None, "ambiguous_duplicate"),
|
||||
("200", "matched"),
|
||||
]
|
||||
|
|
@ -212,6 +212,7 @@ class FakePlanRepository(PlanRepository):
|
|||
|
||||
def __init__(self) -> None:
|
||||
self.saved: dict[tuple[int, int], Plan] = {}
|
||||
self.default_property_ids: set[int] = set()
|
||||
self._next_id = 1
|
||||
|
||||
def save(
|
||||
|
|
@ -224,6 +225,8 @@ class FakePlanRepository(PlanRepository):
|
|||
is_default: bool,
|
||||
) -> int:
|
||||
self.saved[(property_id, scenario_id)] = plan
|
||||
if is_default:
|
||||
self.default_property_ids.add(property_id)
|
||||
plan_id = self._next_id
|
||||
self._next_id += 1
|
||||
return plan_id
|
||||
|
|
@ -240,6 +243,9 @@ class FakePlanRepository(PlanRepository):
|
|||
for r in requests
|
||||
]
|
||||
|
||||
def property_ids_with_default_plans(self, property_ids: list[int]) -> set[int]:
|
||||
return self.default_property_ids.intersection(property_ids)
|
||||
|
||||
|
||||
class _UnsetProductRepo(ProductRepository):
|
||||
"""Default for a `FakeUnitOfWork` built without a catalogue — raises if a
|
||||
|
|
|
|||
|
|
@ -73,10 +73,11 @@ def test_resolve_uprn_returns_unambiguous_match():
|
|||
|
||||
# Assert
|
||||
assert result is not None
|
||||
uprn, address, score = result
|
||||
uprn, address, score, certificate_number = result
|
||||
assert uprn == "100"
|
||||
assert address == "47 GORDON ROAD"
|
||||
assert score > 0
|
||||
assert certificate_number == ""
|
||||
|
||||
|
||||
def test_resolve_uprn_is_none_when_postcode_has_no_data():
|
||||
|
|
|
|||
|
|
@ -206,3 +206,56 @@ def test_rerun_as_non_default_does_not_demote_the_prior_default(
|
|||
|
||||
assert len(plan_rows) == 2
|
||||
assert by_id[first_id].is_default is True
|
||||
|
||||
|
||||
def test_reports_which_properties_already_have_a_default_plan(
|
||||
db_engine: Engine,
|
||||
) -> None:
|
||||
# Arrange — property 20 has a default Plan, 21 only a non-default one,
|
||||
# 22 has no Plans at all
|
||||
with Session(db_engine) as session:
|
||||
repo = PlanPostgresRepository(session)
|
||||
repo.save(_plan(), property_id=20, scenario_id=7, portfolio_id=1, is_default=True)
|
||||
repo.save(_plan(), property_id=21, scenario_id=7, portfolio_id=1, is_default=False)
|
||||
session.commit()
|
||||
|
||||
# Act
|
||||
with Session(db_engine) as session:
|
||||
have_default = PlanPostgresRepository(session).property_ids_with_default_plans(
|
||||
[20, 21, 22]
|
||||
)
|
||||
|
||||
# Assert — only the property with a default Plan is reported
|
||||
assert have_default == {20}
|
||||
|
||||
|
||||
def test_a_new_default_demotes_the_prior_default_across_scenarios(
|
||||
db_engine: Engine,
|
||||
) -> None:
|
||||
"""One default Plan per property: a promoted first Plan (non-default
|
||||
scenario) must lose the default when the default scenario's Plan lands —
|
||||
readers filter portfolio + is_default only, never scenario (#1490)."""
|
||||
# Arrange — the property's first Plan, promoted under non-default scenario 7
|
||||
with Session(db_engine) as session:
|
||||
first_id = PlanPostgresRepository(session).save(
|
||||
_plan(), property_id=30, scenario_id=7, portfolio_id=1, is_default=True
|
||||
)
|
||||
session.commit()
|
||||
|
||||
# Act — the default scenario (8) models the same property
|
||||
with Session(db_engine) as session:
|
||||
second_id = PlanPostgresRepository(session).save(
|
||||
_plan(), property_id=30, scenario_id=8, portfolio_id=1, is_default=True
|
||||
)
|
||||
session.commit()
|
||||
|
||||
# Assert — exactly one default for the property, the newest plan
|
||||
with Session(db_engine) as session:
|
||||
plan_rows = session.exec(
|
||||
select(PlanModel).where(col(PlanModel.property_id) == 30)
|
||||
).all()
|
||||
by_id = {p.id: p for p in plan_rows}
|
||||
|
||||
assert by_id[first_id].is_default is False # demoted across scenarios
|
||||
assert by_id[second_id].is_default is True
|
||||
assert sum(1 for p in plan_rows if p.is_default) == 1
|
||||
|
|
|
|||
|
|
@ -0,0 +1,102 @@
|
|||
"""Integration tests for ``PropertyOverridePostgresRepository.upsert_all``.
|
||||
|
||||
The conflict handling lives entirely in SQL (``INSERT ... ON CONFLICT
|
||||
(property_id, override_component, building_part) DO UPDATE``), so it can only be
|
||||
verified against a real Postgres -- the ``db_engine`` fixture in
|
||||
``tests/conftest.py`` spins one up per test. The batch-dedup below is exactly
|
||||
the case an in-memory fake cannot catch: only a real ``ON CONFLICT`` statement
|
||||
raises ``CardinalityViolation`` when a batch would touch one target row twice
|
||||
(the ADR-0057 backstop).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterator
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import Engine
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from infrastructure.postgres.property_override_table import PropertyOverrideRow
|
||||
from repositories.property.property_override_postgres_repository import (
|
||||
PropertyOverridePostgresRepository,
|
||||
)
|
||||
from repositories.property.property_override_repository import PropertyOverrideInsert
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def session(db_engine: Engine) -> Iterator[Session]:
|
||||
with Session(db_engine) as s:
|
||||
yield s
|
||||
|
||||
|
||||
def _insert(
|
||||
property_id: int,
|
||||
override_value: str,
|
||||
description: str,
|
||||
*,
|
||||
building_part: int = 0,
|
||||
component: str = "property_type",
|
||||
) -> PropertyOverrideInsert:
|
||||
return PropertyOverrideInsert(
|
||||
property_id=property_id,
|
||||
portfolio_id=820,
|
||||
building_part=building_part,
|
||||
override_component=component,
|
||||
override_value=override_value,
|
||||
original_spreadsheet_description=description,
|
||||
)
|
||||
|
||||
|
||||
def _all_rows(session: Session) -> list[PropertyOverrideRow]:
|
||||
return list(session.exec(select(PropertyOverrideRow)).all())
|
||||
|
||||
|
||||
def test_duplicate_conflict_key_in_one_batch_collapses_to_last(
|
||||
session: Session,
|
||||
) -> None:
|
||||
# arrange: two batch rows sharing the conflict key
|
||||
# (property_id, override_component, building_part) — before the backstop this
|
||||
# raised CardinalityViolation ("cannot affect row a second time").
|
||||
repo = PropertyOverridePostgresRepository(session)
|
||||
|
||||
# act
|
||||
repo.upsert_all(
|
||||
[
|
||||
_insert(742961, "House", "House: Semi-Detached"),
|
||||
_insert(742961, "House", "House: Mid-Terrace"),
|
||||
]
|
||||
)
|
||||
session.commit()
|
||||
|
||||
# assert: one row, last occurrence wins (matches ON CONFLICT DO UPDATE).
|
||||
rows = _all_rows(session)
|
||||
assert len(rows) == 1
|
||||
assert rows[0].override_value == "House"
|
||||
assert rows[0].original_spreadsheet_description == "House: Mid-Terrace"
|
||||
|
||||
|
||||
def test_distinct_conflict_keys_are_all_written(session: Session) -> None:
|
||||
# arrange: dedup must only collapse exact conflict-key duplicates — a
|
||||
# different property_id, building_part, or component is a distinct row.
|
||||
repo = PropertyOverridePostgresRepository(session)
|
||||
|
||||
# act
|
||||
repo.upsert_all(
|
||||
[
|
||||
_insert(742961, "House", "House: Semi-Detached"),
|
||||
_insert(742962, "House", "House: Mid-Terrace"), # different property
|
||||
_insert(742961, "House", "ext", building_part=1), # different part
|
||||
_insert(742961, "SolidBrick", "solid", component="wall_type"), # component
|
||||
]
|
||||
)
|
||||
session.commit()
|
||||
|
||||
# assert
|
||||
assert len(_all_rows(session)) == 4
|
||||
|
||||
|
||||
def test_empty_batch_is_a_noop(session: Session) -> None:
|
||||
repo = PropertyOverridePostgresRepository(session)
|
||||
assert repo.upsert_all([]) == 0
|
||||
assert _all_rows(session) == []
|
||||
Loading…
Add table
Reference in a new issue