Model/scripts/reclassify_flat_roof_thickness.py

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