diff --git a/domain/epc/property_overlays/roof_type_overlay.py b/domain/epc/property_overlays/roof_type_overlay.py index aadcc3878..a2ba1811b 100644 --- a/domain/epc/property_overlays/roof_type_overlay.py +++ b/domain/epc/property_overlays/roof_type_overlay.py @@ -27,6 +27,8 @@ from datatypes.epc.domain.epc_property_data import BuildingPartIdentifier from domain.modelling.simulation import BuildingPartOverlay, EpcSimulation _LOFT_MM = re.compile(r"(\d+)\+?\s*mm loft insulation") +# A flat roof carries no loft, so its depth reads "N mm insulation" (no "loft"). +_FLAT_MM = re.compile(r"(\d+)\+?\s*mm insulation") def roof_overlay_for( @@ -49,8 +51,18 @@ def _overlay_for(roof_type_value: str) -> Optional[BuildingPartOverlay]: if match is not None: return BuildingPartOverlay(roof_insulation_thickness=int(match.group(1))) if roof_type_value.startswith("Flat,"): - # Flat roof: U-value is the age-band default; flag the shape and let the - # age-band overlay drive `_FLAT_ROOF_BY_AGE` (ADR-0033). + flat_depth = _FLAT_MM.search(roof_type_value) + if flat_depth is not None: + # A flat roof with a known depth: RdSAP 10 Table 16 col (1) ("joists at + # ceiling level and flat roof") scores it by thickness, so carry the + # depth (the calculator routes flat + thickness through Table 16, not + # the age-band default). ADR-0041. + return BuildingPartOverlay( + roof_construction_type="Flat", + roof_insulation_thickness=int(flat_depth.group(1)), + ) + # Flat roof with no stated depth: U-value is the age-band default; flag the + # shape and let the age-band overlay drive `_FLAT_ROOF_BY_AGE` (ADR-0033). return BuildingPartOverlay(roof_construction_type="Flat") if roof_type_value == "Pitched, Unknown loft insulation": # Unknown loft depth: U-value is the pitched age-band default. Assert the diff --git a/domain/epc/property_overrides/roof_type.py b/domain/epc/property_overrides/roof_type.py index 56ef9e8ed..bc3073f86 100644 --- a/domain/epc/property_overrides/roof_type.py +++ b/domain/epc/property_overrides/roof_type.py @@ -22,6 +22,28 @@ class RoofType(Enum): FLAT_NO_INSULATION = "Flat, no insulation" FLAT_NO_INSULATION_ASSUMED = "Flat, no insulation (assumed)" + # A flat roof with a known insulation depth. RdSAP 10 Table 16 col (1) + # ("joists at ceiling level and flat roof") scores flat roofs by thickness, so + # these carry the depth into the overlay (mirroring the PITCHED_LOFT_* ladder; + # no "loft" — a flat roof has none). ADR-0041. The value column is an FE-owned + # pgEnum, so these members are added to the Drizzle enum by the FE owner. + FLAT_12MM = "Flat, 12 mm insulation" + FLAT_25MM = "Flat, 25 mm insulation" + FLAT_50MM = "Flat, 50 mm insulation" + FLAT_75MM = "Flat, 75 mm insulation" + FLAT_100MM = "Flat, 100 mm insulation" + FLAT_125MM = "Flat, 125 mm insulation" + FLAT_150MM = "Flat, 150 mm insulation" + FLAT_175MM = "Flat, 175 mm insulation" + FLAT_200MM = "Flat, 200 mm insulation" + FLAT_225MM = "Flat, 225 mm insulation" + FLAT_250MM = "Flat, 250 mm insulation" + FLAT_270MM = "Flat, 270 mm insulation" + FLAT_300MM = "Flat, 300 mm insulation" + FLAT_350MM = "Flat, 350 mm insulation" + FLAT_400MM = "Flat, 400 mm insulation" + FLAT_400_PLUS = "Flat, 400+ mm insulation" + PITCHED_INSULATED = "Pitched, insulated" PITCHED_INSULATED_ASSUMED = "Pitched, insulated (assumed)" PITCHED_INSULATED_AT_RAFTERS = "Pitched, insulated at rafters" diff --git a/scripts/reclassify_flat_roof_thickness.py b/scripts/reclassify_flat_roof_thickness.py new file mode 100644 index 000000000..487ee88bc --- /dev/null +++ b/scripts/reclassify_flat_roof_thickness.py @@ -0,0 +1,169 @@ +"""One-time re-classification of flat-roof overrides that discarded a known depth. + +#1376 / ADR-0041: a landlord flat-roof override carrying a real insulation depth +(``flat: 150mm``) was classified to ``"Flat, insulated (assumed)"`` — which the +roof Simulation Overlay resolves to the flat **age-band default**, discarding the +depth. RdSAP 10 Table 16 col (1) ("joists at ceiling level and flat roof") scores +flat roofs **by thickness**, so this re-maps those rows onto the new +``"Flat, N mm insulation"`` members (nearest tabulated rung ≤ the stated depth, +per Table 16), which the overlay now carries into the calculator. + +Only rows whose raw description is a flat roof with a numeric depth are touched; +``flat: unknown`` / ``flat: as built`` / ``flat: no insulation`` (no depth) keep +their age-band-default classification. + +Updates the TEXT ``property_overrides.override_value`` (what the modelling reads — +the immediate fix) always. The ``landlord_roof_type_overrides.value`` classifier +cache is a ``roof_type`` **pgEnum**: the new ``FLAT_*MM`` values are FE-owned and +must be added to the Drizzle enum first, so cache writes for a value the live enum +does not yet carry are **deferred** and reported (the Class-A/B pattern). + +DRY-RUN BY DEFAULT: prints the counts it would change and writes nothing. Pass +``--apply`` to execute inside a transaction. Idempotent — only rows whose stored +value differs from the target member are touched. +""" + +from __future__ import annotations + +import argparse +import re +from collections.abc import Iterable + +from sqlalchemy import Connection, text + +from scripts.e2e_common import build_engine, load_env + +# The flat-roof thickness rungs that have a taxonomy member (RdSAP 10 Table 16), +# ascending; a depth above the top maps to the "400+" member. +_FLAT_RUNGS: tuple[int, ...] = ( + 12, 25, 50, 75, 100, 125, 150, 175, 200, 225, 250, 270, 300, 350, 400, +) +_FLAT_DEPTH_RE = re.compile(r"^flat:\s*(\d+)\+?\s*mm") + + +def _flat_member_value(depth_mm: int) -> str: + """The ``"Flat, N mm insulation"`` member value for a stated depth — the + nearest tabulated rung ≤ the depth (Table 16), or the "400+" member above 400.""" + if depth_mm > 400: + return "Flat, 400+ mm insulation" + rung = _FLAT_RUNGS[0] + for candidate in _FLAT_RUNGS: + if depth_mm >= candidate: + rung = candidate + return f"Flat, {rung} mm insulation" + + +def flat_thickness_corrections( + stored: Iterable[tuple[str, str]], +) -> dict[str, str]: + """``(description, stored override_value)`` → the corrected flat-thickness member + value, for descriptions that are a flat roof with a numeric depth whose stored + value is not already that member. Depthless flat descriptions and non-flat + descriptions are omitted, so re-running against corrected data is a no-op.""" + corrections: dict[str, str] = {} + for description, value in stored: + match = _FLAT_DEPTH_RE.match(description.strip().lower()) + if match is None: + continue + target = _flat_member_value(int(match.group(1))) + if value != target: + corrections[description] = target + return corrections + + +_DISTINCT = text( + """ + SELECT DISTINCT lower(original_spreadsheet_description) AS description, + override_value AS value + FROM property_overrides + WHERE override_component = 'roof_type' + """ +) +_OVERRIDES_UPDATE = text( + """ + UPDATE property_overrides + SET override_value = :new_value + WHERE override_component = 'roof_type' + AND lower(original_spreadsheet_description) = :description + AND override_value <> :new_value + """ +) +_OVERRIDES_COUNT = text( + """ + SELECT count(*) FROM property_overrides + WHERE override_component = 'roof_type' + AND lower(original_spreadsheet_description) = :description + AND override_value <> :new_value + """ +) +_CACHE_UPDATE = text( + """ + UPDATE landlord_roof_type_overrides + SET value = :new_value, updated_at = now() + WHERE lower(description) = :description + AND value::text <> :new_value + """ +) +_ENUM_VALUES = text( + "SELECT e.enumlabel FROM pg_enum e JOIN pg_type t ON t.oid = e.enumtypid " + "WHERE t.typname = 'roof_type'" +) + + +def reclassify(conn: Connection, *, apply: bool) -> tuple[int, set[str]]: + """Re-map flat-depth roof overrides onto their thickness member. Returns the + number of ``property_overrides`` rows found and the set of target values the + live ``roof_type`` pgEnum does not yet carry (cache-deferred until the FE + migration).""" + stored = [(r.description, r.value) for r in conn.execute(_DISTINCT)] + enum_values = {r[0] for r in conn.execute(_ENUM_VALUES)} + total = 0 + deferred: set[str] = set() + for description, new_value in flat_thickness_corrections(stored).items(): + params = {"description": description, "new_value": new_value} + total += conn.execute(_OVERRIDES_COUNT, params).scalar() or 0 + in_enum = new_value in enum_values + if not in_enum: + deferred.add(new_value) + if apply: + conn.execute(_OVERRIDES_UPDATE, params) + if in_enum: + conn.execute(_CACHE_UPDATE, params) + return total, deferred + + +def main() -> None: + load_env() + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--apply", + action="store_true", + help="execute the updates (default: dry-run, writes nothing)", + ) + args = parser.parse_args() + + engine = build_engine() + with engine.begin() as conn: + conn.execute(text("SET statement_timeout = 120000")) + total, deferred = reclassify(conn, apply=args.apply) + + verb = "re-classified" if args.apply else "would re-classify" + print( + f"{verb} {total} flat-roof override row(s) from the age-band default onto " + "their Table 16 thickness member (property_overrides / TEXT)." + ) + if deferred: + print( + f"\n{len(deferred)} target value(s) NOT yet in the roof_type pgEnum — " + "their classifier-cache rows are deferred until the FE-repo enum " + "migration adds these members (property_overrides was still updated, " + "which is what the modelling reads):" + ) + for value in sorted(deferred): + print(f" {value!r}") + if not args.apply: + print("\nDRY-RUN — nothing written. Re-run with --apply to execute.") + + +if __name__ == "__main__": + main() diff --git a/tests/domain/epc/test_roof_type_overlay.py b/tests/domain/epc/test_roof_type_overlay.py index 050d76fa2..1c39ec9f7 100644 --- a/tests/domain/epc/test_roof_type_overlay.py +++ b/tests/domain/epc/test_roof_type_overlay.py @@ -11,6 +11,26 @@ import pytest from datatypes.epc.domain.epc_property_data import BuildingPartIdentifier from domain.epc.property_overlays.roof_type_overlay import roof_overlay_for +from domain.epc.property_overrides.roof_type import RoofType + +_FLAT_THICKNESS_MEMBERS: list[tuple[RoofType, int]] = [ + (RoofType.FLAT_12MM, 12), + (RoofType.FLAT_25MM, 25), + (RoofType.FLAT_50MM, 50), + (RoofType.FLAT_75MM, 75), + (RoofType.FLAT_100MM, 100), + (RoofType.FLAT_125MM, 125), + (RoofType.FLAT_150MM, 150), + (RoofType.FLAT_175MM, 175), + (RoofType.FLAT_200MM, 200), + (RoofType.FLAT_225MM, 225), + (RoofType.FLAT_250MM, 250), + (RoofType.FLAT_270MM, 270), + (RoofType.FLAT_300MM, 300), + (RoofType.FLAT_350MM, 350), + (RoofType.FLAT_400MM, 400), + (RoofType.FLAT_400_PLUS, 400), +] def test_pitched_loft_depth_maps_to_roof_insulation_thickness() -> None: @@ -42,6 +62,38 @@ def test_each_loft_depth_is_parsed(roof_type_value: str, expected_mm: int) -> No ].roof_insulation_thickness == expected_mm +def test_flat_roof_depth_maps_to_roof_insulation_thickness() -> None: + # Act — a flat roof with a known depth. RdSAP 10 Table 16 col (1) scores flat + # roofs by insulation thickness ("joists at ceiling level and flat roof"), so + # the depth must reach the overlay, not be discarded for the age-band default + # (ADR-0041). + simulation = roof_overlay_for("Flat, 150 mm insulation", 0) + + # Assert — the flat shape and the real depth both ride on the overlay. + assert simulation is not None + overlay = simulation.building_parts[BuildingPartIdentifier.MAIN] + assert overlay.roof_construction_type == "Flat" + assert overlay.roof_insulation_thickness == 150 + + +@pytest.mark.parametrize(("member", "expected_mm"), _FLAT_THICKNESS_MEMBERS) +def test_every_flat_thickness_member_resolves_to_its_depth( + member: RoofType, expected_mm: int +) -> None: + # Each FLAT_*MM taxonomy member must stay lock-step with the overlay: its value + # resolves to the depth it names (so a landlord-stored member scores via Table + # 16 col (1), not the age-band default). ADR-0041. + + # Act + simulation = roof_overlay_for(member.value, 0) + + # Assert + assert simulation is not None + overlay = simulation.building_parts[BuildingPartIdentifier.MAIN] + assert overlay.roof_construction_type == "Flat" + assert overlay.roof_insulation_thickness == expected_mm + + @pytest.mark.parametrize( "roof_type_value", [ diff --git a/tests/scripts/test_reclassify_flat_roof_thickness.py b/tests/scripts/test_reclassify_flat_roof_thickness.py new file mode 100644 index 000000000..5a0ae87c4 --- /dev/null +++ b/tests/scripts/test_reclassify_flat_roof_thickness.py @@ -0,0 +1,44 @@ +"""The flat-roof reclassify maps a stated depth onto its Table 16 thickness member, +leaves depthless / non-flat rows alone, and is idempotent (#1376 / ADR-0041).""" + +from __future__ import annotations + +from scripts.reclassify_flat_roof_thickness import flat_thickness_corrections + + +def test_flat_depth_rows_are_corrected_to_the_thickness_member() -> None: + # Arrange — flat roofs with a real depth (currently on the age-band-default + # value), alongside depthless flat rows and a pitched row that must be left. + stored = [ + ("flat: 150mm", "Flat, insulated (assumed)"), + ("flat: 50mm", "Flat, insulated (assumed)"), + ("flat: unknown", "Flat, insulated (assumed)"), + ("flat: as built", "Flat, insulated (assumed)"), + ("flat: no insulation", "Flat, no insulation"), + ("pitchednormalloftaccess: 100mm", "Pitched, 100 mm loft insulation"), + ] + + # Act + corrections = flat_thickness_corrections(stored) + + # Assert — only the flat-depth rows are re-mapped, to their thickness member. + assert corrections == { + "flat: 150mm": "Flat, 150 mm insulation", + "flat: 50mm": "Flat, 50 mm insulation", + } + + +def test_a_non_rung_depth_snaps_to_the_nearest_tabulated_rung() -> None: + # RdSAP Table 16 uses the nearest tabulated thickness <= the stated depth. + + # Act / Assert + assert flat_thickness_corrections( + [("flat: 60mm", "Flat, insulated (assumed)")] + ) == {"flat: 60mm": "Flat, 50 mm insulation"} + + +def test_already_corrected_flat_rows_need_no_change() -> None: + # Act / Assert — idempotent. + assert ( + flat_thickness_corrections([("flat: 150mm", "Flat, 150 mm insulation")]) == {} + )