Merge pull request #1379 from Hestia-Homes/fix/1376-flat-roof-insulation-thickness

Score flat roofs by their known insulation depth, not the age-band default (#1376)

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

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"

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()

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",
     [

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")]) == {}
+    )