diff --git a/domain/epc/property_overlays/glazing_overlay.py b/domain/epc/property_overlays/glazing_overlay.py index d692181d8..81733461c 100644 --- a/domain/epc/property_overlays/glazing_overlay.py +++ b/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, } diff --git a/domain/epc/property_overrides/glazing_mix_guard.py b/domain/epc/property_overrides/glazing_mix_guard.py index 5d59e4fa1..05b0fb2b0 100644 --- a/domain/epc/property_overrides/glazing_mix_guard.py +++ b/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. diff --git a/domain/epc/property_overrides/glazing_type.py b/domain/epc/property_overrides/glazing_type.py index 8724f5b5e..4a4ef96b3 100644 --- a/domain/epc/property_overrides/glazing_type.py +++ b/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" diff --git a/scripts/lisasrequest/reclassify_secondary_glazing.py b/scripts/lisasrequest/reclassify_secondary_glazing.py new file mode 100644 index 000000000..984d75c86 --- /dev/null +++ b/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()) diff --git a/tests/domain/epc/test_glazing_mix_guard.py b/tests/domain/epc/test_glazing_mix_guard.py index de1faccf0..8d6e349cd 100644 --- a/tests/domain/epc/test_glazing_mix_guard.py +++ b/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 diff --git a/tests/domain/epc/test_glazing_overlay.py b/tests/domain/epc/test_glazing_overlay.py index 81c35e6ca..1c58fe590 100644 --- a/tests/domain/epc/test_glazing_overlay.py +++ b/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