diff --git a/backend/app/modelling/property_filters.py b/backend/app/modelling/property_filters.py index a09aedb94..ce398cb25 100644 --- a/backend/app/modelling/property_filters.py +++ b/backend/app/modelling/property_filters.py @@ -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 + }