junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-30 13:10:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Type-clean the override_component consistency guard 🟪

Browse source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jun-te Kim 2026-06-19 18:19:07 +00:00
parent f7ee47118b
commit e03fd27357
3 changed files with 533 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

437
scripts/hyde/build_property_overrides.py Normal file
View file

@ -0,0 +1,437 @@
"""Build ``property_overrides`` for a portfolio from the Hyde Excel, bypassing the
frontend + lambdas, using the ``landlord_*_overrides`` tables as the durable
classification ledger.
Why the ledger (not a throwaway cache): ``landlord_*_overrides`` stores
``(portfolio_id, description) -> value`` with a ``source`` (classifier|user).
* Re-runs classify only descriptions NOT already stored -> saves ChatGPT calls.
* Human corrections are stored as ``source=user`` and the classifier is
forbidden from overwriting them (ADR-0003) -> edits are permanent.
Then we resolve the vocab + match each row to a ``property.id`` by **org_ref**
(Excel "Organisation Reference" -> property.landlord_property_id) and upsert
``property_overrides`` (the fact layer the SAP overlay reads).
Subcommands:
list-values print each component's valid override values (reference)
classify --excel f --portfolio-id 795
PASS 1: classify cache-misses via ChatGPT,
upsert to landlord tables, write
overrides_unknowns.csv (with allowed_values)
validate --edits overrides_edits.csv
check a hand-edited file: every corrected_value
must be a valid enum value (suggests fixes)
apply-edits --edits overrides_edits.csv --portfolio-id 795 [--apply]
upsert validated corrections as source=user
write --excel f --portfolio-id 795 [--apply]
PASS 2: build + upsert property_overrides from vocab
Env: POSTGRES_* (PostgresConfig.from_env) and OPENAI_API_KEY (ChatGPT).
"""
from __future__ import annotations
import argparse
import csv
import difflib
import logging
import os
from collections import Counter
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Optional
import pandas as pd
from sqlalchemy import Table, text
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlmodel import SQLModel
from domain.epc.property_overrides.built_form_type import BuiltFormType
from domain.epc.property_overrides.construction_age_band import ConstructionAgeBand
from domain.epc.property_overrides.glazing_type import GlazingType
from domain.epc.property_overrides.main_fuel_type import MainFuelType
from domain.epc.property_overrides.main_heating_system_type import MainHeatingSystemType
from domain.epc.property_overrides.property_type import PropertyType
from domain.epc.property_overrides.roof_type import RoofType
from domain.epc.property_overrides.wall_type import WallType
from domain.epc.property_overrides.wall_type_construction_dates import (
wall_type_construction_date_prompt_hint,
)
from domain.epc.property_overrides.water_heating_type import WaterHeatingType
from infrastructure.chatgpt.chatgpt import ChatGPT
from infrastructure.chatgpt.chatgpt_column_classifier import ChatGptColumnClassifier
from infrastructure.landlord_overrides.landlord_override_reader_postgres_repository import (
LandlordOverrideReaderPostgresRepository,
)
from infrastructure.landlord_overrides.landlord_overrides_postgres_repository import (
LandlordOverridesRepository,
)
from infrastructure.postgres.config import PostgresConfig
from infrastructure.postgres.engine import commit_scope, make_engine, make_session
from infrastructure.postgres.landlord_built_form_type_override_table import (
LandlordBuiltFormTypeOverrideRow,
)
from infrastructure.postgres.landlord_construction_age_band_override_table import (
LandlordConstructionAgeBandOverrideRow,
)
from infrastructure.postgres.landlord_glazing_override_table import (
LandlordGlazingOverrideRow,
)
from infrastructure.postgres.landlord_main_fuel_override_table import (
LandlordMainFuelOverrideRow,
)
from infrastructure.postgres.landlord_main_heating_system_override_table import (
LandlordMainHeatingSystemOverrideRow,
)
from infrastructure.postgres.landlord_override_enums import OverrideSource
from infrastructure.postgres.landlord_property_type_override_table import (
LandlordPropertyTypeOverrideRow,
)
from infrastructure.postgres.landlord_roof_type_override_table import (
LandlordRoofTypeOverrideRow,
)
from infrastructure.postgres.landlord_wall_type_override_table import (
LandlordWallTypeOverrideRow,
)
from infrastructure.postgres.landlord_water_heating_override_table import (
LandlordWaterHeatingOverrideRow,
)
from repositories.property.landlord_override_overlays import overlays_from
from repositories.property.property_override_postgres_repository import (
PropertyOverridePostgresRepository,
)
from repositories.property.property_override_repository import PropertyOverrideInsert
from repositories.property.property_overrides_postgres_reader import (
PropertyOverridesPostgresReader,
)
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("build_property_overrides")
ORG_REF_COLUMN = "Organisation Reference"
UNKNOWNS_PATH = "overrides_unknowns.csv"
@dataclass(frozen=True)
class ComponentSpec:
component: str
enum_cls: type[Enum]
unknown: Enum
row_type: type[SQLModel]
excel_header: str
per_building_part: bool # comma = building parts (wall/roof/age) vs whole-dwelling
extra_instructions: Optional[str] = None
def allowed_values(self) -> list[str]:
"""Valid override values a human may pick (excludes UNKNOWN)."""
return sorted(m.value for m in self.enum_cls if m is not self.unknown)
def _component_specs() -> list[ComponentSpec]:
return [
ComponentSpec("property_type", PropertyType, PropertyType.UNKNOWN, LandlordPropertyTypeOverrideRow, "Property Type", False),
ComponentSpec("built_form_type", BuiltFormType, BuiltFormType.UNKNOWN, LandlordBuiltFormTypeOverrideRow, "Property Type", False),
ComponentSpec("wall_type", WallType, WallType.UNKNOWN, LandlordWallTypeOverrideRow, "Walls", True, wall_type_construction_date_prompt_hint()),
ComponentSpec("roof_type", RoofType, RoofType.UNKNOWN, LandlordRoofTypeOverrideRow, "Roofs", True),
ComponentSpec("construction_age_band", ConstructionAgeBand, ConstructionAgeBand.UNKNOWN, LandlordConstructionAgeBandOverrideRow, "Age", True),
ComponentSpec("main_fuel", MainFuelType, MainFuelType.UNKNOWN, LandlordMainFuelOverrideRow, "Main Fuel", False),
ComponentSpec("glazing", GlazingType, GlazingType.UNKNOWN, LandlordGlazingOverrideRow, "Glazing", False),
ComponentSpec("water_heating", WaterHeatingType, WaterHeatingType.UNKNOWN, LandlordWaterHeatingOverrideRow, "Hot Water", False),
ComponentSpec("main_heating_system", MainHeatingSystemType, MainHeatingSystemType.UNKNOWN, LandlordMainHeatingSystemOverrideRow, "Heating", False),
]
def _specs_by_component() -> dict[str, ComponentSpec]:
return {s.component: s for s in _component_specs()}
def _norm(s: Any) -> str:
"""Vocab key normalisation — mirrors the orchestrator (strip + lower)."""
return str(s or "").strip().lower()
def _split_entries(cell: Any, per_building_part: bool) -> list[str]:
raw = "" if cell is None else str(cell)
if not raw.strip():
return []
if not per_building_part:
return [raw.strip()]
return [part.strip() for part in raw.split(",") if part.strip()]
def _load_rows(excel: str, sheet: str) -> list[dict[str, Any]]:
return pd.read_excel(excel, sheet_name=sheet).to_dict(orient="records") # type: ignore[return-value]
def _filter_rows(rows: list[dict[str, Any]], org_ref: Optional[str],
limit: Optional[int]) -> list[dict[str, Any]]:
"""Narrow to one property (--org-ref) or the first N rows (--limit) for a
cheap smoke test before the full run."""
if org_ref:
rows = [r for r in rows if str(r.get(ORG_REF_COLUMN, "")).strip() == org_ref]
if limit:
rows = rows[:limit]
return rows
def _distinct_entries(rows: list[dict[str, Any]], spec: ComponentSpec) -> Counter[str]:
counts: Counter[str] = Counter()
for row in rows:
for entry in _split_entries(row.get(spec.excel_header), spec.per_building_part):
counts[entry] += 1
return counts
# --------------------------------------------------------------------------- #
def list_values(_: argparse.Namespace) -> None:
"""Print the valid override values per component (the reference for edits)."""
for spec in _component_specs():
print(f"\n## {spec.component} (Excel: {spec.excel_header})")
for v in spec.allowed_values():
print(f" {v}")
def validate(args: argparse.Namespace) -> None:
"""Check a hand-edited CSV: every corrected_value must be a valid enum value."""
specs = _specs_by_component()
bad = 0
with open(args.edits, newline="") as f:
for i, r in enumerate(csv.DictReader(f), start=2):
val = (r.get("corrected_value") or "").strip()
if not val:
continue
comp = (r.get("component") or "").strip()
spec = specs.get(comp)
if spec is None:
logger.error("row %d: unknown component %r", i, comp)
bad += 1
continue
if val not in spec.allowed_values():
hint = difflib.get_close_matches(val, spec.allowed_values(), n=2)
logger.error("row %d [%s]: %r is not a valid value.%s",
i, comp, val,
f" Did you mean: {hint}?" if hint else
" Run 'list-values' for the allowed set.")
bad += 1
if bad:
raise SystemExit(f"{bad} invalid corrected_value(s) — fix them before apply-edits.")
logger.info("All corrected values are valid enum values. ✓")
def _db_session() -> Any:
return make_session(make_engine(PostgresConfig.from_env(os.environ)))
def classify(args: argparse.Namespace) -> None:
rows = _filter_rows(_load_rows(args.excel, args.sheet), args.org_ref, args.limit)
logger.info("Classifying over %d row(s).", len(rows))
chat_gpt = ChatGPT()
session = _db_session()
reader = LandlordOverrideReaderPostgresRepository(session)
try:
vocab = reader.load_for_portfolio(args.portfolio_id) # {component: {desc: value}}
unknown_rows: list[tuple[str, str, int, str]] = []
for spec in _component_specs():
counts = _distinct_entries(rows, spec)
known = vocab.get(spec.component, {}) # already-classified (cache)
to_classify = {d for d in counts if _norm(d) not in known}
logger.info("%-22s %4d distinct | %4d cached | %4d to classify",
spec.component, len(counts), len(counts) - len(to_classify), len(to_classify))
resolved: dict[str, Enum] = {}
if to_classify:
classifier: ChatGptColumnClassifier[Any] = ChatGptColumnClassifier(
chat_gpt, spec.enum_cls, spec.unknown, extra_instructions=spec.extra_instructions)
resolved = classifier.classify(to_classify)
repo: LandlordOverridesRepository[Any] = LandlordOverridesRepository(session, spec.row_type)
with commit_scope(session):
# store keyed on the normalised description (matches the reader/finaliser lookup)
repo.upsert_all(args.portfolio_id, {_norm(d): m for d, m in resolved.items()})
# collect UNKNOWNs (freshly classified + anything cached as UNKNOWN) for review
unk = spec.unknown.value
for desc, n in counts.items():
v = resolved.get(desc).value if desc in resolved and resolved[desc] else known.get(_norm(desc)) # type: ignore[union-attr]
if v is None or v == unk:
allowed = " | ".join(spec.allowed_values())
unknown_rows.append((spec.component, desc, n, allowed))
with open(UNKNOWNS_PATH, "w", newline="") as f:
w = csv.writer(f)
w.writerow(["component", "description", "count", "corrected_value", "allowed_values"])
for comp, desc, n, allowed in sorted(unknown_rows, key=lambda r: (-r[2])):
w.writerow([comp, desc, n, "", allowed])
logger.info("\nWrote %s — fill 'corrected_value' (must match 'allowed_values'), "
"then: validate -> apply-edits -> write.", UNKNOWNS_PATH)
finally:
session.close()
def _upsert_user_corrections(session: Any, portfolio_id: int,
by_component: dict[str, dict[str, str]]) -> int:
"""Upsert validated human corrections as source=user (always wins on conflict)."""
specs = _specs_by_component()
n = 0
now = datetime.now(timezone.utc)
for comp, mapping in by_component.items():
spec = specs[comp]
table: Table = getattr(spec.row_type, "__table__")
rows = [{"portfolio_id": portfolio_id, "description": _norm(d), "value": v,
"source": OverrideSource.USER, "created_at": now, "updated_at": now}
for d, v in mapping.items()]
if not rows:
continue
stmt = pg_insert(table).values(rows)
stmt = stmt.on_conflict_do_update(
index_elements=["portfolio_id", "description"],
set_={"value": stmt.excluded.value, "source": stmt.excluded.source,
"updated_at": stmt.excluded.updated_at})
session.execute(stmt)
n += len(rows)
return n
def apply_edits(args: argparse.Namespace) -> None:
validate(args) # fail before touching the DB
specs = _specs_by_component()
by_component: dict[str, dict[str, str]] = {}
with open(args.edits, newline="") as f:
for r in csv.DictReader(f):
val = (r.get("corrected_value") or "").strip()
if val and r["component"] in specs:
by_component.setdefault(r["component"], {})[r["description"]] = val
session = _db_session()
try:
if not args.apply:
total = sum(len(m) for m in by_component.values())
logger.info("DRY RUN — %d user corrections ready. Re-run with --apply.", total)
return
with commit_scope(session):
n = _upsert_user_corrections(session, args.portfolio_id, by_component)
logger.info("Upserted %d user corrections (source=user).", n)
finally:
session.close()
def _org_ref_to_property_id(session: Any, portfolio_id: int) -> dict[str, int]:
stmt = text("SELECT landlord_property_id, id FROM property "
"WHERE portfolio_id = :pid AND landlord_property_id IS NOT NULL")
return {str(ref).strip(): int(pid) for ref, pid in session.execute(stmt, {"pid": portfolio_id})}
def write(args: argparse.Namespace) -> None:
rows = _filter_rows(_load_rows(args.excel, args.sheet), args.org_ref, args.limit)
logger.info("Writing over %d row(s).", len(rows))
session = _db_session()
reader = LandlordOverrideReaderPostgresRepository(session)
try:
vocab = reader.load_for_portfolio(args.portfolio_id)
org_ref_map = _org_ref_to_property_id(session, args.portfolio_id)
logger.info("Portfolio %d: %d properties with org_ref.", args.portfolio_id, len(org_ref_map))
inserts: list[PropertyOverrideInsert] = []
unmatched: Counter[str] = Counter()
unresolved: Counter[str] = Counter()
for row in rows:
org_ref = str(row.get(ORG_REF_COLUMN, "")).strip()
property_id = org_ref_map.get(org_ref)
if property_id is None:
unmatched[org_ref] += 1
continue
for spec in _component_specs():
comp_vocab = vocab.get(spec.component, {})
for building_part, entry in enumerate(
_split_entries(row.get(spec.excel_header), spec.per_building_part)):
value = comp_vocab.get(_norm(entry))
if not value or value == spec.unknown.value:
unresolved[f"{spec.component}: {entry}"] += 1
continue
inserts.append(PropertyOverrideInsert(
property_id=property_id, portfolio_id=args.portfolio_id,
building_part=building_part, override_component=spec.component,
override_value=value, original_spreadsheet_description=entry))
logger.info("Built %d rows | %d unmatched org_refs | %d unresolved",
len(inserts), sum(unmatched.values()), sum(unresolved.values()))
if unresolved:
logger.info("Top unresolved (need apply-edits): %s", unresolved.most_common(10))
if not args.apply:
logger.info("DRY RUN — not writing. Re-run with --apply.")
for ins in inserts[:10]:
logger.info(" %s", ins)
return
with commit_scope(session):
affected = PropertyOverridePostgresRepository(session).upsert_all(inserts)
logger.info("Upserted %d property_overrides.", affected)
finally:
session.close()
def verify(args: argparse.Namespace) -> None:
"""For one property (by org_ref): show the persisted property_overrides rows
and the EpcSimulation overlays they produce — the end-to-end proof that the
chain reaches the SAP overlay surface."""
session = _db_session()
try:
org_ref_map = _org_ref_to_property_id(session, args.portfolio_id)
property_id = org_ref_map.get(args.org_ref)
if property_id is None:
raise SystemExit(f"org_ref {args.org_ref!r} not found in portfolio {args.portfolio_id}.")
reader = PropertyOverridesPostgresReader(lambda: session)
resolved = reader.overrides_for(property_id)
logger.info("property_id %d — %d property_overrides rows:", property_id, len(resolved.rows))
for r in resolved.rows:
logger.info(" part %d | %-22s = %s", r.building_part, r.override_component, r.override_value)
overlays = overlays_from(resolved)
logger.info("\n-> %d EpcSimulation overlay(s) produced (what the SAP calc applies):", len(overlays))
for o in overlays:
logger.info(" %s", o)
finally:
session.close()
def main() -> None:
p = argparse.ArgumentParser(description=__doc__)
sub = p.add_subparsers(dest="cmd", required=True)
sub.add_parser("list-values").set_defaults(func=list_values)
v = sub.add_parser("validate")
v.add_argument("--edits", required=True)
v.set_defaults(func=validate)
c = sub.add_parser("classify")
c.add_argument("--excel", required=True)
c.add_argument("--sheet", default="AddressProfilingResults")
c.add_argument("--portfolio-id", type=int, required=True)
c.add_argument("--org-ref", default=None, help="smoke test: only this property's org_ref")
c.add_argument("--limit", type=int, default=None, help="smoke test: first N rows")
c.set_defaults(func=classify)
a = sub.add_parser("apply-edits")
a.add_argument("--edits", required=True)
a.add_argument("--portfolio-id", type=int, required=True)
a.add_argument("--apply", action="store_true")
a.set_defaults(func=apply_edits)
w = sub.add_parser("write")
w.add_argument("--excel", required=True)
w.add_argument("--sheet", default="AddressProfilingResults")
w.add_argument("--portfolio-id", type=int, required=True)
w.add_argument("--org-ref", default=None, help="smoke test: only this property's org_ref")
w.add_argument("--limit", type=int, default=None, help="smoke test: first N rows")
w.add_argument("--apply", action="store_true")
w.set_defaults(func=write)
vf = sub.add_parser("verify")
vf.add_argument("--portfolio-id", type=int, required=True)
vf.add_argument("--org-ref", required=True)
vf.set_defaults(func=verify)
args = p.parse_args()
args.func(args)
if __name__ == "__main__":
main()

13
tests/repositories/property/test_override_component_consistency.py
View file

@ -11,12 +11,14 @@ main_fuel / glazing / construction_age_band reader entries).
from __future__ import annotations
from infrastructure.landlord_overrides.landlord_override_reader_postgres_repository import ( # pyright: ignore[reportPrivateUsage]
_ROW_TYPES,
from typing import cast
from infrastructure.landlord_overrides.landlord_override_reader_postgres_repository import (
_ROW_TYPES, # pyright: ignore[reportPrivateUsage]
)
from infrastructure.postgres.property_override_table import override_component_sa_enum
from repositories.property.landlord_override_overlays import ( # pyright: ignore[reportPrivateUsage]
_COMPONENT_OVERLAYS,
from repositories.property.landlord_override_overlays import (
_COMPONENT_OVERLAYS, # pyright: ignore[reportPrivateUsage]
)
@ -29,4 +31,5 @@ def test_override_component_pgenum_covers_every_component() -> None:
# The property_overrides.override_component pgEnum mirror must list every
# component, or writing/reading a new-component row through it throws a
# LookupError against Postgres (caught live on the Hyde portfolio-796 run).
assert set(override_component_sa_enum.enums) == set(_COMPONENT_OVERLAYS)
pgenum_values = cast(list[str], override_component_sa_enum.enums)
assert set(pgenum_values) == set(_COMPONENT_OVERLAYS)

88
tests/scripts/test_build_property_overrides_smoke.py Normal file
View file

@ -0,0 +1,88 @@
"""End-to-end smoke of the Hyde override script for ONE property, against a real
(ephemeral) Postgres. Seeds the landlord vocab (simulating post-classify, so no
ChatGPT) + a minimal ``property`` row, then runs the script's real
``write`` + ``verify`` paths and asserts property_overrides + overlays land.
"""
from __future__ import annotations
import argparse
from typing import Any
from sqlalchemy import Engine, text
from sqlmodel import Session
import scripts.hyde.build_property_overrides as b
from domain.epc.property_overrides.built_form_type import BuiltFormType
from domain.epc.property_overrides.construction_age_band import ConstructionAgeBand
from domain.epc.property_overrides.glazing_type import GlazingType
from domain.epc.property_overrides.main_fuel_type import MainFuelType
from domain.epc.property_overrides.main_heating_system_type import MainHeatingSystemType
from domain.epc.property_overrides.property_type import PropertyType
from domain.epc.property_overrides.roof_type import RoofType
from domain.epc.property_overrides.wall_type import WallType
from domain.epc.property_overrides.water_heating_type import WaterHeatingType
from infrastructure.landlord_overrides.landlord_overrides_postgres_repository import (
LandlordOverridesRepository,
)
PORTFOLIO = 795
ORG_REF = "55180004001"
EXCEL = "scripts/hyde/hyde_property_overrides.xlsx"
# What ChatGPT WOULD resolve this property's 9 descriptions to (component ->
# (raw Excel entry, enum member)). Seeded into the landlord ledger.
SEED = {
"property_type": ("House: MidTerrace", PropertyType.HOUSE),
"built_form_type": ("House: MidTerrace", BuiltFormType.MID_TERRACE),
"wall_type": ("TimberFrame: AsBuilt", WallType.TIMBER_FRAME_AS_BUILT_NO_INSULATION_ASSUMED),
"roof_type": ("PitchedNormalLoftAccess: 300mm", RoofType.PITCHED_300_MM),
"construction_age_band": ("L: 2012-2022", ConstructionAgeBand.L_2012_2022),
"main_fuel": ("Gas: Mains Gas", MainFuelType.MAINS_GAS),
"glazing": ("100% Double glazing 2002 or later", GlazingType.DOUBLE_POST_2002),
"water_heating": ("From main heating system: Mains Gas", WaterHeatingType.FROM_MAIN_MAINS_GAS),
"main_heating_system": ("Boiler: C rated Combi", MainHeatingSystemType.GAS_COMBI),
}
def test_one_property_end_to_end(db_engine: Engine, monkeypatch: Any, capsys: Any) -> None:
specs = b._specs_by_component()
# minimal FE-owned `property` table + the one row we'll match by org_ref
with Session(db_engine) as s:
s.execute(text(
"CREATE TABLE property (id bigint PRIMARY KEY, portfolio_id bigint, "
"landlord_property_id text)"))
s.execute(text("INSERT INTO property VALUES (1, :p, :ref)"),
{"p": PORTFOLIO, "ref": ORG_REF})
# seed the classifier ledger (keyed on normalised description)
for comp, (raw, member) in SEED.items():
LandlordOverridesRepository(s, specs[comp].row_type).upsert_all(
PORTFOLIO, {b._norm(raw): member})
s.commit()
# point the script at the ephemeral engine
monkeypatch.setattr(b, "_db_session", lambda: Session(db_engine))
# --- run the real write() for this one property ---
b.write(argparse.Namespace(excel=EXCEL, sheet="AddressProfilingResults",
portfolio_id=PORTFOLIO, org_ref=ORG_REF, limit=None, apply=True))
with Session(db_engine) as s:
rows = list(s.execute(text(
"SELECT override_component, building_part, override_value "
"FROM property_overrides WHERE property_id = 1 ORDER BY override_component")))
got = {c: v for c, _, v in rows}
# every seeded component produced a property_overrides row with the resolved value
assert got["main_fuel"] == "mains gas"
assert got["glazing"] == "Double glazing, 2002 or later"
assert got["construction_age_band"] == "L"
assert got["main_heating_system"] == "Gas boiler, combi"
assert got["water_heating"] == "From main system, mains gas"
assert len(rows) == 9 # all 9 components
# --- run the real verify() and confirm overlays are produced ---
b.verify(argparse.Namespace(portfolio_id=PORTFOLIO, org_ref=ORG_REF))
out = capsys.readouterr().out
print(out) # surfaced with -s so we can eyeball the one-property result
assert "EpcSimulation overlay" in out