mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
Both filterable components resolve through one precedence engine 🟪
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar
parent
513977a6fd
commit
19d9e758ad
1 changed files with 78 additions and 100 deletions
|
|
@ -6,6 +6,7 @@ Next.js app from this same rule, so any change here must land in both
|
|||
codebases and amend the ADR.
|
||||
"""
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
|
|
@ -38,13 +39,60 @@ class FilteredProperty:
|
|||
postcode: Optional[str]
|
||||
|
||||
|
||||
# "Unknown" is itself a selectable filter value: exactly the bucket of
|
||||
# properties whose value resolves at no precedence level (ADR-0056).
|
||||
_UNKNOWN = "Unknown"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _ComponentResolution:
|
||||
"""How one filterable component (property_type / built_form) resolves at
|
||||
each ADR-0056 precedence level: which override row names it, how its RdSAP
|
||||
numeric codes map (text labels pass through as-is), and where it lives on
|
||||
the EPC and the legacy property row."""
|
||||
|
||||
override_component: str
|
||||
codes: dict[str, str]
|
||||
epc_value: Callable[[EpcPropertyModel], Optional[str]]
|
||||
legacy_value: Callable[[PropertyRow], Optional[str]]
|
||||
|
||||
|
||||
_PROPERTY_TYPE = _ComponentResolution(
|
||||
override_component="property_type",
|
||||
codes={
|
||||
"0": "House",
|
||||
"1": "Bungalow",
|
||||
"2": "Flat",
|
||||
"3": "Maisonette",
|
||||
"4": "Park home",
|
||||
},
|
||||
# A cert without the property_type column falls back to its dwelling_type.
|
||||
epc_value=lambda epc: epc.property_type or epc.dwelling_type,
|
||||
legacy_value=lambda row: row.property_type,
|
||||
)
|
||||
|
||||
_BUILT_FORM = _ComponentResolution(
|
||||
override_component="built_form_type",
|
||||
codes={
|
||||
"1": "Detached",
|
||||
"2": "Semi-Detached",
|
||||
"3": "End-Terrace",
|
||||
"4": "Mid-Terrace",
|
||||
"5": "Enclosed End-Terrace",
|
||||
"6": "Enclosed Mid-Terrace",
|
||||
},
|
||||
epc_value=lambda epc: epc.built_form,
|
||||
legacy_value=lambda row: row.built_form,
|
||||
)
|
||||
|
||||
|
||||
def resolve_filtered_property_ids(
|
||||
session: Session, portfolio_id: int, filters: PropertyGroupFilters
|
||||
) -> list[FilteredProperty]:
|
||||
"""Resolve *filters* against *portfolio_id* per ADR-0056: base set is the
|
||||
portfolio's rows with ``marked_for_deletion = false``; property_type /
|
||||
built_form resolve override → lodged EPC → predicted EPC → legacy columns
|
||||
→ "Unknown"."""
|
||||
→ "Unknown". Filters AND-combine; an absent key is unconstrained."""
|
||||
statement = (
|
||||
select(PropertyRow)
|
||||
.where(PropertyRow.portfolio_id == portfolio_id)
|
||||
|
|
@ -55,25 +103,20 @@ def resolve_filtered_property_ids(
|
|||
)
|
||||
if filters.postcodes is not None:
|
||||
statement = statement.where(col(PropertyRow.postcode).in_(filters.postcodes))
|
||||
rows = session.exec(statement).all()
|
||||
candidates = [row for row in rows if row.id is not None]
|
||||
candidates = [row for row in session.exec(statement).all() if row.id is not None]
|
||||
|
||||
if filters.property_types is not None:
|
||||
wanted = set(filters.property_types)
|
||||
types = _resolved_property_types(session, candidates)
|
||||
for wanted_values, component in (
|
||||
(filters.property_types, _PROPERTY_TYPE),
|
||||
(filters.built_forms, _BUILT_FORM),
|
||||
):
|
||||
if wanted_values is None:
|
||||
continue
|
||||
wanted = set(wanted_values)
|
||||
resolved = _resolved_component_values(session, candidates, component)
|
||||
candidates = [
|
||||
row
|
||||
for row in candidates
|
||||
if row.id is not None and types.get(row.id) in wanted
|
||||
]
|
||||
|
||||
if filters.built_forms is not None:
|
||||
wanted_forms = set(filters.built_forms)
|
||||
forms = _resolved_built_forms(session, candidates)
|
||||
candidates = [
|
||||
row
|
||||
for row in candidates
|
||||
if row.id is not None and forms.get(row.id) in wanted_forms
|
||||
if row.id is not None and resolved.get(row.id) in wanted
|
||||
]
|
||||
|
||||
return [
|
||||
|
|
@ -83,46 +126,19 @@ def resolve_filtered_property_ids(
|
|||
]
|
||||
|
||||
|
||||
# "Unknown" is itself a selectable filter value: exactly the bucket of
|
||||
# properties whose value resolves at no precedence level (ADR-0056).
|
||||
_UNKNOWN = "Unknown"
|
||||
|
||||
# RdSAP numeric codes as lodged on some certs; text labels pass through as-is
|
||||
# (ADR-0056 — the mapping is part of the shared preview contract).
|
||||
_PROPERTY_TYPE_CODES: dict[str, str] = {
|
||||
"0": "House",
|
||||
"1": "Bungalow",
|
||||
"2": "Flat",
|
||||
"3": "Maisonette",
|
||||
"4": "Park home",
|
||||
}
|
||||
|
||||
_BUILT_FORM_CODES: dict[str, str] = {
|
||||
"1": "Detached",
|
||||
"2": "Semi-Detached",
|
||||
"3": "End-Terrace",
|
||||
"4": "Mid-Terrace",
|
||||
"5": "Enclosed End-Terrace",
|
||||
"6": "Enclosed Mid-Terrace",
|
||||
}
|
||||
|
||||
|
||||
def _resolved_built_forms(
|
||||
session: Session, candidates: list[PropertyRow]
|
||||
def _resolved_component_values(
|
||||
session: Session, candidates: list[PropertyRow], component: _ComponentResolution
|
||||
) -> dict[int, str]:
|
||||
"""Each property's effective built_form per the ADR-0056 precedence:
|
||||
override (building_part 0) → lodged EPC → predicted EPC → legacy column."""
|
||||
"""Each candidate's effective value for *component* per the ADR-0056
|
||||
precedence: override (building_part 0) → lodged EPC → predicted EPC →
|
||||
legacy column → "Unknown"."""
|
||||
property_ids = [row.id for row in candidates if row.id is not None]
|
||||
by_legacy_column = {
|
||||
row.id: row.built_form
|
||||
for row in candidates
|
||||
if row.id is not None and row.built_form is not None
|
||||
}
|
||||
|
||||
overrides = session.exec(
|
||||
select(PropertyOverrideRow)
|
||||
.where(col(PropertyOverrideRow.property_id).in_(property_ids))
|
||||
.where(PropertyOverrideRow.building_part == 0)
|
||||
.where(PropertyOverrideRow.override_component == "built_form_type")
|
||||
.where(PropertyOverrideRow.override_component == component.override_component)
|
||||
).all()
|
||||
by_override = {o.property_id: o.override_value for o in overrides}
|
||||
|
||||
|
|
@ -133,63 +149,25 @@ def _resolved_built_forms(
|
|||
).all()
|
||||
by_epc_source: dict[str, dict[int, str]] = {"lodged": {}, "predicted": {}}
|
||||
for epc in epcs:
|
||||
if epc.property_id is not None and epc.built_form:
|
||||
value = epc.built_form
|
||||
by_epc_source[epc.source][epc.property_id] = _BUILT_FORM_CODES.get(
|
||||
value, value
|
||||
)
|
||||
|
||||
resolved: dict[int, str] = {}
|
||||
for pid in property_ids:
|
||||
resolved[pid] = (
|
||||
by_override.get(pid)
|
||||
or by_epc_source["lodged"].get(pid)
|
||||
or by_epc_source["predicted"].get(pid)
|
||||
or by_legacy_column.get(pid)
|
||||
or _UNKNOWN
|
||||
)
|
||||
return resolved
|
||||
|
||||
|
||||
def _resolved_property_types(
|
||||
session: Session, candidates: list[PropertyRow]
|
||||
) -> dict[int, str]:
|
||||
"""Each property's effective property_type per the ADR-0056 precedence:
|
||||
override (building_part 0) → lodged EPC → predicted EPC → legacy column."""
|
||||
property_ids = [row.id for row in candidates if row.id is not None]
|
||||
by_legacy_column = {
|
||||
row.id: row.property_type
|
||||
for row in candidates
|
||||
if row.id is not None and row.property_type is not None
|
||||
}
|
||||
overrides = session.exec(
|
||||
select(PropertyOverrideRow)
|
||||
.where(col(PropertyOverrideRow.property_id).in_(property_ids))
|
||||
.where(PropertyOverrideRow.building_part == 0)
|
||||
.where(PropertyOverrideRow.override_component == "property_type")
|
||||
).all()
|
||||
by_override = {o.property_id: o.override_value for o in overrides}
|
||||
|
||||
epcs = session.exec(
|
||||
select(EpcPropertyModel)
|
||||
.where(col(EpcPropertyModel.property_id).in_(property_ids))
|
||||
.where(col(EpcPropertyModel.source).in_(("lodged", "predicted")))
|
||||
).all()
|
||||
by_epc_source: dict[str, dict[int, str]] = {"lodged": {}, "predicted": {}}
|
||||
for epc in epcs:
|
||||
value = epc.property_type or epc.dwelling_type
|
||||
value = component.epc_value(epc)
|
||||
if epc.property_id is not None and value:
|
||||
by_epc_source[epc.source][epc.property_id] = _PROPERTY_TYPE_CODES.get(
|
||||
by_epc_source[epc.source][epc.property_id] = component.codes.get(
|
||||
value, value
|
||||
)
|
||||
|
||||
resolved: dict[int, str] = {}
|
||||
for pid in property_ids:
|
||||
resolved[pid] = (
|
||||
by_legacy_column = {
|
||||
row.id: value
|
||||
for row in candidates
|
||||
if row.id is not None and (value := component.legacy_value(row)) is not None
|
||||
}
|
||||
|
||||
return {
|
||||
pid: (
|
||||
by_override.get(pid)
|
||||
or by_epc_source["lodged"].get(pid)
|
||||
or by_epc_source["predicted"].get(pid)
|
||||
or by_legacy_column.get(pid)
|
||||
or _UNKNOWN
|
||||
)
|
||||
return resolved
|
||||
for pid in property_ids
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue