feat(modelling): surface each fired measure's trigger attributes

Section 3 of the report: build_property_report now runs the Modelling stage
and, for every Plan Measure, records the EPC attribute(s) that caused its
generator to fire (MeasureTrigger) — wall_construction/insulation for cavity
fill, roof thickness for loft, floor thickness/construction for floors, the
absent mechanical kind for ventilation. Modelling raises are captured as
plan_error, independent of the calculator-error capture.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-06-04 11:10:43 +00:00
parent ecebb07c9e
commit 2b04dddb06
2 changed files with 164 additions and 7 deletions

View file

@ -22,25 +22,47 @@ from __future__ import annotations
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Final, Optional
from typing import Any, Final, Optional
from datatypes.epc.domain.epc_property_data import (
BuildingPartIdentifier,
EpcPropertyData,
SapBuildingPart,
)
from datatypes.epc.domain.mapper import EpcPropertyDataMapper
from domain.modelling.plan import Plan
from domain.sap10_calculator.calculator import Sap10Calculator
from harness.console import DEFAULT_CATALOGUE, run_modelling
# A lodged-vs-calculated SAP gap beyond this many points is flagged for
# investigation (the ADR-0010/0013 shadow-validation design target).
SAP_ERROR_THRESHOLD: Final[float] = 0.5
@dataclass(frozen=True)
class MeasureTrigger:
"""One fired measure and the EPC attribute(s) that triggered its generator
the "why" behind the recommendation (e.g. cavity fill fired because
`wall_construction == 4` and `wall_insulation_type == 4`)."""
measure_type: str
triggers: dict[str, Any]
@dataclass(frozen=True)
class PropertyReport:
"""One property's inspection result. `calculator_error` records a raise
from mapping or scoring the cert (then the SAP figures are None)."""
from mapping or scoring the cert (then the SAP figures are None);
`plan_error` records a raise from the Modelling stage (then `plan` is None
and no triggers are surfaced)."""
name: str
lodged_sap: Optional[int]
calculated_sap: Optional[float]
calculator_error: Optional[str] = None
plan: Optional[Plan] = None
plan_error: Optional[str] = None
measure_triggers: tuple[MeasureTrigger, ...] = ()
@property
def sap_error(self) -> Optional[float]:
@ -57,10 +79,58 @@ class PropertyReport:
return error is not None and abs(error) > SAP_ERROR_THRESHOLD
def build_property_report(path: Path) -> PropertyReport:
"""Build one `PropertyReport` from an API-shaped EPC JSON file, comparing
its lodged SAP to our calculator's. A mapping/scoring raise is captured as
`calculator_error` rather than propagated."""
def _main_part(epc: EpcPropertyData) -> SapBuildingPart:
"""The MAIN building part the fabric generators read."""
return next(
part
for part in epc.sap_building_parts
if part.identifier is BuildingPartIdentifier.MAIN
)
def _triggers_for(epc: EpcPropertyData, measure_type: str) -> dict[str, Any]:
"""The EPC attribute(s) that caused `measure_type`'s generator to fire.
Mirrors each generator's guard so the report can explain the "why":
- cavity_wall_insulation : wall_recommendation.py (wall_construction == 4
and wall_insulation_type == 4)
- loft_insulation : roof_recommendation.py (roof_insulation_thickness == 0)
- {solid,suspended}_floor_insulation : floor_recommendation.py
(uninsulated floor_insulation_thickness + floor_construction_type)
- mechanical_ventilation : ventilation_recommendation.py (no lodged kind)
"""
main: SapBuildingPart = _main_part(epc)
if measure_type == "cavity_wall_insulation":
return {
"wall_construction": main.wall_construction,
"wall_insulation_type": main.wall_insulation_type,
}
if measure_type == "loft_insulation":
return {"roof_insulation_thickness": main.roof_insulation_thickness}
if measure_type in ("solid_floor_insulation", "suspended_floor_insulation"):
return {
"floor_insulation_thickness": main.floor_insulation_thickness,
"floor_construction_type": main.floor_construction_type,
}
if measure_type == "mechanical_ventilation":
kind: Optional[str] = (
None
if epc.sap_ventilation is None
else epc.sap_ventilation.mechanical_ventilation_kind
)
return {"mechanical_ventilation_kind": kind}
return {}
def build_property_report(
path: Path,
*,
goal_band: str = "C",
catalogue_path: Path = DEFAULT_CATALOGUE,
) -> PropertyReport:
"""Build one `PropertyReport` from an API-shaped EPC JSON file: the
lodged-vs-calculated SAP comparison, the optimised Plan, and each fired
measure's trigger attributes. A mapping/scoring raise is captured as
`calculator_error`; a Modelling raise as `plan_error`; neither propagates."""
name: str = path.stem
try:
epc = EpcPropertyDataMapper.from_api_response(json.loads(path.read_text()))
@ -73,8 +143,32 @@ def build_property_report(path: Path) -> PropertyReport:
calculated_sap=None,
calculator_error=f"{type(error).__name__}: {error}",
)
plan: Optional[Plan] = None
plan_error: Optional[str] = None
measure_triggers: tuple[MeasureTrigger, ...] = ()
try:
plan = run_modelling(
epc,
goal_band=goal_band,
catalogue_path=catalogue_path,
print_table=False,
)
measure_triggers = tuple(
MeasureTrigger(
measure_type=measure.measure_type,
triggers=_triggers_for(epc, measure.measure_type),
)
for measure in plan.measures
)
except Exception as error: # noqa: BLE001 — modelling raise must not abort the report
plan_error = f"{type(error).__name__}: {error}"
return PropertyReport(
name=name,
lodged_sap=lodged_sap,
calculated_sap=calculated_sap,
plan=plan,
plan_error=plan_error,
measure_triggers=measure_triggers,
)

View file

@ -5,7 +5,11 @@ from __future__ import annotations
import json
from pathlib import Path
from harness.report import PropertyReport, build_property_report
from harness.report import (
MeasureTrigger,
PropertyReport,
build_property_report,
)
_GOLDEN = (
Path(__file__).resolve().parents[1]
@ -18,6 +22,14 @@ _GOLDEN = (
_WITHIN_TOLERANCE = "0036-6325-1100-0063-1226"
_DIVERGENT = "0240-0200-5706-2365-8010"
# 0330 fires all three trigger kinds: an uninsulated cavity wall (cavity fill),
# its dependent mechanical ventilation, and an uninsulated solid floor.
_THREE_MEASURES = "0330-2249-8150-2326-4121"
def _triggers_by_measure(report: PropertyReport) -> dict[str, MeasureTrigger]:
return {trigger.measure_type: trigger for trigger in report.measure_triggers}
def test_calculator_error_is_lodged_minus_calculated_and_within_tolerance() -> None:
# Arrange
@ -50,6 +62,54 @@ def test_calculator_error_flags_divergence_beyond_half_a_sap_point() -> None:
assert report.sap_error_exceeds_threshold is True
def test_each_fired_measure_carries_the_attributes_that_triggered_it() -> None:
# Arrange
path: Path = _GOLDEN / f"{_THREE_MEASURES}.json"
# Act
report: PropertyReport = build_property_report(path)
# Assert — the Plan ran and every fired measure names its trigger fields.
assert report.plan is not None
assert report.plan_error is None
triggers: dict[str, MeasureTrigger] = _triggers_by_measure(report)
assert set(triggers) == {
"cavity_wall_insulation",
"mechanical_ventilation",
"solid_floor_insulation",
}
# Cavity fill fired because the main wall is an uninsulated cavity.
assert triggers["cavity_wall_insulation"].triggers == {
"wall_construction": 4,
"wall_insulation_type": 4,
}
# Ventilation fired because the dwelling lodges no mechanical kind.
assert triggers["mechanical_ventilation"].triggers == {
"mechanical_ventilation_kind": None,
}
# Solid-floor insulation fired off an uninsulated solid ground floor.
assert triggers["solid_floor_insulation"].triggers == {
"floor_insulation_thickness": None,
"floor_construction_type": "Solid",
}
def test_single_measure_cert_surfaces_only_that_measures_trigger() -> None:
# Arrange
path: Path = _GOLDEN / f"{_WITHIN_TOLERANCE}.json"
# Act
report: PropertyReport = build_property_report(path)
# Assert — 0036 fires the solid-floor measure alone.
triggers: dict[str, MeasureTrigger] = _triggers_by_measure(report)
assert set(triggers) == {"solid_floor_insulation"}
assert triggers["solid_floor_insulation"].triggers == {
"floor_insulation_thickness": None,
"floor_construction_type": "Solid",
}
def test_unparseable_cert_is_captured_not_raised(tmp_path: Path) -> None:
# Arrange — a payload the mapper rejects must not abort the report.
bad: Path = tmp_path / "broken.json"
@ -66,3 +126,6 @@ def test_unparseable_cert_is_captured_not_raised(tmp_path: Path) -> None:
assert report.sap_error_exceeds_threshold is False
assert report.calculator_error is not None
assert "ValueError" in report.calculator_error
# No Plan either — but it is recorded, not raised.
assert report.plan is None
assert report.measure_triggers == ()