mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-27 23:35:01 +00:00
Source export fabric from structured SAP fields; fix stale-cert selection
The "latest EPC per UPRN" selection ordered by e.id DESC, but row-id does not track recency: a gov-EPC cert re-ingested after a PasHub survey lands with a higher e.id, so the header/perf reads silently picked the stale gov cert (wrong TFA, lodgement, property_type, current band) on 3 of portfolio 838. Order by the effective lodgement date (registration -> completion -> inspection) DESC NULLS LAST, e.id DESC tiebreak. Compose the descriptive fabric from the structured SAP fields a re-survey lodges (no gov-EPC prose): walls/roof/floor from the main epc_building_part, heating and hot water from epc_main_heating_detail + the water-heating codes, decoded with the SAP engine's own code space (fabric_description.py). Structured fabric wins where present and falls back to the gov-EPC prose otherwise; windows and lighting stay on prose (their per-element codes do not reconstruct the dwelling phrase). Add built_form (epc_property, code-decoded) and property_age_band (building part construction_age_band -> RdSAP date range) columns. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar
parent
a820274d4f
commit
35a0c6e972
5 changed files with 790 additions and 8 deletions
284
domain/scenario_export/fabric_description.py
Normal file
284
domain/scenario_export/fabric_description.py
Normal file
|
|
@ -0,0 +1,284 @@
|
|||
"""Human-readable fabric descriptions from an EPC's structured SAP fields
|
||||
(ADR-0065).
|
||||
|
||||
The PasHub re-survey lodges *structured* fabric — SAP building parts, a main
|
||||
heating detail, coded fuel/water-heating — but no gov-EPC descriptive prose, so
|
||||
the export's descriptive columns would be blank for a re-surveyed Property. This
|
||||
module composes the client-sheet text for walls / roof / floor (from the main
|
||||
``epc_building_part``) and heating / hot water (from ``epc_main_heating_detail``
|
||||
and the ``epc_property`` water-heating fields), decoding the SAP integer codes
|
||||
with the same code space the SAP engine uses. Windows and lighting are left on
|
||||
the gov-EPC prose fallback (their per-element codes do not cleanly reconstruct
|
||||
the dwelling-level phrase).
|
||||
|
||||
Pure logic: the repository reads the structured rows and calls these; each
|
||||
renderer returns ``None`` when it has nothing to say, so the caller can fall
|
||||
back to the gov-EPC prose.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Optional
|
||||
|
||||
# --- SAP integer-code → display-text maps ---------------------------------
|
||||
# wall_construction (SAP10 codes, domain/sap10_ml/rdsap_uvalues.py:127-142).
|
||||
_WALL_CONSTRUCTION: dict[int, str] = {
|
||||
1: "Granite or whinstone",
|
||||
2: "Sandstone",
|
||||
3: "Solid brick",
|
||||
4: "Cavity",
|
||||
5: "Timber frame",
|
||||
6: "System build",
|
||||
7: "Cob",
|
||||
8: "Park home",
|
||||
9: "Curtain wall",
|
||||
10: "Unknown",
|
||||
11: "Cavity (filled party wall)",
|
||||
}
|
||||
# wall_insulation_type (RdSAP schema codes, rdsap_uvalues.py:145-158).
|
||||
_WALL_INSULATION: dict[int, str] = {
|
||||
1: "external insulation",
|
||||
2: "filled cavity",
|
||||
3: "internal insulation",
|
||||
4: "as built",
|
||||
5: "no insulation",
|
||||
6: "filled cavity + external insulation",
|
||||
7: "filled cavity + internal insulation",
|
||||
}
|
||||
# heat_emitter_type (SAP10.2, mapper.py _ELMHURST_HEAT_EMITTER_TO_SAP10).
|
||||
_HEAT_EMITTER: dict[int, str] = {
|
||||
1: "radiators",
|
||||
2: "underfloor heating (in screed)",
|
||||
3: "underfloor heating (timber floor)",
|
||||
4: "warm air",
|
||||
5: "fan coils",
|
||||
}
|
||||
# main_fuel_type / water_heating_fuel (RdSAP "(not community)" family; invert of
|
||||
# main_fuel_overlay._FUEL_CODES).
|
||||
_FUEL: dict[int, str] = {
|
||||
3: "bottled LPG",
|
||||
6: "wood logs",
|
||||
10: "dual fuel (mineral and wood)",
|
||||
15: "smokeless coal",
|
||||
17: "LPG (special condition)",
|
||||
20: "mains gas (community)",
|
||||
25: "electricity (community)",
|
||||
26: "mains gas",
|
||||
27: "LPG (bulk)",
|
||||
28: "oil",
|
||||
29: "electricity",
|
||||
31: "biomass (community)",
|
||||
33: "house coal",
|
||||
}
|
||||
# main_heating_control (SAP10.2 Table 4e Group 1; comments in
|
||||
# heating_recommendation._CONTROL_FEATURES_BY_CODE).
|
||||
_HEATING_CONTROL: dict[int, str] = {
|
||||
2101: "no time or thermostatic control",
|
||||
2102: "programmer, no room thermostat",
|
||||
2103: "room thermostat only",
|
||||
2104: "programmer and room thermostat",
|
||||
2105: "programmer and at least two room thermostats",
|
||||
2106: "programmer, room thermostat and TRVs",
|
||||
2107: "programmer, TRVs and bypass",
|
||||
2108: "programmer, TRVs and flow switch",
|
||||
2109: "programmer, TRVs and boiler energy manager",
|
||||
2111: "TRVs and bypass",
|
||||
2113: "room thermostat and TRVs",
|
||||
}
|
||||
# water_heating_code (SAP Table 4a; water_heating_overlay docstring).
|
||||
_WATER_HEATING_SYSTEM: dict[int, str] = {
|
||||
901: "from main system",
|
||||
903: "electric immersion",
|
||||
911: "gas boiler / circulator",
|
||||
}
|
||||
# RdSAP England-&-Wales construction age-band letters → the date range the EPC
|
||||
# prints (domain/epc/property_overrides/construction_age_band.py). The lodged
|
||||
# `epc_building_part.construction_age_band` carries the bare letter.
|
||||
_AGE_BAND: dict[str, str] = {
|
||||
"A": "before 1900",
|
||||
"B": "1900-1929",
|
||||
"C": "1930-1949",
|
||||
"D": "1950-1966",
|
||||
"E": "1967-1975",
|
||||
"F": "1976-1982",
|
||||
"G": "1983-1990",
|
||||
"H": "1991-1995",
|
||||
"I": "1996-2002",
|
||||
"J": "2003-2006",
|
||||
"K": "2007-2011",
|
||||
"L": "2012-2022",
|
||||
"M": "2023 onwards",
|
||||
}
|
||||
|
||||
|
||||
def _as_int(value: Any) -> Optional[int]:
|
||||
"""A SAP code as ``int``. Accepts the raw int, a numeric string, or ``None``;
|
||||
a non-numeric label (e.g. an insulation ``"As built"`` string in a numeric
|
||||
slot) yields ``None``."""
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, bool):
|
||||
return None
|
||||
if isinstance(value, int):
|
||||
return value
|
||||
try:
|
||||
return int(str(value).strip())
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _clean_text(value: Any) -> Optional[str]:
|
||||
"""A lodged free-text field trimmed to a non-empty string, else ``None``.
|
||||
A bare integer (a code lodged in a text slot) is not display text."""
|
||||
if value is None or isinstance(value, (int, float, bool)):
|
||||
return None
|
||||
text = str(value).strip()
|
||||
return text or None
|
||||
|
||||
|
||||
# built_form codes (epc_codes.csv; override_code_mapping._BUILT_FORM_CODES).
|
||||
# The lodged value is usually the display text already; a minority of records
|
||||
# carry the bare integer code instead.
|
||||
_BUILT_FORM: dict[int, str] = {
|
||||
1: "Detached",
|
||||
2: "Semi-detached",
|
||||
3: "End-terrace",
|
||||
4: "Mid-terrace",
|
||||
5: "Enclosed end-terrace",
|
||||
6: "Enclosed mid-terrace",
|
||||
}
|
||||
|
||||
|
||||
def built_form_description(built_form: Any) -> Optional[str]:
|
||||
"""The dwelling's built form. Decodes the integer code where a record
|
||||
lodges one, otherwise passes the lodged display text through unchanged."""
|
||||
code = _as_int(built_form)
|
||||
if code is not None:
|
||||
return _BUILT_FORM.get(code)
|
||||
return _clean_text(built_form)
|
||||
|
||||
|
||||
def age_band_description(band: Any) -> Optional[str]:
|
||||
"""The construction age band as its date range, e.g. ``"1900-1929 (band B)"``.
|
||||
The lodged value is a bare RdSAP letter; an unrecognised or ``"Unknown"``
|
||||
band renders nothing."""
|
||||
letter = _clean_text(band)
|
||||
if letter is None:
|
||||
return None
|
||||
date_range = _AGE_BAND.get(letter.strip().upper())
|
||||
return f"{date_range} (band {letter.strip().upper()})" if date_range else None
|
||||
|
||||
|
||||
def wall_description(construction: Any, insulation: Any) -> Optional[str]:
|
||||
"""A wall part's text, e.g. ``"Cavity wall, filled cavity"``. The
|
||||
construction names the wall; the insulation qualifies it when known."""
|
||||
cons = _WALL_CONSTRUCTION.get(_as_int(construction) or 0)
|
||||
if cons is None:
|
||||
return None
|
||||
ins = _WALL_INSULATION.get(_as_int(insulation) or 0)
|
||||
return f"{cons} wall, {ins}" if ins else f"{cons} wall"
|
||||
|
||||
|
||||
def _insulation_thickness_mm(thickness: Any) -> Optional[int]:
|
||||
"""A roof-insulation thickness in mm from a numeric code or a ``"225mm"``
|
||||
string; a non-numeric label (``"As built"``, ``"ND"``) or ``None`` yields
|
||||
``None``. Zero is a real value (no insulation) and is preserved."""
|
||||
as_int = _as_int(thickness)
|
||||
if as_int is not None:
|
||||
return as_int
|
||||
text = _clean_text(thickness)
|
||||
if text is None:
|
||||
return None
|
||||
digits = "".join(ch for ch in text if ch.isdigit())
|
||||
return int(digits) if digits else None
|
||||
|
||||
|
||||
def roof_description(
|
||||
construction_type: Any, insulation_location: Any, insulation_thickness: Any
|
||||
) -> Optional[str]:
|
||||
"""A roof part's text, e.g. ``"Pitched roof (Slates or tiles), Access to
|
||||
loft; 270mm insulation at joists"``. The construction is already lodged as
|
||||
prose; the insulation clause is appended when a thickness or a named
|
||||
location is present."""
|
||||
cons = _clean_text(construction_type)
|
||||
if cons is None:
|
||||
return None
|
||||
thickness_mm = _insulation_thickness_mm(insulation_thickness)
|
||||
location = _clean_text(insulation_location)
|
||||
if location is not None and location.lower() in {"unknown", "none"}:
|
||||
location = None
|
||||
if thickness_mm == 0:
|
||||
return f"{cons}; no insulation"
|
||||
if thickness_mm is None and location is None:
|
||||
return cons
|
||||
clause = "insulation" if thickness_mm is None else f"{thickness_mm}mm insulation"
|
||||
if location is not None:
|
||||
clause = f"{clause} at {location.lower()}"
|
||||
return f"{cons}; {clause}"
|
||||
|
||||
|
||||
def floor_description(
|
||||
floor_type: Any, construction_type: Any, insulation: Any
|
||||
) -> Optional[str]:
|
||||
"""A floor part's text, e.g. ``"Ground floor, suspended timber, insulation
|
||||
as built"``. Composed from the lodged floor type, construction, and
|
||||
insulation status (all lodged as prose)."""
|
||||
parts: list[str] = []
|
||||
location = _clean_text(floor_type)
|
||||
if location is not None:
|
||||
parts.append(location[0].upper() + location[1:].lower())
|
||||
construction = _clean_text(construction_type)
|
||||
if construction is not None:
|
||||
parts.append(construction.lower())
|
||||
ins = _clean_text(insulation)
|
||||
if ins is not None:
|
||||
parts.append(f"insulation {ins.lower()}")
|
||||
return ", ".join(parts) or None
|
||||
|
||||
|
||||
def heating_description(
|
||||
fuel: Any, emitter: Any, control: Any, condensing: Any
|
||||
) -> Optional[str]:
|
||||
"""The main heating system's text from its SAP codes, e.g. ``"Mains gas,
|
||||
radiators (condensing); programmer, room thermostat and TRVs"``. Reports
|
||||
only what the codes state — the appliance archetype is not lodged on a
|
||||
re-survey, so no boiler/heat-pump type is asserted."""
|
||||
fuel_text = _FUEL.get(_as_int(fuel) or 0)
|
||||
emitter_text = _HEAT_EMITTER.get(_as_int(emitter) or 0)
|
||||
lead_parts = [p for p in (fuel_text, emitter_text) if p]
|
||||
if not lead_parts:
|
||||
return None
|
||||
lead = ", ".join(p[0].upper() + p[1:] if i == 0 else p for i, p in enumerate(lead_parts))
|
||||
if condensing is True:
|
||||
lead = f"{lead} (condensing)"
|
||||
control_text = _HEATING_CONTROL.get(_as_int(control) or 0)
|
||||
return f"{lead}; {control_text}" if control_text else lead
|
||||
|
||||
|
||||
def hot_water_description(
|
||||
water_heating_code: Any,
|
||||
water_heating_fuel: Any,
|
||||
main_fuel: Any,
|
||||
has_cylinder: Any,
|
||||
solar: Any,
|
||||
) -> Optional[str]:
|
||||
"""The hot-water system's text from its SAP codes, e.g. ``"From main
|
||||
system, mains gas"``. Where the water-heating code is not lodged on the
|
||||
re-survey, it falls back to the SAP default (a dwelling with no cylinder
|
||||
draws hot water from the main system) using the main heating fuel."""
|
||||
system = _WATER_HEATING_SYSTEM.get(_as_int(water_heating_code) or 0)
|
||||
fuel = _FUEL.get(_as_int(water_heating_fuel) or 0) or _FUEL.get(_as_int(main_fuel) or 0)
|
||||
if system is None:
|
||||
# No lodged water-heating code: a dwelling without a cylinder inherits
|
||||
# hot water from the main system (SAP Table 4a code 901).
|
||||
if has_cylinder is True:
|
||||
system = "hot water cylinder"
|
||||
elif has_cylinder is False:
|
||||
system = "from main system"
|
||||
else:
|
||||
return None
|
||||
text = f"{system}, {fuel}" if fuel else system
|
||||
text = text[0].upper() + text[1:]
|
||||
if solar is True:
|
||||
text = f"{text}; with solar water heating"
|
||||
return text
|
||||
|
|
@ -59,6 +59,8 @@ _ID_COLUMNS: tuple[str, ...] = (
|
|||
"address",
|
||||
"postcode",
|
||||
"property_type",
|
||||
"built_form",
|
||||
"property_age_band",
|
||||
"walls",
|
||||
"roof",
|
||||
"floor",
|
||||
|
|
@ -147,6 +149,8 @@ class PropertyScenarioData:
|
|||
address: Optional[str] = None
|
||||
postcode: Optional[str] = None
|
||||
property_type: Optional[str] = None
|
||||
built_form: Optional[str] = None
|
||||
property_age_band: Optional[str] = None
|
||||
walls: Optional[str] = None
|
||||
roof: Optional[str] = None
|
||||
floor: Optional[str] = None
|
||||
|
|
@ -196,6 +200,8 @@ def _shape_row(prop: PropertyScenarioData) -> dict[str, Any]:
|
|||
"address": _blank(prop.address),
|
||||
"postcode": _blank(prop.postcode),
|
||||
"property_type": _blank(prop.property_type),
|
||||
"built_form": _blank(prop.built_form),
|
||||
"property_age_band": _blank(prop.property_age_band),
|
||||
"walls": _blank(prop.walls),
|
||||
"roof": _blank(prop.roof),
|
||||
"floor": _blank(prop.floor),
|
||||
|
|
|
|||
|
|
@ -6,6 +6,15 @@ from typing import Any, Optional, Sequence
|
|||
from sqlalchemy import text
|
||||
from sqlmodel import Session
|
||||
|
||||
from domain.scenario_export.fabric_description import (
|
||||
age_band_description,
|
||||
built_form_description,
|
||||
floor_description,
|
||||
heating_description,
|
||||
hot_water_description,
|
||||
roof_description,
|
||||
wall_description,
|
||||
)
|
||||
from domain.scenario_export.scenario_sheet import ExportMeasure, PropertyScenarioData
|
||||
from repositories.scenario_export.scenario_export_repository import (
|
||||
ScenarioExportRepository,
|
||||
|
|
@ -60,19 +69,26 @@ _MEASURES_SQL = text(
|
|||
# performance are matched by UPRN (the latest epc_property for the source),
|
||||
# because the PasHub re-fetch lands records with property_id = NULL — a
|
||||
# property_id join would silently fall back to a stale gov-EPC cert (wrong
|
||||
# lodgement dates / property_type) or nothing. The descriptive prose elements
|
||||
# have no PasHub equivalent (a survey lodges structured fields, not gov-EPC
|
||||
# prose), so they stay on the property_id join as the best-available gov-EPC
|
||||
# fallback. All three are read separately and composed per Property.
|
||||
# lodgement dates / property_type) or nothing. "Latest" is ordered by the cert's
|
||||
# effective lodgement date (registration → completion → inspection) DESC, NOT by
|
||||
# `e.id`: row-id order does not track recency — a gov-EPC cert re-ingested after
|
||||
# the PasHub survey lands with a HIGHER `e.id`, so an `e.id` sort silently picks
|
||||
# the stale gov cert over the fresh 2026 survey (observed on 3 of portfolio 838).
|
||||
# `e.id DESC` remains the tiebreak within one date (duplicate survey loads carry
|
||||
# identical facts). The descriptive prose elements have no PasHub equivalent (a
|
||||
# survey lodges structured fields, not gov-EPC prose), so they stay on the
|
||||
# property_id join as the best-available gov-EPC fallback. All three are read
|
||||
# separately and composed per Property.
|
||||
_EPC_HEADER_SQL = text(
|
||||
"WITH latest AS ("
|
||||
" SELECT DISTINCT ON (p.id) p.id AS property_id, e.id AS epc_id"
|
||||
" FROM property p JOIN epc_property e ON e.uprn = p.uprn"
|
||||
" WHERE p.id = ANY(:property_ids) AND e.source = :source"
|
||||
" ORDER BY p.id, e.id DESC)"
|
||||
" ORDER BY p.id, COALESCE(e.registration_date, e.completion_date,"
|
||||
" e.inspection_date) DESC NULLS LAST, e.id DESC)"
|
||||
" SELECT l.property_id, e.property_type, e.total_floor_area_m2,"
|
||||
" COALESCE(e.registration_date, e.completion_date, e.inspection_date),"
|
||||
" e.habitable_rooms_count"
|
||||
" e.habitable_rooms_count, e.built_form"
|
||||
" FROM latest l JOIN epc_property e ON e.id = l.epc_id"
|
||||
)
|
||||
_EPC_ELEMENTS_SQL = text(
|
||||
|
|
@ -86,13 +102,57 @@ _EPC_PERF_SQL = text(
|
|||
" SELECT DISTINCT ON (p.id) p.id AS property_id, e.id AS epc_id"
|
||||
" FROM property p JOIN epc_property e ON e.uprn = p.uprn"
|
||||
" WHERE p.id = ANY(:property_ids) AND e.source = :source"
|
||||
" ORDER BY p.id, e.id DESC)"
|
||||
" ORDER BY p.id, COALESCE(e.registration_date, e.completion_date,"
|
||||
" e.inspection_date) DESC NULLS LAST, e.id DESC)"
|
||||
" SELECT l.property_id, perf.energy_rating_current,"
|
||||
" perf.current_energy_efficiency_band"
|
||||
" FROM latest l"
|
||||
" JOIN epc_property_energy_performance perf ON perf.epc_property_id = l.epc_id"
|
||||
)
|
||||
|
||||
# The structured SAP fabric on the same canonical (latest-by-date, UPRN-matched)
|
||||
# cert the header reads: the MAIN building part (walls / roof / floor), the main
|
||||
# heating detail (fuel / emitter / control / condensing) and the water-heating
|
||||
# fields. A PasHub re-survey lodges these coded facts but no gov-EPC prose, so
|
||||
# they supply walls/roof/floor/heating/hot_water where the prose fallback is
|
||||
# blank. The main building part is chosen identifier='main' first, then the
|
||||
# lowest part number / id (matching the SAP engine's `parts[0]` primary part).
|
||||
# Windows and lighting are not read here (their per-element codes do not cleanly
|
||||
# reconstruct the dwelling phrase); they stay on the gov-EPC prose fallback.
|
||||
_EPC_STRUCTURED_FABRIC_SQL = text(
|
||||
"WITH latest AS ("
|
||||
" SELECT DISTINCT ON (p.id) p.id AS property_id, e.id AS epc_id"
|
||||
" FROM property p JOIN epc_property e ON e.uprn = p.uprn"
|
||||
" WHERE p.id = ANY(:property_ids) AND e.source = :source"
|
||||
" ORDER BY p.id, COALESCE(e.registration_date, e.completion_date,"
|
||||
" e.inspection_date) DESC NULLS LAST, e.id DESC),"
|
||||
" main_part AS ("
|
||||
" SELECT DISTINCT ON (l.property_id) l.property_id, bp.construction_age_band,"
|
||||
" bp.wall_construction, bp.wall_insulation_type, bp.roof_construction_type,"
|
||||
" bp.roof_insulation_location, bp.roof_insulation_thickness, bp.floor_type,"
|
||||
" bp.floor_construction_type, bp.floor_insulation_type_str"
|
||||
" FROM latest l JOIN epc_building_part bp ON bp.epc_property_id = l.epc_id"
|
||||
" ORDER BY l.property_id,"
|
||||
" (CASE WHEN lower(bp.identifier) = 'main' THEN 0 ELSE 1 END),"
|
||||
" bp.building_part_number NULLS FIRST, bp.id),"
|
||||
" main_heat AS ("
|
||||
" SELECT DISTINCT ON (h.epc_property_id) h.epc_property_id, h.main_fuel_type,"
|
||||
" h.heat_emitter_type, h.main_heating_control, h.condensing"
|
||||
" FROM epc_main_heating_detail h"
|
||||
" ORDER BY h.epc_property_id, h.main_heating_number NULLS FIRST, h.id)"
|
||||
" SELECT l.property_id, mp.wall_construction, mp.wall_insulation_type,"
|
||||
" mp.roof_construction_type, mp.roof_insulation_location,"
|
||||
" mp.roof_insulation_thickness, mp.floor_type, mp.floor_construction_type,"
|
||||
" mp.floor_insulation_type_str, mh.main_fuel_type, mh.heat_emitter_type,"
|
||||
" mh.main_heating_control, mh.condensing, e.heating_water_heating_code,"
|
||||
" e.heating_water_heating_fuel, e.has_hot_water_cylinder,"
|
||||
" e.solar_water_heating, mp.construction_age_band"
|
||||
" FROM latest l"
|
||||
" JOIN epc_property e ON e.id = l.epc_id"
|
||||
" LEFT JOIN main_part mp ON mp.property_id = l.property_id"
|
||||
" LEFT JOIN main_heat mh ON mh.epc_property_id = l.epc_id"
|
||||
)
|
||||
|
||||
# A landlord property_type override wins over any EPC-derived value (ADR-0065);
|
||||
# the descriptive prose fields have no override prose, so only property_type is
|
||||
# resolved this way (the override enum carries coded fabric facts, not prose).
|
||||
|
|
@ -183,6 +243,8 @@ class ScenarioExportPostgresRepository(ScenarioExportRepository):
|
|||
energy_consumption_savings=_float(row[11]),
|
||||
valuation_increase=_float(row[12]),
|
||||
property_type=facts.get("property_type"),
|
||||
built_form=facts.get("built_form"),
|
||||
property_age_band=facts.get("property_age_band"),
|
||||
walls=facts.get("walls"),
|
||||
roof=facts.get("roof"),
|
||||
floor=facts.get("floor"),
|
||||
|
|
@ -261,6 +323,7 @@ class ScenarioExportPostgresRepository(ScenarioExportRepository):
|
|||
"lodgement_date": registration_date,
|
||||
"is_expired": _is_expired(registration_date),
|
||||
"number_of_rooms": row[4],
|
||||
"built_form": built_form_description(row[5]),
|
||||
}
|
||||
)
|
||||
|
||||
|
|
@ -287,4 +350,23 @@ class ScenarioExportPostgresRepository(ScenarioExportRepository):
|
|||
}
|
||||
)
|
||||
|
||||
# Structured SAP fabric (walls/roof/floor/heating/hot_water) composed
|
||||
# from the canonical cert's coded fields; overrides the gov-EPC prose
|
||||
# where present (a re-survey lodges these but no prose), and leaves the
|
||||
# prose in place where a field renders to nothing.
|
||||
for row in connection.execute(_EPC_STRUCTURED_FABRIC_SQL, source_params).all():
|
||||
rendered = {
|
||||
"property_age_band": age_band_description(row[17]),
|
||||
"walls": wall_description(row[1], row[2]),
|
||||
"roof": roof_description(row[3], row[4], row[5]),
|
||||
"floor": floor_description(row[6], row[7], row[8]),
|
||||
"heating": heating_description(row[9], row[10], row[11], row[12]),
|
||||
"hot_water": hot_water_description(
|
||||
row[13], row[14], row[9], row[15], row[16]
|
||||
),
|
||||
}
|
||||
facts.setdefault(row[0], {}).update(
|
||||
{field: value for field, value in rendered.items() if value is not None}
|
||||
)
|
||||
|
||||
return facts
|
||||
|
|
|
|||
174
tests/domain/scenario_export/test_fabric_description.py
Normal file
174
tests/domain/scenario_export/test_fabric_description.py
Normal file
|
|
@ -0,0 +1,174 @@
|
|||
"""Behaviour of the structured-fabric renderers (ADR-0065): composing client
|
||||
sheet text from an EPC's SAP integer codes, decoding with the SAP engine's own
|
||||
code space and normalising the ragged shapes a re-survey lodges."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from domain.scenario_export.fabric_description import (
|
||||
age_band_description,
|
||||
built_form_description,
|
||||
floor_description,
|
||||
heating_description,
|
||||
hot_water_description,
|
||||
roof_description,
|
||||
wall_description,
|
||||
)
|
||||
|
||||
|
||||
class TestBuiltFormDescription:
|
||||
def test_lodged_display_text_passes_through(self) -> None:
|
||||
assert built_form_description("Mid-terrace") == "Mid-terrace"
|
||||
assert built_form_description("Semi-Detached") == "Semi-Detached"
|
||||
|
||||
def test_integer_code_is_decoded(self) -> None:
|
||||
assert built_form_description(4) == "Mid-terrace"
|
||||
assert built_form_description("1") == "Detached"
|
||||
|
||||
def test_absent_or_unknown_code_renders_nothing(self) -> None:
|
||||
assert built_form_description(None) is None
|
||||
assert built_form_description(99) is None
|
||||
|
||||
|
||||
class TestAgeBandDescription:
|
||||
def test_rdsap_letter_decodes_to_date_range_and_band(self) -> None:
|
||||
assert age_band_description("B") == "1900-1929 (band B)"
|
||||
assert age_band_description("A") == "before 1900 (band A)"
|
||||
assert age_band_description("M") == "2023 onwards (band M)"
|
||||
|
||||
def test_lowercase_and_whitespace_are_tolerated(self) -> None:
|
||||
assert age_band_description(" k ") == "2007-2011 (band K)"
|
||||
|
||||
def test_unknown_or_absent_band_renders_nothing(self) -> None:
|
||||
assert age_band_description("Unknown") is None
|
||||
assert age_band_description(None) is None
|
||||
assert age_band_description("Z") is None
|
||||
|
||||
|
||||
class TestWallDescription:
|
||||
def test_construction_and_insulation_codes_decode_to_text(self) -> None:
|
||||
assert wall_description(4, 2) == "Cavity wall, filled cavity"
|
||||
assert wall_description(3, 4) == "Solid brick wall, as built"
|
||||
assert wall_description(5, 1) == "Timber frame wall, external insulation"
|
||||
|
||||
def test_construction_only_when_insulation_unknown(self) -> None:
|
||||
assert wall_description(4, None) == "Cavity wall"
|
||||
assert wall_description(4, 99) == "Cavity wall"
|
||||
|
||||
def test_numeric_string_codes_are_accepted(self) -> None:
|
||||
# JSONB can round-trip a code as a string; it still decodes.
|
||||
assert wall_description("4", "2") == "Cavity wall, filled cavity"
|
||||
|
||||
def test_unknown_construction_renders_nothing(self) -> None:
|
||||
assert wall_description(None, 2) is None
|
||||
assert wall_description(99, 2) is None
|
||||
|
||||
|
||||
class TestRoofDescription:
|
||||
def test_construction_prose_with_thickness_and_location(self) -> None:
|
||||
assert (
|
||||
roof_description("Pitched roof, Access to loft", "Joists", 270)
|
||||
== "Pitched roof, Access to loft; 270mm insulation at joists"
|
||||
)
|
||||
|
||||
def test_thickness_string_with_unit_is_normalised(self) -> None:
|
||||
# A minority of records lodge the thickness as "225mm" rather than 225.
|
||||
assert (
|
||||
roof_description("Pitched", "Joists", "225mm")
|
||||
== "Pitched; 225mm insulation at joists"
|
||||
)
|
||||
|
||||
def test_non_numeric_thickness_and_coded_location_are_dropped(self) -> None:
|
||||
# thickness "As built" is not a measurement; a bare int location is a
|
||||
# code, not a place — with neither, only the construction survives.
|
||||
assert roof_description("Pitched", 2, "As built") == "Pitched"
|
||||
|
||||
def test_named_location_without_a_thickness_still_notes_insulation(self) -> None:
|
||||
assert (
|
||||
roof_description("Pitched", "Joists", "As built")
|
||||
== "Pitched; insulation at joists"
|
||||
)
|
||||
|
||||
def test_zero_thickness_reads_as_no_insulation(self) -> None:
|
||||
assert roof_description("Flat roof", "Joists", 0) == "Flat roof; no insulation"
|
||||
|
||||
def test_unknown_location_is_omitted(self) -> None:
|
||||
assert roof_description("Pitched", "Unknown", 100) == "Pitched; 100mm insulation"
|
||||
|
||||
def test_construction_only_when_no_insulation_detail(self) -> None:
|
||||
assert roof_description("Pitched", None, None) == "Pitched"
|
||||
|
||||
def test_no_construction_renders_nothing(self) -> None:
|
||||
assert roof_description(None, "Joists", 270) is None
|
||||
|
||||
|
||||
class TestFloorDescription:
|
||||
def test_composes_type_construction_and_insulation(self) -> None:
|
||||
assert (
|
||||
floor_description("Ground floor", "Suspended, timber", "As Built")
|
||||
== "Ground floor, suspended, timber, insulation as built"
|
||||
)
|
||||
|
||||
def test_normalises_casing_of_the_floor_type(self) -> None:
|
||||
assert floor_description("Ground Floor", "Solid", None) == "Ground floor, solid"
|
||||
|
||||
def test_retrofitted_insulation_is_carried(self) -> None:
|
||||
assert (
|
||||
floor_description("Ground floor", "Solid", "Retro-fitted")
|
||||
== "Ground floor, solid, insulation retro-fitted"
|
||||
)
|
||||
|
||||
def test_empty_when_nothing_lodged(self) -> None:
|
||||
assert floor_description(None, None, None) is None
|
||||
|
||||
|
||||
class TestHeatingDescription:
|
||||
def test_fuel_emitter_control_and_condensing(self) -> None:
|
||||
assert (
|
||||
heating_description(26, 1, 2106, True)
|
||||
== "Mains gas, radiators (condensing); programmer, room thermostat and TRVs"
|
||||
)
|
||||
|
||||
def test_condensing_false_or_absent_omits_the_flag(self) -> None:
|
||||
assert heating_description(26, 1, 2104, None) == (
|
||||
"Mains gas, radiators; programmer and room thermostat"
|
||||
)
|
||||
assert heating_description(26, 1, None, False) == "Mains gas, radiators"
|
||||
|
||||
def test_unknown_control_is_dropped(self) -> None:
|
||||
assert heating_description(26, 1, 9999, True) == "Mains gas, radiators (condensing)"
|
||||
|
||||
def test_empty_when_no_fuel_or_emitter(self) -> None:
|
||||
assert heating_description(None, None, 2106, True) is None
|
||||
|
||||
|
||||
class TestHotWaterDescription:
|
||||
def test_lodged_code_and_fuel_decode(self) -> None:
|
||||
assert (
|
||||
hot_water_description(901, 26, 26, False, False) == "From main system, mains gas"
|
||||
)
|
||||
assert (
|
||||
hot_water_description(903, 29, 26, True, False)
|
||||
== "Electric immersion, electricity"
|
||||
)
|
||||
|
||||
def test_falls_back_to_main_system_when_no_code_and_no_cylinder(self) -> None:
|
||||
# SAP Table 4a: a dwelling with no cylinder draws hot water from the main
|
||||
# system; the main heating fuel names it.
|
||||
assert (
|
||||
hot_water_description(None, None, 26, False, False)
|
||||
== "From main system, mains gas"
|
||||
)
|
||||
|
||||
def test_cylinder_without_a_code(self) -> None:
|
||||
assert (
|
||||
hot_water_description(None, None, 29, True, False)
|
||||
== "Hot water cylinder, electricity"
|
||||
)
|
||||
|
||||
def test_solar_water_heating_is_appended(self) -> None:
|
||||
assert hot_water_description(901, 26, 26, False, True) == (
|
||||
"From main system, mains gas; with solar water heating"
|
||||
)
|
||||
|
||||
def test_empty_when_nothing_lodged(self) -> None:
|
||||
assert hot_water_description(None, None, None, None, False) is None
|
||||
|
|
@ -4,14 +4,16 @@ read-model, from persisted data only (no live EPC calls)."""
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
from typing import Optional, Union
|
||||
|
||||
from sqlalchemy import Engine
|
||||
from sqlmodel import Session
|
||||
|
||||
from datatypes.epc.domain.epc import Epc
|
||||
from infrastructure.postgres.epc_property_table import ( # registers epc tables
|
||||
EpcBuildingPartModel,
|
||||
EpcEnergyElementModel,
|
||||
EpcMainHeatingDetailModel,
|
||||
EpcPropertyEnergyPerformanceModel,
|
||||
EpcPropertyModel,
|
||||
)
|
||||
|
|
@ -37,6 +39,7 @@ def _add_epc(
|
|||
uprn: int = 100001,
|
||||
source: str = "lodged",
|
||||
property_type: Optional[str] = None,
|
||||
built_form: Optional[str] = None,
|
||||
total_floor_area_m2: float = 90.0,
|
||||
registration_date: str = "2005-01-01",
|
||||
habitable_rooms_count: int = 4,
|
||||
|
|
@ -54,6 +57,7 @@ def _add_epc(
|
|||
uprn=uprn,
|
||||
source=source,
|
||||
property_type=property_type,
|
||||
built_form=built_form,
|
||||
total_floor_area_m2=total_floor_area_m2,
|
||||
registration_date=registration_date,
|
||||
habitable_rooms_count=habitable_rooms_count,
|
||||
|
|
@ -103,6 +107,66 @@ def _add_epc(
|
|||
)
|
||||
|
||||
|
||||
def _add_building_part(
|
||||
session: Session,
|
||||
*,
|
||||
epc_id: int,
|
||||
identifier: str = "main",
|
||||
building_part_number: Optional[int] = None,
|
||||
wall_construction: int = 4,
|
||||
wall_insulation_type: int = 2,
|
||||
roof_construction_type: Optional[str] = "Pitched roof, Access to loft",
|
||||
roof_insulation_location: Optional[str] = "Joists",
|
||||
roof_insulation_thickness: Optional[Union[str, int]] = 270,
|
||||
floor_type: Optional[str] = "Ground floor",
|
||||
floor_construction_type: Optional[str] = "Solid",
|
||||
floor_insulation_type_str: Optional[str] = "As Built",
|
||||
) -> None:
|
||||
"""Seed one SAP building part on an EPC slot (the source of the structured
|
||||
walls / roof / floor text)."""
|
||||
session.add(
|
||||
EpcBuildingPartModel(
|
||||
epc_property_id=epc_id,
|
||||
identifier=identifier,
|
||||
construction_age_band="D",
|
||||
wall_construction=wall_construction,
|
||||
wall_insulation_type=wall_insulation_type,
|
||||
wall_thickness_measured=False,
|
||||
building_part_number=building_part_number,
|
||||
roof_construction_type=roof_construction_type,
|
||||
roof_insulation_location=roof_insulation_location,
|
||||
roof_insulation_thickness=roof_insulation_thickness,
|
||||
floor_type=floor_type,
|
||||
floor_construction_type=floor_construction_type,
|
||||
floor_insulation_type_str=floor_insulation_type_str,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _add_main_heating(
|
||||
session: Session,
|
||||
*,
|
||||
epc_id: int,
|
||||
main_fuel_type: int = 26,
|
||||
heat_emitter_type: int = 1,
|
||||
main_heating_control: int = 2106,
|
||||
condensing: Optional[bool] = True,
|
||||
) -> None:
|
||||
"""Seed the main heating detail on an EPC slot (the source of the structured
|
||||
heating text)."""
|
||||
session.add(
|
||||
EpcMainHeatingDetailModel(
|
||||
epc_property_id=epc_id,
|
||||
has_fghrs=False,
|
||||
main_fuel_type=main_fuel_type,
|
||||
heat_emitter_type=heat_emitter_type,
|
||||
emitter_temperature=1,
|
||||
main_heating_control=main_heating_control,
|
||||
condensing=condensing,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def test_returns_a_row_for_a_property_with_a_default_plan_under_the_scenario(
|
||||
db_engine: Engine,
|
||||
) -> None:
|
||||
|
|
@ -361,6 +425,178 @@ def test_effective_epc_descriptive_fields_come_from_the_lodged_epc(
|
|||
assert row.is_expired is True
|
||||
|
||||
|
||||
def test_header_picks_the_freshest_cert_by_date_not_by_row_id(
|
||||
db_engine: Engine,
|
||||
) -> None:
|
||||
# arrange — two lodged certs on one UPRN: a fresh 2026 survey lodged with a
|
||||
# LOWER epc_id, and a stale 2013 gov cert re-ingested afterwards with a
|
||||
# HIGHER epc_id. An `e.id DESC` sort would wrongly pick the stale gov cert;
|
||||
# the freshest lodgement date must win (the portfolio-838 TFA regression).
|
||||
with Session(db_engine) as session:
|
||||
session.add(
|
||||
PropertyRow(id=1, portfolio_id=100, landlord_property_id="LP1", uprn=100001)
|
||||
)
|
||||
session.add(
|
||||
PlanModel(portfolio_id=100, property_id=1, scenario_id=5, is_default=True)
|
||||
)
|
||||
_add_epc(
|
||||
session,
|
||||
property_id=1,
|
||||
epc_id=10, # fresh survey, LOWER id
|
||||
uprn=100001,
|
||||
total_floor_area_m2=76.62,
|
||||
registration_date="2026-05-18",
|
||||
)
|
||||
_add_epc(
|
||||
session,
|
||||
property_id=1,
|
||||
epc_id=20, # stale gov cert re-ingested later, HIGHER id
|
||||
uprn=100001,
|
||||
total_floor_area_m2=77.0,
|
||||
registration_date="2013-10-09",
|
||||
)
|
||||
session.commit()
|
||||
|
||||
# act
|
||||
with Session(db_engine) as session:
|
||||
rows = ScenarioExportPostgresRepository(session).rows_for(
|
||||
portfolio_id=100, scenario_id=5, property_ids=[1]
|
||||
)
|
||||
|
||||
# assert — the fresh 2026 survey's floor area and lodgement win, not the
|
||||
# higher-id 2013 gov cert.
|
||||
assert rows[0].total_floor_area == 76.62
|
||||
assert rows[0].lodgement_date == "2026-05-18"
|
||||
|
||||
|
||||
def test_structured_fabric_supplies_walls_roof_floor_heating_and_hot_water(
|
||||
db_engine: Engine,
|
||||
) -> None:
|
||||
# arrange — a lodged EPC carrying gov prose for every descriptive field AND
|
||||
# the structured SAP fabric (building part + heating detail) a re-survey
|
||||
# lodges. The structured fabric must win for walls/roof/floor/heating/
|
||||
# hot_water; windows/lighting have no structured source and keep the prose.
|
||||
with Session(db_engine) as session:
|
||||
session.add(
|
||||
PropertyRow(id=1, portfolio_id=100, landlord_property_id="LP1", uprn=100001)
|
||||
)
|
||||
session.add(
|
||||
PlanModel(portfolio_id=100, property_id=1, scenario_id=5, is_default=True)
|
||||
)
|
||||
_add_epc(
|
||||
session,
|
||||
property_id=1,
|
||||
epc_id=1,
|
||||
source="lodged",
|
||||
built_form="Mid-terrace",
|
||||
elements={
|
||||
"wall": "PROSE walls",
|
||||
"roof": "PROSE roof",
|
||||
"floor": "PROSE floor",
|
||||
"main_heating": "PROSE heating",
|
||||
"hot_water": "PROSE hot water",
|
||||
"window": "Fully double glazed",
|
||||
"lighting": "Low energy lighting in all fixed outlets",
|
||||
},
|
||||
)
|
||||
_add_building_part(session, epc_id=1)
|
||||
_add_main_heating(session, epc_id=1)
|
||||
session.commit()
|
||||
|
||||
# act
|
||||
with Session(db_engine) as session:
|
||||
rows = ScenarioExportPostgresRepository(session).rows_for(
|
||||
portfolio_id=100, scenario_id=5, property_ids=[1]
|
||||
)
|
||||
|
||||
# assert — structured fabric replaces the prose for its five fields; windows
|
||||
# and lighting (no structured source) keep the gov prose.
|
||||
row = rows[0]
|
||||
assert row.built_form == "Mid-terrace"
|
||||
assert row.property_age_band == "1950-1966 (band D)"
|
||||
assert row.walls == "Cavity wall, filled cavity"
|
||||
assert row.roof == "Pitched roof, Access to loft; 270mm insulation at joists"
|
||||
assert row.floor == "Ground floor, solid, insulation as built"
|
||||
assert row.heating == (
|
||||
"Mains gas, radiators (condensing); programmer, room thermostat and TRVs"
|
||||
)
|
||||
assert row.hot_water == "From main system, mains gas"
|
||||
assert row.windows == "Fully double glazed"
|
||||
assert row.lighting == "Low energy lighting in all fixed outlets"
|
||||
|
||||
|
||||
def test_structured_fabric_reads_the_main_building_part_over_an_extension(
|
||||
db_engine: Engine,
|
||||
) -> None:
|
||||
# arrange — two building parts on the cert: an extension (solid brick) and
|
||||
# the main dwelling (cavity). The main part must supply the wall text.
|
||||
with Session(db_engine) as session:
|
||||
session.add(
|
||||
PropertyRow(id=1, portfolio_id=100, landlord_property_id="LP1", uprn=100001)
|
||||
)
|
||||
session.add(
|
||||
PlanModel(portfolio_id=100, property_id=1, scenario_id=5, is_default=True)
|
||||
)
|
||||
_add_epc(session, property_id=1, epc_id=1, source="lodged")
|
||||
_add_building_part(
|
||||
session,
|
||||
epc_id=1,
|
||||
identifier="extension_1",
|
||||
building_part_number=2,
|
||||
wall_construction=3, # solid brick — the extension
|
||||
)
|
||||
_add_building_part(
|
||||
session,
|
||||
epc_id=1,
|
||||
identifier="main",
|
||||
building_part_number=1,
|
||||
wall_construction=4, # cavity — the main dwelling
|
||||
)
|
||||
session.commit()
|
||||
|
||||
# act
|
||||
with Session(db_engine) as session:
|
||||
rows = ScenarioExportPostgresRepository(session).rows_for(
|
||||
portfolio_id=100, scenario_id=5, property_ids=[1]
|
||||
)
|
||||
|
||||
# assert — the main (cavity) part wins, not the extension (solid brick).
|
||||
assert rows[0].walls == "Cavity wall, filled cavity"
|
||||
|
||||
|
||||
def test_prose_remains_where_the_structured_fabric_is_absent(
|
||||
db_engine: Engine,
|
||||
) -> None:
|
||||
# arrange — a lodged EPC with gov prose but no building part and no heating
|
||||
# detail (an older gov cert, not a re-survey).
|
||||
with Session(db_engine) as session:
|
||||
session.add(
|
||||
PropertyRow(id=1, portfolio_id=100, landlord_property_id="LP1", uprn=100001)
|
||||
)
|
||||
session.add(
|
||||
PlanModel(portfolio_id=100, property_id=1, scenario_id=5, is_default=True)
|
||||
)
|
||||
_add_epc(
|
||||
session,
|
||||
property_id=1,
|
||||
epc_id=1,
|
||||
source="lodged",
|
||||
elements={"wall": "Cavity wall, as built", "roof": "Pitched, 250mm"},
|
||||
)
|
||||
session.commit()
|
||||
|
||||
# act
|
||||
with Session(db_engine) as session:
|
||||
rows = ScenarioExportPostgresRepository(session).rows_for(
|
||||
portfolio_id=100, scenario_id=5, property_ids=[1]
|
||||
)
|
||||
|
||||
# assert — with no structured fabric, the gov prose is the fallback.
|
||||
row = rows[0]
|
||||
assert row.walls == "Cavity wall, as built"
|
||||
assert row.roof == "Pitched, 250mm"
|
||||
|
||||
|
||||
def test_property_type_override_beats_the_lodged_epc(db_engine: Engine) -> None:
|
||||
# arrange — the lodged EPC says Flat, but a landlord override says House
|
||||
# (Effective-EPC precedence: override → lodged, ADR-0065).
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue