From 35a0c6e972e094dc004d0984ab27f8a0150ab543 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Sat, 18 Jul 2026 19:38:12 +0000 Subject: [PATCH] Source export fabric from structured SAP fields; fix stale-cert selection The "latest EPC per UPRN" selection ordered by e.id DESC, but row-id does not track recency: a gov-EPC cert re-ingested after a PasHub survey lands with a higher e.id, so the header/perf reads silently picked the stale gov cert (wrong TFA, lodgement, property_type, current band) on 3 of portfolio 838. Order by the effective lodgement date (registration -> completion -> inspection) DESC NULLS LAST, e.id DESC tiebreak. Compose the descriptive fabric from the structured SAP fields a re-survey lodges (no gov-EPC prose): walls/roof/floor from the main epc_building_part, heating and hot water from epc_main_heating_detail + the water-heating codes, decoded with the SAP engine's own code space (fabric_description.py). Structured fabric wins where present and falls back to the gov-EPC prose otherwise; windows and lighting stay on prose (their per-element codes do not reconstruct the dwelling phrase). Add built_form (epc_property, code-decoded) and property_age_band (building part construction_age_band -> RdSAP date range) columns. Co-Authored-By: Claude Opus 4.8 (1M context) --- domain/scenario_export/fabric_description.py | 284 ++++++++++++++++++ domain/scenario_export/scenario_sheet.py | 6 + .../scenario_export_postgres_repository.py | 96 +++++- .../test_fabric_description.py | 174 +++++++++++ ...est_scenario_export_postgres_repository.py | 238 ++++++++++++++- 5 files changed, 790 insertions(+), 8 deletions(-) create mode 100644 domain/scenario_export/fabric_description.py create mode 100644 tests/domain/scenario_export/test_fabric_description.py diff --git a/domain/scenario_export/fabric_description.py b/domain/scenario_export/fabric_description.py new file mode 100644 index 000000000..54529f212 --- /dev/null +++ b/domain/scenario_export/fabric_description.py @@ -0,0 +1,284 @@ +"""Human-readable fabric descriptions from an EPC's structured SAP fields +(ADR-0065). + +The PasHub re-survey lodges *structured* fabric — SAP building parts, a main +heating detail, coded fuel/water-heating — but no gov-EPC descriptive prose, so +the export's descriptive columns would be blank for a re-surveyed Property. This +module composes the client-sheet text for walls / roof / floor (from the main +``epc_building_part``) and heating / hot water (from ``epc_main_heating_detail`` +and the ``epc_property`` water-heating fields), decoding the SAP integer codes +with the same code space the SAP engine uses. Windows and lighting are left on +the gov-EPC prose fallback (their per-element codes do not cleanly reconstruct +the dwelling-level phrase). + +Pure logic: the repository reads the structured rows and calls these; each +renderer returns ``None`` when it has nothing to say, so the caller can fall +back to the gov-EPC prose. +""" + +from __future__ import annotations + +from typing import Any, Optional + +# --- SAP integer-code → display-text maps --------------------------------- +# wall_construction (SAP10 codes, domain/sap10_ml/rdsap_uvalues.py:127-142). +_WALL_CONSTRUCTION: dict[int, str] = { + 1: "Granite or whinstone", + 2: "Sandstone", + 3: "Solid brick", + 4: "Cavity", + 5: "Timber frame", + 6: "System build", + 7: "Cob", + 8: "Park home", + 9: "Curtain wall", + 10: "Unknown", + 11: "Cavity (filled party wall)", +} +# wall_insulation_type (RdSAP schema codes, rdsap_uvalues.py:145-158). +_WALL_INSULATION: dict[int, str] = { + 1: "external insulation", + 2: "filled cavity", + 3: "internal insulation", + 4: "as built", + 5: "no insulation", + 6: "filled cavity + external insulation", + 7: "filled cavity + internal insulation", +} +# heat_emitter_type (SAP10.2, mapper.py _ELMHURST_HEAT_EMITTER_TO_SAP10). +_HEAT_EMITTER: dict[int, str] = { + 1: "radiators", + 2: "underfloor heating (in screed)", + 3: "underfloor heating (timber floor)", + 4: "warm air", + 5: "fan coils", +} +# main_fuel_type / water_heating_fuel (RdSAP "(not community)" family; invert of +# main_fuel_overlay._FUEL_CODES). +_FUEL: dict[int, str] = { + 3: "bottled LPG", + 6: "wood logs", + 10: "dual fuel (mineral and wood)", + 15: "smokeless coal", + 17: "LPG (special condition)", + 20: "mains gas (community)", + 25: "electricity (community)", + 26: "mains gas", + 27: "LPG (bulk)", + 28: "oil", + 29: "electricity", + 31: "biomass (community)", + 33: "house coal", +} +# main_heating_control (SAP10.2 Table 4e Group 1; comments in +# heating_recommendation._CONTROL_FEATURES_BY_CODE). +_HEATING_CONTROL: dict[int, str] = { + 2101: "no time or thermostatic control", + 2102: "programmer, no room thermostat", + 2103: "room thermostat only", + 2104: "programmer and room thermostat", + 2105: "programmer and at least two room thermostats", + 2106: "programmer, room thermostat and TRVs", + 2107: "programmer, TRVs and bypass", + 2108: "programmer, TRVs and flow switch", + 2109: "programmer, TRVs and boiler energy manager", + 2111: "TRVs and bypass", + 2113: "room thermostat and TRVs", +} +# water_heating_code (SAP Table 4a; water_heating_overlay docstring). +_WATER_HEATING_SYSTEM: dict[int, str] = { + 901: "from main system", + 903: "electric immersion", + 911: "gas boiler / circulator", +} +# RdSAP England-&-Wales construction age-band letters → the date range the EPC +# prints (domain/epc/property_overrides/construction_age_band.py). The lodged +# `epc_building_part.construction_age_band` carries the bare letter. +_AGE_BAND: dict[str, str] = { + "A": "before 1900", + "B": "1900-1929", + "C": "1930-1949", + "D": "1950-1966", + "E": "1967-1975", + "F": "1976-1982", + "G": "1983-1990", + "H": "1991-1995", + "I": "1996-2002", + "J": "2003-2006", + "K": "2007-2011", + "L": "2012-2022", + "M": "2023 onwards", +} + + +def _as_int(value: Any) -> Optional[int]: + """A SAP code as ``int``. Accepts the raw int, a numeric string, or ``None``; + a non-numeric label (e.g. an insulation ``"As built"`` string in a numeric + slot) yields ``None``.""" + if value is None: + return None + if isinstance(value, bool): + return None + if isinstance(value, int): + return value + try: + return int(str(value).strip()) + except ValueError: + return None + + +def _clean_text(value: Any) -> Optional[str]: + """A lodged free-text field trimmed to a non-empty string, else ``None``. + A bare integer (a code lodged in a text slot) is not display text.""" + if value is None or isinstance(value, (int, float, bool)): + return None + text = str(value).strip() + return text or None + + +# built_form codes (epc_codes.csv; override_code_mapping._BUILT_FORM_CODES). +# The lodged value is usually the display text already; a minority of records +# carry the bare integer code instead. +_BUILT_FORM: dict[int, str] = { + 1: "Detached", + 2: "Semi-detached", + 3: "End-terrace", + 4: "Mid-terrace", + 5: "Enclosed end-terrace", + 6: "Enclosed mid-terrace", +} + + +def built_form_description(built_form: Any) -> Optional[str]: + """The dwelling's built form. Decodes the integer code where a record + lodges one, otherwise passes the lodged display text through unchanged.""" + code = _as_int(built_form) + if code is not None: + return _BUILT_FORM.get(code) + return _clean_text(built_form) + + +def age_band_description(band: Any) -> Optional[str]: + """The construction age band as its date range, e.g. ``"1900-1929 (band B)"``. + The lodged value is a bare RdSAP letter; an unrecognised or ``"Unknown"`` + band renders nothing.""" + letter = _clean_text(band) + if letter is None: + return None + date_range = _AGE_BAND.get(letter.strip().upper()) + return f"{date_range} (band {letter.strip().upper()})" if date_range else None + + +def wall_description(construction: Any, insulation: Any) -> Optional[str]: + """A wall part's text, e.g. ``"Cavity wall, filled cavity"``. The + construction names the wall; the insulation qualifies it when known.""" + cons = _WALL_CONSTRUCTION.get(_as_int(construction) or 0) + if cons is None: + return None + ins = _WALL_INSULATION.get(_as_int(insulation) or 0) + return f"{cons} wall, {ins}" if ins else f"{cons} wall" + + +def _insulation_thickness_mm(thickness: Any) -> Optional[int]: + """A roof-insulation thickness in mm from a numeric code or a ``"225mm"`` + string; a non-numeric label (``"As built"``, ``"ND"``) or ``None`` yields + ``None``. Zero is a real value (no insulation) and is preserved.""" + as_int = _as_int(thickness) + if as_int is not None: + return as_int + text = _clean_text(thickness) + if text is None: + return None + digits = "".join(ch for ch in text if ch.isdigit()) + return int(digits) if digits else None + + +def roof_description( + construction_type: Any, insulation_location: Any, insulation_thickness: Any +) -> Optional[str]: + """A roof part's text, e.g. ``"Pitched roof (Slates or tiles), Access to + loft; 270mm insulation at joists"``. The construction is already lodged as + prose; the insulation clause is appended when a thickness or a named + location is present.""" + cons = _clean_text(construction_type) + if cons is None: + return None + thickness_mm = _insulation_thickness_mm(insulation_thickness) + location = _clean_text(insulation_location) + if location is not None and location.lower() in {"unknown", "none"}: + location = None + if thickness_mm == 0: + return f"{cons}; no insulation" + if thickness_mm is None and location is None: + return cons + clause = "insulation" if thickness_mm is None else f"{thickness_mm}mm insulation" + if location is not None: + clause = f"{clause} at {location.lower()}" + return f"{cons}; {clause}" + + +def floor_description( + floor_type: Any, construction_type: Any, insulation: Any +) -> Optional[str]: + """A floor part's text, e.g. ``"Ground floor, suspended timber, insulation + as built"``. Composed from the lodged floor type, construction, and + insulation status (all lodged as prose).""" + parts: list[str] = [] + location = _clean_text(floor_type) + if location is not None: + parts.append(location[0].upper() + location[1:].lower()) + construction = _clean_text(construction_type) + if construction is not None: + parts.append(construction.lower()) + ins = _clean_text(insulation) + if ins is not None: + parts.append(f"insulation {ins.lower()}") + return ", ".join(parts) or None + + +def heating_description( + fuel: Any, emitter: Any, control: Any, condensing: Any +) -> Optional[str]: + """The main heating system's text from its SAP codes, e.g. ``"Mains gas, + radiators (condensing); programmer, room thermostat and TRVs"``. Reports + only what the codes state — the appliance archetype is not lodged on a + re-survey, so no boiler/heat-pump type is asserted.""" + fuel_text = _FUEL.get(_as_int(fuel) or 0) + emitter_text = _HEAT_EMITTER.get(_as_int(emitter) or 0) + lead_parts = [p for p in (fuel_text, emitter_text) if p] + if not lead_parts: + return None + lead = ", ".join(p[0].upper() + p[1:] if i == 0 else p for i, p in enumerate(lead_parts)) + if condensing is True: + lead = f"{lead} (condensing)" + control_text = _HEATING_CONTROL.get(_as_int(control) or 0) + return f"{lead}; {control_text}" if control_text else lead + + +def hot_water_description( + water_heating_code: Any, + water_heating_fuel: Any, + main_fuel: Any, + has_cylinder: Any, + solar: Any, +) -> Optional[str]: + """The hot-water system's text from its SAP codes, e.g. ``"From main + system, mains gas"``. Where the water-heating code is not lodged on the + re-survey, it falls back to the SAP default (a dwelling with no cylinder + draws hot water from the main system) using the main heating fuel.""" + system = _WATER_HEATING_SYSTEM.get(_as_int(water_heating_code) or 0) + fuel = _FUEL.get(_as_int(water_heating_fuel) or 0) or _FUEL.get(_as_int(main_fuel) or 0) + if system is None: + # No lodged water-heating code: a dwelling without a cylinder inherits + # hot water from the main system (SAP Table 4a code 901). + if has_cylinder is True: + system = "hot water cylinder" + elif has_cylinder is False: + system = "from main system" + else: + return None + text = f"{system}, {fuel}" if fuel else system + text = text[0].upper() + text[1:] + if solar is True: + text = f"{text}; with solar water heating" + return text diff --git a/domain/scenario_export/scenario_sheet.py b/domain/scenario_export/scenario_sheet.py index 018b89e11..ae3860d23 100644 --- a/domain/scenario_export/scenario_sheet.py +++ b/domain/scenario_export/scenario_sheet.py @@ -59,6 +59,8 @@ _ID_COLUMNS: tuple[str, ...] = ( "address", "postcode", "property_type", + "built_form", + "property_age_band", "walls", "roof", "floor", @@ -147,6 +149,8 @@ class PropertyScenarioData: address: Optional[str] = None postcode: Optional[str] = None property_type: Optional[str] = None + built_form: Optional[str] = None + property_age_band: Optional[str] = None walls: Optional[str] = None roof: Optional[str] = None floor: Optional[str] = None @@ -196,6 +200,8 @@ def _shape_row(prop: PropertyScenarioData) -> dict[str, Any]: "address": _blank(prop.address), "postcode": _blank(prop.postcode), "property_type": _blank(prop.property_type), + "built_form": _blank(prop.built_form), + "property_age_band": _blank(prop.property_age_band), "walls": _blank(prop.walls), "roof": _blank(prop.roof), "floor": _blank(prop.floor), diff --git a/repositories/scenario_export/scenario_export_postgres_repository.py b/repositories/scenario_export/scenario_export_postgres_repository.py index a2674b20f..50e21dd66 100644 --- a/repositories/scenario_export/scenario_export_postgres_repository.py +++ b/repositories/scenario_export/scenario_export_postgres_repository.py @@ -6,6 +6,15 @@ from typing import Any, Optional, Sequence from sqlalchemy import text from sqlmodel import Session +from domain.scenario_export.fabric_description import ( + age_band_description, + built_form_description, + floor_description, + heating_description, + hot_water_description, + roof_description, + wall_description, +) from domain.scenario_export.scenario_sheet import ExportMeasure, PropertyScenarioData from repositories.scenario_export.scenario_export_repository import ( ScenarioExportRepository, @@ -60,19 +69,26 @@ _MEASURES_SQL = text( # performance are matched by UPRN (the latest epc_property for the source), # because the PasHub re-fetch lands records with property_id = NULL — a # property_id join would silently fall back to a stale gov-EPC cert (wrong -# lodgement dates / property_type) or nothing. The descriptive prose elements -# have no PasHub equivalent (a survey lodges structured fields, not gov-EPC -# prose), so they stay on the property_id join as the best-available gov-EPC -# fallback. All three are read separately and composed per Property. +# lodgement dates / property_type) or nothing. "Latest" is ordered by the cert's +# effective lodgement date (registration → completion → inspection) DESC, NOT by +# `e.id`: row-id order does not track recency — a gov-EPC cert re-ingested after +# the PasHub survey lands with a HIGHER `e.id`, so an `e.id` sort silently picks +# the stale gov cert over the fresh 2026 survey (observed on 3 of portfolio 838). +# `e.id DESC` remains the tiebreak within one date (duplicate survey loads carry +# identical facts). The descriptive prose elements have no PasHub equivalent (a +# survey lodges structured fields, not gov-EPC prose), so they stay on the +# property_id join as the best-available gov-EPC fallback. All three are read +# separately and composed per Property. _EPC_HEADER_SQL = text( "WITH latest AS (" " SELECT DISTINCT ON (p.id) p.id AS property_id, e.id AS epc_id" " FROM property p JOIN epc_property e ON e.uprn = p.uprn" " WHERE p.id = ANY(:property_ids) AND e.source = :source" - " ORDER BY p.id, e.id DESC)" + " ORDER BY p.id, COALESCE(e.registration_date, e.completion_date," + " e.inspection_date) DESC NULLS LAST, e.id DESC)" " SELECT l.property_id, e.property_type, e.total_floor_area_m2," " COALESCE(e.registration_date, e.completion_date, e.inspection_date)," - " e.habitable_rooms_count" + " e.habitable_rooms_count, e.built_form" " FROM latest l JOIN epc_property e ON e.id = l.epc_id" ) _EPC_ELEMENTS_SQL = text( @@ -86,13 +102,57 @@ _EPC_PERF_SQL = text( " SELECT DISTINCT ON (p.id) p.id AS property_id, e.id AS epc_id" " FROM property p JOIN epc_property e ON e.uprn = p.uprn" " WHERE p.id = ANY(:property_ids) AND e.source = :source" - " ORDER BY p.id, e.id DESC)" + " ORDER BY p.id, COALESCE(e.registration_date, e.completion_date," + " e.inspection_date) DESC NULLS LAST, e.id DESC)" " SELECT l.property_id, perf.energy_rating_current," " perf.current_energy_efficiency_band" " FROM latest l" " JOIN epc_property_energy_performance perf ON perf.epc_property_id = l.epc_id" ) +# The structured SAP fabric on the same canonical (latest-by-date, UPRN-matched) +# cert the header reads: the MAIN building part (walls / roof / floor), the main +# heating detail (fuel / emitter / control / condensing) and the water-heating +# fields. A PasHub re-survey lodges these coded facts but no gov-EPC prose, so +# they supply walls/roof/floor/heating/hot_water where the prose fallback is +# blank. The main building part is chosen identifier='main' first, then the +# lowest part number / id (matching the SAP engine's `parts[0]` primary part). +# Windows and lighting are not read here (their per-element codes do not cleanly +# reconstruct the dwelling phrase); they stay on the gov-EPC prose fallback. +_EPC_STRUCTURED_FABRIC_SQL = text( + "WITH latest AS (" + " SELECT DISTINCT ON (p.id) p.id AS property_id, e.id AS epc_id" + " FROM property p JOIN epc_property e ON e.uprn = p.uprn" + " WHERE p.id = ANY(:property_ids) AND e.source = :source" + " ORDER BY p.id, COALESCE(e.registration_date, e.completion_date," + " e.inspection_date) DESC NULLS LAST, e.id DESC)," + " main_part AS (" + " SELECT DISTINCT ON (l.property_id) l.property_id, bp.construction_age_band," + " bp.wall_construction, bp.wall_insulation_type, bp.roof_construction_type," + " bp.roof_insulation_location, bp.roof_insulation_thickness, bp.floor_type," + " bp.floor_construction_type, bp.floor_insulation_type_str" + " FROM latest l JOIN epc_building_part bp ON bp.epc_property_id = l.epc_id" + " ORDER BY l.property_id," + " (CASE WHEN lower(bp.identifier) = 'main' THEN 0 ELSE 1 END)," + " bp.building_part_number NULLS FIRST, bp.id)," + " main_heat AS (" + " SELECT DISTINCT ON (h.epc_property_id) h.epc_property_id, h.main_fuel_type," + " h.heat_emitter_type, h.main_heating_control, h.condensing" + " FROM epc_main_heating_detail h" + " ORDER BY h.epc_property_id, h.main_heating_number NULLS FIRST, h.id)" + " SELECT l.property_id, mp.wall_construction, mp.wall_insulation_type," + " mp.roof_construction_type, mp.roof_insulation_location," + " mp.roof_insulation_thickness, mp.floor_type, mp.floor_construction_type," + " mp.floor_insulation_type_str, mh.main_fuel_type, mh.heat_emitter_type," + " mh.main_heating_control, mh.condensing, e.heating_water_heating_code," + " e.heating_water_heating_fuel, e.has_hot_water_cylinder," + " e.solar_water_heating, mp.construction_age_band" + " FROM latest l" + " JOIN epc_property e ON e.id = l.epc_id" + " LEFT JOIN main_part mp ON mp.property_id = l.property_id" + " LEFT JOIN main_heat mh ON mh.epc_property_id = l.epc_id" +) + # A landlord property_type override wins over any EPC-derived value (ADR-0065); # the descriptive prose fields have no override prose, so only property_type is # resolved this way (the override enum carries coded fabric facts, not prose). @@ -183,6 +243,8 @@ class ScenarioExportPostgresRepository(ScenarioExportRepository): energy_consumption_savings=_float(row[11]), valuation_increase=_float(row[12]), property_type=facts.get("property_type"), + built_form=facts.get("built_form"), + property_age_band=facts.get("property_age_band"), walls=facts.get("walls"), roof=facts.get("roof"), floor=facts.get("floor"), @@ -261,6 +323,7 @@ class ScenarioExportPostgresRepository(ScenarioExportRepository): "lodgement_date": registration_date, "is_expired": _is_expired(registration_date), "number_of_rooms": row[4], + "built_form": built_form_description(row[5]), } ) @@ -287,4 +350,23 @@ class ScenarioExportPostgresRepository(ScenarioExportRepository): } ) + # Structured SAP fabric (walls/roof/floor/heating/hot_water) composed + # from the canonical cert's coded fields; overrides the gov-EPC prose + # where present (a re-survey lodges these but no prose), and leaves the + # prose in place where a field renders to nothing. + for row in connection.execute(_EPC_STRUCTURED_FABRIC_SQL, source_params).all(): + rendered = { + "property_age_band": age_band_description(row[17]), + "walls": wall_description(row[1], row[2]), + "roof": roof_description(row[3], row[4], row[5]), + "floor": floor_description(row[6], row[7], row[8]), + "heating": heating_description(row[9], row[10], row[11], row[12]), + "hot_water": hot_water_description( + row[13], row[14], row[9], row[15], row[16] + ), + } + facts.setdefault(row[0], {}).update( + {field: value for field, value in rendered.items() if value is not None} + ) + return facts diff --git a/tests/domain/scenario_export/test_fabric_description.py b/tests/domain/scenario_export/test_fabric_description.py new file mode 100644 index 000000000..6d0397c1f --- /dev/null +++ b/tests/domain/scenario_export/test_fabric_description.py @@ -0,0 +1,174 @@ +"""Behaviour of the structured-fabric renderers (ADR-0065): composing client +sheet text from an EPC's SAP integer codes, decoding with the SAP engine's own +code space and normalising the ragged shapes a re-survey lodges.""" + +from __future__ import annotations + +from domain.scenario_export.fabric_description import ( + age_band_description, + built_form_description, + floor_description, + heating_description, + hot_water_description, + roof_description, + wall_description, +) + + +class TestBuiltFormDescription: + def test_lodged_display_text_passes_through(self) -> None: + assert built_form_description("Mid-terrace") == "Mid-terrace" + assert built_form_description("Semi-Detached") == "Semi-Detached" + + def test_integer_code_is_decoded(self) -> None: + assert built_form_description(4) == "Mid-terrace" + assert built_form_description("1") == "Detached" + + def test_absent_or_unknown_code_renders_nothing(self) -> None: + assert built_form_description(None) is None + assert built_form_description(99) is None + + +class TestAgeBandDescription: + def test_rdsap_letter_decodes_to_date_range_and_band(self) -> None: + assert age_band_description("B") == "1900-1929 (band B)" + assert age_band_description("A") == "before 1900 (band A)" + assert age_band_description("M") == "2023 onwards (band M)" + + def test_lowercase_and_whitespace_are_tolerated(self) -> None: + assert age_band_description(" k ") == "2007-2011 (band K)" + + def test_unknown_or_absent_band_renders_nothing(self) -> None: + assert age_band_description("Unknown") is None + assert age_band_description(None) is None + assert age_band_description("Z") is None + + +class TestWallDescription: + def test_construction_and_insulation_codes_decode_to_text(self) -> None: + assert wall_description(4, 2) == "Cavity wall, filled cavity" + assert wall_description(3, 4) == "Solid brick wall, as built" + assert wall_description(5, 1) == "Timber frame wall, external insulation" + + def test_construction_only_when_insulation_unknown(self) -> None: + assert wall_description(4, None) == "Cavity wall" + assert wall_description(4, 99) == "Cavity wall" + + def test_numeric_string_codes_are_accepted(self) -> None: + # JSONB can round-trip a code as a string; it still decodes. + assert wall_description("4", "2") == "Cavity wall, filled cavity" + + def test_unknown_construction_renders_nothing(self) -> None: + assert wall_description(None, 2) is None + assert wall_description(99, 2) is None + + +class TestRoofDescription: + def test_construction_prose_with_thickness_and_location(self) -> None: + assert ( + roof_description("Pitched roof, Access to loft", "Joists", 270) + == "Pitched roof, Access to loft; 270mm insulation at joists" + ) + + def test_thickness_string_with_unit_is_normalised(self) -> None: + # A minority of records lodge the thickness as "225mm" rather than 225. + assert ( + roof_description("Pitched", "Joists", "225mm") + == "Pitched; 225mm insulation at joists" + ) + + def test_non_numeric_thickness_and_coded_location_are_dropped(self) -> None: + # thickness "As built" is not a measurement; a bare int location is a + # code, not a place — with neither, only the construction survives. + assert roof_description("Pitched", 2, "As built") == "Pitched" + + def test_named_location_without_a_thickness_still_notes_insulation(self) -> None: + assert ( + roof_description("Pitched", "Joists", "As built") + == "Pitched; insulation at joists" + ) + + def test_zero_thickness_reads_as_no_insulation(self) -> None: + assert roof_description("Flat roof", "Joists", 0) == "Flat roof; no insulation" + + def test_unknown_location_is_omitted(self) -> None: + assert roof_description("Pitched", "Unknown", 100) == "Pitched; 100mm insulation" + + def test_construction_only_when_no_insulation_detail(self) -> None: + assert roof_description("Pitched", None, None) == "Pitched" + + def test_no_construction_renders_nothing(self) -> None: + assert roof_description(None, "Joists", 270) is None + + +class TestFloorDescription: + def test_composes_type_construction_and_insulation(self) -> None: + assert ( + floor_description("Ground floor", "Suspended, timber", "As Built") + == "Ground floor, suspended, timber, insulation as built" + ) + + def test_normalises_casing_of_the_floor_type(self) -> None: + assert floor_description("Ground Floor", "Solid", None) == "Ground floor, solid" + + def test_retrofitted_insulation_is_carried(self) -> None: + assert ( + floor_description("Ground floor", "Solid", "Retro-fitted") + == "Ground floor, solid, insulation retro-fitted" + ) + + def test_empty_when_nothing_lodged(self) -> None: + assert floor_description(None, None, None) is None + + +class TestHeatingDescription: + def test_fuel_emitter_control_and_condensing(self) -> None: + assert ( + heating_description(26, 1, 2106, True) + == "Mains gas, radiators (condensing); programmer, room thermostat and TRVs" + ) + + def test_condensing_false_or_absent_omits_the_flag(self) -> None: + assert heating_description(26, 1, 2104, None) == ( + "Mains gas, radiators; programmer and room thermostat" + ) + assert heating_description(26, 1, None, False) == "Mains gas, radiators" + + def test_unknown_control_is_dropped(self) -> None: + assert heating_description(26, 1, 9999, True) == "Mains gas, radiators (condensing)" + + def test_empty_when_no_fuel_or_emitter(self) -> None: + assert heating_description(None, None, 2106, True) is None + + +class TestHotWaterDescription: + def test_lodged_code_and_fuel_decode(self) -> None: + assert ( + hot_water_description(901, 26, 26, False, False) == "From main system, mains gas" + ) + assert ( + hot_water_description(903, 29, 26, True, False) + == "Electric immersion, electricity" + ) + + def test_falls_back_to_main_system_when_no_code_and_no_cylinder(self) -> None: + # SAP Table 4a: a dwelling with no cylinder draws hot water from the main + # system; the main heating fuel names it. + assert ( + hot_water_description(None, None, 26, False, False) + == "From main system, mains gas" + ) + + def test_cylinder_without_a_code(self) -> None: + assert ( + hot_water_description(None, None, 29, True, False) + == "Hot water cylinder, electricity" + ) + + def test_solar_water_heating_is_appended(self) -> None: + assert hot_water_description(901, 26, 26, False, True) == ( + "From main system, mains gas; with solar water heating" + ) + + def test_empty_when_nothing_lodged(self) -> None: + assert hot_water_description(None, None, None, None, False) is None diff --git a/tests/repositories/scenario_export/test_scenario_export_postgres_repository.py b/tests/repositories/scenario_export/test_scenario_export_postgres_repository.py index 8d6f77bdd..0210502ad 100644 --- a/tests/repositories/scenario_export/test_scenario_export_postgres_repository.py +++ b/tests/repositories/scenario_export/test_scenario_export_postgres_repository.py @@ -4,14 +4,16 @@ read-model, from persisted data only (no live EPC calls).""" from __future__ import annotations -from typing import Optional +from typing import Optional, Union from sqlalchemy import Engine from sqlmodel import Session from datatypes.epc.domain.epc import Epc from infrastructure.postgres.epc_property_table import ( # registers epc tables + EpcBuildingPartModel, EpcEnergyElementModel, + EpcMainHeatingDetailModel, EpcPropertyEnergyPerformanceModel, EpcPropertyModel, ) @@ -37,6 +39,7 @@ def _add_epc( uprn: int = 100001, source: str = "lodged", property_type: Optional[str] = None, + built_form: Optional[str] = None, total_floor_area_m2: float = 90.0, registration_date: str = "2005-01-01", habitable_rooms_count: int = 4, @@ -54,6 +57,7 @@ def _add_epc( uprn=uprn, source=source, property_type=property_type, + built_form=built_form, total_floor_area_m2=total_floor_area_m2, registration_date=registration_date, habitable_rooms_count=habitable_rooms_count, @@ -103,6 +107,66 @@ def _add_epc( ) +def _add_building_part( + session: Session, + *, + epc_id: int, + identifier: str = "main", + building_part_number: Optional[int] = None, + wall_construction: int = 4, + wall_insulation_type: int = 2, + roof_construction_type: Optional[str] = "Pitched roof, Access to loft", + roof_insulation_location: Optional[str] = "Joists", + roof_insulation_thickness: Optional[Union[str, int]] = 270, + floor_type: Optional[str] = "Ground floor", + floor_construction_type: Optional[str] = "Solid", + floor_insulation_type_str: Optional[str] = "As Built", +) -> None: + """Seed one SAP building part on an EPC slot (the source of the structured + walls / roof / floor text).""" + session.add( + EpcBuildingPartModel( + epc_property_id=epc_id, + identifier=identifier, + construction_age_band="D", + wall_construction=wall_construction, + wall_insulation_type=wall_insulation_type, + wall_thickness_measured=False, + building_part_number=building_part_number, + roof_construction_type=roof_construction_type, + roof_insulation_location=roof_insulation_location, + roof_insulation_thickness=roof_insulation_thickness, + floor_type=floor_type, + floor_construction_type=floor_construction_type, + floor_insulation_type_str=floor_insulation_type_str, + ) + ) + + +def _add_main_heating( + session: Session, + *, + epc_id: int, + main_fuel_type: int = 26, + heat_emitter_type: int = 1, + main_heating_control: int = 2106, + condensing: Optional[bool] = True, +) -> None: + """Seed the main heating detail on an EPC slot (the source of the structured + heating text).""" + session.add( + EpcMainHeatingDetailModel( + epc_property_id=epc_id, + has_fghrs=False, + main_fuel_type=main_fuel_type, + heat_emitter_type=heat_emitter_type, + emitter_temperature=1, + main_heating_control=main_heating_control, + condensing=condensing, + ) + ) + + def test_returns_a_row_for_a_property_with_a_default_plan_under_the_scenario( db_engine: Engine, ) -> None: @@ -361,6 +425,178 @@ def test_effective_epc_descriptive_fields_come_from_the_lodged_epc( assert row.is_expired is True +def test_header_picks_the_freshest_cert_by_date_not_by_row_id( + db_engine: Engine, +) -> None: + # arrange — two lodged certs on one UPRN: a fresh 2026 survey lodged with a + # LOWER epc_id, and a stale 2013 gov cert re-ingested afterwards with a + # HIGHER epc_id. An `e.id DESC` sort would wrongly pick the stale gov cert; + # the freshest lodgement date must win (the portfolio-838 TFA regression). + with Session(db_engine) as session: + session.add( + PropertyRow(id=1, portfolio_id=100, landlord_property_id="LP1", uprn=100001) + ) + session.add( + PlanModel(portfolio_id=100, property_id=1, scenario_id=5, is_default=True) + ) + _add_epc( + session, + property_id=1, + epc_id=10, # fresh survey, LOWER id + uprn=100001, + total_floor_area_m2=76.62, + registration_date="2026-05-18", + ) + _add_epc( + session, + property_id=1, + epc_id=20, # stale gov cert re-ingested later, HIGHER id + uprn=100001, + total_floor_area_m2=77.0, + registration_date="2013-10-09", + ) + session.commit() + + # act + with Session(db_engine) as session: + rows = ScenarioExportPostgresRepository(session).rows_for( + portfolio_id=100, scenario_id=5, property_ids=[1] + ) + + # assert — the fresh 2026 survey's floor area and lodgement win, not the + # higher-id 2013 gov cert. + assert rows[0].total_floor_area == 76.62 + assert rows[0].lodgement_date == "2026-05-18" + + +def test_structured_fabric_supplies_walls_roof_floor_heating_and_hot_water( + db_engine: Engine, +) -> None: + # arrange — a lodged EPC carrying gov prose for every descriptive field AND + # the structured SAP fabric (building part + heating detail) a re-survey + # lodges. The structured fabric must win for walls/roof/floor/heating/ + # hot_water; windows/lighting have no structured source and keep the prose. + with Session(db_engine) as session: + session.add( + PropertyRow(id=1, portfolio_id=100, landlord_property_id="LP1", uprn=100001) + ) + session.add( + PlanModel(portfolio_id=100, property_id=1, scenario_id=5, is_default=True) + ) + _add_epc( + session, + property_id=1, + epc_id=1, + source="lodged", + built_form="Mid-terrace", + elements={ + "wall": "PROSE walls", + "roof": "PROSE roof", + "floor": "PROSE floor", + "main_heating": "PROSE heating", + "hot_water": "PROSE hot water", + "window": "Fully double glazed", + "lighting": "Low energy lighting in all fixed outlets", + }, + ) + _add_building_part(session, epc_id=1) + _add_main_heating(session, epc_id=1) + session.commit() + + # act + with Session(db_engine) as session: + rows = ScenarioExportPostgresRepository(session).rows_for( + portfolio_id=100, scenario_id=5, property_ids=[1] + ) + + # assert — structured fabric replaces the prose for its five fields; windows + # and lighting (no structured source) keep the gov prose. + row = rows[0] + assert row.built_form == "Mid-terrace" + assert row.property_age_band == "1950-1966 (band D)" + assert row.walls == "Cavity wall, filled cavity" + assert row.roof == "Pitched roof, Access to loft; 270mm insulation at joists" + assert row.floor == "Ground floor, solid, insulation as built" + assert row.heating == ( + "Mains gas, radiators (condensing); programmer, room thermostat and TRVs" + ) + assert row.hot_water == "From main system, mains gas" + assert row.windows == "Fully double glazed" + assert row.lighting == "Low energy lighting in all fixed outlets" + + +def test_structured_fabric_reads_the_main_building_part_over_an_extension( + db_engine: Engine, +) -> None: + # arrange — two building parts on the cert: an extension (solid brick) and + # the main dwelling (cavity). The main part must supply the wall text. + with Session(db_engine) as session: + session.add( + PropertyRow(id=1, portfolio_id=100, landlord_property_id="LP1", uprn=100001) + ) + session.add( + PlanModel(portfolio_id=100, property_id=1, scenario_id=5, is_default=True) + ) + _add_epc(session, property_id=1, epc_id=1, source="lodged") + _add_building_part( + session, + epc_id=1, + identifier="extension_1", + building_part_number=2, + wall_construction=3, # solid brick — the extension + ) + _add_building_part( + session, + epc_id=1, + identifier="main", + building_part_number=1, + wall_construction=4, # cavity — the main dwelling + ) + session.commit() + + # act + with Session(db_engine) as session: + rows = ScenarioExportPostgresRepository(session).rows_for( + portfolio_id=100, scenario_id=5, property_ids=[1] + ) + + # assert — the main (cavity) part wins, not the extension (solid brick). + assert rows[0].walls == "Cavity wall, filled cavity" + + +def test_prose_remains_where_the_structured_fabric_is_absent( + db_engine: Engine, +) -> None: + # arrange — a lodged EPC with gov prose but no building part and no heating + # detail (an older gov cert, not a re-survey). + with Session(db_engine) as session: + session.add( + PropertyRow(id=1, portfolio_id=100, landlord_property_id="LP1", uprn=100001) + ) + session.add( + PlanModel(portfolio_id=100, property_id=1, scenario_id=5, is_default=True) + ) + _add_epc( + session, + property_id=1, + epc_id=1, + source="lodged", + elements={"wall": "Cavity wall, as built", "roof": "Pitched, 250mm"}, + ) + session.commit() + + # act + with Session(db_engine) as session: + rows = ScenarioExportPostgresRepository(session).rows_for( + portfolio_id=100, scenario_id=5, property_ids=[1] + ) + + # assert — with no structured fabric, the gov prose is the fallback. + row = rows[0] + assert row.walls == "Cavity wall, as built" + assert row.roof == "Pitched, 250mm" + + def test_property_type_override_beats_the_lodged_epc(db_engine: Engine) -> None: # arrange — the lodged EPC says Flat, but a landlord override says House # (Effective-EPC precedence: override → lodged, ADR-0065).