diff --git a/domain/epc_prediction/comparable_properties.py b/domain/epc_prediction/comparable_properties.py index 3e111bb36..2a612f02f 100644 --- a/domain/epc_prediction/comparable_properties.py +++ b/domain/epc_prediction/comparable_properties.py @@ -31,6 +31,26 @@ FLOOR_AREA_TOLERANCE = 0.20 # RdSAP Table S1 band letters in chronological order, for band-distance checks. _AGE_BAND_ORDER = "ABCDEFGHIJKLM" +# roof_construction codes by FORM family. Pinned empirically — a 7,974-cert +# co-occurrence sweep of code x roofs[0].description over single-building-part +# certs (scripts/roof_construction_code_sweep.py): 1=Flat 98%, 4=Pitched 99%, +# 5/8=Pitched 88%, 3="(another dwelling above)" 100% — plus 7/9 = "(another +# premises above)", established by the #1452 heat-loss suppression fix. +ROOF_FORM_BY_CONSTRUCTION: dict[int, str] = { + 1: "flat", + 3: "dwelling_above", + 4: "pitched", + 5: "pitched", + 7: "premises_above", + 8: "pitched", + 9: "premises_above", +} + + +def roof_form_of_construction(code: object) -> Optional[str]: + """The form family of a roof_construction code, or None when unknown.""" + return ROOF_FORM_BY_CONSTRUCTION.get(code) if isinstance(code, int) else None + def age_bands_within_one(candidate: object, target_band: object) -> bool: """Whether two Table S1 band letters are at most one band apart. Assessors @@ -109,6 +129,12 @@ def select_comparables( active=target.wall_construction is not None, minimum_cohort=minimum_cohort, ) + cohort = _maybe_filter( + cohort, + lambda c: _main_roof_form(c) == target.roof_form, + active=target.roof_form is not None, + minimum_cohort=minimum_cohort, + ) cohort = _maybe_filter( cohort, lambda c: age_bands_within_one( @@ -170,6 +196,12 @@ def _main_wall_construction(comparable: ComparableProperty) -> object: return parts[0].wall_construction if parts else None +def _main_roof_form(comparable: ComparableProperty) -> Optional[str]: + """The main building part's roof-form family, or None when unresolvable.""" + parts = comparable.epc.sap_building_parts + return roof_form_of_construction(parts[0].roof_construction) if parts else None + + def _main_construction_age_band(comparable: ComparableProperty) -> object: """The main building part's Table S1 band letter, or None when no part lodged.""" parts = comparable.epc.sap_building_parts diff --git a/domain/epc_prediction/historic_conditioning.py b/domain/epc_prediction/historic_conditioning.py index 5e92762c5..dca4b4052 100644 --- a/domain/epc_prediction/historic_conditioning.py +++ b/domain/epc_prediction/historic_conditioning.py @@ -82,6 +82,16 @@ _FUEL_CODES: dict[str, int] = { "biomass (community)": 31, } +# Roof FORM families by the description's pre-comma half. Only forms whose +# API code grouping is pinned (see comparable_properties.ROOF_FORM_BY_CONSTRUCTION +# for the empirical evidence); roof rooms and thatch stay None — never guess. +_ROOF_FORM_BY_PREFIX: dict[str, str] = { + "Pitched": "pitched", + "Flat": "flat", + "(another dwelling above)": "dwelling_above", + "(another premises above)": "premises_above", +} + _NOT_COMMUNITY_SUFFIX = " (not community)" # Pre-RdSAP-17 lodgements carry a deprecation rider after the fuel name; the @@ -117,6 +127,11 @@ def _wall_construction(description: str) -> Optional[int]: return _WALL_MATERIAL_CONSTRUCTION.get(material) +def _roof_form(description: str) -> Optional[str]: + form = description.split(",", 1)[0].strip() + return _ROOF_FORM_BY_PREFIX.get(form) + + def _main_fuel(description: str) -> Optional[int]: base = description.strip().removesuffix(_NOT_COMMUNITY_SUFFIX) base = base.removesuffix(_LEGACY_SUFFIX) @@ -140,6 +155,7 @@ def conditioning_from_historic(record: HistoricEpc) -> HistoricConditioning: construction_age_band=_AGE_BAND_LETTERS.get(record.construction_age_band), main_fuel=_main_fuel(record.main_fuel), total_floor_area_m2=_floor_area(record.total_floor_area), + roof_form=_roof_form(record.roof_description), ) @@ -173,4 +189,5 @@ def target_with_conditioning( construction_age_band=conditioning.construction_age_band, main_fuel=conditioning.main_fuel, total_floor_area_m2=conditioning.total_floor_area_m2, + roof_form=conditioning.roof_form, ) diff --git a/scripts/expired_prediction_pairs_harness.py b/scripts/expired_prediction_pairs_harness.py index 3b113a449..d301987d7 100644 --- a/scripts/expired_prediction_pairs_harness.py +++ b/scripts/expired_prediction_pairs_harness.py @@ -46,6 +46,7 @@ from domain.epc_prediction.comparable_properties import ( # noqa: E402 FLOOR_AREA_TOLERANCE, ComparableProperty, age_bands_within_one, + roof_form_of_construction, select_comparables, ) from domain.epc_prediction.epc_prediction import EpcPrediction # noqa: E402 @@ -210,6 +211,11 @@ def _part0(epc: EpcPropertyData) -> Optional[object]: return parts[0] if parts else None +def _actual_roof_form(epc: EpcPropertyData) -> Optional[str]: + part = _part0(epc) + return roof_form_of_construction(getattr(part, "roof_construction", None)) + + def _actual_age_band(epc: EpcPropertyData) -> Optional[object]: part = _part0(epc) return getattr(part, "construction_age_band", None) @@ -247,6 +253,7 @@ def simulate_conditioning_ladder( age_band: Optional[str], main_fuel: Optional[int], total_floor_area_m2: Optional[float], + roof_form: Optional[str] = None, minimum_cohort: int = 5, ) -> dict[str, Optional[LadderStep]]: """Replay select_comparables' age->fuel->TFA conditioning sequence over the @@ -266,6 +273,11 @@ def simulate_conditioning_ladder( if engaged: cohort = matches + apply( + "roof_form", + roof_form is not None, + [c for c in cohort if _actual_roof_form(c.epc) == roof_form], + ) apply( "construction_age_band", age_band is not None, @@ -296,7 +308,7 @@ def format_diagnosis(rows: list[dict[str, object]]) -> str: with the newly lodged one (the staleness measurement).""" if not rows: return "" - filters = ("construction_age_band", "main_fuel", "total_floor_area") + filters = ("roof_form", "construction_age_band", "main_fuel", "total_floor_area") lines = [ "", "## Diagnosis — filter engagement (conditioned arm)", @@ -321,6 +333,7 @@ def format_diagnosis(rows: list[dict[str, object]]) -> str: "property_type", "built_form", "wall_construction", + "roof_form", "construction_age_band", "main_fuel", "tfa_within_band", @@ -478,6 +491,7 @@ def run( # pragma: no cover - live IO composition age_band=conditioning.construction_age_band, main_fuel=conditioning.main_fuel, total_floor_area_m2=conditioning.total_floor_area_m2, + roof_form=conditioning.roof_form, ) emit_telemetry( { @@ -504,6 +518,11 @@ def run( # pragma: no cover - live IO composition if conditioning.built_form is None else conditioning.built_form == actual.built_form ), + "agrees_roof_form": ( + None + if conditioning.roof_form is None + else conditioning.roof_form == _actual_roof_form(actual) + ), "agrees_wall_construction": ( None if conditioning.wall_construction is None diff --git a/tests/scripts/test_expired_prediction_pairs_harness.py b/tests/scripts/test_expired_prediction_pairs_harness.py index 2bf4e5268..70601ee87 100644 --- a/tests/scripts/test_expired_prediction_pairs_harness.py +++ b/tests/scripts/test_expired_prediction_pairs_harness.py @@ -142,6 +142,7 @@ def test_ladder_simulation_skips_unresolved_attributes(): ) assert steps == { + "roof_form": None, "construction_age_band": None, "main_fuel": None, "total_floor_area": None,