Model/domain/epc_prediction/prediction_comparison.py
Khalim Conn-Kowlessar 4afab2c3d8 feat(epc-prediction): roof-insulation +/-1-bucket reporting
Adds roof_insulation_thickness_pm1 (mirrors construction_age_band_pm1, issue
#1222): adjacent RdSAP thickness buckets (0/NI,12mm..400mm+) carry near-
identical roof U-values, so an off-by-one bucket is a SAP-neutral hit. 'ND'
(no-data) is off the ordered scale, so only an exact match counts there.
Honest measurement of SAP-relevant roof-insulation quality.

Corpus (150pc/514): exact 49.3% -> +/-1 53.7% (the misses are often multiple
buckets or ND, so the band gain is smaller than age's). Fixture: exact ==
+/-1 (0.4118) — its misses are all >1 bucket; gate floor added at 0.4118.

Also fixes two pre-existing pyright errors in the touched test file
(_epc main_fuel_type/main_heating_control were Optional but the
MainHeatingDetail attributes are non-optional Union[int, str]).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-15 14:04:18 +00:00

287 lines
10 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Per-Property prediction comparison for the EPC Prediction validation harness
(ADR-0029).
`compare_prediction` scores a predicted `EpcPropertyData` against the actual one
on the accuracy signals the leave-one-out harness aggregates: classification
matches on the key categoricals (wall / roof / floor construction + insulation,
construction age band) and residuals on the geometry (window area + count,
building-parts count, floor area). Pure — the SAP residual is computed in the
runner, which has the calculator and the lodged SAP.
"""
from __future__ import annotations
from collections import Counter
from dataclasses import dataclass
from typing import Optional
from datatypes.epc.domain.epc_property_data import (
EpcPropertyData,
MainHeatingDetail,
SapBuildingPart,
)
@dataclass(frozen=True)
class PredictionComparison:
"""One Property's prediction accuracy: per-component classification hits +
geometry residuals (predicted actual). `categorical_hits` maps a component
name to its hit: True / False, or `None` ("not applicable") when the actual
lodges no value there, so the harness can keep it out of the
classification-rate denominator rather than score a free win. Keyed by name
(not flat fields) so the component set can grow without reshaping the
runner — see ADR-0030 Component Accuracy."""
categorical_hits: dict[str, Optional[bool]]
floor_area_residual: float
building_parts_residual: int
window_count_residual: int
total_window_area_residual: float
door_count_residual: int
def _main(epc: EpcPropertyData) -> SapBuildingPart:
return epc.sap_building_parts[0]
def _main_floor_construction(epc: EpcPropertyData) -> Optional[int]:
"""The main building part's ground-floor construction code, or None when no
floor dimension is lodged."""
dims = _main(epc).sap_floor_dimensions
return dims[0].floor_construction if dims else None
def _classify(predicted: object, actual: object) -> Optional[bool]:
"""A categorical hit: None ("not applicable") when the actual is absent,
else whether the predicted value matches it."""
if actual is None:
return None
return predicted == actual
# RdSAP construction age bands, oldest → newest. Adjacent bands carry near-
# identical U-values, so an off-by-one is treated as a (SAP-neutral) ±1 hit.
_AGE_BAND_ORDER: str = "ABCDEFGHIJKL"
def _age_band_within_one(predicted: object, actual: object) -> Optional[bool]:
"""A ±1-band age hit: None when the actual is absent, True on an exact or
adjacent-band match, else False (issue #1222 — exact match overstates the
SAP impact of age-band misses)."""
if actual is None:
return None
if predicted == actual:
return True
if (
isinstance(predicted, str)
and isinstance(actual, str)
and predicted in _AGE_BAND_ORDER
and actual in _AGE_BAND_ORDER
):
return (
abs(_AGE_BAND_ORDER.index(predicted) - _AGE_BAND_ORDER.index(actual))
<= 1
)
return False
# RdSAP roof-insulation thickness buckets, thinnest → thickest. Uninsulated is
# lodged as either 0 or "NI" (not insulated), so both map to the bottom rung;
# "ND" (no data) is off the scale entirely. Adjacent buckets carry near-identical
# roof U-values, so an off-by-one bucket is treated as a (SAP-neutral) ±1 hit —
# the same measurement honesty as the construction age band (issue #1222).
_ROOF_THICKNESS_ORDINAL: dict[object, int] = {
0: 0,
"NI": 0,
"12mm": 1,
"25mm": 2,
"50mm": 3,
"75mm": 4,
"100mm": 5,
"125mm": 6,
"150mm": 7,
"175mm": 8,
"200mm": 9,
"225mm": 10,
"250mm": 11,
"270mm": 12,
"300mm": 13,
"350mm": 14,
"400mm+": 15,
}
def _roof_insulation_within_one(
predicted: object, actual: object
) -> Optional[bool]:
"""A ±1-bucket roof-insulation hit: None when the actual is absent, True on an
exact or adjacent-bucket match, else False. Off the ordered scale (e.g. the
"ND" no-data category) only an exact match counts."""
if actual is None:
return None
if predicted == actual:
return True
pred_rung = _ROOF_THICKNESS_ORDINAL.get(predicted)
actual_rung = _ROOF_THICKNESS_ORDINAL.get(actual)
if pred_rung is None or actual_rung is None:
return False
return abs(pred_rung - actual_rung) <= 1
def _main_heating_detail(epc: EpcPropertyData) -> Optional[MainHeatingDetail]:
"""The primary heating system's detail row, or None when none is lodged."""
details = epc.sap_heating.main_heating_details
return details[0] if details else None
def _heating_hits(
predicted: EpcPropertyData, actual: EpcPropertyData
) -> dict[str, Optional[bool]]:
"""Classification hits for the heating components — the dominant SAP lever
(ADR-0030). Main-system fields come off the primary `MainHeatingDetail`;
hot-water + secondary fields off `SapHeating`."""
pred_main = _main_heating_detail(predicted)
actual_main = _main_heating_detail(actual)
pred_h = predicted.sap_heating
actual_h = actual.sap_heating
return {
"heating_main_fuel": _classify(
getattr(pred_main, "main_fuel_type", None),
getattr(actual_main, "main_fuel_type", None),
),
"heating_main_category": _classify(
getattr(pred_main, "main_heating_category", None),
getattr(actual_main, "main_heating_category", None),
),
"heating_main_control": _classify(
getattr(pred_main, "main_heating_control", None),
getattr(actual_main, "main_heating_control", None),
),
"water_heating_fuel": _classify(
pred_h.water_heating_fuel, actual_h.water_heating_fuel
),
"water_heating_code": _classify(
pred_h.water_heating_code, actual_h.water_heating_code
),
"has_hot_water_cylinder": _classify(
predicted.has_hot_water_cylinder, actual.has_hot_water_cylinder
),
"cylinder_insulation_type": _classify(
pred_h.cylinder_insulation_type, actual_h.cylinder_insulation_type
),
"secondary_heating_type": _classify(
pred_h.secondary_heating_type, actual_h.secondary_heating_type
),
}
def _modal_glazing_type(epc: EpcPropertyData) -> Optional[object]:
"""The most common glazing type across the dwelling's windows, or None when
none are lodged. A single dwelling-level glazing signal, robust to one
odd window."""
types = [w.glazing_type for w in epc.sap_windows]
return Counter(types).most_common(1)[0][0] if types else None
def _has_pv(epc: EpcPropertyData) -> bool:
"""True iff the dwelling lodges any photovoltaic supply (either path)."""
source = epc.sap_energy_source
return source.photovoltaic_supply is not None or bool(
source.photovoltaic_arrays
)
def _renewables_and_fabric_hits(
predicted: EpcPropertyData, actual: EpcPropertyData
) -> dict[str, Optional[bool]]:
"""Hits for the remaining fabric-insulation, glazing and renewables
components (ADR-0030). Presence flags (room-in-roof, PV, solar) are always
applicable — predicting absence when present is a real miss."""
return {
"roof_insulation_thickness": _classify(
_main(predicted).roof_insulation_thickness,
_main(actual).roof_insulation_thickness,
),
"roof_insulation_thickness_pm1": _roof_insulation_within_one(
_main(predicted).roof_insulation_thickness,
_main(actual).roof_insulation_thickness,
),
"floor_insulation": _classify(
_main_floor_insulation(predicted), _main_floor_insulation(actual)
),
"has_room_in_roof": _classify(
_main(predicted).sap_room_in_roof is not None,
_main(actual).sap_room_in_roof is not None,
),
"modal_glazing_type": _classify(
_modal_glazing_type(predicted), _modal_glazing_type(actual)
),
"has_pv": _classify(_has_pv(predicted), _has_pv(actual)),
"solar_water_heating": _classify(
predicted.solar_water_heating, actual.solar_water_heating
),
}
def _main_floor_insulation(epc: EpcPropertyData) -> Optional[int]:
"""The main building part's ground-floor insulation code, or None when no
floor dimension is lodged."""
dims = _main(epc).sap_floor_dimensions
return dims[0].floor_insulation if dims else None
def _total_window_area(epc: EpcPropertyData) -> float:
return sum(w.window_width * w.window_height for w in epc.sap_windows)
def compare_prediction(
predicted: EpcPropertyData, actual: EpcPropertyData
) -> PredictionComparison:
"""Compare a predicted picture against the actual one, field by field. All
residuals are signed, predicted actual."""
fabric_hits: dict[str, Optional[bool]] = {
"wall_construction": _classify(
_main(predicted).wall_construction,
_main(actual).wall_construction,
),
"wall_insulation_type": _classify(
_main(predicted).wall_insulation_type,
_main(actual).wall_insulation_type,
),
"construction_age_band": _classify(
_main(predicted).construction_age_band,
_main(actual).construction_age_band,
),
"construction_age_band_pm1": _age_band_within_one(
_main(predicted).construction_age_band,
_main(actual).construction_age_band,
),
"roof_construction": _classify(
_main(predicted).roof_construction,
_main(actual).roof_construction,
),
"floor_construction": _classify(
_main_floor_construction(predicted),
_main_floor_construction(actual),
),
}
return PredictionComparison(
categorical_hits={
**fabric_hits,
**_heating_hits(predicted, actual),
**_renewables_and_fabric_hits(predicted, actual),
},
floor_area_residual=(
predicted.total_floor_area_m2 - actual.total_floor_area_m2
),
building_parts_residual=(
len(predicted.sap_building_parts) - len(actual.sap_building_parts)
),
window_count_residual=(
len(predicted.sap_windows) - len(actual.sap_windows)
),
total_window_area_residual=(
_total_window_area(predicted) - _total_window_area(actual)
),
door_count_residual=predicted.door_count - actual.door_count,
)