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

Merge remote-tracking branch 'origin/main' into feature/hyde_make_it_more_accurate_with_tests

Browse source 
# Conflicts:
#	datatypes/epc/domain/mapper.py
This commit is contained in:
Jun-te Kim 2026-06-17 09:05:37 +00:00
commit d87718f316
46 changed files with 2413 additions and 193 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
backend/documents_parser/elmhurst_extractor.py
View file

@ -7,6 +7,7 @@ from datatypes.epc.surveys.elmhurst_site_notes import (
BathsAndShowers,
BuildingPartDimensions,
CommunityHeating,
Conservatory,
ElmhurstSiteNotes,
ExtensionPart,
FloorDetails,
@ -30,6 +31,17 @@ from datatypes.epc.surveys.elmhurst_site_notes import (
)
def _parse_conservatory_storeys(raw: Optional[str]) -> float:
"""Parse the §5 "Room Height" lodgement ("1 Storey", "1.5 Storey",
"1½ Storey") into the equivalent-storey count RdSAP 10 §6.1 translates
to a metre height. Defaults to 1.0 (single storey) when unparseable."""
if not raw:
return 1.0
text = raw.replace("½", ".5")
m = re.search(r"\d+(?:\.\d+)?", text)
return float(m.group(0)) if m else 1.0
def _parse_solar_pitch_deg(raw: Optional[str]) -> Optional[int]:
"""Parse the §16.0 "Collector elevation" lodgement (e.g. "30°", "60°",
or a bare integer). Returns None when absent or unparseable."""
@ -81,6 +93,13 @@ class ElmhurstSiteNotesExtractor:
except (ValueError, IndexError):
return 0
def _float_val(self, label: str) -> Optional[float]:
v = self._next_val(label)
if not v:
return None
m = re.search(r"-?\d+(?:\.\d+)?", v)
return float(m.group(0)) if m else None
def _date_val(self, label: str) -> date:
v = self._next_val(label)
if not v:
@ -179,8 +198,39 @@ class ElmhurstSiteNotesExtractor:
v = self._local_val(lines, label)
return v is not None and v.lower() == "yes"
def _local_float(self, lines: List[str], label: str) -> Optional[float]:
v = self._local_val(lines, label)
if not v:
return None
m = re.search(r"-?\d+(?:\.\d+)?", v)
return float(m.group(0)) if m else None
# --- section extractors ---
def _extract_conservatory(self) -> Optional[Conservatory]:
"""Summary §5.0 — geometry of a conservatory (RdSAP 10 §6, PDF
p.49). Returns None when none is lodged. Scoped to the §5 block
so the generic labels ("Floor Area", "Room Height") can't collide
with §4 dimensions. A separated conservatory is still returned
(with `thermally_separated=True`); the mapper drops it per §6.2."""
if not self._bool_val("Is there a conservatory?"):
return None
lines = self._section_lines_first_end(
"5.0 Conservatory", ("7.0 Walls", "6.0 ", "Summary Information"),
)
return Conservatory(
thermally_separated=self._local_bool(
lines, "Is it thermally separated?"
),
floor_area_m2=self._local_float(lines, "Floor Area [m2]") or 0.0,
double_glazed=self._local_bool(lines, "Double Glazed"),
glazed_perimeter_m=self._local_float(lines, "Glazed Perimeter [m]")
or 0.0,
room_height_storeys=_parse_conservatory_storeys(
self._local_val(lines, "Room Height")
),
)
def _extract_surveyor_info(self) -> SurveyorInfo:
return SurveyorInfo(
surveyor_code=self._str_val("Surveyor"),
@ -1302,6 +1352,10 @@ class ElmhurstSiteNotesExtractor:
air_permeability_ap4_m3_h_m2 = float(ap4_raw.split()[0])
except (ValueError, IndexError):
air_permeability_ap4_m3_h_m2 = None
# SAP 10.2 §2 (17) "Measured/design AP50" from a Blower Door test.
# Routes the cascade's (18) via `AP50 / 20 + (8)` (preferred over
# AP4). Absent when the test method is "Not available".
ap50_raw = self._local_float(pressure_lines, "Pressure Test Result (AP50)")
# Summary §12.1 "Mechanical Ventilation Type" — scoped to §12.1
# body so the global "Type" labels in §14 / §15 can't shadow it.
mv_lines = self._section_lines(
@ -1350,6 +1404,7 @@ class ElmhurstSiteNotesExtractor:
mechanical_ventilation=self._bool_val("Mechanical Ventilation"),
pressure_test_method=self._str_val("Test Method"),
air_permeability_ap4_m3_h_m2=air_permeability_ap4_m3_h_m2,
air_permeability_ap50_m3_h_m2=ap50_raw,
mechanical_ventilation_type=mechanical_ventilation_type,
mechanical_ventilation_pcdf_reference=mev_pcdf_reference,
wet_rooms_count=wet_rooms_count,
@ -1820,6 +1875,7 @@ class ElmhurstSiteNotesExtractor:
construction_age_band=self._extract_main_age_band(),
dimensions=self._extract_dimensions(),
has_conservatory=self._bool_val("Is there a conservatory?"),
conservatory=self._extract_conservatory(),
walls=self._extract_walls(),
roof=self._extract_roof(),
floor=self._extract_floor(),

BIN
backend/documents_parser/tests/fixtures/Summary_001431_case44.pdf vendored Normal file
View file

Binary file not shown.

72
backend/epc_api/json_samples/real_life_examples/SAP-Schema-17.1/uprn_10093116324/elmhurst_inputs.md Normal file
View file

@ -0,0 +1,72 @@
# Elmhurst RdSAP inputs — UPRN 10093116324 (SAP-Schema-17.1)
**Lodged SAP:** 80 **Our engine:** 79 ← compare Elmhurst against the engine
**Property:** semi-detached BUNGALOW (single storey), 2017, mains-gas combi, TFA 52 m²
**Known divergences (same full-SAP→RdSAP pattern as 10093116543/529):**
- Cert lodges measured U (wall 0.19, floor 0.12, roof 0.12); engine uses them, Elmhurst RdSAP forces band-L defaults. Expect engine a few points ABOVE Elmhurst.
- Boiler PCDB 17505 (88.5%) → Elmhurst generic BGW combi 84% (PCDB search disabled).
- Age band engine 'M' → Elmhurst on-screen **L (2012-2022)** (2017 build).
## Property Description
| Field | Value | Notes |
|---|---|---|
| Property Type | House | dwelling_type "Semi-detached bungalow" |
| Built form | Semi-Detached | built_form 2 |
| Age band | **L (2012-2022)** | 2017 |
| Storeys | **1** | bungalow (single storey) |
| Habitable rooms | 2 | |
## Dimensions (single storey)
| Field | Value |
|---|---|
| Lowest-floor area | 51.90 m² |
| Room height | 2.40 m |
| Heat-loss perimeter | 45.53 m |
| Party-wall length | 6.37 m |
## Walls
| Field | Value | Notes |
|---|---|---|
| Construction | Cavity / As Built | wall_construction 4; cert measured U 0.19 (Elmhurst RdSAP default) |
| Party wall | Present, **U Unable to determine** | type-4 party wall area 15.29; party_w/k 3.82 |
## Roofs
| Field | Value | Notes |
|---|---|---|
| Type | Pitched, access to loft / Joists / Unknown thickness | roof measured U 0.12; accept band-L default |
## Floors
| Field | Value | Notes |
|---|---|---|
| Ground / Solid / As built | — | floor measured U 0.12 |
## Openings
| Field | Value | Notes |
|---|---|---|
| Windows (combined) | **6.11 m²**, East, Double post-2022 | 2 synth windows; U 1.4 / g 0.72 |
| Doors | 2 insulated, U 1.19 | door_count 2 |
## Ventilation & Lighting
| Field | Value | Notes |
|---|---|---|
| Air Pressure Test | **Blower Door 2.51** (+ cert number) | AP50 2.51 |
| Extract fans | 2 | |
| Sheltered sides | 2 | |
| Lighting | 100% low-energy | 10/10 |
## Space Heating / Water
| Field | Value | Notes |
|---|---|---|
| Main heating | Mains gas condensing combi (generic BGW) | PCDB 17505 (88.5%, can't enter), radiators, fan flue |
| Water heating | From main (combi), no cylinder | has_hot_water_cylinder False |
| Secondary | None | |
## Fields to clear (do NOT map)
| Field | Set to | Why |
|---|---|---|
| 1st-floor dimensions | 0 / blank | single-storey bungalow (prior cert was 2-storey/end-terrace) |
| Flats page | (House) | — |
| Conservatory | none/uncheck | not lodged |
| Cylinder | none | combi |
| PV / Wind / secondary | none | none lodged |

35
datatypes/epc/domain/epc_property_data.py
View file

@ -208,6 +208,10 @@ class SapVentilation:
# Pulse pressure test, m³/h per m² of envelope area. When present the
# cascade routes (18) via the AP4 formula `0.263 × AP4^0.924 + (8)`.
air_permeability_ap4_m3_h_m2: Optional[float] = None
# SAP 10.2 §2 (17) — air permeability at 50 Pa from a Blower Door test,
# m³/h per m² of envelope area. When present the cascade routes (18)
# via `AP50 / 20 + (8)` (preferred over AP4).
air_permeability_ap50_m3_h_m2: Optional[float] = None
# SAP 10.2 §2 (23a)/(24a..d) — Elmhurst "Mechanical Ventilation Type"
# string mapped to the `MechanicalVentilationKind` enum name (e.g.
# "EXTRACT_OR_PIV_OUTSIDE" for MEV decentralised). The cascade uses
@ -267,6 +271,33 @@ class SapRoofWindow:
window_location: Union[int, str] = 0
@dataclass(frozen=True)
class SapConservatory:
"""RdSAP 10 §6.1 (PDF p.49) — a NON-SEPARATED (heated) conservatory.
Its floor area and volume are added to the dwelling total (TFA (4),
volume (5)); its fully-glazed walls bill as a window (27) and its
fully-glazed roof as a rooflight (27a); the floor adds a ground-loss
term (28a). U-values come from RdSAP 10 Table 25 (p.51): double 6 mm
window 3.1 / roof 3.4 / g 0.76; single window 4.8 / roof 5.3 / g 0.85.
`room_height_storeys` is the equivalent number of storey heights of
the dwelling to the nearest half (Summary §5 "Room Height", gov-API
glazed building part), translated to a metre height per §6.1:
1 storey → ground-floor room height; 1½ → ground + 0.25 + 0.5×first;
2 → ground + 0.25 + first; etc.
A SEPARATED conservatory (§6.2) is disregarded entirely and is never
represented here (`thermally_separated` stays a guard for the cascade).
"""
floor_area_m2: float
glazed_perimeter_m: float
double_glazed: bool
thermally_separated: bool
room_height_storeys: float
@dataclass
class SapWindow:
frame_material: Optional[str]
@ -780,6 +811,10 @@ class EpcPropertyData:
# has no roof windows; for cert-cascade fixtures the bootstrap path
# lodges per-window area + raw U.
sap_roof_windows: Optional[List[SapRoofWindow]] = None
# RdSAP 10 §6.1 — geometry of a non-separated (heated) conservatory.
# None when no conservatory is lodged or it is thermally separated
# (§6.2 disregards separated conservatories).
sap_conservatory: Optional[SapConservatory] = None
calculation_software_version: Optional[str] = None # Do we care about this?
mechanical_vent_duct_placement: Optional[int] = None
mechanical_vent_duct_insulation: Optional[int] = None

311
datatypes/epc/domain/mapper.py
View file

@ -6,6 +6,7 @@ from decimal import ROUND_HALF_UP, Decimal
from typing import Any, Dict, Final, List, Optional, Sequence, Tuple, TypeVar, Union, cast
from datatypes.epc.schema.helpers import from_dict
from datatypes.epc.domain.epc import Epc
from datatypes.epc.domain.epc_property_data import (
BASEMENT_WALL_CONSTRUCTION_CODE,
Addendum,
@ -25,6 +26,7 @@ from datatypes.epc.domain.epc_property_data import (
SapEnergySource,
SapFlatDetails,
SapFloorDimension,
SapConservatory,
SapHeating,
SapRoofWindow,
SapRoomInRoof,
@ -111,6 +113,7 @@ from datatypes.epc.surveys.elmhurst_site_notes import (
AlternativeWall as ElmhurstAlternativeWall,
BuildingPartDimensions as ElmhurstBuildingPartDimensions,
CommunityHeating,
Conservatory as ElmhurstConservatory,
ElmhurstSiteNotes,
FloorDetails as ElmhurstFloorDetails,
MainHeating as ElmhurstMainHeating,
@ -481,12 +484,25 @@ class EpcPropertyDataMapper:
ext.room_in_roof.floor_area_m2
for ext in survey.extensions
if ext.room_in_roof is not None
)
# RdSAP 10 §6.1 (PDF p.49) — a non-separated conservatory's
# floor area is added to the dwelling total floor area. TFA
# drives occupancy → §5 internal gains + §4 hot-water demand,
# so it must include the conservatory (the worksheet's (4) =
# 95.38 carries it). Separated conservatories (§6.2) are
# disregarded.
+ (
survey.conservatory.floor_area_m2
if survey.conservatory is not None
and not survey.conservatory.thermally_separated
else 0.0
),
2,
),
built_form=built_form,
property_type=property_type,
has_conservatory=survey.has_conservatory,
sap_conservatory=_map_elmhurst_conservatory(survey.conservatory),
blocked_chimneys_count=survey.ventilation.blocked_chimneys_count,
number_of_storeys=survey.number_of_storeys,
hydro=survey.renewables.hydro_electricity_generated_kwh > 0,
@ -2323,8 +2339,14 @@ class EpcPropertyDataMapper:
else None
),
)
# RdSAP 10 §6.1 — exclude the glazed conservatory BP from the
# fabric loop; it is carried as `sap_conservatory` below and
# billed by the §6.1 cascade (window/rooflight/floor), not as
# a dwelling building part.
for bp in schema.sap_building_parts
if getattr(bp, "glazed_perimeter", None) is None
],
sap_conservatory=_api_sap_conservatory(schema.sap_building_parts),
renewable_heat_incentive=RenewableHeatIncentive(
space_heating_kwh=float(
schema.renewable_heat_incentive.space_heating_existing_dwelling
@ -2517,71 +2539,56 @@ class EpcPropertyDataMapper:
if schema == "RdSAP-Schema-21.0.1":
from datatypes.epc.schema.rdsap_schema_21_0_1 import RdSapSchema21_0_1
return _clear_basement_flag_when_system_built(
EpcPropertyDataMapper.from_rdsap_schema_21_0_1(
from_dict(RdSapSchema21_0_1, data)
)
mapped = EpcPropertyDataMapper.from_rdsap_schema_21_0_1(
from_dict(RdSapSchema21_0_1, data)
)
if schema == "RdSAP-Schema-21.0.0":
elif schema == "RdSAP-Schema-21.0.0":
from datatypes.epc.schema.rdsap_schema_21_0_0 import RdSapSchema21_0_0
return _clear_basement_flag_when_system_built(
EpcPropertyDataMapper.from_rdsap_schema_21_0_0(
from_dict(RdSapSchema21_0_0, data)
)
mapped = EpcPropertyDataMapper.from_rdsap_schema_21_0_0(
from_dict(RdSapSchema21_0_0, data)
)
if schema == "RdSAP-Schema-20.0.0":
elif schema == "RdSAP-Schema-20.0.0":
from datatypes.epc.schema.rdsap_schema_20_0_0 import RdSapSchema20_0_0
return _clear_basement_flag_when_system_built(
EpcPropertyDataMapper.from_rdsap_schema_20_0_0(
from_dict(RdSapSchema20_0_0, data)
)
mapped = EpcPropertyDataMapper.from_rdsap_schema_20_0_0(
from_dict(RdSapSchema20_0_0, data)
)
if schema == "RdSAP-Schema-19.0":
elif schema == "RdSAP-Schema-19.0":
from datatypes.epc.schema.rdsap_schema_19_0 import RdSapSchema19_0
return _clear_basement_flag_when_system_built(
EpcPropertyDataMapper.from_rdsap_schema_19_0(
from_dict(RdSapSchema19_0, data)
)
mapped = EpcPropertyDataMapper.from_rdsap_schema_19_0(
from_dict(RdSapSchema19_0, data)
)
if schema == "RdSAP-Schema-18.0":
elif schema == "RdSAP-Schema-18.0":
from datatypes.epc.schema.rdsap_schema_18_0 import RdSapSchema18_0
return _clear_basement_flag_when_system_built(
EpcPropertyDataMapper.from_rdsap_schema_18_0(
from_dict(RdSapSchema18_0, data)
)
mapped = EpcPropertyDataMapper.from_rdsap_schema_18_0(
from_dict(RdSapSchema18_0, data)
)
if schema == "RdSAP-Schema-17.1":
elif schema == "RdSAP-Schema-17.1":
from datatypes.epc.schema.rdsap_schema_17_1 import RdSapSchema17_1
return _clear_basement_flag_when_system_built(
EpcPropertyDataMapper.from_rdsap_schema_17_1(
from_dict(RdSapSchema17_1, data)
)
mapped = EpcPropertyDataMapper.from_rdsap_schema_17_1(
from_dict(RdSapSchema17_1, data)
)
if schema == "RdSAP-Schema-17.0":
elif schema == "RdSAP-Schema-17.0":
from datatypes.epc.schema.rdsap_schema_17_0 import RdSapSchema17_0
return _clear_basement_flag_when_system_built(
EpcPropertyDataMapper.from_rdsap_schema_17_0(
from_dict(RdSapSchema17_0, data)
)
mapped = EpcPropertyDataMapper.from_rdsap_schema_17_0(
from_dict(RdSapSchema17_0, data)
)
if schema in ("SAP-Schema-17.1", "SAP-Schema-17.0", "SAP-Schema-18.0.0"):
elif schema in ("SAP-Schema-17.1", "SAP-Schema-17.0", "SAP-Schema-18.0.0"):
# Full SAP (not RdSAP). SAP-Schema-17.0 / 18.0.0 are structurally
# identical to 17.1 (same measured sap_opening_types / building
# parts), so they parse with the 17.1 dataclass and reuse the same
# mapper. D8:
# _clear_basement_flag_when_system_built is an RdSAP code-6
# disambiguation; full SAP lodges explicit wall types (no code-6
# basement ambiguity), so it's a no-op and is skipped.
return EpcPropertyDataMapper.from_sap_schema_17_1(
# mapper. The common post-processing below (incl.
# _clear_basement_flag_when_system_built) is a no-op for full SAP
# (explicit wall types — no RdSAP code-6 basement ambiguity).
mapped = EpcPropertyDataMapper.from_sap_schema_17_1(
from_dict(SapSchema17_1, data)
)
if schema in ("SAP-Schema-16.2", "SAP-Schema-16.3", "SAP-Schema-16.0"):
elif schema in ("SAP-Schema-16.2", "SAP-Schema-16.3", "SAP-Schema-16.0"):
# The SAP-Schema-16.x family is structurally RdSAP-17.1 (reduced
# fields, glazed_area band, construction-code building parts) under a
# different name + a handful of renamed/omitted fields — normalise
@ -2589,13 +2596,17 @@ class EpcPropertyDataMapper:
# `_normalize_sap_schema_16_x`.
from datatypes.epc.schema.rdsap_schema_17_1 import RdSapSchema17_1
return _clear_basement_flag_when_system_built(
EpcPropertyDataMapper.from_rdsap_schema_17_1(
from_dict(RdSapSchema17_1, _normalize_sap_schema_16_x(data))
)
mapped = EpcPropertyDataMapper.from_rdsap_schema_17_1(
from_dict(RdSapSchema17_1, _normalize_sap_schema_16_x(data))
)
else:
raise ValueError(f"Unsupported EPC schema: {schema!r}")
raise ValueError(f"Unsupported EPC schema: {schema!r}")
return _clear_basement_flag_when_system_built(
_with_renewable_heat_incentive(
_with_recorded_performance(mapped, data), data
)
)
# ---------------------------------------------------------------------------
@ -2708,6 +2719,41 @@ def _sap_back_solved_habitable_rooms(schema: SapSchema17_1) -> int:
)
def _with_recorded_performance(
epc: EpcPropertyData, data: Dict[str, Any]
) -> EpcPropertyData:
"""Overlay the recorded current-performance scalars from the raw API payload.
The current SAP rating, EPC band, Primary Energy Intensity
(``energy_consumption_current``) and CO2 are top-level fields on every RdSAP
schema response, but only a couple of the per-schema mappers copy them
through (and none map the band). Baseline's Lodged Performance reads all four
off the EPC, so map them here, once, for every schema version. An absent key
leaves the mapped value untouched.
"""
band = data.get("current_energy_efficiency_band")
co2 = data.get("co2_emissions_current")
consumption = data.get("energy_consumption_current")
rating = data.get("energy_rating_current")
return replace(
epc,
current_energy_efficiency_band=(
Epc(band) if band is not None else epc.current_energy_efficiency_band
),
co2_emissions_current=(
float(co2) if co2 is not None else epc.co2_emissions_current
),
energy_consumption_current=(
int(consumption)
if consumption is not None
else epc.energy_consumption_current
),
energy_rating_current=(
int(rating) if rating is not None else epc.energy_rating_current
),
)
def _sap_17_1_building_part(
bp: SapBuildingPart_SAP_17_1, index: int
) -> SapBuildingPart:
@ -2809,6 +2855,50 @@ def _sap_door_aggregates(schema: SapSchema17_1) -> Tuple[int, Optional[float]]:
return count, u_value
def _with_renewable_heat_incentive(
epc: EpcPropertyData, data: Dict[str, Any]
) -> EpcPropertyData:
"""Gap-fill the RHI block (baseline space/water-heating kWh) from the raw
payload.
The ``renewable_heat_incentive`` object is present on every schema response,
but only the 21.x mappers copy it through; Baseline reads
``space_heating_kwh`` / ``water_heating_kwh`` off it. Only fills when a mapper
left it unset, and only when the block carries both required kWh figures —
otherwise the EPC is returned untouched.
"""
if epc.renewable_heat_incentive is not None:
return epc
rhi = data.get("renewable_heat_incentive")
if not isinstance(rhi, dict):
return epc
rhi_obj = cast(Dict[str, Any], rhi)
space = rhi_obj.get("space_heating_existing_dwelling")
water = rhi_obj.get("water_heating")
if space is None or water is None:
return epc
return replace(
epc,
renewable_heat_incentive=RenewableHeatIncentive(
space_heating_kwh=float(space),
water_heating_kwh=float(water),
impact_of_loft_insulation_kwh=_optional_float(
rhi_obj.get("impact_of_loft_insulation")
),
impact_of_cavity_insulation_kwh=_optional_float(
rhi_obj.get("impact_of_cavity_insulation")
),
impact_of_solid_wall_insulation_kwh=_optional_float(
rhi_obj.get("impact_of_solid_wall_insulation")
),
),
)
def _optional_float(value: Any) -> Optional[float]:
return float(value) if value is not None else None
def _clear_basement_flag_when_system_built(
epc: EpcPropertyData,
) -> EpcPropertyData:
@ -2890,6 +2980,33 @@ def _measurement_value(field: Any) -> float:
return float(field)
def _api_sap_conservatory(building_parts: Any) -> Optional[SapConservatory]:
"""Build the domain `SapConservatory` from the gov-API glazed
conservatory building part — the part the API uses for a NON-SEPARATED
conservatory (RdSAP 10 §6.1, PDF p.49), identified by a lodged
`glazed_perimeter` (real dwelling parts carry fabric + floor dimensions
instead, never `glazed_perimeter`). Only type-4 (non-separated)
conservatories lodge this BP; separated ones (§6.2) lodge nothing, so
its presence is the §6.1 signal. Mirror of `_map_elmhurst_conservatory`
for the API path — proven equivalent by cross-mapper parity (the cascade
reads `epc.sap_conservatory` identically). Returns None when absent."""
if not building_parts:
return None
for bp in building_parts:
if getattr(bp, "glazed_perimeter", None) is None:
continue
return SapConservatory(
floor_area_m2=_measurement_value(bp.floor_area),
glazed_perimeter_m=_measurement_value(bp.glazed_perimeter),
double_glazed=bp.double_glazed == "Y",
# The gov API only lodges this glazed BP for NON-separated
# (type-4) conservatories; separated ones (§6.2) lodge no BP.
thermally_separated=False,
room_height_storeys=float(_measurement_value(bp.room_height)),
)
return None
def _total_floor_area_from_building_parts(building_parts: Any) -> Optional[float]:
"""Sum per-bp `sap_floor_dimensions[*].total_floor_area` (plus each bp's
`sap_room_in_roof.floor_area` when present) to recover the precise
@ -2914,6 +3031,13 @@ def _total_floor_area_from_building_parts(building_parts: Any) -> Optional[float
total = 0.0
found = False
for bp in building_parts:
# RdSAP 10 §6.1 — a non-separated conservatory's glazed BP (no floor
# dimensions) adds its floor area to the dwelling TFA. TFA drives
# occupancy → §4/§5 demand, so the conservatory must be in the sum.
if getattr(bp, "glazed_perimeter", None) is not None:
total += _measurement_value(bp.floor_area)
found = True
continue
floor_dims: Any = bp.sap_floor_dimensions or []
for fd in floor_dims:
total += _measurement_value(fd.total_floor_area)
@ -4492,12 +4616,88 @@ def _api_build_room_in_roof(
# §3.9.1 default RR storey height (2.45 m); the type code routes
# the U-value (Exposed → main-wall U, Party → 0.25).
rir.detailed_surfaces = _api_type_1_gable_surfaces(type_1)
type_2 = getattr(bp_rir, "room_in_roof_type_2", None)
if type_2 is not None:
rir.detailed_surfaces = _api_type_2_surfaces(type_2)
details = getattr(bp_rir, "room_in_roof_details", None)
if details is not None:
rir.detailed_surfaces = _api_rir_detailed_surfaces(details, is_flat=is_flat)
return rir
def _api_type_2_surfaces(
type_2: Any,
) -> Optional[List[SapRoomInRoofSurface]]:
"""Translate the §3.9.2 Simplified Type 2 block into the per-surface
list the cascade's Detailed-RR branch consumes — MIRRORING the
worksheet-validated Summary path (`_map_elmhurst_rir_surface`,
is_simplified, validated to 1e-4 by cohort cert 000565). Unlike the
Type 1 block (gable lengths only, billed raw L × 2.45), Type 2 lodges
gable + common-wall lengths AND heights, so the spec's §3.9.2 areas
apply:
common wall → `L × (0.25 + H)` (billed at uw)
gable → `L × (0.25 + H_gable)
− Σ_each_common (H_gable − H_common,n)² / 2`
The gable correction is taken over ALL common walls for an exposed/
party/sheltered gable (the worksheet evaluates it literally, incl. the
H_gable=0 absent-gable case → a negative area that deducts from the
A_RR residual without billing a physical wall); a Connected gable
(Table 4 row 4, U=0) sums only the common walls it overtops, matching
the Summary's connected-gable branch. The `gable_wall_type_*` code
routes the kind (0 Party / 1 Exposed / 2 Sheltered / 3 Connected) via
`_api_type_1_gable_kind`; U-values are left to the cascade (no per-
gable U is lodged on the API path)."""
cw_heights = [
float(h)
for length, h in (
(type_2.common_wall_length_1, type_2.common_wall_height_1),
(type_2.common_wall_length_2, type_2.common_wall_height_2),
)
if length is not None and h is not None and length > 0 and h > 0
]
surfaces: List[SapRoomInRoofSurface] = []
gable_specs = (
(type_2.gable_wall_type_1, type_2.gable_wall_length_1,
type_2.gable_wall_height_1),
(type_2.gable_wall_type_2, type_2.gable_wall_length_2,
type_2.gable_wall_height_2),
)
for gable_type, length, height in gable_specs:
# Length is mandatory; H may be 0 for the §3.9.2 absent-gable
# quadratic (only when common walls drive the correction).
if length is None or length <= 0 or height is None:
continue
if height <= 0 and not cw_heights:
continue
kind = _api_type_1_gable_kind(gable_type)
length_m, height_m = float(length), float(height)
if cw_heights:
if kind == "connected_wall":
correction = sum(
((height_m - h) ** 2) / 2.0 for h in cw_heights if height_m > h
)
else:
correction = sum(((height_m - h) ** 2) / 2.0 for h in cw_heights)
area = _round_half_up_2dp(1.0, length_m * (0.25 + height_m) - correction)
else:
area = _round_half_up_2dp(length_m, height_m)
surfaces.append(SapRoomInRoofSurface(kind=kind, area_m2=area))
common_specs = (
(type_2.common_wall_length_1, type_2.common_wall_height_1),
(type_2.common_wall_length_2, type_2.common_wall_height_2),
)
for length, height in common_specs:
if length is None or height is None or length <= 0 or height <= 0:
continue
surfaces.append(
SapRoomInRoofSurface(
kind="common_wall",
area_m2=_round_half_up_2dp(float(length), 0.25 + float(height)),
)
)
return surfaces or None
def _api_rir_detailed_surfaces(
details: Any,
*,
@ -5711,6 +5911,24 @@ def _elmhurst_roof_window_u_value(w: ElmhurstWindow) -> float:
return w.u_value + _ELMHURST_ROOF_WINDOW_INCLINATION_ADJUSTMENT_W_PER_M2K
def _map_elmhurst_conservatory(
cons: Optional[ElmhurstConservatory],
) -> Optional[SapConservatory]:
"""RdSAP 10 §6 — translate the Summary §5 conservatory geometry into
the domain `SapConservatory`. A SEPARATED conservatory (§6.2, PDF
p.49) is disregarded entirely, so it maps to None (the cascade adds
nothing). Returns None when no conservatory is lodged."""
if cons is None or cons.thermally_separated:
return None
return SapConservatory(
floor_area_m2=cons.floor_area_m2,
glazed_perimeter_m=cons.glazed_perimeter_m,
double_glazed=cons.double_glazed,
thermally_separated=cons.thermally_separated,
room_height_storeys=cons.room_height_storeys,
)
def _map_elmhurst_roof_window(w: ElmhurstWindow) -> SapRoofWindow:
return SapRoofWindow(
area_m2=w.area_m2,
@ -7382,5 +7600,6 @@ def _map_elmhurst_ventilation(
else (False if has_suspended_timber_floor else None)
),
air_permeability_ap4_m3_h_m2=v.air_permeability_ap4_m3_h_m2,
air_permeability_ap50_m3_h_m2=v.air_permeability_ap50_m3_h_m2,
mechanical_ventilation_kind=_elmhurst_mv_kind(v.mechanical_ventilation_type),
)

129
datatypes/epc/domain/tests/test_from_rdsap_schema.py
View file

@ -2228,3 +2228,132 @@ class TestRoomInRoofDetailedSlopeAndStudWall:
assert studs[0].insulation_thickness_mm == 75
assert abs(commons[0].area_m2 - 10.32) <= 1e-9
assert commons[0].insulation_thickness_mm is None
class TestRoomInRoofType2SimplifiedQuadratic:
"""RdSAP 10 §3.9.2 Simplified Type 2 RR — the gov API lodges gable +
common-wall lengths AND heights under `room_in_roof_type_2`. The block
was undeclared → dropped → the cascade billed the whole A_RR shell at
the Table-18-col-4 default (over-count → under-rate, 7 corpus certs at
signed −5.02). The mapper now MIRRORS the worksheet-validated Summary
§3.9.2 areas (cross-mapper parity, proven identical on cohort cert
000565): common walls L×(0.25+H), gables L×(0.25+H) − Σ(H−H_cw)²/2."""
def test_from_api_response_applies_3_9_2_gable_quadratic(self) -> None:
# Arrange — two common walls (L=8, H=1 → cw_heights [1,1]); an
# exposed gable (L=10, H=2) and a party gable (L=6, H=2).
# common wall = round(8 × (0.25+1)) = 10.00
# exposed gable= round(10 × (0.25+2) − 2×(2−1)²/2) = round(22.5−1) = 21.50
# party gable = round(6 × (0.25+2) − 1.0) = round(13.5−1) = 12.50
cert = load("21_0_1.json")
rir = cert["sap_building_parts"][0]["sap_room_in_roof"]
rir.pop("room_in_roof_type_1", None)
rir["room_in_roof_type_2"] = {
"gable_wall_type_1": 1, "gable_wall_length_1": 10.0,
"gable_wall_height_1": 2.0,
"gable_wall_type_2": 0, "gable_wall_length_2": 6.0,
"gable_wall_height_2": 2.0,
"common_wall_length_1": 8.0, "common_wall_height_1": 1.0,
"common_wall_length_2": 8.0, "common_wall_height_2": 1.0,
}
# Act
result = EpcPropertyDataMapper.from_api_response(cert)
# Assert
rir_part = result.sap_building_parts[0].sap_room_in_roof
assert rir_part is not None
surfaces = rir_part.detailed_surfaces
assert surfaces is not None
ext = [s for s in surfaces if s.kind == "gable_wall_external"]
party = [s for s in surfaces if s.kind == "gable_wall"]
commons = [s for s in surfaces if s.kind == "common_wall"]
assert len(ext) == 1 and abs(ext[0].area_m2 - 21.50) <= 1e-9
assert len(party) == 1 and abs(party[0].area_m2 - 12.50) <= 1e-9
assert len(commons) == 2
assert abs(commons[0].area_m2 - 10.00) <= 1e-9
class TestNonSeparatedConservatoryApiMirror:
"""RdSAP 10 §6.1 (PDF p.49) — the gov API lodges a NON-SEPARATED
conservatory (conservatory_type=4) as a glazed "building part" carrying
only {floor_area, room_height, double_glazed, glazed_perimeter}. The
block was undeclared → `from_dict` dropped it → the conservatory was
silently lost (5 corpus certs over-rating). The mapper now splits it
into `EpcPropertyData.sap_conservatory`, excludes it from the fabric
building-part loop, and adds its floor area to TFA.
Validation is cross-mapper parity, NOT a corpus back-solve: the API
mapper feeds the SAME worksheet-validated §6.1 cascade
(`conservatory_geometry`, pinned to 1e-4 against the case-44 Summary)
as the Elmhurst path — so the API conservatory fabric is correct by
construction."""
def test_from_api_response_splits_out_conservatory_building_part(
self,
) -> None:
# Arrange — a 1-BP dwelling (ground-floor room height 2.45 m) plus a
# non-separated double-glazed conservatory glazed BP.
from datatypes.epc.domain.epc_property_data import SapConservatory
from domain.sap10_calculator.worksheet.conservatory import (
conservatory_geometry,
)
baseline_tfa = EpcPropertyDataMapper.from_api_response(
load("21_0_1.json")
).total_floor_area_m2
cert = load("21_0_1.json")
cert["conservatory_type"] = 4
cert["sap_building_parts"].append(
{
"floor_area": 12.0,
"room_height": 1,
"double_glazed": "Y",
"glazed_perimeter": 9.0,
}
)
# Act
epc = EpcPropertyDataMapper.from_api_response(cert)
# Assert — conservatory split out; the glazed BP is NOT a fabric part.
assert epc.sap_conservatory == SapConservatory(
floor_area_m2=12.0,
glazed_perimeter_m=9.0,
double_glazed=True,
thermally_separated=False,
room_height_storeys=1.0,
)
assert len(epc.sap_building_parts) == 1
# §6.1: the conservatory floor area joins TFA (drives occupancy).
assert abs(epc.total_floor_area_m2 - (baseline_tfa + 12.0)) <= 1e-9
# Cross-mapper parity: the shared §6.1 cascade derives the same
# surfaces it does for the case-44 Summary — glazed wall = exposed
# perimeter × ground-floor room height (9.0 × 2.45 = 22.05); glazed
# roof = floor / cos(20°) (12.0 / 0.9397 = 12.77); Table 25 double
# U_eff = 1/(1/3.1 + 0.04) = 2.758 (wall) / 1/(1/3.4 + 0.04) = 2.993.
geom = conservatory_geometry(epc)
assert geom is not None
assert abs(geom.glazed_wall_area_m2 - 22.05) <= 1e-4
assert abs(geom.glazed_roof_area_m2 - 12.77) <= 1e-4
assert abs(geom.wall_u_eff - 2.7580) <= 1e-4
assert abs(geom.roof_u_eff - 2.9930) <= 1e-4
def test_separated_conservatory_lodges_no_glazed_building_part(self) -> None:
# Arrange — a separated conservatory (type 2/3) lodges NO glazed BP
# (verified across the gov corpus); the dwelling is unchanged.
from domain.sap10_calculator.worksheet.conservatory import (
conservatory_geometry,
)
cert = load("21_0_1.json")
cert["conservatory_type"] = 2
# Act
epc = EpcPropertyDataMapper.from_api_response(cert)
# Assert — §6.2: disregarded; no conservatory geometry.
assert epc.sap_conservatory is None
assert conservatory_geometry(epc) is None

17
datatypes/epc/schema/rdsap_schema_21_0_0.py
View file

@ -236,12 +236,29 @@ class RoomInRoofDetails:
common_wall_height_2: Optional[float] = None
@dataclass
class RoomInRoofType2:
"""RdSAP §3.9.2 Simplified Type 2 RR — gable + common-wall geometry.
See `rdsap_schema_21_0_1.RoomInRoofType2`. Previously dropped."""
gable_wall_type_1: Optional[int] = None
gable_wall_type_2: Optional[int] = None
gable_wall_length_1: Optional[float] = None
gable_wall_length_2: Optional[float] = None
gable_wall_height_1: Optional[float] = None
gable_wall_height_2: Optional[float] = None
common_wall_length_1: Optional[float] = None
common_wall_length_2: Optional[float] = None
common_wall_height_1: Optional[float] = None
common_wall_height_2: Optional[float] = None
@dataclass
class SapRoomInRoof:
"""Room-in-roof details. insulation and roof_room_connected removed in schema 21.0.0."""
floor_area: Union[int, float]
construction_age_band: str
room_in_roof_type_1: Optional[RoomInRoofType1] = None
room_in_roof_type_2: Optional[RoomInRoofType2] = None
room_in_roof_details: Optional[RoomInRoofDetails] = None

40
datatypes/epc/schema/rdsap_schema_21_0_1.py
View file

@ -278,14 +278,38 @@ class RoomInRoofDetails:
common_wall_height_2: Optional[float] = None
@dataclass
class RoomInRoofType2:
"""RdSAP §3.9.2 Simplified Type 2 RR — a room-in-roof bounded by
continuous common walls (accessible common-wall height < 1.8 m, so the
space counts as RR not a separate storey). Lodges gable + common-wall
lengths AND heights (unlike Type 1, gable lengths only). `gable_wall_
type_*` is the Table 4 variant (0 Party / 1 Exposed / 2 Sheltered /
3 Connected). Previously undeclared → dropped by `from_dict`, so the
cascade billed the whole A_RR shell at the Table-18-col-4 default
(over-count → under-rate)."""
gable_wall_type_1: Optional[int] = None
gable_wall_type_2: Optional[int] = None
gable_wall_length_1: Optional[float] = None
gable_wall_length_2: Optional[float] = None
gable_wall_height_1: Optional[float] = None
gable_wall_height_2: Optional[float] = None
common_wall_length_1: Optional[float] = None
common_wall_length_2: Optional[float] = None
common_wall_height_1: Optional[float] = None
common_wall_height_2: Optional[float] = None
@dataclass
class SapRoomInRoof:
floor_area: Union[int, float]
construction_age_band: str
# Two real-API shapes coexist: older certs (cohort 6035, 0240, test
# fixture 21_0_1.json) lodge the Simplified Type 1 wrapper; newer
# certs (9501) lodge the Detailed-RR block. Accept both.
# Three real-API shapes coexist: older certs (cohort 6035, 0240, test
# fixture 21_0_1.json) lodge the Simplified Type 1 wrapper; some lodge
# the §3.9.2 Simplified Type 2 wrapper (gable + common-wall geometry);
# newer certs (9501) lodge the Detailed-RR block. Accept all three.
room_in_roof_type_1: Optional[RoomInRoofType1] = None
room_in_roof_type_2: Optional[RoomInRoofType2] = None
room_in_roof_details: Optional[RoomInRoofDetails] = None
@ -358,6 +382,16 @@ class SapBuildingPart:
# redacts the backing insulation. Previously undeclared → dropped.
wall_u_value: Optional[float] = None
floor_u_value: Optional[float] = None
# RdSAP 10 §6.1 (PDF p.49) — a NON-SEPARATED conservatory is lodged by
# the gov API as a glazed "building part" carrying ONLY these four
# fields (no fabric, no floor dimensions); `conservatory_type == 4` at
# the property level. Previously undeclared → dropped by `from_dict`,
# so the conservatory was silently lost on the API path. The mapper
# splits this BP out into `EpcPropertyData.sap_conservatory`.
floor_area: Optional[Union[Measurement, int, float]] = None
room_height: Optional[Union[Measurement, int, float]] = None
double_glazed: Optional[str] = None
glazed_perimeter: Optional[Union[Measurement, int, float]] = None
@dataclass

24
datatypes/epc/surveys/elmhurst_site_notes.py
View file

@ -202,6 +202,9 @@ class VentilationAndCooling:
# SAP 10.2 §2 (17a) AP4 reading from §12.2 "Pressure Test Result
# (AP4)" — only present when `pressure_test_method == "Pulse"`.
air_permeability_ap4_m3_h_m2: Optional[float] = None
# SAP 10.2 §2 (17) AP50 reading from §12.2 "Pressure Test Result
# (AP50)" — present for a Blower Door test. Routes (18) via AP50/20.
air_permeability_ap50_m3_h_m2: Optional[float] = None
# Summary §12.1 "Mechanical Ventilation Type" — e.g. "Mechanical
# extract, decentralised (MEV dc)". None when `mechanical_ventilation
# is False` (no MV system).
@ -478,6 +481,21 @@ class ExtensionPart:
room_in_roof: Optional[RoomInRoof] = None
@dataclass
class Conservatory:
"""Summary §5 geometry of a NON-SEPARATED conservatory (RdSAP 10
§6.1). `room_height_storeys` is the lodged equivalent-storey count
("1 Storey" → 1.0, "1.5 Storey" → 1.5); the mapper/cascade translate
it to a metre height. A SEPARATED conservatory (§6.2) is disregarded,
so `thermally_separated=True` records are dropped before the cascade."""
thermally_separated: bool
floor_area_m2: float
double_glazed: bool
glazed_perimeter_m: float
room_height_storeys: float
@dataclass
class ElmhurstSiteNotes:
surveyor_info: SurveyorInfo
@ -560,3 +578,9 @@ class ElmhurstSiteNotes:
# cold loft instead of a room-in-roof). The mapper translates the
# surface table into a `SapRoomInRoof` attached to the Main bp.
room_in_roof: Optional[RoomInRoof] = None
# §5.0 Conservatory geometry — None when the dwelling has no
# conservatory (`has_conservatory=False`). Populated (incl. for
# separated conservatories) so the mapper can apply the §6.1/§6.2
# rule; the mapper drops separated ones.
conservatory: Optional[Conservatory] = None

14
domain/modelling/considered_measures.py
View file

@ -20,6 +20,20 @@ from domain.modelling.measure_type import MeasureType
from domain.modelling.recommendation import Recommendation
def combine_considered_measures(
a: Optional[frozenset[MeasureType]],
b: Optional[frozenset[MeasureType]],
) -> Optional[frozenset[MeasureType]]:
"""Intersect two allowlists, treating ``None`` as "all measures". Used to
layer an explicit override over the allowlist a Scenario's exclusions imply:
None ∧ x = x, and both present narrows to their intersection."""
if a is None:
return b
if b is None:
return a
return a & b
def restrict_to_considered_measures(
recommendations: Iterable[Recommendation],
considered_measures: Optional[frozenset[MeasureType]],

8
domain/modelling/generators/heating_recommendation.py
View file

@ -25,6 +25,7 @@ from domain.modelling.products import (
TuneUpCostInputs,
)
from domain.modelling.measure_type import MeasureType
from domain.modelling.product import Product
from domain.modelling.recommendation import Cost, MeasureOption, Recommendation
from domain.modelling.simulation import EpcSimulation, HeatingOverlay
from domain.sap10_calculator.tables.table_4b import (
@ -641,8 +642,9 @@ def _ashp_option(
if not _ashp_eligible(epc, restrictions):
return None
# Cost is composed per-dwelling from the rate sheet (ADR-0025), not the
# single catalogue scalar; the catalogue row is still read for its id.
product = products.get(_ASHP_MEASURE_TYPE)
# single catalogue scalar; the catalogue row is read only for its id, so an
# absent ASHP row must not suppress the bundle — it just carries no id.
product: Optional[Product] = products.get_optional(_ASHP_MEASURE_TYPE)
cost: Cost = Products().ashp_bundle_cost(ashp_cost_inputs(epc))
return MeasureOption(
measure_type=_ASHP_MEASURE_TYPE,
@ -652,7 +654,7 @@ def _ashp_option(
),
overlay=EpcSimulation(heating=_ASHP_OVERLAY),
cost=cost,
material_id=product.id,
material_id=product.id if product is not None else None,
)

19
domain/modelling/scenario.py
View file

@ -12,16 +12,33 @@ columns are not modelled. Carries no phases — multi-phase is deferred
from dataclasses import dataclass
from typing import Optional
from domain.modelling.measure_type import MeasureType
_NO_EXCLUSIONS: frozenset[MeasureType] = frozenset()
@dataclass(frozen=True)
class Scenario:
"""A retrofit brief: its goal, optional budget, and whether it is the
Property's default Scenario. `goal` / `goal_value` are the lodged target
(e.g. "INCREASING_EPC" → band "C"); carried for the Optimiser, not yet
enforced."""
enforced.
`exclusions` are the measure types the brief bars from the run (the only
measure-scoping the live ``scenario`` table persists — there is no
inclusions column). Empty means nothing is barred."""
id: int
goal: str
goal_value: str
budget: Optional[float]
is_default: bool
exclusions: frozenset[MeasureType] = _NO_EXCLUSIONS
def considered_measures(self) -> Optional[frozenset[MeasureType]]:
"""The measure-type allowlist the Scenario's exclusions imply: every
modelled measure minus the excluded ones, or None (consider every
measure) when nothing is excluded."""
if not self.exclusions:
return None
return frozenset(MeasureType) - self.exclusions

3
domain/sap10_calculator/rdsap/cert_to_inputs.py
View file

@ -4988,6 +4988,9 @@ def ventilation_from_cert(
# components-based (16) ach.
ap50 = sv.air_permeability_ap50_m3_h_m2 if sv is not None else None
ap4 = sv.air_permeability_ap4_m3_h_m2 if sv is not None else None
# SAP 10.2 §2 (17) — AP50 Blower Door reading routes (18) via
# `AP50 / 20 + (8)`, preferred over AP4 when both are lodged.
ap50 = sv.air_permeability_ap50_m3_h_m2 if sv is not None else None
# SAP 10.2 §2 (23a)/(24a..d) — MV kind dispatch chooses the (25)m
# effective-ach formula. The Elmhurst mapper translates the lodged
# "Mechanical Ventilation Type" string to an enum *name*; resolve

149
domain/sap10_calculator/worksheet/conservatory.py Normal file
View file

@ -0,0 +1,149 @@
"""RdSAP 10 §6.1 — non-separated (heated) conservatory geometry.
A non-separated conservatory is treated as part of the dwelling
(RdSAP 10 Specification, 9th June 2025, §6.1 + Table 25, pages 49-51):
- its floor area and volume are added to TFA (4) and volume (5);
- its fully-glazed walls bill as a window — line (27) — at the Table 25
"U-value of window"; its glazed roof bills as a rooflight — line (27a)
— at the Table 25 "U-value of roof window"; both U-values already
include the §3.2 curtain resistance (R=0.04 m²K/W);
- its floor adds a ground-loss term — line (28a) — via BS EN ISO 13370,
taken as an uninsulated solid floor with 300 mm walls (§5.12 note,
spec p.43), exposed perimeter = glazed perimeter;
- its glazed wall + glazed roof + floor areas count toward the total
exposed area (31) and hence thermal bridging (36); the fully-glazed
"structure" walls/roof themselves add nothing (the glazing IS the
window/rooflight).
Its roof area is the floor area / cos(20°) and its wall area is the
exposed perimeter × height; the height is translated from the lodged
equivalent storey count (§6.1): 1 storey → ground-floor room height;
1½ → ground + 0.25 + 0.5×first; 2 → ground + 0.25 + first; etc.
A SEPARATED conservatory (§6.2) is disregarded entirely — the mapper
maps it to None, so it never reaches this module.
"""
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP
from math import cos, radians
from typing import Final, Optional
from datatypes.epc.domain.epc_property_data import EpcPropertyData
# RdSAP 10 §6.1 — conservatory roof area = floor area / cos(20°); §6.1
# also fixes the rooflight solar pitch at 20°.
CONSERVATORY_ROOF_PITCH_DEG: Final[float] = 20.0
_COS_ROOF_PITCH: Final[float] = cos(radians(CONSERVATORY_ROOF_PITCH_DEG))
# RdSAP 10 Table 25 (PDF p.51) — default conservatory glazing U-values
# (W/m²K, INCLUSIVE of the §3.2 curtain resistance) and g-values. The
# Summary lodges only double vs single (no triple), so a bool selects the
# row: True → double (6 mm gap), False → single.
_TABLE_25_WALL_U: Final[dict[bool, float]] = {True: 3.1, False: 4.8}
_TABLE_25_ROOF_U: Final[dict[bool, float]] = {True: 3.4, False: 5.3}
_TABLE_25_G_VALUE: Final[dict[bool, float]] = {True: 0.76, False: 0.85}
_TABLE_25_FRAME_FACTOR: Final[float] = 0.70 # Table 25 — wood/PVC frame
# SAP 10.2 §3.2 formula (2) curtain/blind resistance. Table 25 U-values
# are "adjusted for curtains" already, so the EFFECTIVE conduction U is
# 1 / (1/U_table25 + 0.04) — the same transform `heat_transmission`
# applies to regular windows/rooflights.
_CURTAIN_RESISTANCE_M2K_PER_W: Final[float] = 0.04
# RdSAP 10 §5.12 (spec p.43) — a non-separated conservatory floor is an
# uninsulated solid ground floor with 300 mm walls.
_CONSERVATORY_WALL_THICKNESS_MM: Final[int] = 300
_AREA_ROUND_DP: Final[int] = 2
def _round2(value: float) -> float:
"""RdSAP 10 §15 (p.66): element areas + conservatory height → 2 d.p."""
return float(
Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
)
@dataclass(frozen=True)
class ConservatoryGeometry:
"""Derived §6.1 geometry for one non-separated conservatory. Areas and
height are rounded to 2 d.p. per RdSAP 10 §15."""
height_m: float
floor_area_m2: float
glazed_wall_area_m2: float
glazed_roof_area_m2: float
glazed_perimeter_m: float
wall_u_raw: float # Table 25 window U, pre-curtain
roof_u_raw: float # Table 25 roof-window U, pre-curtain
wall_u_eff: float # post-curtain conduction U for line (27)
roof_u_eff: float # post-curtain conduction U for line (27a)
g_value: float
frame_factor: float
volume_m3: float
def _conservatory_height_m(epc: EpcPropertyData, storeys: float) -> float:
"""Translate the equivalent storey count into a metre height per
RdSAP 10 §6.1 using the dwelling's per-storey room heights:
1 storey → ground-floor room height
1½ storey → ground + 0.25 + 0.5 × first-floor room height
2 storey → ground + 0.25 + first-floor room height
etc.
Room heights are taken from the Main building part's floor
dimensions (floor 0 = ground, 1 = first, ...). Returns 0.0 when no
storeys are lodged (defensive; the conservatory then bills no walls)."""
parts = epc.sap_building_parts or []
heights: list[float] = []
if parts:
fds = sorted(
parts[0].sap_floor_dimensions,
key=lambda fd: fd.floor if fd.floor is not None else 0,
)
heights = [fd.room_height_m for fd in fds if fd.room_height_m]
if not heights:
return 0.0
n_full = int(storeys)
height = heights[0]
for s in range(1, n_full):
height += 0.25 + heights[min(s, len(heights) - 1)]
if storeys - n_full >= 0.5:
height += 0.25 + 0.5 * heights[min(n_full, len(heights) - 1)]
return _round2(height)
def conservatory_geometry(
epc: EpcPropertyData,
) -> Optional[ConservatoryGeometry]:
"""Build the §6.1 conservatory geometry, or None when there is no
(non-separated) conservatory."""
cons = epc.sap_conservatory
if cons is None or cons.thermally_separated:
return None
height = _conservatory_height_m(epc, cons.room_height_storeys)
floor_area = cons.floor_area_m2
glazed_perimeter = cons.glazed_perimeter_m
glazed_wall = _round2(glazed_perimeter * height)
glazed_roof = _round2(floor_area / _COS_ROOF_PITCH)
dg = cons.double_glazed
wall_u_raw = _TABLE_25_WALL_U[dg]
roof_u_raw = _TABLE_25_ROOF_U[dg]
return ConservatoryGeometry(
height_m=height,
floor_area_m2=floor_area,
glazed_wall_area_m2=glazed_wall,
glazed_roof_area_m2=glazed_roof,
glazed_perimeter_m=glazed_perimeter,
wall_u_raw=wall_u_raw,
roof_u_raw=roof_u_raw,
wall_u_eff=1.0 / (1.0 / wall_u_raw + _CURTAIN_RESISTANCE_M2K_PER_W),
roof_u_eff=1.0 / (1.0 / roof_u_raw + _CURTAIN_RESISTANCE_M2K_PER_W),
g_value=_TABLE_25_G_VALUE[dg],
frame_factor=_TABLE_25_FRAME_FACTOR,
volume_m3=floor_area * height,
)

16
domain/sap10_calculator/worksheet/dimensions.py
View file

@ -21,6 +21,7 @@ from dataclasses import dataclass
from typing import Final
from datatypes.epc.domain.epc_property_data import EpcPropertyData, SapBuildingPart
from domain.sap10_calculator.worksheet.conservatory import conservatory_geometry
_DEFAULT_STOREY_HEIGHT_M: Final[float] = 2.5
@ -145,17 +146,28 @@ def dimensions_from_cert(epc: EpcPropertyData) -> Dimensions:
total_storey_count = max(part_storey_counts) if part_storey_counts else 0
has_storeys = sum_per_storey_area_m2 > 0
# `avg_height` (used by §2 (9) dwelling height → infiltration) is a
# property of the dwelling's storeys, so the conservatory is excluded
# from it. The conservatory IS added to TFA (4) and volume (5) per
# RdSAP 10 §6.1 ("The floor area and volume of a non-separated
# conservatory are added to the total floor area and volume of the
# dwelling") — it just doesn't form a storey.
avg_height = (
sum_per_storey_volume_m3 / sum_per_storey_area_m2
if has_storeys
else _DEFAULT_STOREY_HEIGHT_M
)
cons = conservatory_geometry(epc)
cons_floor_area_m2 = cons.floor_area_m2 if cons is not None else 0.0
cons_volume_m3 = cons.volume_m3 if cons is not None else 0.0
return Dimensions(
total_floor_area_m2=(
sum_per_storey_area_m2 if has_storeys else epc.total_floor_area_m2
sum_per_storey_area_m2 + cons_floor_area_m2
if has_storeys
else epc.total_floor_area_m2
),
volume_m3=(
sum_per_storey_volume_m3
sum_per_storey_volume_m3 + cons_volume_m3
if has_storeys
else epc.total_floor_area_m2 * _DEFAULT_STOREY_HEIGHT_M
),

91
domain/sap10_calculator/worksheet/heat_transmission.py
View file

@ -71,6 +71,7 @@ from domain.sap10_ml.rdsap_uvalues import (
u_wall,
u_window,
)
from domain.sap10_calculator.worksheet.conservatory import conservatory_geometry
from math import cos, floor, radians, sqrt
@ -123,6 +124,9 @@ _DEFAULT_DOOR_AREA_M2: Final[float] = 1.85
# deducts from that wall, not the main wall.
_CORRIDOR_DOOR_U_W_PER_M2K: Final[float] = 1.4
_DEFAULT_STOREY_HEIGHT_M: Final[float] = 2.5
# RdSAP 10 §5.12 (spec p.43) — a non-separated conservatory floor is an
# uninsulated solid ground floor with 300 mm walls.
_CONSERVATORY_WALL_THICKNESS_MM: Final[int] = 300
# SAP10.2 §3.2 curtain/blind thermal resistance applied to windows (and
# roof windows) — turns raw window U into the worksheet's (27) effective U.
_WINDOW_CURTAIN_RESISTANCE_M2K_PER_W: Final[float] = 0.04
@ -452,6 +456,32 @@ def _joined_main_roof_descriptions(roofs: list[Any]) -> Optional[str]:
return " | ".join(parts)
def _main_roof_descriptions_by_kind(
roofs: list[Any],
) -> tuple[Optional[str], Optional[str]]:
"""Partition the non-RR roof descriptions into ``(pitched, flat)`` joins.
The deduplicated ``epc.roofs[]`` list cannot be indexed 1:1 against the
building parts (190/329 multi-part certs have len(roofs) != len(parts)),
so each part's ``u_roof`` historically consumed the SINGLE join of every
roof description. That leaks one part's insulation state onto another: a
"Flat, no insulation" extension dragged a "Pitched, insulated (assumed)"
main roof to the uninsulated 2.30, ~3x over-stating its heat loss (cert
100010129331: roof 110.5 -> ~28 W/K, +13 SAP). Splitting by flat vs
pitched/sloping lets each part match its own kind; the global join
(`_joined_main_roof_descriptions`) stays the fallback when a part's kind
has no matching entry. "Roof room(s)" entries are dropped (they carry
their own §3.9/§3.10 shell cascade)."""
pitched: list[str] = []
flat: list[str] = []
for e in roofs:
d = getattr(e, "description", "")
if not d or "roof room" in d.lower():
continue
(flat if "flat" in d.lower() else pitched).append(d)
return (" | ".join(pitched) or None, " | ".join(flat) or None)
def _part_geometry(part: SapBuildingPart) -> dict[str, float]:
if not part.sap_floor_dimensions:
# A part with no floor dimensions has no derivable RR shell or
@ -669,6 +699,9 @@ def heat_transmission_from_cert(
country = Country.from_code(epc.country_code)
roof_description = _joined_main_roof_descriptions(epc.roofs)
pitched_roof_description, flat_roof_description = (
_main_roof_descriptions_by_kind(epc.roofs)
)
wall_description = _joined_descriptions(epc.walls)
floor_description = _joined_descriptions(epc.floors)
@ -942,8 +975,19 @@ def heat_transmission_from_cert(
roof_thickness_explicitly_zero = (
isinstance(raw_roof_thickness, int) and raw_roof_thickness == 0
)
# RdSAP 10 §5.11 — match THIS part's roof to its own kind's lodged
# description (flat vs pitched/sloping) rather than the global join,
# so a flat "no insulation" part does not drag a pitched insulated
# part to the uninsulated 2.30. Fall back to the global join when the
# part's kind has no matching `epc.roofs[]` entry.
part_roof_is_flat = "flat" in (part.roof_construction_type or "").lower()
matched_roof_description = (
flat_roof_description if part_roof_is_flat else pitched_roof_description
)
if matched_roof_description is None:
matched_roof_description = roof_description
effective_roof_description = (
None if roof_thickness_explicitly_zero else roof_description
None if roof_thickness_explicitly_zero else matched_roof_description
)
# RdSAP 10 §5.11 Table 18 page 45: column (3) "Flat roof" applies
# when the per-bp roof construction lodges as a flat roof and the
@ -1429,6 +1473,51 @@ def heat_transmission_from_cert(
# door line.
doors += _CORRIDOR_DOOR_U_W_PER_M2K * corridor_door_area
roof_windows_w_per_k = roof_windows_w_per_k_total
# RdSAP 10 §6.1 (PDF p.49) + Table 25 (p.51) — a non-separated
# conservatory. Its fully-glazed walls bill as a window (27), its
# glazed roof as a rooflight (27a), and its floor adds a ground-loss
# term (28a) via BS EN ISO 13370 (uninsulated solid floor, 300 mm
# walls per §5.12; exposed perimeter = glazed perimeter). The glazed
# wall + roof + floor areas join (31)/(36) external area; the fully-
# glazed "structure" walls/roof add nothing (the glazing IS the
# window/rooflight). A separated conservatory (§6.2) is mapped to
# None upstream and never reaches here.
cons_geom = conservatory_geometry(epc)
cons_windows_w_per_k: float = 0.0
if cons_geom is not None:
cons_windows_w_per_k = (
cons_geom.glazed_wall_area_m2 * cons_geom.wall_u_eff
)
roof_windows_w_per_k += (
cons_geom.glazed_roof_area_m2 * cons_geom.roof_u_eff
)
u_cons_floor = u_floor(
country=country,
age_band=primary_age,
construction=None,
insulation_thickness_mm=0,
area_m2=cons_geom.floor_area_m2,
perimeter_m=cons_geom.glazed_perimeter_m,
wall_thickness_mm=_CONSERVATORY_WALL_THICKNESS_MM,
# Force the solid-floor branch of BS EN ISO 13370 regardless of
# age band (§5.12: conservatory floor is an uninsulated SOLID
# ground floor — the A/B suspended-timber default must not fire).
description="Solid",
)
floor += u_cons_floor * cons_geom.floor_area_m2
cons_external_area = (
cons_geom.glazed_wall_area_m2
+ cons_geom.glazed_roof_area_m2
+ cons_geom.floor_area_m2
)
total_external_area += cons_external_area
bridging += dwelling_y * cons_external_area
# Fold the conservatory glazed wall into the (27) window readout. The
# `windows` accumulator is partially-typed upstream (the per-window
# `u_value` arrives as `Any`); `float(...)` re-asserts the strict float
# type as we add the strictly-typed conservatory term.
windows = float(windows) + cons_windows_w_per_k
fabric_heat_loss = (
walls + roof + floor + party + windows + roof_windows_w_per_k + doors # (33)
)

46
domain/sap10_calculator/worksheet/solar_gains.py
View file

@ -36,6 +36,10 @@ from math import cos, radians, sin
from typing import Final
from datatypes.epc.domain.epc_property_data import EpcPropertyData, SapWindow
from domain.sap10_calculator.worksheet.conservatory import (
CONSERVATORY_ROOF_PITCH_DEG,
conservatory_geometry,
)
from domain.sap10_calculator.tables.pcdb.postcode_weather import PostcodeClimate
from domain.sap10_calculator.climate.appendix_u import (
horizontal_solar_irradiance_w_per_m2,
@ -435,6 +439,48 @@ def solar_gains_from_cert(
o: _sum_tuples(*per_orientation[o]) for o in Orientation
}
# RdSAP 10 §6.1 (PDF p.49) + Table 25 note (p.51): "The orientation of
# windows in a conservatory is not recorded, thus solar gains are
# calculated using the default solar flux (East/West orientation, with
# 20° pitch for roof windows)." Average overshading per §7 (Table 6d).
# The glazed wall bills onto the (76) East line (vertical, Z=z_vertical);
# the glazed roof onto the (82) roof-window line (20° pitch, Z=1.0).
cons = conservatory_geometry(epc)
if cons is not None:
cons_wall_monthly = tuple(
window_solar_gain_w(
area_m2=cons.glazed_wall_area_m2,
surface_flux_w_per_m2=surface_solar_flux_w_per_m2(
orientation=Orientation.E, pitch_deg=90.0,
region=region, month=m,
),
g_perpendicular=cons.g_value,
frame_factor=cons.frame_factor,
overshading_factor=z_vertical,
)
for m in _MONTHS
)
cons_roof_monthly = tuple(
window_solar_gain_w(
area_m2=cons.glazed_roof_area_m2,
surface_flux_w_per_m2=surface_solar_flux_w_per_m2(
orientation=Orientation.E,
pitch_deg=CONSERVATORY_ROOF_PITCH_DEG,
region=region, month=m,
),
g_perpendicular=cons.g_value,
frame_factor=cons.frame_factor,
overshading_factor=_HORIZONTAL_Z,
)
for m in _MONTHS
)
per_orientation_summed[Orientation.E] = _sum_tuples(
per_orientation_summed[Orientation.E], cons_wall_monthly,
)
roof_windows_monthly = _sum_tuples(
roof_windows_monthly, cons_roof_monthly,
)
total = _sum_tuples(
*per_orientation_summed.values(),
roof_windows_monthly,

29
domain/sap10_ml/rdsap_uvalues.py
View file

@ -589,6 +589,35 @@ def u_wall(
):
u0 = _u_stone_thin_wall_age_a_to_e(construction, wall_thickness_mm)
if u0 is not None:
# RdSAP 10 §5.8 + Table 14 (PDF p.41-42) — added External/Internal
# insulation on a stone wall: U = 1/(1/U₀ + R_ins), with U₀ the
# RAW §5.6 stone result (the Table-6 footnote (a) 1.7 cap does NOT
# apply to the insulated path — same rule the brick branch below
# and the dry-lined granite pin 000565 follow). λ defaults to
# 0.04 W/m·K per §5.8 final note. Mirrors the WALL_SOLID_BRICK
# insulated branch; without it a stone wall lodging code 1/3 +
# a thickness was billed at its UNINSULATED U (e.g. sandstone
# 520 mm + 100 mm internal: 1.64 → 0.30), the dominant cause of
# the wall_insulation_type=3 corpus under-rate cluster.
if (
wall_insulation_type in (
_WALL_INSULATION_EXTERNAL, _WALL_INSULATION_INTERNAL,
)
and insulation_thickness_mm is not None
and insulation_thickness_mm > 0
):
r_ins = _r_insulation_table_14(
insulation_thickness_mm,
_resolve_wall_insulation_lambda_w_per_mk(
wall_insulation_thermal_conductivity
),
)
u_unrounded = 1.0 / (1.0 / u0 + r_ins)
return float(
Decimal(str(u_unrounded)).quantize(
Decimal("0.01"), rounding=ROUND_HALF_UP
)
)
if dry_lined:
# Round to 2 d.p. — worksheet (29a) A×U product uses
# the 2-d.p.-displayed U (cf. 000565 Main alt_wall_1:

74
domain/sap10_ml/tests/test_rdsap_uvalues.py
View file

@ -725,6 +725,80 @@ def test_u_wall_stone_sandstone_thin_wall_age_a_120mm_uses_5_6_sandstone_formula
assert abs(result - 3.7408) <= 1e-3
def test_u_wall_stone_sandstone_with_internal_insulation_applies_5_8_table_14_r_value() -> None:
# Arrange — RdSAP 10 §5.8 + Table 14 (PDF p.41-42): a stone wall lodging
# External/Internal insulation (wall_insulation_type 1/3) + a thickness
# gets the same R-value adjustment as solid brick, applied to the RAW §5.6
# U₀. Mirrors corpus cert 100052159386 (Sandstone, 520 mm, 100 mm internal):
# U₀ = 54.876 × 520^(-0.561) = 1.6433
# R = 0.025 × 100 + 0.25 = 2.75 (Table 14, λ = 0.04)
# U = 1 / (1/1.6433 + 2.75) = 0.2977 → 0.30 (2 d.p.)
# Before this branch the wall was billed at its UNINSULATED U (≈1.64),
# the dominant cause of the wall_insulation_type=3 corpus under-rate cluster.
# Act
result = u_wall(
country=Country.ENG,
age_band="A",
construction=WALL_STONE_SANDSTONE,
insulation_thickness_mm=100,
insulation_present=True,
wall_insulation_type=3,
dry_lined=False,
wall_thickness_mm=520,
)
# Assert
assert abs(result - 0.30) <= 1e-4
def test_u_wall_stone_sandstone_insulated_feeds_raw_u0_not_table_6_cap() -> None:
# Arrange — the Table-6 footnote (a) 1.7 cap applies ONLY to the as-built
# row; the insulated §5.8 path takes the RAW §5.6 U₀ (same rule the brick
# branch and the dry-lined granite pin 000565 follow). At W=120 mm the raw
# sandstone U₀ = 3.7408 (> 1.7), so the 100 mm internal result must be
# 1 / (1/3.7408 + 2.75) = 0.331 → 0.33 (raw),
# NOT the capped 1 / (1/1.7 + 2.75) = 0.30. The 0.33 vs 0.30 split proves
# the cap is bypassed on the insulated path.
# Act
result = u_wall(
country=Country.ENG,
age_band="A",
construction=WALL_STONE_SANDSTONE,
insulation_thickness_mm=100,
insulation_present=True,
wall_insulation_type=3,
dry_lined=False,
wall_thickness_mm=120,
)
# Assert
assert abs(result - 0.33) <= 1e-4
def test_u_wall_stone_granite_with_external_insulation_applies_5_8_table_14_r_value() -> None:
# Arrange — granite/whinstone §5.6 formula + §5.8 external insulation:
# U₀ = 45.315 × 120^(-0.513) = 3.8871
# R = 0.025 × 50 + 0.25 = 1.50 (Table 14, λ = 0.04)
# U = 1 / (1/3.8871 + 1.50) = 0.567 → 0.57 (2 d.p.)
# Act
result = u_wall(
country=Country.ENG,
age_band="A",
construction=WALL_STONE_GRANITE,
insulation_thickness_mm=50,
insulation_present=True,
wall_insulation_type=1,
dry_lined=False,
wall_thickness_mm=120,
)
# Assert
assert abs(result - 0.57) <= 1e-4
def test_u_wall_stone_granite_age_g_with_wall_thickness_ignores_5_6_formula_per_age_a_to_e_gate() -> None:
# Arrange — §5.6 (PDF p.40) heading explicitly scopes the formula
# to "age bands A to E". For age F onwards Table 6 gives literal

35
harness/console.py
View file

@ -25,14 +25,19 @@ from domain.geospatial.coordinates import Coordinates
from domain.geospatial.planning_restrictions import PlanningRestrictions
from domain.modelling.measure_type import MeasureType
from domain.modelling.plan import Plan
from domain.modelling.recommendation import Recommendation
from domain.modelling.scenario import Scenario
from domain.modelling.solar_potential import SolarPotential
from domain.property.property import Property, PropertyIdentity
from domain.property_baseline.rebaseliner import StubRebaseliner
from domain.sap10_calculator.calculator import Sap10Calculator
from harness.plan_table import format_plan_table
from orchestration.ara_first_run_pipeline import AraFirstRunPipeline
from orchestration.ingestion_orchestrator import IngestionOrchestrator
from orchestration.modelling_orchestrator import ModellingOrchestrator
from orchestration.modelling_orchestrator import (
ModellingOrchestrator,
_candidate_recommendations, # pyright: ignore[reportPrivateUsage]
)
from orchestration.property_baseline_orchestrator import PropertyBaselineOrchestrator
from repositories.fuel_rates.fuel_rates_static_file_repository import (
FuelRatesStaticFileRepository,
@ -247,3 +252,31 @@ def run_modelling(
if print_table:
print("\n" + format_plan_table(plan))
return plan
def candidate_recommendations(
epc: EpcPropertyData,
*,
catalogue_path: Path = DEFAULT_CATALOGUE,
planning_restrictions: PlanningRestrictions = PlanningRestrictions(),
solar_insights: Optional[dict[str, Any]] = None,
considered_measures: Optional[frozenset[MeasureType]] = None,
products: Optional[ProductRepository] = None,
) -> list[Recommendation]:
"""Every candidate Recommendation the Generators produce for ``epc`` — the
full menu of Measure Options with their per-Option cost, *before* the
Optimiser selects a Plan. Use this to inspect measures (and their cost) that
a Plan does not end up selecting, e.g. an ASHP the Optimiser passed over for
a cheaper route to the target band. Inputs mirror `run_modelling`."""
solar_potential: Optional[SolarPotential] = (
SolarPotential.from_building_insights(solar_insights)
if solar_insights is not None and "solarPotential" in solar_insights
else None
)
return _candidate_recommendations(
epc,
products or ProductJsonRepository(catalogue_path),
planning_restrictions,
solar_potential,
considered_measures,
)

26
infrastructure/postgres/modelling/scenario_table.py
View file

@ -8,10 +8,35 @@ from sqlalchemy import Enum as SAEnum
from sqlalchemy.sql import func
from sqlmodel import Field, SQLModel
from domain.modelling.measure_type import MeasureType
from domain.modelling.portfolio_goal import PortfolioGoal
from domain.modelling.scenario import Scenario
def _parse_exclusions(raw: Optional[str]) -> frozenset[MeasureType]:
"""Parse the live ``scenario.exclusions`` column — a Postgres text-array
literal like ``{solar_pv,internal_wall_insulation}`` — into the excluded
MeasureTypes. Each token must be an exact MeasureType value (no high-level
category expansion); an unknown token is a data error and raises, matching
the repo's strict-enum convention."""
if not raw:
return frozenset()
inner = raw.strip()
if inner.startswith("{") and inner.endswith("}"):
inner = inner[1:-1]
tokens = [token.strip().strip('"') for token in inner.split(",") if token.strip()]
excluded: set[MeasureType] = set()
for token in tokens:
try:
excluded.add(MeasureType(token))
except ValueError as error:
raise ValueError(
f"scenario excludes unknown measure type {token!r}; the "
f"exclusions column must hold exact MeasureType values"
) from error
return frozenset(excluded)
class ScenarioModel(SQLModel, table=True):
"""The single SQLModel definition of the live ``scenario`` table (ADR-0017
amendment). Full legacy column parity; ``goal`` is the ``PortfolioGoal``
@ -89,4 +114,5 @@ class ScenarioModel(SQLModel, table=True):
goal_value=self.goal_value,
budget=self.budget,
is_default=self.is_default,
exclusions=_parse_exclusions(self.exclusions),
)

63
infrastructure/postgres/property_baseline_performance_table.py
View file

@ -72,26 +72,31 @@ class PropertyBaselinePerformanceModel(SQLModel, table=True):
space_heating_kwh: float
water_heating_kwh: float
# The Fuel Rates snapshot period the bill was priced against (FE-owned column,
# nullable). Not yet threaded through Bill Derivation, so left None for now.
fuel_rates_period: Optional[str] = Field(default=None)
# Bill Derivation block (ADR-0014 §6). Nullable: all None when no calculator
# ran (stub path). The ``bill_`` prefix avoids clashing with the
# ran (stub path). Column names are unprefixed to mirror the FE-owned table —
# the per-section ``heating_kwh`` / ``hot_water_kwh`` do not clash with the
# recorded-demand ``space_heating_kwh`` / ``water_heating_kwh`` above.
bill_heating_kwh: Optional[float] = Field(default=None)
bill_heating_cost_gbp: Optional[float] = Field(default=None)
bill_hot_water_kwh: Optional[float] = Field(default=None)
bill_hot_water_cost_gbp: Optional[float] = Field(default=None)
bill_lighting_kwh: Optional[float] = Field(default=None)
bill_lighting_cost_gbp: Optional[float] = Field(default=None)
bill_appliances_kwh: Optional[float] = Field(default=None)
bill_appliances_cost_gbp: Optional[float] = Field(default=None)
bill_cooking_kwh: Optional[float] = Field(default=None)
bill_cooking_cost_gbp: Optional[float] = Field(default=None)
bill_pumps_fans_kwh: Optional[float] = Field(default=None)
bill_pumps_fans_cost_gbp: Optional[float] = Field(default=None)
bill_cooling_kwh: Optional[float] = Field(default=None)
bill_cooling_cost_gbp: Optional[float] = Field(default=None)
bill_standing_charges_gbp: Optional[float] = Field(default=None)
bill_seg_credit_gbp: Optional[float] = Field(default=None)
bill_total_annual_bill_gbp: Optional[float] = Field(default=None)
heating_kwh: Optional[float] = Field(default=None)
heating_cost_gbp: Optional[float] = Field(default=None)
hot_water_kwh: Optional[float] = Field(default=None)
hot_water_cost_gbp: Optional[float] = Field(default=None)
lighting_kwh: Optional[float] = Field(default=None)
lighting_cost_gbp: Optional[float] = Field(default=None)
appliances_kwh: Optional[float] = Field(default=None)
appliances_cost_gbp: Optional[float] = Field(default=None)
cooking_kwh: Optional[float] = Field(default=None)
cooking_cost_gbp: Optional[float] = Field(default=None)
pumps_fans_kwh: Optional[float] = Field(default=None)
pumps_fans_cost_gbp: Optional[float] = Field(default=None)
cooling_kwh: Optional[float] = Field(default=None)
cooling_cost_gbp: Optional[float] = Field(default=None)
standing_charges_gbp: Optional[float] = Field(default=None)
seg_credit_gbp: Optional[float] = Field(default=None)
total_annual_bill_gbp: Optional[float] = Field(default=None)
@classmethod
def from_domain(
@ -122,15 +127,15 @@ class PropertyBaselinePerformanceModel(SQLModel, table=True):
return
for section, stem in _SECTION_COLUMN_STEM.items():
cost = bill.sections.get(section)
setattr(self, f"bill_{stem}_kwh", cost.kwh if cost is not None else None)
setattr(self, f"{stem}_kwh", cost.kwh if cost is not None else None)
setattr(
self,
f"bill_{stem}_cost_gbp",
f"{stem}_cost_gbp",
cost.cost_gbp if cost is not None else None,
)
self.bill_standing_charges_gbp = bill.standing_charges_gbp
self.bill_seg_credit_gbp = bill.seg_credit_gbp
self.bill_total_annual_bill_gbp = bill.total_gbp
self.standing_charges_gbp = bill.standing_charges_gbp
self.seg_credit_gbp = bill.seg_credit_gbp
self.total_annual_bill_gbp = bill.total_gbp
def to_domain(self) -> PropertyBaselinePerformance:
return PropertyBaselinePerformance(
@ -157,18 +162,18 @@ class PropertyBaselinePerformanceModel(SQLModel, table=True):
not-None discriminator: a persisted bill always sets it, so its absence
means no calculator ran and the bill was None. A section is rebuilt only
when its kWh column is not None (paired with its cost)."""
if self.bill_total_annual_bill_gbp is None:
if self.total_annual_bill_gbp is None:
return None
sections: dict[BillSection, BillSectionCost] = {}
for section, stem in _SECTION_COLUMN_STEM.items():
kwh = cast(Optional[float], getattr(self, f"bill_{stem}_kwh"))
kwh = cast(Optional[float], getattr(self, f"{stem}_kwh"))
if kwh is None:
continue
cost_gbp = cast(float, getattr(self, f"bill_{stem}_cost_gbp"))
cost_gbp = cast(float, getattr(self, f"{stem}_cost_gbp"))
sections[section] = BillSectionCost(kwh=kwh, cost_gbp=cost_gbp)
return Bill(
sections=sections,
standing_charges_gbp=cast(float, self.bill_standing_charges_gbp),
seg_credit_gbp=cast(float, self.bill_seg_credit_gbp),
total_gbp=self.bill_total_annual_bill_gbp,
standing_charges_gbp=cast(float, self.standing_charges_gbp),
seg_credit_gbp=cast(float, self.seg_credit_gbp),
total_gbp=self.total_annual_bill_gbp,
)

8
infrastructure/postgres/property_table.py
View file

@ -1,5 +1,6 @@
from __future__ import annotations
from datetime import datetime
from typing import ClassVar, Optional
from sqlalchemy import Column
@ -48,3 +49,10 @@ class PropertyRow(SQLModel, table=True):
user_inputted_address: Optional[str] = Field(default=None)
user_inputted_postcode: Optional[str] = Field(default=None)
lexiscore: Optional[float] = Field(default=None)
# FE-owned columns the modelling pipeline now WRITES to record a run: the old
# engine set `has_recommendations` (engine.py); we mirror that, and bump
# `updated_at` so a run is datable (a first-run under the new process is
# `updated_at >= 2026-06-01`, the cutoff the old pipeline predates).
has_recommendations: Optional[bool] = Field(default=None)
updated_at: Optional[datetime] = Field(default=None)

116
orchestration/modelling_orchestrator.py
View file

@ -7,7 +7,10 @@ from datatypes.epc.domain.epc import Epc
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.billing.bill import Bill, EnergyBreakdown
from domain.billing.bill_derivation import BillDerivation
from domain.modelling.considered_measures import restrict_to_considered_measures
from domain.modelling.considered_measures import (
combine_considered_measures,
restrict_to_considered_measures,
)
from domain.modelling.generators.floor_recommendation import recommend_floor_insulation
from domain.modelling.measure_type import MeasureType
from domain.modelling.optimisation.measure_dependency import ventilation_dependency
@ -123,6 +126,7 @@ class ModellingOrchestrator:
solar_potential: Optional[SolarPotential] = _solar_potential_for(
uow.solar, prop.identity.uprn
)
has_recommendations = False
for scenario in scenarios:
plan = self._plan_for(
scorer,
@ -142,6 +146,13 @@ class ModellingOrchestrator:
portfolio_id=portfolio_id,
is_default=scenario.is_default,
)
has_recommendations = has_recommendations or bool(plan.measures)
# Record the run on the Property: the old engine's per-Property
# `has_recommendations` marker (true if any Scenario yielded a
# measure), with `updated_at` bumped so the run is datable.
uow.property.mark_modelled(
property_id, has_recommendations=has_recommendations
)
uow.commit()
def _plan_for(
@ -159,18 +170,23 @@ class ModellingOrchestrator:
) -> Plan:
"""Generate → score → optimise → re-score/repair → attribute → bill →
assemble the Plan for one Property + Scenario."""
# The Scenario's own exclusions scope the run; an explicit
# ``considered_measures`` (e.g. from a harness) narrows it further.
considered: Optional[frozenset[MeasureType]] = combine_considered_measures(
scenario.considered_measures(), considered_measures
)
groups: list[list[ScoredOption]] = _scored_candidate_groups(
scorer,
effective_epc,
products,
planning_restrictions,
solar_potential,
considered_measures,
considered,
)
# Forced Measure Dependencies (ventilation) are excluded from the pool
# but injected into the package before the re-score (ADR-0016).
dependencies: list[MeasureDependency] = _measure_dependencies(
effective_epc, products, considered_measures
effective_epc, products, considered
)
package: OptimisedPackage = optimise_package(
groups=groups,
@ -249,24 +265,86 @@ def _candidate_recommendations(
solar_potential: Optional[SolarPotential],
considered_measures: Optional[frozenset[MeasureType]],
) -> list[Recommendation]:
"""Run every Recommendation Generator; keep the ones that apply. Solid-wall
insulation, glazing, heating and solar are additionally gated by the
Property's planning protections (ADR-0019 / ADR-0022 / ADR-0024 / ADR-0026);
solar also needs the Property's Google solar potential. ``considered_measures``
then restricts the survivors to the run's allowlist (None = all)."""
found = (
recommend_cavity_wall(effective_epc, products),
recommend_solid_wall(effective_epc, products, planning_restrictions),
recommend_roof_insulation(effective_epc, products),
recommend_floor_insulation(effective_epc, products),
recommend_glazing(effective_epc, products, planning_restrictions),
recommend_lighting(effective_epc, products),
recommend_heating(effective_epc, products, planning_restrictions),
recommend_secondary_heating_removal(effective_epc, products),
recommend_solar(
effective_epc, products, solar_potential, planning_restrictions
"""Run the applicable Recommendation Generators; keep the ones that apply.
Solid-wall insulation, glazing, heating and solar are additionally gated by
the Property's planning protections (ADR-0019 / ADR-0022 / ADR-0024 /
ADR-0026); solar also needs the Property's Google solar potential.
``considered_measures`` gates generation *up front*: a generator runs only
when the allowlist admits at least one of the measure types it can emit
(None = every measure), so an excluded measure never reaches the catalogue —
which matters when the live ``material.type`` enum cannot even represent it
(e.g. ``secondary_heating_removal``). ``restrict_to_considered_measures``
then trims any disallowed Options off the multi-Option survivors."""
def admitted(*emits: MeasureType) -> bool:
return considered_measures is None or any(
measure in considered_measures for measure in emits
)
# Each generator paired with the measure types it can emit, so the allowlist
# can skip a generator whose every type is excluded before it is invoked.
generators: tuple[
tuple[bool, Callable[[], Optional[Recommendation]]], ...
] = (
(
admitted(MeasureType.CAVITY_WALL_INSULATION),
lambda: recommend_cavity_wall(effective_epc, products),
),
(
admitted(
MeasureType.INTERNAL_WALL_INSULATION,
MeasureType.EXTERNAL_WALL_INSULATION,
),
lambda: recommend_solid_wall(
effective_epc, products, planning_restrictions
),
),
(
admitted(
MeasureType.LOFT_INSULATION,
MeasureType.SLOPING_CEILING_INSULATION,
MeasureType.FLAT_ROOF_INSULATION,
),
lambda: recommend_roof_insulation(effective_epc, products),
),
(
admitted(
MeasureType.SUSPENDED_FLOOR_INSULATION,
MeasureType.SOLID_FLOOR_INSULATION,
),
lambda: recommend_floor_insulation(effective_epc, products),
),
(
admitted(MeasureType.DOUBLE_GLAZING, MeasureType.SECONDARY_GLAZING),
lambda: recommend_glazing(effective_epc, products, planning_restrictions),
),
(
admitted(MeasureType.LOW_ENERGY_LIGHTING),
lambda: recommend_lighting(effective_epc, products),
),
(
admitted(
MeasureType.HIGH_HEAT_RETENTION_STORAGE_HEATERS,
MeasureType.AIR_SOURCE_HEAT_PUMP,
MeasureType.GAS_BOILER_UPGRADE,
MeasureType.SYSTEM_TUNE_UP,
MeasureType.SYSTEM_TUNE_UP_ZONED,
),
lambda: recommend_heating(effective_epc, products, planning_restrictions),
),
(
admitted(MeasureType.SECONDARY_HEATING_REMOVAL),
lambda: recommend_secondary_heating_removal(effective_epc, products),
),
(
admitted(MeasureType.SOLAR_PV),
lambda: recommend_solar(
effective_epc, products, solar_potential, planning_restrictions
),
),
)
found = [thunk() for is_admitted, thunk in generators if is_admitted]
applicable = [
recommendation for recommendation in found if recommendation is not None
]

28
repositories/epc/epc_postgres_repository.py
View file

@ -1,7 +1,7 @@
from __future__ import annotations
from collections.abc import Sequence
from datetime import date
from datetime import date, datetime
from typing import Optional, Protocol, TypeVar
from sqlmodel import Session, col, delete, select
@ -57,6 +57,24 @@ def _require(value: Optional[_T], field: str) -> _T:
return value
def _as_date(value: object) -> date:
"""Normalise an ``epc_property`` date column value to a ``date``.
The FE-owned date columns (``inspection_date`` / ``completion_date`` /
``registration_date``) are Postgres ``timestamp``s even though the SQLModel
mirror types them ``str`` (it stores the writer's ``isoformat()`` string).
So a read hands back a ``datetime``, while a value still in flight may be
the ISO string — accept both.
"""
if isinstance(value, datetime):
return value.date()
if isinstance(value, date):
return value
if isinstance(value, str):
return date.fromisoformat(value)
raise TypeError(f"unexpected inspection_date value: {value!r}")
class _HasEpcPropertyId(Protocol):
epc_property_id: int
@ -425,7 +443,7 @@ class EpcPostgresRepository(EpcRepository):
return EpcPropertyData(
dwelling_type=p.dwelling_type,
inspection_date=date.fromisoformat(p.inspection_date),
inspection_date=_as_date(p.inspection_date),
tenure=p.tenure,
transaction_type=p.transaction_type,
address_line_1=_require(p.address_line_1, "address_line_1"),
@ -480,12 +498,10 @@ class EpcPostgresRepository(EpcRepository):
pressure_test=p.pressure_test,
language_code=p.language_code,
completion_date=(
date.fromisoformat(p.completion_date) if p.completion_date else None
_as_date(p.completion_date) if p.completion_date else None
),
registration_date=(
date.fromisoformat(p.registration_date)
if p.registration_date
else None
_as_date(p.registration_date) if p.registration_date else None
),
measurement_type=p.measurement_type,
conservatory_type=p.conservatory_type,

4
repositories/product/product_json_repository.py
View file

@ -6,7 +6,7 @@ from typing import Any, cast
from domain.modelling.contingencies import contingency_rate
from domain.modelling.product import Product
from repositories.product.product_repository import ProductRepository
from repositories.product.product_repository import ProductNotFound, ProductRepository
class ProductJsonRepository(ProductRepository):
@ -33,7 +33,7 @@ class ProductJsonRepository(ProductRepository):
def get(self, measure_type: str) -> Product:
entry: Any = self._entries.get(measure_type)
if entry is None:
raise ValueError(f"no product for measure type {measure_type!r}")
raise ProductNotFound(f"no product for measure type {measure_type!r}")
if not isinstance(entry, dict):
raise ValueError(f"product {measure_type!r} entry is not an object")
typed_entry: dict[str, Any] = cast("dict[str, Any]", entry)

6
repositories/product/product_postgres_repository.py
View file

@ -5,7 +5,7 @@ from sqlmodel import Session, col, select
from domain.modelling.contingencies import contingency_rate
from domain.modelling.product import Product
from infrastructure.postgres.product_table import MaterialRow
from repositories.product.product_repository import ProductRepository
from repositories.product.product_repository import ProductNotFound, ProductRepository
# The domain ``MeasureType`` vocabulary and the catalogue's ``material.type``
@ -47,7 +47,9 @@ class ProductPostgresRepository(ProductRepository):
.order_by(col(MaterialRow.id))
).first()
if row is None:
raise ValueError(f"no active product for measure type {measure_type!r}")
raise ProductNotFound(
f"no active product for measure type {measure_type!r}"
)
if row.total_cost is None:
raise ValueError(f"product {measure_type!r} has no total_cost")
return Product(

23
repositories/product/product_repository.py
View file

@ -1,10 +1,18 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Optional
from domain.modelling.product import Product
class ProductNotFound(ValueError):
"""Raised when the catalogue has no active entry for a Measure Type. A
subclass of ``ValueError`` so existing callers that catch ``ValueError``
keep working, while callers that only want to know *whether* a row exists
(see ``get_optional``) can catch this case alone."""
class ProductRepository(ABC):
"""Loads Products from the catalogue, abstracting the data source (a
Postgres-backed materials table today; a JSON file for costs the ETL does
@ -13,6 +21,17 @@ class ProductRepository(ABC):
@abstractmethod
def get(self, measure_type: str) -> Product:
"""Return the Product for a Measure Type, raising if there is no active
catalogue entry."""
"""Return the Product for a Measure Type, raising ``ProductNotFound``
if there is no active catalogue entry."""
...
def get_optional(self, measure_type: str) -> Optional[Product]:
"""Return the Product for a Measure Type, or None when the catalogue has
no active entry. For measures whose cost is composed off-catalogue (e.g.
ASHP, priced from the rate sheet per ADR-0025) the catalogue row is read
only for its id, so a missing row is not an error — the measure is still
offered, just without a ``material_id``."""
try:
return self.get(measure_type)
except ProductNotFound:
return None

14
repositories/property/property_postgres_repository.py
View file

@ -2,8 +2,9 @@ from __future__ import annotations
from typing import Optional, cast
from sqlalchemy import Table
from sqlalchemy import Table, func
from sqlalchemy import select as sa_select
from sqlalchemy import update as sa_update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlmodel import Session, col, select
@ -116,6 +117,17 @@ class PropertyPostgresRepository(PropertyRepository):
return {}
return self._spatial_repo.get_for_uprns(uprns)
def mark_modelled(self, property_id: int, *, has_recommendations: bool) -> None:
# The old engine set `has_recommendations` per Property; we mirror it and
# bump `updated_at` (DB clock) so a new-process run is datable against the
# 2026-06-01 cutoff. Does not commit — the Unit of Work owns the txn.
stmt = (
sa_update(self._table)
.where(self._table.c.id == property_id)
.values(has_recommendations=has_recommendations, updated_at=func.now())
)
self._session.execute(stmt) # pyright: ignore[reportDeprecated]
def insert_all(self, rows: list[PropertyIdentityInsert]) -> int:
if not rows:
return 0

9
repositories/property/property_repository.py
View file

@ -47,6 +47,15 @@ class PropertyRepository(ABC):
input ids."""
...
@abstractmethod
def mark_modelled(self, property_id: int, *, has_recommendations: bool) -> None:
"""Record that a Property has been run through the modelling pipeline:
set ``has_recommendations`` (the old engine's per-Property marker — true
when the Plan carries measures) and bump ``updated_at`` so the run is
datable (a first-run under the new process is ``updated_at >=
2026-06-01``). Idempotent — re-running overwrites the same row."""
...
@abstractmethod
def insert_all(self, rows: list[PropertyIdentityInsert]) -> int:
"""Bulk-insert identity rows, skipping any whose ``(portfolio_id, uprn)``

68
scripts/e2e_common.py Normal file
View file

@ -0,0 +1,68 @@
"""Shared configuration + client plumbing for the local e2e runner scripts
(``run_modelling_e2e`` and ``run_first_run_e2e``).
Loads ``backend/.env`` and builds the DB engine from the FastAPI-layer ``DB_*``
vars (the ``infrastructure/postgres`` layer reads ``POSTGRES_*``, which the .env
does not carry), plus an S3-backed ``ParquetReader`` for the geospatial
repository. Secrets live in the .env and the ambient ``~/.aws`` profile; this
module never hard-codes them.
"""
from __future__ import annotations
import io
import os
from pathlib import Path
from typing import Any, cast
import boto3
import pandas as pd
from sqlalchemy import Engine, create_engine
from repositories.geospatial.geospatial_s3_repository import ParquetReader
_REPO_ROOT = Path(__file__).resolve().parents[1]
ENV_PATH = _REPO_ROOT / "backend" / ".env"
def load_env(path: Path = ENV_PATH) -> None:
"""Load `KEY=value` lines from `backend/.env` into the environment (without
overriding anything already set), so the DB creds + API tokens are present."""
if not path.exists():
return
for raw in path.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
def db_url() -> str:
"""The connection string from the FastAPI-layer `DB_*` env vars."""
env = os.environ
return (
f"postgresql+psycopg2://{env['DB_USERNAME']}:{env['DB_PASSWORD']}"
f"@{env['DB_HOST']}:{env['DB_PORT']}/{env['DB_NAME']}"
)
def build_engine() -> Engine:
"""A connection-pooled engine to the target DB (DB_* creds)."""
return create_engine(
db_url(), pool_pre_ping=True, connect_args={"connect_timeout": 10}
)
def s3_parquet_reader(bucket: str) -> ParquetReader:
"""A `ParquetReader` (key -> DataFrame) backed by `bucket` in S3, for the
`GeospatialS3Repository`. AWS creds come from the ambient `~/.aws` profile;
pyarrow reads the parquet bytes (s3fs is not installed here)."""
# boto3 ships only partial type stubs, so the client is an untyped boundary.
client = cast(Any, boto3.client("s3")) # pyright: ignore[reportUnknownMemberType]
def read(key: str) -> pd.DataFrame:
body = cast(bytes, client.get_object(Bucket=bucket, Key=key)["Body"].read())
return pd.read_parquet(io.BytesIO(body))
return read

162
scripts/run_first_run_e2e.py Normal file
View file

@ -0,0 +1,162 @@
"""Run the **full** ``AraFirstRunPipeline`` (Ingestion → Baseline → Modelling)
end-to-end against the real database, locally.
This is the production pipeline the ``ara_first_run`` Lambda runs, driven from a
shell instead of an SQS event. The Lambda ``handler`` itself cannot run locally —
``applications/ara_first_run/handler.py::_source_clients_from_env`` deliberately
raises until the deploy/Terraform wiring lands (#1136). So this script composes
the same pipeline directly via the existing ``build_first_run_pipeline`` seam,
supplying the three source clients that ``run_modelling_e2e`` already proves out
(EPC API, geospatial S3, Google Solar), then calls ``dispatch_first_run``.
How it differs from ``run_modelling_e2e``:
* It runs the **real Ingestion stage** — fetches each Property's EPC by UPRN,
resolves spatial + Google Solar, and **persists** them (``epc_property`` /
``property_details_spatial`` / ``solar``) — then Baseline, then Modelling.
``run_modelling_e2e`` does ingestion inline and only models.
* **There is no inspect-only mode**: the stages persist as they go (ADR-0012),
so any run writes to the DB. This script is gated behind ``--confirm``; without
it the script previews what it would do and exits.
* **The modelling batch is all-or-nothing**: each stage commits once per batch,
so one Property raising aborts the whole batch (no per-Property recovery like
``run_modelling_e2e``). Make sure the inputs are clean first.
Measure scoping comes **only from the Scenario's exclusions** — the pipeline
threads no ``--measures`` override (issue #1130). So if the live ``material``
catalogue cannot price/represent a measure a Property is eligible for (today:
``secondary_heating_removal``, absent from the ``material.type`` enum), that
Property's modelling raises and aborts the batch. Exclude it on the Scenario
first, e.g.::
UPDATE scenario SET exclusions = '{secondary_heating_removal}' WHERE id = 1266;
EPC Prediction (ADR-0031) is left **off** — its Landlord-Override attributes
reader is not wired here, so an EPC-less Property is not gap-filled.
Config + secrets are loaded exactly as ``run_modelling_e2e`` does: ``backend/.env``
for the DB creds (``DB_*``), the EPC Bearer token (``OPEN_EPC_API_TOKEN``), the
Google Solar key (``GOOGLE_SOLAR_API_KEY``) and the S3 bucket (``DATA_BUCKET``);
AWS creds from the ambient ``~/.aws`` profile. Run from the worktree root::
# preview only (no writes): print what would run, then exit
python -m scripts.run_first_run_e2e --scenario-ids 1266 --portfolio-id 785 \
709634 709635 709636
# actually run the full pipeline and persist (Ingestion -> Baseline -> Modelling)
python -m scripts.run_first_run_e2e --scenario-ids 1266 --portfolio-id 785 \
--confirm 709634 709635 709636
"""
from __future__ import annotations
import argparse
import os
import sys
from pathlib import Path
from uuid import uuid4
_REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(_REPO_ROOT)) # worktree root first — avoid the import trap
from applications.ara_first_run.ara_first_run_trigger_body import ( # noqa: E402
AraFirstRunTriggerBody,
)
from applications.ara_first_run.handler import ( # noqa: E402
build_first_run_pipeline,
dispatch_first_run,
)
from infrastructure.epc_client.epc_client_service import EpcClientService # noqa: E402
from infrastructure.solar.google_solar_api_client import ( # noqa: E402
GoogleSolarApiClient,
)
from repositories.geospatial.geospatial_s3_repository import ( # noqa: E402
GeospatialS3Repository,
)
from repositories.postgres_unit_of_work import PostgresUnitOfWork # noqa: E402
from scripts.e2e_common import ( # noqa: E402
ENV_PATH,
build_engine,
load_env,
s3_parquet_reader,
)
from sqlmodel import Session # noqa: E402
def _parse_ids(raw: str) -> list[int]:
"""Parse a comma-separated id list (e.g. ``--scenario-ids 1266,1270``)."""
return [int(token.strip()) for token in raw.split(",") if token.strip()]
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"property_ids", type=int, nargs="+", help="Property ids to run"
)
parser.add_argument(
"--scenario-ids",
required=True,
help="comma-separated Scenario ids to model against (exclusions come "
"from each Scenario)",
)
parser.add_argument(
"--portfolio-id", type=int, required=True, help="portfolio id for the run"
)
parser.add_argument(
"--confirm",
action="store_true",
default=False,
help="actually run the pipeline and WRITE to the DB (default: preview only)",
)
args = parser.parse_args()
scenario_ids = _parse_ids(args.scenario_ids)
load_env(ENV_PATH)
engine = build_engine()
body = AraFirstRunTriggerBody(
# task/sub_task drive the Lambda SubTask lifecycle only; running the
# pipeline directly bypasses the @subtask_handler decorator, so synthetic
# ids satisfy validation without touching the task tables.
task_id=uuid4(),
sub_task_id=uuid4(),
portfolio_id=args.portfolio_id,
property_ids=args.property_ids,
scenario_ids=scenario_ids,
)
print(
f"full AraFirstRunPipeline (Ingestion -> Baseline -> Modelling) · "
f"{len(args.property_ids)} propertie(s) · scenarios {scenario_ids} · "
f"portfolio {args.portfolio_id}"
)
if not args.confirm:
print(
"\nPREVIEW ONLY — no writes. This run WOULD fetch + persist EPC/"
"spatial/solar, rebaseline, and model+persist Plans for:\n"
f" properties: {args.property_ids}\n"
"Re-run with --confirm to execute. NOTE: the modelling batch is "
"all-or-nothing; ensure each Scenario excludes any measure the live "
"catalogue cannot price (e.g. secondary_heating_removal)."
)
return
epc_fetcher = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"])
geospatial_repo = GeospatialS3Repository(
s3_parquet_reader(os.environ["DATA_BUCKET"])
)
solar_fetcher = GoogleSolarApiClient(os.environ["GOOGLE_SOLAR_API_KEY"])
pipeline = build_first_run_pipeline(
unit_of_work=lambda: PostgresUnitOfWork(lambda: Session(engine)),
epc_fetcher=epc_fetcher,
geospatial_repo=geospatial_repo,
solar_fetcher=solar_fetcher,
)
print("running... (Ingestion -> Baseline -> Modelling, persisting per stage)\n")
dispatch_first_run(body.model_dump(), pipeline=pipeline)
print("done — EPC/spatial/solar + Baseline + Plans persisted for the batch.")
if __name__ == "__main__":
main()

220
scripts/run_modelling_e2e.py
View file

@ -14,10 +14,19 @@ persists **nothing** (the run is for inspecting recommendations); pass
To keep the inspected recommendations identical to what gets stored, **both
modes price against the live ``material`` catalogue (read-only)** and model
against a real **Scenario** read from the DB — not the JSON sample catalogue.
Pass `--scenario-id` to target a real Scenario (its ``goal_value`` drives the
band); without it the run synthesises an Increasing-EPC-to-``--goal`` Scenario.
``--measures`` restricts the run to a comma-separated set of measure types
(mirroring the legacy `inclusions`) — e.g. only HHRSH + Solar PV.
Pass `--scenario-id` to target a real Scenario; its ``goal_value`` drives the
band and **its ``exclusions`` drive which measures the run considers** (the live
scenario table persists exclusions only, no inclusions). Without `--scenario-id`
the run synthesises an Increasing-EPC-to-``--goal`` Scenario with no exclusions.
`--measures` / `--exclude-measures` are optional overlays layered on top of the
Scenario's own exclusions.
KNOWN GOTCHA: the live ``material.type`` enum does not yet carry
``secondary_heating_removal``, so a Property with a lodged secondary heater
crashes the catalogue read for that measure. Until the catalogue stocks it, pass
``--exclude-measures secondary_heating_removal`` (an ASHP row is also absent, but
ASHP is priced off the rate sheet so it degrades to ``material_id=None`` rather
than crashing — no flag needed).
Config: loads `backend/.env` for the DB creds (`DB_*`), the EPC API token
(`OPEN_EPC_API_TOKEN` — the Bearer token for the new gov API), the Google Solar
@ -25,13 +34,13 @@ key (`GOOGLE_SOLAR_API_KEY`) and the S3
reference bucket (`DATA_BUCKET`) — the agent never sees the secrets. AWS creds
come from the ambient `~/.aws` profile. Run from the worktree root:
# inspect only (no DB writes), HHRSH + Solar PV, against Scenario 1263:
python -m scripts.run_modelling_e2e --scenario-id 1263 \
--measures high_heat_retention_storage_heaters,solar_pv 115 116 117
# same run, but persist the Plans (needs --portfolio-id):
python -m scripts.run_modelling_e2e --scenario-id 1263 --portfolio-id 4 \
--measures high_heat_retention_storage_heaters,solar_pv --persist 115 116 117
python -m scripts.run_modelling_e2e --no-solar 115 116 # skip the Google leg
# inspect only (no DB writes), Scenario 1266, measures from the Scenario:
python -m scripts.run_modelling_e2e --scenario-id 1266 \
--exclude-measures secondary_heating_removal 709634 709635 709636
# same run, but persist EPC + spatial + solar + Plan (needs --portfolio-id):
python -m scripts.run_modelling_e2e --scenario-id 1266 --portfolio-id 785 \
--persist --exclude-measures secondary_heating_removal 709634 709635
python -m scripts.run_modelling_e2e --no-solar 709634 709635 # skip Google leg
Per Property the spatial reference (S3 Open-UPRN parquet) gives the planning
protections (conservation/listed/heritage — gate the wall + solar measures) and
@ -44,14 +53,10 @@ Google leg.
from __future__ import annotations
import argparse
import io
import os
import sys
from pathlib import Path
from typing import Any, Optional, cast
import boto3
import pandas as pd
from typing import Any, Optional
_REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(_REPO_ROOT)) # worktree root first — avoid the import trap
@ -59,10 +64,14 @@ sys.path.insert(0, str(_REPO_ROOT)) # worktree root first — avoid the import
from datatypes.epc.domain.epc_property_data import EpcPropertyData # noqa: E402
from domain.geospatial.planning_restrictions import PlanningRestrictions # noqa: E402
from domain.geospatial.spatial_reference import SpatialReference # noqa: E402
from domain.modelling.considered_measures import ( # noqa: E402
combine_considered_measures,
)
from domain.modelling.measure_type import MeasureType # noqa: E402
from domain.modelling.plan import Plan, PlanMeasure # noqa: E402
from domain.modelling.recommendation import Recommendation # noqa: E402
from domain.modelling.scenario import Scenario # noqa: E402
from harness.console import run_modelling # noqa: E402
from harness.console import candidate_recommendations, run_modelling # noqa: E402
from harness.plan_table import format_plan_table # noqa: E402
from infrastructure.epc_client.epc_client_service import EpcClientService # noqa: E402
from infrastructure.solar.google_solar_api_client import ( # noqa: E402
@ -71,7 +80,6 @@ from infrastructure.solar.google_solar_api_client import ( # noqa: E402
)
from repositories.geospatial.geospatial_s3_repository import ( # noqa: E402
GeospatialS3Repository,
ParquetReader,
)
from repositories.product.product_postgres_repository import ( # noqa: E402
ProductPostgresRepository,
@ -80,48 +88,18 @@ from repositories.postgres_unit_of_work import PostgresUnitOfWork # noqa: E402
from repositories.scenario.scenario_postgres_repository import ( # noqa: E402
ScenarioPostgresRepository,
)
from sqlalchemy import Engine, create_engine, text # noqa: E402
from scripts.e2e_common import ( # noqa: E402
ENV_PATH,
build_engine,
load_env,
s3_parquet_reader,
)
from sqlalchemy import Engine, text # noqa: E402
from sqlmodel import Session # noqa: E402
_ENV_PATH = _REPO_ROOT / "backend" / ".env"
_MARKDOWN_PATH = Path("modelling_e2e.md")
_CSV_PATH = Path("modelling_e2e.csv")
def _load_env(path: Path) -> None:
"""Load `KEY=value` lines from `backend/.env` into the environment (without
overriding anything already set), so the DB creds + EPC token are present."""
if not path.exists():
return
for raw in path.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
def _db_url() -> str:
"""The connection string from the FastAPI-layer `DB_*` env vars."""
env = os.environ
return (
f"postgresql+psycopg2://{env['DB_USERNAME']}:{env['DB_PASSWORD']}"
f"@{env['DB_HOST']}:{env['DB_PORT']}/{env['DB_NAME']}"
)
def _s3_parquet_reader(bucket: str) -> ParquetReader:
"""A `ParquetReader` (key -> DataFrame) backed by `bucket` in S3, for the
`GeospatialS3Repository`. AWS creds come from the ambient `~/.aws` profile;
pyarrow reads the parquet bytes (s3fs is not installed here)."""
# boto3 ships only partial type stubs, so the client is an untyped boundary.
client = cast(Any, boto3.client("s3")) # pyright: ignore[reportUnknownMemberType]
def read(key: str) -> pd.DataFrame:
body = cast(bytes, client.get_object(Bucket=bucket, Key=key)["Body"].read())
return pd.read_parquet(io.BytesIO(body))
return read
_CANDIDATES_CSV_PATH = Path("modelling_e2e_candidates.csv")
def _spatial_for(repo: GeospatialS3Repository, uprn: int) -> Optional[SpatialReference]:
@ -152,13 +130,6 @@ def _solar_insights_for(
return None # no Google solar coverage at this point — model without it
def _engine() -> Engine:
"""A connection-pooled engine to DevAssessmentModelDB (DB_* creds)."""
return create_engine(
_db_url(), pool_pre_ping=True, connect_args={"connect_timeout": 10}
)
def _uprns_for(engine: Engine, property_ids: list[int]) -> dict[int, Optional[int]]:
"""Read each Property's UPRN from the DB (read-only)."""
with engine.connect() as conn:
@ -191,6 +162,21 @@ def _parse_measures(raw: Optional[str]) -> Optional[frozenset[MeasureType]]:
)
def _resolve_considered(
allowlist: Optional[frozenset[MeasureType]],
excluded: Optional[frozenset[MeasureType]],
) -> Optional[frozenset[MeasureType]]:
"""Combine the `--measures` allowlist with the `--exclude-measures` set. With
no exclusions the allowlist is returned unchanged (None = every measure).
With exclusions the result is (the allowlist, or every measure) minus the
excluded types — so `--exclude-measures secondary_heating_removal` considers
every measure except that one, without enumerating the rest."""
if not excluded:
return allowlist
base = allowlist if allowlist is not None else frozenset(MeasureType)
return base - excluded
def _context_summary(
spatial: Optional[SpatialReference], solar_insights: Optional[dict[str, Any]]
) -> str:
@ -221,6 +207,53 @@ def _measure_summary(measure: PlanMeasure) -> str:
)
def _candidate_lines(
recommendations: list[Recommendation], selected: set[MeasureType]
) -> list[str]:
"""Render every candidate Option (the full menu the Generators produced,
not just the Plan the Optimiser selected) with its per-Option cost, flagging
the Options that made it into the Plan — so measures the Optimiser passed
over (e.g. an ASHP it found too costly for the target band) are visible."""
lines: list[str] = []
for recommendation in recommendations:
for option in recommendation.options:
cost = option.cost
cost_note = (
f"£{cost.total:,.0f} (+{cost.contingency_rate * 100:.0f}% cont.)"
if cost is not None
else "no cost"
)
flag = " ✓ SELECTED" if option.measure_type in selected else ""
lines.append(
f" [{recommendation.surface}] {option.measure_type} · "
f"{cost_note}{flag} — {option.description}"
)
return lines
def _candidate_csv_rows(
property_id: int,
uprn: Optional[int],
recommendations: list[Recommendation],
selected: set[MeasureType],
) -> list[str]:
"""One CSV row per candidate Option: the full measure menu with cost,
contingency, and whether the Optimiser selected it."""
rows: list[str] = []
for recommendation in recommendations:
for option in recommendation.options:
cost = option.cost
total = f"{cost.total:.2f}" if cost is not None else ""
contingency = f"{cost.contingency_rate:.4f}" if cost is not None else ""
chosen = "yes" if option.measure_type in selected else "no"
description = option.description.replace(",", ";")
rows.append(
f"{property_id},{uprn or ''},{recommendation.surface},"
f"{option.measure_type},{total},{contingency},{chosen},{description}"
)
return rows
def _persist(
engine: Engine,
*,
@ -258,6 +291,12 @@ def _persist(
portfolio_id=portfolio_id,
is_default=scenario.is_default,
)
# Mark the Property as run under the new process (old engine's
# `has_recommendations` marker + a bumped `updated_at`); the modelling
# compute above runs on in-memory fakes, so this DB UoW must set it.
uow.property.mark_modelled(
property_id, has_recommendations=bool(plan.measures)
)
uow.commit()
@ -275,7 +314,15 @@ def main() -> None:
parser.add_argument(
"--measures",
default=None,
help="comma-separated measure types to consider (default: all)",
help="optional override: comma-separated measure types to consider. The "
"Scenario's exclusions already drive this; the flag narrows it further.",
)
parser.add_argument(
"--exclude-measures",
default=None,
help="optional override: comma-separated measure types to exclude on top "
"of the Scenario's own exclusions (e.g. secondary_heating_removal, which "
"the live catalogue does not yet stock)",
)
parser.add_argument(
"--portfolio-id",
@ -299,15 +346,17 @@ def main() -> None:
if args.persist and (args.scenario_id is None or args.portfolio_id is None):
parser.error("--persist requires --scenario-id and --portfolio-id")
_load_env(_ENV_PATH)
load_env(ENV_PATH)
# The new gov EPC API (Bearer) authenticates with OPEN_EPC_API_TOKEN — the
# name is misleading; EPC_AUTH_TOKEN is dead (403). Verified against the
# /api/domestic/search endpoint.
epc_client = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"])
geospatial = GeospatialS3Repository(_s3_parquet_reader(os.environ["DATA_BUCKET"]))
geospatial = GeospatialS3Repository(s3_parquet_reader(os.environ["DATA_BUCKET"]))
solar_client = GoogleSolarApiClient(os.environ["GOOGLE_SOLAR_API_KEY"])
engine = _engine()
considered = _parse_measures(args.measures)
engine = build_engine()
cli_considered = _resolve_considered(
_parse_measures(args.measures), _parse_measures(args.exclude_measures)
)
uprns = _uprns_for(engine, args.property_ids)
# One read-only session for the live `material` catalogue, reused across the
# batch so both store and no-store runs price against the same DB rows.
@ -318,6 +367,12 @@ def main() -> None:
if args.scenario_id is not None
else None
)
# The Scenario's own exclusions drive which measures the run considers; the
# --measures/--exclude-measures flags are an optional override layered on top.
considered = combine_considered_measures(
scenario.considered_measures() if scenario is not None else None,
cli_considered,
)
target = (
f"scenario {scenario.id} (band {scenario.goal_value})"
@ -335,6 +390,10 @@ def main() -> None:
csv_rows: list[str] = [
"property_id,uprn,baseline_sap,post_sap,measures,measure_types,cost_of_works"
]
candidate_csv_rows: list[str] = [
"property_id,uprn,surface,measure_type,cost_total,contingency_rate,"
"selected,description"
]
for property_id in args.property_ids:
uprn = uprns.get(property_id)
@ -361,6 +420,15 @@ def main() -> None:
scenario=scenario,
print_table=False,
)
# The full candidate menu (every Generator Option + its cost), so
# measures the Optimiser did not select are still visible.
candidates: list[Recommendation] = candidate_recommendations(
epc,
planning_restrictions=restrictions,
solar_insights=solar_insights,
considered_measures=considered,
products=products,
)
if args.persist:
assert scenario is not None # guaranteed by the --persist guard
_persist(
@ -389,7 +457,9 @@ def main() -> None:
continue
measure_types = [m.measure_type for m in plan.measures]
selected: set[MeasureType] = {m.measure_type for m in plan.measures}
context = _context_summary(spatial, solar_insights)
candidate_lines = _candidate_lines(candidates, selected)
header = (
f"=== Property {property_id} (uprn {uprn}) === "
f"SAP {plan.baseline.sap_continuous:.1f} -> {plan.post_sap_continuous:.1f} "
@ -397,6 +467,9 @@ def main() -> None:
)
print(header)
print(format_plan_table(plan))
print(f" candidate measures considered ({len(candidate_lines)} option(s)):")
for candidate_line in candidate_lines:
print(candidate_line)
print()
md_lines.append(f"## Property {property_id} (uprn {uprn})\n")
@ -405,19 +478,30 @@ def main() -> None:
f"· {len(plan.measures)} measure(s) · cost £{plan.cost_of_works:,.0f} "
f"· {context}\n"
)
md_lines.append("**Selected Plan**\n")
md_lines.extend(_measure_summary(m) for m in plan.measures)
md_lines.append("")
md_lines.append("**All candidate measures (cost per measure)**\n")
md_lines.extend(candidate_lines)
md_lines.append("")
csv_rows.append(
f"{property_id},{uprn},{plan.baseline.sap_continuous:.2f},"
f"{plan.post_sap_continuous:.2f},{len(plan.measures)},"
f"{'|'.join(measure_types)},{plan.cost_of_works:.0f}"
)
candidate_csv_rows.extend(
_candidate_csv_rows(property_id, uprn, candidates, selected)
)
catalogue_session.close()
_MARKDOWN_PATH.write_text("\n".join(md_lines) + "\n", encoding="utf-8")
_CSV_PATH.write_text("\n".join(csv_rows) + "\n", encoding="utf-8")
_CANDIDATES_CSV_PATH.write_text(
"\n".join(candidate_csv_rows) + "\n", encoding="utf-8"
)
print(f"wrote {_MARKDOWN_PATH.resolve()}")
print(f"wrote {_CSV_PATH.resolve()}")
print(f"wrote {_CANDIDATES_CSV_PATH.resolve()}")
if __name__ == "__main__":

17
tests/domain/modelling/test_considered_measures.py
View file

@ -8,7 +8,10 @@ Options; a Recommendation left with no allowed Option is dropped entirely. A
None allowlist means "consider everything" (today's unrestricted behaviour).
"""
from domain.modelling.considered_measures import restrict_to_considered_measures
from domain.modelling.considered_measures import (
combine_considered_measures,
restrict_to_considered_measures,
)
from domain.modelling.measure_type import MeasureType
from domain.modelling.recommendation import MeasureOption, Recommendation
from domain.modelling.simulation import EpcSimulation
@ -68,6 +71,18 @@ def test_drops_recommendations_with_no_allowed_option() -> None:
assert surfaces == {"Heating & Hot Water", "Solar PV"}
def test_combine_treats_none_as_all_and_intersects_two_allowlists() -> None:
# Arrange
a = frozenset({MeasureType.SOLAR_PV, MeasureType.LOFT_INSULATION})
b = frozenset({MeasureType.SOLAR_PV, MeasureType.CAVITY_WALL_INSULATION})
# Act / Assert — None means "all", so it never narrows; two sets intersect.
assert combine_considered_measures(None, b) == b
assert combine_considered_measures(a, None) == a
assert combine_considered_measures(None, None) is None
assert combine_considered_measures(a, b) == frozenset({MeasureType.SOLAR_PV})
def test_filters_options_within_a_kept_recommendation() -> None:
# Arrange — HHRSH is allowed but the competing ASHP bundle is not.
considered = frozenset(

41
tests/domain/modelling/test_heating_recommendation.py
View file

@ -8,10 +8,14 @@ later slices. Detection + pricing only; impact is produced by scoring (ADR-0016)
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.geospatial.planning_restrictions import PlanningRestrictions
from domain.modelling.generators.heating_recommendation import recommend_heating
from domain.modelling.measure_type import MeasureType
from domain.modelling.product import Product
from domain.modelling.recommendation import Recommendation
from domain.modelling.simulation import HeatingOverlay
from repositories.product.product_repository import ProductRepository
from repositories.product.product_repository import (
ProductNotFound,
ProductRepository,
)
from tests.domain.modelling._elmhurst_recommendation import (
parse_recommendation_summary,
)
@ -170,6 +174,41 @@ def test_gas_boiler_house_yields_an_ashp_bundle() -> None:
)
class _StubProductsWithoutAshp(ProductRepository):
"""A catalogue with no ASHP row. ASHP's cost is composed from the rate sheet
(ADR-0025) and the catalogue row is read only for its id, so a missing row
must not suppress the bundle — it just carries no material_id."""
def get(self, measure_type: str) -> Product:
if measure_type == MeasureType.AIR_SOURCE_HEAT_PUMP:
raise ProductNotFound(f"no active product for {measure_type!r}")
return Product(
measure_type=measure_type, unit_cost_per_m2=3500.0, contingency_rate=0.26
)
def test_ashp_bundle_offered_when_catalogue_lacks_an_ashp_product() -> None:
# Arrange — a mains-gas house (ASHP-eligible) priced against a catalogue with
# no ASHP row; ASHP is costed from the rate sheet, so the bundle must still
# be offered, just without a material id.
baseline: EpcPropertyData = _gas_boiler_house()
# Act
recommendation: Recommendation | None = recommend_heating(
baseline, _StubProductsWithoutAshp()
)
# Assert — the ASHP bundle is still offered, carrying its composite cost and
# no material id.
assert recommendation is not None
option = next(
o for o in recommendation.options if o.measure_type == "air_source_heat_pump"
)
assert option.material_id is None
assert option.cost is not None
assert option.cost.total > 0.0
def test_ashp_bundle_carries_the_composite_per_dwelling_cost() -> None:
# Arrange — a mains-gas regular boiler with a cylinder (90 m2, 7 habitable
# rooms): the ASHP reuses the existing wet system (ADR-0025).

43
tests/domain/modelling/test_scenario.py Normal file
View file

@ -0,0 +1,43 @@
"""The Scenario's measure scoping: its exclusions imply the allowlist the run
considers (the live `scenario` table persists exclusions only — no inclusions)."""
from domain.modelling.measure_type import MeasureType
from domain.modelling.scenario import Scenario
def _scenario(exclusions: frozenset[MeasureType]) -> Scenario:
return Scenario(
id=1,
goal="Increasing EPC",
goal_value="C",
budget=None,
is_default=True,
exclusions=exclusions,
)
def test_no_exclusions_considers_every_measure() -> None:
# Arrange
scenario = _scenario(frozenset())
# Act
considered = scenario.considered_measures()
# Assert — None means "consider all" (the unrestricted default).
assert considered is None
def test_exclusions_imply_the_complement_allowlist() -> None:
# Arrange — exclude solar PV and ASHP.
scenario = _scenario(
frozenset({MeasureType.SOLAR_PV, MeasureType.AIR_SOURCE_HEAT_PUMP})
)
# Act
considered = scenario.considered_measures()
# Assert — every modelled measure survives except the two excluded ones.
assert considered is not None
assert MeasureType.SOLAR_PV not in considered
assert MeasureType.AIR_SOURCE_HEAT_PUMP not in considered
assert considered == frozenset(MeasureType) - scenario.exclusions

144
tests/domain/sap10_calculator/worksheet/_elmhurst_worksheet_001431_case44.py Normal file
View file

@ -0,0 +1,144 @@
"""Mapper-driven cascade pin against the Elmhurst P960-0001-001431
"simulated case 44" worksheet — a 2-storey mid-terrace with a NON-SEPARATED
(heated, type-4) DOUBLE-glazed CONSERVATORY.
Case 44 is the 1e-4 oracle for RdSAP 10 §6.1 (PDF p.49) + Table 25 (p.51).
The Summary §5 lodges: Floor Area 12.00 m², Glazed Perimeter 9.00 m,
Double Glazed Yes, thermally separated No, Room Height 1 Storey. From that
the §6.1 cascade derives (all verified against the P960 §3 to 1e-4):
- conservatory height = ground-floor room height = 2.60 m (1 storey);
- glazed WALL → window (27): A = perimeter × height = 9.0 × 2.60 = 23.40,
U = 1/(1/3.1 + 0.04) = 2.758 (Table 25 double 3.1 + §3.2 curtain);
- glazed ROOF → rooflight (27a): A = floor_area / cos(20°) = 12.77,
U = 1/(1/3.4 + 0.04) = 2.993 (Table 25 roof 3.4 + curtain);
- FLOOR → ground floor (28a): A = 12.00, U = 0.89 via BS EN ISO 13370
(uninsulated solid, 300 mm walls, P = glazed perimeter 9.0);
- the fully-glazed structure walls/roof bill at U=0 (the glazing IS the
window/rooflight) — they contribute nothing but DO count their glazed
area toward (31)/(36);
- TFA (4) += 12.00 → 95.38; volume (5) += 12.00 × 2.60 = 31.20 → 257.16.
Like the other `_elmhurst_worksheet_001431_case*` fixtures this does NOT
hand-build the EpcPropertyData: it routes the Summary PDF through
ElmhurstSiteNotesExtractor + from_elmhurst_site_notes so the pin exercises
the WHOLE extractor + mapper + calculator pipeline.
Source: user-simulated PDFs at `sap worksheets/golden fixture debugging/
simulated case 44/`. The Summary is mirrored into the tracked
`backend/documents_parser/tests/fixtures/Summary_001431_case44.pdf` so the
test runs without depending on the unstaged workspace.
Worksheet pin targets (P960-0001-001431, "11a. SAP rating" UK-average
rating block our cascade reproduces):
- (4) TFA, m² = 95.3800
- (5) Dwelling volume, m³ = 257.1630
- (27) Windows (31.5795 main + 64.5374 cons) = 96.1169
- (27a) Roof windows (conservatory glazed roof) = 38.2201
- (28a) Ground floor (10.7364 main + 10.6800) = 21.4164
- (29a) External walls = 35.5852
- (30) External roof = 7.4688
- (31) Total net area of external elements = 294.2900
- (33) Fabric heat loss, W/K = 207.3274
- (36) Thermal bridges (0.080 × (31)) = 23.5432
Per [[feedback-zero-error-strict]]: pins are abs <= 1e-4 against the PDF.
"""
from __future__ import annotations
import re
import subprocess
from pathlib import Path
from typing import Final
from backend.documents_parser.elmhurst_extractor import ElmhurstSiteNotesExtractor
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from datatypes.epc.domain.mapper import EpcPropertyDataMapper
# parents[0]=worksheet/, [1]=sap10_calculator/, [2]=domain/, [3]=tests/,
# [4]=repo root.
_SUMMARY_PDF: Final[Path] = (
Path(__file__).resolve().parents[4]
/ "backend" / "documents_parser" / "tests" / "fixtures"
/ "Summary_001431_case44.pdf"
)
LINE_4_TFA_M2: Final[float] = 95.3800
LINE_5_VOLUME_M3: Final[float] = 257.1630
LINE_27_WINDOWS_W_PER_K: Final[float] = 96.1169
LINE_27A_ROOF_WINDOWS_W_PER_K: Final[float] = 38.2201
LINE_28A_FLOOR_W_PER_K: Final[float] = 21.4164
LINE_29A_WALLS_W_PER_K: Final[float] = 35.5852
LINE_30_ROOF_W_PER_K: Final[float] = 7.4688
LINE_31_EXTERNAL_AREA_M2: Final[float] = 294.2900
LINE_33_FABRIC_W_PER_K: Final[float] = 207.3274
LINE_36_THERMAL_BRIDGING_W_PER_K: Final[float] = 23.5432
# Demand-side line refs (Jan column, UK-average rating block). These
# integrate the WHOLE §6.1 conservatory chain end-to-end:
# - (73) internal gains — the conservatory floor area enters TFA (4),
# which drives occupancy → §5 appliance/cooking/metabolic gains;
# - (83) solar gains — the glazed wall (E/W flux, 90° pitch) + glazed
# roof (E/W flux, 20° pitch) at Table 25 g=0.76, FF=0.70;
# - (95) useful gains = (84) total gains × the §7 utilisation factor —
# matches only when fabric (33), ventilation (38) AND gains (84) all
# agree, so it is the single tightest end-to-end conservatory pin;
# - (99) space heating per m² = (98c)/(4) — the integrated demand.
LINE_73_INTERNAL_GAINS_JAN_W: Final[float] = 625.1759
LINE_83_SOLAR_GAINS_JAN_W: Final[float] = 495.8655
LINE_95_USEFUL_GAINS_JAN_W: Final[float] = 1079.6510
LINE_99_SPACE_HEATING_PER_M2_KWH: Final[float] = 89.8073
# NB — the full SAP value (72.9517) + (272) CO2 (3241.8656) are NOT pinned
# here. The case-44 Summary PDF omits the House-Coal secondary heater
# (SAP 633, 60% eff) that the P960 worksheet's descriptor block carries
# (the same secondary as case 43); routed through the extractor the
# Summary therefore yields NO secondary system, and the residual SAP/CO2
# gap is exactly that missing secondary (main+secondary CO2 1927.31 +
# 563.92 = 2491.23 vs cascade 2141.46 → +349.77 ≈ the 350 kg deficit).
# This is a Summary-input defect, independent of §6.1 — every
# conservatory-affected line ref above reproduces the P960 EXACTLY.
def _summary_pdf_to_textract_style_pages(pdf_path: Path) -> list[str]:
"""Convert a Summary PDF into the per-page text format the
ElmhurstSiteNotesExtractor expects (label/value token sequences).
Mirror of the helper in the other `_elmhurst_worksheet_*` fixtures.
"""
info = subprocess.run(
["pdfinfo", str(pdf_path)], capture_output=True, text=True, check=True,
).stdout
m = re.search(r"Pages:\s+(\d+)", info)
if m is None:
raise RuntimeError(f"Could not parse page count from {pdf_path}")
page_count = int(m.group(1))
pages: list[str] = []
for i in range(1, page_count + 1):
layout = subprocess.run(
[
"pdftotext", "-layout", "-f", str(i), "-l", str(i),
str(pdf_path), "-",
],
capture_output=True, text=True, check=True,
).stdout
tokens: list[str] = []
for line in layout.splitlines():
if not line.strip():
tokens.append("")
continue
parts = [p for p in re.split(r"\s{2,}", line.strip()) if p]
tokens.extend(parts)
pages.append("\n".join(tokens))
return pages
def build_epc() -> EpcPropertyData:
"""Route the simulated case-44 Summary through extractor + mapper.
No hand-built EpcPropertyData — the extractor and mapper are part of
the test target. This module is a pin PROVIDER (build_epc + LINE_*
constants); the collected assertions live in
`test_section_cascade_pins`."""
pages = _summary_pdf_to_textract_style_pages(_SUMMARY_PDF)
site_notes = ElmhurstSiteNotesExtractor(pages).extract()
return EpcPropertyDataMapper.from_elmhurst_site_notes(site_notes)

75
tests/domain/sap10_calculator/worksheet/test_heat_transmission.py
View file

@ -38,6 +38,7 @@ from domain.sap10_calculator.worksheet.heat_transmission import (
from domain.sap10_calculator.worksheet.heat_transmission import (
_alt_wall_w_per_k, # pyright: ignore[reportPrivateUsage]
_joined_main_roof_descriptions, # pyright: ignore[reportPrivateUsage]
_main_roof_descriptions_by_kind, # pyright: ignore[reportPrivateUsage]
_part_geometry, # pyright: ignore[reportPrivateUsage]
_round_half_up, # pyright: ignore[reportPrivateUsage]
_window_bp_index, # pyright: ignore[reportPrivateUsage]
@ -82,6 +83,80 @@ def test_joined_main_roof_descriptions_keeps_pure_rr_fallback() -> None:
assert result == "Roof room(s), no insulation (assumed)"
def test_main_roof_descriptions_by_kind_splits_flat_from_pitched() -> None:
# Arrange — a cert with a pitched insulated main roof + a flat
# uninsulated extension. The deduplicated epc.roofs[] cannot be indexed
# 1:1 against the parts, so each part must match its own KIND's
# description: the flat part's "no insulation" must not leak onto the
# pitched part (which would force the whole pitched roof to U=2.30).
roofs = [
_Desc("Pitched, insulated (assumed)"),
_Desc("Flat, no insulation"),
_Desc("Roof room(s), no insulation (assumed)"),
]
# Act
pitched, flat = _main_roof_descriptions_by_kind(roofs)
# Assert — RR dropped; flat and pitched kept apart.
assert pitched == "Pitched, insulated (assumed)"
assert flat == "Flat, no insulation"
def test_mixed_flat_pitched_roof_does_not_contaminate_pitched_u_value() -> None:
# Arrange — 2-part dwelling: a 100 m² pitched insulated-assumed main
# roof (U=0.40) + a 2 m² flat uninsulated extension (U=2.30). Before the
# per-kind split, the joined "Pitched, insulated (assumed) | Flat, no
# insulation" description leaked the flat's "no insulation" onto the
# pitched part, billing the WHOLE roof at 2.30 (100×2.30 + 2×2.30 =
# 234.6 W/K). Correct: 100×0.40 + 2×2.30 = 44.6 W/K. Mirrors corpus
# cert 100010129331 (roof 110.5 -> 31.3 W/K, +13 -> 0 SAP).
main = make_building_part(
identifier=BuildingPartIdentifier.MAIN,
construction_age_band="C",
roof_construction=4,
floor_dimensions=[
make_floor_dimension(
total_floor_area_m2=100.0, room_height_m=2.5,
party_wall_length_m=0.0, heat_loss_perimeter_m=40.0, floor=0,
),
],
)
ext = make_building_part(
identifier=BuildingPartIdentifier.EXTENSION_1,
construction_age_band="C",
roof_construction=5,
floor_dimensions=[
make_floor_dimension(
total_floor_area_m2=2.0, room_height_m=2.5,
party_wall_length_m=0.0, heat_loss_perimeter_m=6.0, floor=0,
),
],
)
ext.roof_construction_type = "Flat"
epc = make_minimal_sap10_epc(
total_floor_area_m2=102.0,
country_code="ENG",
sap_building_parts=[main, ext],
)
epc.roofs = [
EnergyElement(
description="Pitched, insulated (assumed)",
energy_efficiency_rating=4, environmental_efficiency_rating=4,
),
EnergyElement(
description="Flat, no insulation",
energy_efficiency_rating=1, environmental_efficiency_rating=1,
),
]
# Act
result = heat_transmission_from_cert(epc)
# Assert — pitched main billed at its insulated U, not the flat's 2.30.
assert abs(result.roof_w_per_k - 44.6) <= 2.0
def test_part_geometry_floorless_part_honours_full_key_contract() -> None:
# Arrange — a building part lodged with NO sap_floor_dimensions (e.g.
# a party-wall-only or RR-only extension; observed on 5 certs in a

121
tests/domain/sap10_calculator/worksheet/test_section_cascade_pins.py
View file

@ -45,6 +45,7 @@ from tests.domain.sap10_calculator.worksheet import (
_elmhurst_worksheet_001431_case6 as _w001431_case6,
_elmhurst_worksheet_001431_case21 as _w001431_case21,
_elmhurst_worksheet_001431_case43 as _w001431_case43,
_elmhurst_worksheet_001431_case44 as _w001431_case44,
)
@ -370,6 +371,126 @@ def test_case43_detailed_rr_dryline_and_mixed_roof_match_pdf() -> None:
)
def test_case44_non_separated_conservatory_fabric_matches_pdf() -> None:
"""§3 fabric pin for simulated case 44 — a non-separated DOUBLE-glazed
conservatory (RdSAP 10 §6.1 + Table 25). The conservatory's glazed wall
bills as a window (27), its glazed roof as a rooflight (27a), its floor
adds a ground-loss term (28a), and its glazed wall + roof + floor areas
join (31)/(36); TFA (4) and volume (5) absorb its floor area + volume.
The main dwelling's walls (29a) / roof (30) are untouched — pinned to
guard against the conservatory leaking into the wrong element."""
# Arrange
epc = _w001431_case44.build_epc()
# Act
ht = heat_transmission_section_from_cert(epc)
dim = dimensions_from_cert(epc)
# Assert — §1 totals + §3 fabric, each at abs=1e-4.
_pin(dim.total_floor_area_m2, _w001431_case44.LINE_4_TFA_M2, "§1 (4) case44")
_pin(dim.volume_m3, _w001431_case44.LINE_5_VOLUME_M3, "§1 (5) case44")
_pin(
ht.windows_w_per_k,
_w001431_case44.LINE_27_WINDOWS_W_PER_K,
"§3 (27) case44",
)
_pin(
ht.roof_windows_w_per_k,
_w001431_case44.LINE_27A_ROOF_WINDOWS_W_PER_K,
"§3 (27a) case44",
)
_pin(
ht.floor_w_per_k,
_w001431_case44.LINE_28A_FLOOR_W_PER_K,
"§3 (28a) case44",
)
_pin(
ht.walls_w_per_k,
_w001431_case44.LINE_29A_WALLS_W_PER_K,
"§3 (29a) case44",
)
_pin(ht.roof_w_per_k, _w001431_case44.LINE_30_ROOF_W_PER_K, "§3 (30) case44")
_pin(
ht.total_external_element_area_m2,
_w001431_case44.LINE_31_EXTERNAL_AREA_M2,
"§3 (31) case44",
)
_pin(
ht.fabric_heat_loss_w_per_k,
_w001431_case44.LINE_33_FABRIC_W_PER_K,
"§3 (33) case44",
)
_pin(
ht.thermal_bridging_w_per_k,
_w001431_case44.LINE_36_THERMAL_BRIDGING_W_PER_K,
"§3 (36) case44",
)
def test_case44_conservatory_demand_side_matches_pdf() -> None:
"""End-to-end §6.1 conservatory demand pin for simulated case 44.
Beyond the §3 fabric, the conservatory ripples through the demand
cascade: its floor area enters TFA (4) → occupancy → §5 internal
gains (73); its glazing contributes §6 solar gains (83) at the
default E/W flux (Table 25 g=0.76, FF=0.70, 20° roof pitch); fabric
+ ventilation + gains combine into the §7 useful gains (95) and the
space-heating demand (99). Every line ref reproduces the P960 to 1e-4.
The full SAP/CO2 is NOT asserted: the case-44 Summary omits the
House-Coal secondary heater the P960 carries (see the provider's NB) —
a Summary-input defect downstream of, and independent of, §6.1."""
# Arrange
epc = _w001431_case44.build_epc()
# Act
ig = internal_gains_section_from_cert(epc)
sg = solar_gains_section_from_cert(epc)
sh = space_heating_section_from_cert(epc)
assert ig is not None # TFA present ⇒ §5 helper returns a result
# Assert — §5/§6/§7 demand line refs, each at abs=1e-4.
_pin(
ig.total_internal_gains_monthly_w[0],
_w001431_case44.LINE_73_INTERNAL_GAINS_JAN_W,
"§5 (73) case44",
)
_pin(
sg.total_solar_gains_monthly_w[0],
_w001431_case44.LINE_83_SOLAR_GAINS_JAN_W,
"§6 (83) case44",
)
_pin(
sh.useful_gains_monthly_w[0],
_w001431_case44.LINE_95_USEFUL_GAINS_JAN_W,
"§7 (95) case44",
)
_pin(
sh.space_heating_per_m2_kwh,
_w001431_case44.LINE_99_SPACE_HEATING_PER_M2_KWH,
"§7 (99) case44",
)
def test_case44_blower_door_pressure_test_matches_pdf() -> None:
"""Simulated case 44 lodges a Blower Door air-pressure test
(§12.2 "Pressure Test Result (AP50) 4.50"). SAP 10.2 §2 (17)-(18):
the AP50 reading routes infiltration via `(18) = AP50/20 + (8)` =
4.5/20 + 0.1167 = 0.3417, in preference to the components-based (16)
estimate. The extractor previously read only the AP4 (Pulse) column,
so a Blower Door result fell through to the structural-infiltration
default (effective ach 0.81 vs the worksheet's 0.58 → ventilation
heat loss over-counted by ~38%)."""
# Arrange
epc = _w001431_case44.build_epc()
# Act
vent = ventilation_from_cert(epc)
# Assert — (18) infiltration + (25) Jan effective ach, at abs=1e-4.
_pin(vent.pressure_test_ach, 0.3417, "§2 (18) case44")
_pin(vent.effective_monthly_ach[0], 0.5812, "§2 (25) Jan case44")
def test_case6_main_2_emitter_and_control_extracted() -> None:
"""Simulated case 6's §14.1 Main Heating2 lodges its OWN emitter
("Underfloor Heating") and control ("SAP code 2110, ...") — the two

57
tests/harness/test_console.py
View file

@ -10,9 +10,16 @@ from datatypes.epc.domain.epc import Epc
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.geospatial.planning_restrictions import PlanningRestrictions
from domain.modelling.contingencies import contingency_rate
from domain.modelling.measure_type import MeasureType
from domain.modelling.scenario import Scenario
from domain.modelling.generators.heating_recommendation import recommend_heating
from domain.modelling.generators.solid_wall_recommendation import recommend_solid_wall
from harness.console import DEFAULT_CATALOGUE, run_modelling, run_one
from harness.console import (
DEFAULT_CATALOGUE,
candidate_recommendations,
run_modelling,
run_one,
)
from repositories.product.product_json_repository import ProductJsonRepository
from tests.domain.modelling._elmhurst_recommendation import (
parse_recommendation_summary,
@ -224,6 +231,54 @@ def test_run_modelling_recommends_hhr_storage_for_an_electric_dwelling() -> None
assert "air_source_heat_pump" in {m.measure_type for m in plan.measures}
def test_candidate_recommendations_surface_unselected_options_with_cost() -> None:
# Arrange — an electric dwelling whose heating Recommendation offers both an
# ASHP and an HHR-storage bundle; the Optimiser selects only one of them.
epc: EpcPropertyData = _electric_storage_lit_epc()
# Act — the full candidate menu (every Generator Option, pre-optimisation)
# alongside the optimised Plan.
candidates = candidate_recommendations(epc)
plan = run_modelling(epc, goal_band="C", print_table=False)
# Assert — the menu carries every offered Option (so a measure the Plan did
# not select, like the passed-over HHR bundle, is still inspectable), and
# every Option carries a cost so "cost per measure" is always available.
offered = {
option.measure_type for rec in candidates for option in rec.options
}
selected = {measure.measure_type for measure in plan.measures}
assert "high_heat_retention_storage_heaters" in offered
assert "air_source_heat_pump" in offered
assert offered - selected # at least one offered measure was not selected
assert all(
option.cost is not None for rec in candidates for option in rec.options
)
def test_run_modelling_honours_a_scenarios_exclusions() -> None:
# Arrange — an electric dwelling whose optimised Plan normally leads with the
# ASHP bundle; a Scenario that excludes ASHP must keep it out of the Plan.
epc: EpcPropertyData = _electric_storage_lit_epc()
scenario = Scenario(
id=1,
goal="Increasing EPC",
goal_value="C",
budget=None,
is_default=True,
exclusions=frozenset({MeasureType.AIR_SOURCE_HEAT_PUMP}),
)
# Act — the orchestrator derives the allowlist from the Scenario's exclusions.
plan = run_modelling(epc, scenario=scenario, print_table=False)
# Assert — ASHP is excluded; the Plan still improves the dwelling via other
# measures (e.g. the HHR storage bundle).
measure_types = {measure.measure_type for measure in plan.measures}
assert MeasureType.AIR_SOURCE_HEAT_PUMP not in measure_types
assert measure_types # a non-empty Plan still came back
def test_sample_catalogue_prices_every_generator_measure_type() -> None:
# Arrange — the default offline catalogue.
products: ProductJsonRepository = ProductJsonRepository(DEFAULT_CATALOGUE)

32
tests/infrastructure/epc_client/test_sap_accuracy_corpus.py
View file

@ -95,10 +95,34 @@ _CORPUS = Path(
# 67.3% -> 67.5% (MAE 1.020 -> 0.987). The follow-on `common_wall_*` Detailed-RR
# surfaces (billed at main-wall U, deducted from the §3.10.1 residual) took the
# 6-cert detailed-common-wall cohort 2.43 -> 1.25; corpus -> 67.6% (MAE 0.979).
_MIN_WITHIN_HALF_SAP = 0.67
_MAX_SAP_MAE = 0.99
_MAX_CO2_MAE_TONNES = 0.35 # t CO2 / yr vs co2_emissions_current
_MAX_PE_PER_M2_MAE = 16.0 # kWh / m2 / yr vs energy_consumption_current
# The §3.9.2 Simplified Type-2 RR mapper (room_in_roof_type_2: gable quadratic +
# common-wall L×(0.25+H), MIRRORING the worksheet-validated Summary path,
# cross-mapper-parity-exact on cert 000565) -> 67.9% (MAE 0.959).
# The §6.1 non-separated conservatory mapper (the gov API's glazed building
# part → SapConservatory → §6.1 window/rooflight/floor cascade + TFA, MIRRORING
# the case-44 Summary path pinned to 1e-4) -> 68.6% (MAE 0.942). 5 type-4
# certs were over-rating (conservatory dropped → too little heat loss).
# STONE WALL + INTERNAL/EXTERNAL INSULATION (RdSAP 10 §5.8 + Table 14, p.41-42):
# the §5.8 added-insulation R-value adjustment was applied ONLY to WALL_SOLID_
# BRICK; a stone (granite/sandstone) wall lodging wall_insulation_type 1/3 + a
# thickness fell through the §5.6 branch and was billed at its UNINSULATED U
# (e.g. sandstone 520 mm + 100 mm internal: 1.64 instead of 0.30 → 5× wall heat
# loss). Mirroring the brick branch into the stone block recovered the worst of
# the wall_insulation_type=3 under-rate cluster (cert 100052159386 -26.2 -> -4.1
# SAP, walls 300 -> 55 W/K). within-0.5 68.6% -> 68.8% (MAE 0.942 -> 0.888;
# PE MAE 14.3 -> 13.9; CO2 MAE 0.27 -> 0.26). Unit-pinned in test_rdsap_uvalues.
# PER-PART ROOF DESCRIPTION (RdSAP 10 §5.11): the deduplicated epc.roofs[] list
# was joined into ONE description fed to EVERY building part's u_roof, so a flat
# "no insulation" extension dragged a pitched "insulated (assumed)" main roof to
# the uninsulated 2.30 (3-part certs systematically under-rated: 56% within,
# -0.79 mean). Matching each part to its own kind (flat vs pitched) fixed cert
# 100010129331 (roof 110.5 -> 31.3 W/K, +13.1 -> -0.05 SAP). within-0.5
# 68.8% -> 69.5% (MAE 0.888 -> 0.859; PE 13.9 -> 13.6); 3-part cohort 56% ->
# 61%. Pinned in test_heat_transmission (by_kind split + no-contamination).
_MIN_WITHIN_HALF_SAP = 0.69
_MAX_SAP_MAE = 0.86
_MAX_CO2_MAE_TONNES = 0.30 # t CO2 / yr vs co2_emissions_current
_MAX_PE_PER_M2_MAE = 14.0 # kWh / m2 / yr vs energy_consumption_current
def _load_corpus() -> list[dict[str, Any]]:

5
tests/orchestration/fakes.py
View file

@ -63,6 +63,11 @@ class FakePropertyRepo(PropertyRepository):
def get_many(self, property_ids: list[int]) -> Properties:
return Properties([self._hydrate(property_id) for property_id in property_ids])
def mark_modelled(self, property_id: int, *, has_recommendations: bool) -> None:
# Record the marker so tests can assert the pipeline set it.
self.modelled: dict[int, bool] = getattr(self, "modelled", {})
self.modelled[property_id] = has_recommendations
def insert_all(self, rows: list[PropertyIdentityInsert]) -> int:
self.inserted: list[PropertyIdentityInsert] = list(rows)
return len(rows)

5
tests/orchestration/test_bulk_upload_finaliser_orchestrator.py
View file

@ -50,6 +50,11 @@ class FakePropertyRepository(PropertyRepository):
def get_many(self, property_ids: list[int]) -> Properties: # pragma: no cover
raise NotImplementedError
def mark_modelled( # pragma: no cover
self, property_id: int, *, has_recommendations: bool
) -> None:
raise NotImplementedError
class FakeStatusWriter(BulkUploadStatusWriter):
def __init__(self) -> None:

47
tests/orchestration/test_modelling_solar_threading.py
View file

@ -116,6 +116,53 @@ def test_candidate_recommendations_excludes_solar_without_potential() -> None:
assert "Solar PV" not in {r.surface for r in recommendations}
class _ProductsRaisingFor(ProductRepository):
"""A catalogue that raises for one measure type — mirroring the live DB,
whose ``material.type`` enum does not carry ``secondary_heating_removal``.
Invoking that measure's generator would raise, so this proves an excluded
generator is never run."""
def __init__(self, forbidden: MeasureType) -> None:
self._forbidden = forbidden
def get(self, measure_type: str) -> Product:
if measure_type == self._forbidden:
raise ValueError(f"catalogue cannot represent {measure_type!r}")
return Product(
measure_type=measure_type,
unit_cost_per_m2=0.0,
contingency_rate=0.15,
id=909,
)
def test_an_excluded_measures_generator_is_not_invoked() -> None:
# Arrange — a dwelling with a lodged secondary heating system (so the
# secondary-heating generator is eligible to fire) priced against a catalogue
# that raises for that type, exactly as the live `material.type` enum does.
epc = _eligible_house()
epc.sap_heating.secondary_heating_type = 631
allowlist = frozenset(MeasureType) - {MeasureType.SECONDARY_HEATING_REMOVAL}
# Act — excluding the measure must stop its generator running at all (it would
# otherwise query the catalogue and raise).
recommendations = _candidate_recommendations(
epc,
_ProductsRaisingFor(MeasureType.SECONDARY_HEATING_REMOVAL),
PlanningRestrictions(),
None,
allowlist,
)
# Assert — the run completes and no secondary-heating option leaks through.
option_types = {
option.measure_type
for recommendation in recommendations
for option in recommendation.options
}
assert MeasureType.SECONDARY_HEATING_REMOVAL not in option_types
def test_considered_measures_restricts_candidates_to_the_allowlist() -> None:
# Arrange — a solar-eligible house, with its solar potential present, so the
# unrestricted run offers Solar PV alongside any fabric/heating candidates.

29
tests/repositories/property/test_property_repository.py
View file

@ -111,3 +111,32 @@ def test_get_many_defaults_to_unrestricted_when_uprn_has_no_spatial_row(
# Assert — an uncovered UPRN means unrestricted, not blocked (per legacy
# `empty_spatial_df`; ADR-0020).
assert properties.items[0].planning_restrictions == PlanningRestrictions()
def test_mark_modelled_sets_has_recommendations_and_bumps_updated_at(
db_engine: Engine,
) -> None:
# Arrange — a freshly-inserted property with no run recorded yet.
with Session(db_engine) as session:
row = PropertyRow(portfolio_id=7, uprn=12345)
session.add(row)
session.commit()
property_id = row.id
assert property_id is not None
assert row.has_recommendations is None
assert row.updated_at is None
# Act — record a run that produced recommendations.
with Session(db_engine) as session:
PropertyPostgresRepository(session).mark_modelled(
property_id, has_recommendations=True
)
session.commit()
# Assert — the marker is set and updated_at is stamped, so the run is datable
# against the 2026-06-01 new-process cutoff.
with Session(db_engine) as session:
refreshed = session.get(PropertyRow, property_id)
assert refreshed is not None
assert refreshed.has_recommendations is True
assert refreshed.updated_at is not None

75
tests/repositories/scenario/test_scenario_postgres_repository.py
View file

@ -14,6 +14,7 @@ import pytest
from sqlalchemy import Engine
from sqlmodel import Session
from domain.modelling.measure_type import MeasureType
from domain.modelling.portfolio_goal import PortfolioGoal
from domain.modelling.scenario import Scenario
from infrastructure.postgres.modelling import ScenarioModel
@ -67,6 +68,80 @@ def test_get_many_maps_live_scenario_rows_to_domain_in_input_order(
)
def test_get_many_parses_the_exclusions_array_into_measure_types(
db_engine: Engine,
) -> None:
# Arrange — the live `exclusions` column is a Postgres text-array literal of
# exact MeasureType values.
with Session(db_engine) as session:
session.add(
ScenarioModel(
id=7,
goal=PortfolioGoal.INCREASING_EPC,
goal_value="C",
is_default=True,
exclusions="{solar_pv,internal_wall_insulation}",
)
)
session.commit()
# Act
with Session(db_engine) as session:
scenario: Scenario = ScenarioPostgresRepository(session).get_many([7])[0]
# Assert
assert scenario.exclusions == frozenset(
{MeasureType.SOLAR_PV, MeasureType.INTERNAL_WALL_INSULATION}
)
def test_get_many_treats_a_null_exclusions_column_as_no_exclusions(
db_engine: Engine,
) -> None:
# Arrange
with Session(db_engine) as session:
session.add(
ScenarioModel(
id=7,
goal=PortfolioGoal.INCREASING_EPC,
goal_value="C",
is_default=True,
exclusions=None,
)
)
session.commit()
# Act
with Session(db_engine) as session:
scenario: Scenario = ScenarioPostgresRepository(session).get_many([7])[0]
# Assert
assert scenario.exclusions == frozenset()
def test_get_many_raises_on_an_exclusion_that_is_not_a_measure_type(
db_engine: Engine,
) -> None:
# Arrange — a legacy high-level category (`heating`) is not an exact
# MeasureType value; exact-only resolution must reject it loudly.
with Session(db_engine) as session:
session.add(
ScenarioModel(
id=7,
goal=PortfolioGoal.INCREASING_EPC,
goal_value="C",
is_default=True,
exclusions="{heating}",
)
)
session.commit()
# Act / Assert
with Session(db_engine) as session:
with pytest.raises(ValueError):
ScenarioPostgresRepository(session).get_many([7])
def test_get_many_raises_when_a_scenario_id_is_missing(db_engine: Engine) -> None:
# Arrange
with Session(db_engine) as session: