Merge pull request #1423 from Hestia-Homes/feat/glazing-secondary-member

Add SECONDARY glazing: GlazingType member, overlay code 5, guard support + 796 backfill

domain/epc/property_overlays/glazing_overlay.py

@@ -24,6 +24,7 @@ _GLAZING_CODES: dict[str, int] = {
     "Double glazing, pre-2002": 3,
     "Triple glazing, pre-2002": 6,
     "Triple glazing, 2002 or later": 9,
+    "Secondary glazing": 5,
 }
 
 

domain/epc/property_overrides/glazing_mix_guard.py

@@ -17,11 +17,12 @@ _PERCENT_TYPE = re.compile(
 # One base type at or above this share is a uniform assertion, not a mix — applied
 # as that type (a near-uniform reglaze), not deferred.
 _UNIFORM_THRESHOLD_PERCENT = 90
-# A dominant base type + stated era → the fully-determined canonical member. Single
-# glazing is era-free, so it needs no era. Double/triple with no (or non-canonical)
-# era is absent here and falls through to the LLM.
+# A dominant base type + stated era → the fully-determined canonical member.
+# Single and secondary glazing are era-free, so they need no era. Double/triple
+# with no (or non-canonical) era is absent here and falls through to the LLM.
 _DOMINANT_MEMBER: dict[tuple[str, Optional[str]], GlazingType] = {
     ("single", None): GlazingType.SINGLE,
+    ("secondary", None): GlazingType.SECONDARY,
     ("double", "2002 or later"): GlazingType.DOUBLE_POST_2002,
     ("double", "pre-2002"): GlazingType.DOUBLE_PRE_2002,
     ("triple", "2002 or later"): GlazingType.TRIPLE_POST_2002,
@@ -51,9 +52,9 @@ def glazing_mix_guard(description: str) -> Optional[GlazingType]:
     * ``MIXED`` for a genuine mix — two or more glazing types present and none
       dominant (≥ the uniform threshold);
     * the fully-determined member for a dominant near-uniform split whose type
-      carries no era ambiguity — single glazing (era-free), or a dominant
-      double/triple whose era is stated ("pre-2002" / "2002 or later"); the LLM
-      would otherwise flatten such a split onto its *minority* type;
+      carries no era ambiguity — single or secondary glazing (era-free), or a
+      dominant double/triple whose era is stated ("pre-2002" / "2002 or later");
+      the LLM would otherwise flatten such a split onto its *minority* type;
     * ``None`` otherwise — an unparseable description, or a dominant double/triple
       with an unstated/non-canonical era ("unknown age") that still carries genuine
       era ambiguity — so the LLM classifier resolves it.

domain/epc/property_overrides/glazing_type.py

@@ -28,4 +28,8 @@ class GlazingType(Enum):
     # which windows are which (ADR-0042 amendment, #1376). The FE-owned pgEnum
     # gains this member via a Drizzle migration.
     MIXED = "Mixed glazing"
+    # Secondary glazing (a pane added inside the original window) is era-free like
+    # SINGLE — SAP10 glazing code 5, U = 2.9 regardless of install year. The
+    # FE-owned pgEnum gains this member via assessment-model#345.
+    SECONDARY = "Secondary glazing"
     UNKNOWN = "Unknown"

scripts/lisasrequest/reclassify_secondary_glazing.py

@@ -0,0 +1,197 @@
+"""Backfill uniform secondary-glazing overrides misfiled as ``Single glazing``.
+
+Hyde-796 landlords assert ``100% secondary glazing (sap 9.94)``; before
+``GlazingType.SECONDARY`` existed the classifier's least-bad option was
+``Single glazing``, which models the windows at U = 4.8 instead of secondary's
+2.9 (SAP10 glazing code 5) — a material distortion (Model#1416).
+
+Fixes BOTH stores in one pass:
+
+* ``property_overrides.override_value`` (TEXT — what the modelling reads), so
+  re-modelling picks the correct code; and
+* the ``landlord_glazing_overrides`` classifier cache, so the stale value can't
+  be re-served to new properties. The cache column is the FE-owned ``glazing``
+  pgEnum — this phase requires the ``Secondary glazing`` label
+  (assessment-model#345 + ``drizzle-kit migrate``) and is skipped with a warning
+  until the label exists.
+
+A *uniform* secondary assertion has exactly one glazing type at 100%, so it is
+fully determined and safe to fix by description match (the live
+``glazing_mix_guard`` deliberately leaves uniform assertions to the LLM, which
+can now target SECONDARY). Dominant secondary *splits* are the guard's job and
+are already covered by ``reclassify_dominant_glazing``.
+
+DRY-RUN BY DEFAULT: prints what it would change and writes nothing. Pass
+``--apply`` to execute inside a transaction; writes audit CSVs of every row
+changed so the change is reversible. Idempotent.
+
+    python -m scripts.lisasrequest.reclassify_secondary_glazing            # dry run
+    python -m scripts.lisasrequest.reclassify_secondary_glazing --apply    # write
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import re
+import sys
+from pathlib import Path
+from typing import Any, Optional
+
+from sqlalchemy import text
+from sqlalchemy.engine import Connection
+
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(_REPO_ROOT))
+
+from domain.epc.property_overrides.glazing_type import GlazingType  # noqa: E402
+from scripts.e2e_common import ENV_PATH, build_engine, load_env  # noqa: E402
+
+# "100% secondary glazing", optionally trailed by "(sap 9.94)" or similar noise.
+# One type, 100% — a fully-determined uniform assertion.
+_UNIFORM_SECONDARY = re.compile(r"^\s*100%\s*secondary\s*glazing\b", re.IGNORECASE)
+
+_SELECT_OVERRIDES = text(
+    """
+    SELECT po.property_id, pr.uprn,
+           po.original_spreadsheet_description AS description,
+           po.override_value AS value
+    FROM property_overrides po
+    JOIN property pr ON pr.id = po.property_id
+    WHERE po.portfolio_id = :portfolio
+      AND po.override_component = 'glazing'
+    """
+)
+_UPDATE_OVERRIDE = text(
+    """
+    UPDATE property_overrides
+       SET override_value = :new_value
+     WHERE portfolio_id = :portfolio
+       AND override_component = 'glazing'
+       AND property_id = :property_id
+       AND override_value <> :new_value
+    """
+)
+_SELECT_CACHE = text(
+    """
+    SELECT id, description, value::text AS value
+    FROM landlord_glazing_overrides
+    WHERE portfolio_id = :portfolio
+    """
+)
+_UPDATE_CACHE = text(
+    """
+    UPDATE landlord_glazing_overrides
+       SET value = CAST(:new_value AS glazing),
+           updated_at = now()
+     WHERE id = :id
+       AND value::text <> :new_value
+    """
+)
+_ENUM_HAS_LABEL = text(
+    """
+    SELECT EXISTS (
+        SELECT 1 FROM pg_enum e
+        JOIN pg_type t ON t.oid = e.enumtypid
+        WHERE t.typname = 'glazing' AND e.enumlabel = :label
+    )
+    """
+)
+
+
+def _target(description: Optional[str]) -> Optional[str]:
+    if description and _UNIFORM_SECONDARY.match(description):
+        return GlazingType.SECONDARY.value
+    return None
+
+
+def _write_audit(name: str, portfolio: int, header: list[str], rows: list[Any]) -> None:
+    out = _REPO_ROOT / "scripts" / "lisasrequest" / f"{name}_{portfolio}_audit.csv"
+    with out.open("w", newline="") as fh:
+        w = csv.writer(fh)
+        w.writerow(header)
+        w.writerows(rows)
+    print(f"audit trail: {out}")
+
+
+def _fix_property_overrides(conn: Connection, portfolio: int, apply: bool) -> int:
+    audit: list[tuple[int, object, str, str, str]] = []
+    for property_id, uprn, description, value in conn.execute(
+        _SELECT_OVERRIDES, {"portfolio": portfolio}
+    ):
+        new_value = _target(description)
+        if new_value is None or value == new_value:
+            continue
+        audit.append((property_id, uprn, description, value, new_value))
+        if apply:
+            conn.execute(
+                _UPDATE_OVERRIDE,
+                {"portfolio": portfolio, "property_id": property_id, "new_value": new_value},
+            )
+    verb = "re-classified" if apply else "would re-classify"
+    print(f"property_overrides: {verb} {len(audit)} row(s)")
+    if apply and audit:
+        _write_audit(
+            "reclassify_secondary_glazing",
+            portfolio,
+            ["property_id", "uprn", "description", "old_value", "new_value"],
+            audit,
+        )
+    return len(audit)
+
+
+def _fix_cache(conn: Connection, portfolio: int, apply: bool) -> int:
+    label = GlazingType.SECONDARY.value
+    if not conn.execute(_ENUM_HAS_LABEL, {"label": label}).scalar():
+        print(
+            f"cache: SKIPPED — the glazing pgEnum has no {label!r} label yet; "
+            "merge assessment-model#345 and run drizzle-kit migrate first, then re-run"
+        )
+        return 0
+    audit: list[tuple[str, str, str, str]] = []
+    for row_id, description, value in conn.execute(_SELECT_CACHE, {"portfolio": portfolio}):
+        new_value = _target(description)
+        if new_value is None or value == new_value:
+            continue
+        audit.append((str(row_id), description, value, new_value))
+        if apply:
+            conn.execute(_UPDATE_CACHE, {"id": row_id, "new_value": new_value})
+    verb = "re-classified" if apply else "would re-classify"
+    print(f"cache: {verb} {len(audit)} row(s)")
+    if apply and audit:
+        _write_audit(
+            "reclassify_secondary_glazing_cache",
+            portfolio,
+            ["id", "description", "old_value", "new_value"],
+            audit,
+        )
+    return len(audit)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--portfolio", type=int, default=796)
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="execute the updates (default: dry-run, writes nothing)",
+    )
+    args = parser.parse_args()
+
+    load_env(ENV_PATH)
+    engine = build_engine()
+
+    with engine.begin() as conn:
+        conn.execute(text("SET statement_timeout = 120000"))
+        changed = _fix_property_overrides(conn, args.portfolio, args.apply)
+        changed += _fix_cache(conn, args.portfolio, args.apply)
+        if not args.apply:
+            conn.rollback()
+
+    if not args.apply:
+        print("\nDRY-RUN — nothing written. Re-run with --apply to execute.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/domain/epc/test_glazing_mix_guard.py

@@ -63,10 +63,31 @@ def test_dominant_era_stated_double_or_triple_resolves_deterministically(
     assert result is expected
 
 
+@pytest.mark.parametrize(
+    "description",
+    [
+        "95% secondary glazing (sap 9.94), 5% single glazing",
+        "90% secondary glazing, 10% double glazing 2002 or later",
+    ],
+)
+def test_dominant_secondary_glazing_resolves_to_secondary(description: str) -> None:
+    # Secondary glazing is era-free like single (SAP10 code 5, U = 2.9 regardless
+    # of install year), so a dominant-secondary split is fully determined — the
+    # guard must claim it rather than let the LLM flatten it onto the minority
+    # type (Model#1416).
+
+    # Act
+    result = glazing_mix_guard(description)
+
+    # Assert
+    assert result is GlazingType.SECONDARY
+
+
 @pytest.mark.parametrize(
     "description",
     [
         "100% double glazing 2002 or later",  # uniform — one type
+        "100% secondary glazing (sap 9.94)",  # uniform — one type
         "90% double glazing unknown age, 10% single glazing",  # dominant but era unstated
         "80% double glazing 2002 or later, 20% double glazing pre-2002",  # same base type
         "some double glazing and a bit of single",  # unparseable — no percentages

tests/domain/epc/test_glazing_overlay.py

@@ -48,6 +48,21 @@ def test_glazing_types_decode_to_their_sap_codes(
     assert simulation.glazing.glazing_type == code
 
 
+def test_secondary_glazing_overlays_its_glazing_code() -> None:
+    # Hyde-796 landlords assert "100% secondary glazing (sap 9.94)"; flattening it
+    # to Single applies U = 4.8 instead of secondary's 2.9 (Model#1416). The
+    # calculator already understands secondary glazing — SAP10 glazing_type code 5
+    # (`_GLAZING_CODE_TO_UWINDOW` in heat_transmission).
+
+    # Act
+    simulation = glazing_overlay_for("Secondary glazing", 0)
+
+    # Assert
+    assert simulation is not None
+    assert simulation.glazing is not None
+    assert simulation.glazing.glazing_type == 5
+
+
 @pytest.mark.parametrize("glazing_value", ["Unknown", ""])
 def test_unresolvable_glazing_produces_no_overlay(glazing_value: str) -> None:
     # Act