junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-27 23:35:01 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Persist room-in-roof geometry via child tables 🟩

Browse source 
SapRoomInRoof.detailed_surfaces (+ the gable/common-wall scalars) had no columns,
so 107 corpus certs lost their whole §3.9/§3.10 room-in-roof geometry on the DB
round-trip and fell back to the Simplified all-elements age-band default —
worst case -15.21 SAP (#1665 / #1664).

Add EpcRoomInRoofModel (0..1 per building part, unique FK) + EpcRoomInRoofSurface
Model (0..n, surface_index order). Save nests via RETURNING (building part -> RIR
-> surfaces); delete clears bottom-up (surfaces -> room_in_roof -> parts) since
the FE FKs are ON DELETE no action; both read paths group + reconstruct in a new
_to_room_in_roof / _to_rir_surface, replacing the two-field flat rebuild in
_to_building_part. The flat room_in_roof_* columns on epc_building_part are
superseded and dropped in a follow-up (#1664). Nullable is preserved on
insulation_thickness_mm / u_value (null vs 0 is the Table 17-vs-18 branch); area
is unconstrained (a §3.9.2 absent-gable adjustment is signed). Drop the 8
_UNPERSISTED_ALLOWLIST entries. Verified: worst RIR cert 10012119141 round-trip
-15.21 -> 0.0000 with detailed_surfaces at deep equality.

Companion FE migration: assessment-model#443 (epc_room_in_roof +
epc_room_in_roof_surface). Deploy gate.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-07-23 09:17:10 +00:00
parent 7c626dda8e
commit 846a3238b3
3 changed files with 253 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

83
infrastructure/postgres/epc_property_table.py
View file

@ -16,6 +16,8 @@ from datatypes.epc.domain.epc_property_data import (
SapFloorDimension,
SapFlatDetails,
SapRoofWindow,
SapRoomInRoof,
SapRoomInRoofSurface,
SapWindow,
)
@ -940,6 +942,87 @@ class EpcRoofWindowModel(SQLModel, table=True):
)
class EpcRoomInRoofModel(SQLModel, table=True):
__tablename__: ClassVar[str] = (
"epc_room_in_roof" # pyright: ignore[reportIncompatibleVariableOverride]
)
id: Optional[int] = Field(default=None, primary_key=True)
# 0..1 per building part (SapBuildingPart.sap_room_in_roof is Optional, not a
# list) — enforced UNIQUE by the FE migration. #1665 + #1664.
epc_building_part_id: int = Field(
foreign_key="epc_building_part.id", nullable=False
)
# floor_area is Union[int, float] in the domain; int-vs-float is not
# semantically load-bearing (the calculator does float(rir.floor_area)).
floor_area: float
construction_age_band: str
# §3.9.2 Simplified Type 2 gable/common-wall scalars (dormant on the API
# path — 0 corpus — but populated by the Elmhurst/site-notes path).
common_wall_length_m: Optional[float] = Field(default=None)
common_wall_height_m: Optional[float] = Field(default=None)
gable_1_length_m: Optional[float] = Field(default=None)
gable_1_height_m: Optional[float] = Field(default=None)
gable_2_length_m: Optional[float] = Field(default=None)
gable_2_height_m: Optional[float] = Field(default=None)
@classmethod
def from_domain(
cls, rir: SapRoomInRoof, epc_building_part_id: int
) -> EpcRoomInRoofModel:
return cls(
epc_building_part_id=epc_building_part_id,
floor_area=float(rir.floor_area),
construction_age_band=rir.construction_age_band,
common_wall_length_m=rir.common_wall_length_m,
common_wall_height_m=rir.common_wall_height_m,
gable_1_length_m=rir.gable_1_length_m,
gable_1_height_m=rir.gable_1_height_m,
gable_2_length_m=rir.gable_2_length_m,
gable_2_height_m=rir.gable_2_height_m,
)
class EpcRoomInRoofSurfaceModel(SQLModel, table=True):
__tablename__: ClassVar[str] = (
"epc_room_in_roof_surface" # pyright: ignore[reportIncompatibleVariableOverride]
)
id: Optional[int] = Field(default=None, primary_key=True)
epc_room_in_roof_id: int = Field(
foreign_key="epc_room_in_roof.id", nullable=False
)
# List position on `detailed_surfaces`, persisted for round-trip order.
surface_index: int
kind: str
# Can be NEGATIVE: a §3.9.2 absent-gable adjustment produces a signed area
# that deducts from the A_RR residual — no CHECK >= 0.
area_m2: float
# Null vs 0 is the U-value branch (Table 17 measured row vs Table 18 col (4)
# age-band default) — keep nullable, never DEFAULT 0.
insulation_thickness_mm: Optional[int] = Field(default=None)
insulation_type: Optional[str] = Field(default=None)
# Assessor U override; null falls through to the Table 4 cascade. Null != 0.
u_value: Optional[float] = Field(default=None)
@classmethod
def from_domain(
cls,
surface: SapRoomInRoofSurface,
surface_index: int,
epc_room_in_roof_id: int,
) -> EpcRoomInRoofSurfaceModel:
return cls(
epc_room_in_roof_id=epc_room_in_roof_id,
surface_index=surface_index,
kind=surface.kind,
area_m2=surface.area_m2,
insulation_thickness_mm=surface.insulation_thickness_mm,
insulation_type=surface.insulation_type,
u_value=surface.u_value,
)
class EpcEnergyElementModel(SQLModel, table=True):
__tablename__: ClassVar[str] = (
"epc_energy_element" # pyright: ignore[reportIncompatibleVariableOverride]

184
repositories/epc/epc_postgres_repository.py
View file

@ -31,6 +31,7 @@ from datatypes.epc.domain.epc_property_data import (
SapHeating,
SapRoofWindow,
SapRoomInRoof,
SapRoomInRoofSurface,
SapVentilation,
SapWindow,
ShowerOutlet,
@ -50,6 +51,8 @@ from infrastructure.postgres.epc_property_table import (
EpcPropertyModel,
EpcRenewableHeatIncentiveModel,
EpcRoofWindowModel,
EpcRoomInRoofModel,
EpcRoomInRoofSurfaceModel,
EpcWindowModel,
)
from repositories.epc.epc_repository import (
@ -330,6 +333,36 @@ class EpcPostgresRepository(EpcRepository):
if floor_rows:
self._session.execute(_sa_insert(EpcFloorDimensionModel), floor_rows) # type: ignore[deprecated]
# Room-in-roof: 0..1 per building part, each with 0..n surfaces.
# Insert the RIR rows with RETURNING, then zip to resolve the
# surface FKs (same positional-zip contract as the parts above).
rir_pairs = [
(part.sap_room_in_roof, bp_row[0])
for (part, _), bp_row in zip(parts_ordered, returned_bps)
if part.sap_room_in_roof is not None
]
if rir_pairs:
rir_rows = [
_col_values(
EpcRoomInRoofModel.from_domain(rir, bp_id), frozenset({"id"})
)
for rir, bp_id in rir_pairs
]
returned_rirs = self._session.execute( # type: ignore[deprecated]
_sa_insert(EpcRoomInRoofModel).returning(EpcRoomInRoofModel.__table__.c["id"]), # type: ignore[attr-defined]
rir_rows,
).all()
surface_rows = [
_col_values(
EpcRoomInRoofSurfaceModel.from_domain(surface, s_idx, rir_row[0]),
frozenset({"id"}),
)
for (rir, _), rir_row in zip(rir_pairs, returned_rirs)
for s_idx, surface in enumerate(rir.detailed_surfaces or [])
]
if surface_rows:
self._session.execute(_sa_insert(EpcRoomInRoofSurfaceModel), surface_rows) # type: ignore[deprecated]
return epc_property_ids
def _delete_for_properties(
@ -359,6 +392,29 @@ class EpcPostgresRepository(EpcRepository):
if i is not None
]
if part_ids:
# Room-in-roof surfaces FK the RIR row, which FKs the building part;
# with ON DELETE no action we must clear the graph bottom-up
# (surfaces -> room_in_roof) before the building parts below.
rir_ids = [
i
for i in self._session.exec(
select(EpcRoomInRoofModel.id).where(
col(EpcRoomInRoofModel.epc_building_part_id).in_(part_ids)
)
).all()
if i is not None
]
if rir_ids:
self._session.exec( # type: ignore[call-overload]
delete(EpcRoomInRoofSurfaceModel).where(
col(EpcRoomInRoofSurfaceModel.epc_room_in_roof_id).in_(rir_ids)
)
)
self._session.exec( # type: ignore[call-overload]
delete(EpcRoomInRoofModel).where(
col(EpcRoomInRoofModel.epc_building_part_id).in_(part_ids)
)
)
self._session.exec( # type: ignore[call-overload]
delete(EpcFloorDimensionModel).where(
col(EpcFloorDimensionModel.epc_building_part_id).in_(part_ids)
@ -407,6 +463,29 @@ class EpcPostgresRepository(EpcRepository):
if i is not None
]
if part_ids:
# Room-in-roof surfaces FK the RIR row, which FKs the building part;
# with ON DELETE no action we must clear the graph bottom-up
# (surfaces -> room_in_roof) before the building parts below.
rir_ids = [
i
for i in self._session.exec(
select(EpcRoomInRoofModel.id).where(
col(EpcRoomInRoofModel.epc_building_part_id).in_(part_ids)
)
).all()
if i is not None
]
if rir_ids:
self._session.exec( # type: ignore[call-overload]
delete(EpcRoomInRoofSurfaceModel).where(
col(EpcRoomInRoofSurfaceModel.epc_room_in_roof_id).in_(rir_ids)
)
)
self._session.exec( # type: ignore[call-overload]
delete(EpcRoomInRoofModel).where(
col(EpcRoomInRoofModel.epc_building_part_id).in_(part_ids)
)
)
self._session.exec( # type: ignore[call-overload]
delete(EpcFloorDimensionModel).where(
col(EpcFloorDimensionModel.epc_building_part_id).in_(part_ids)
@ -578,6 +657,10 @@ class EpcPostgresRepository(EpcRepository):
bp.id for parts in parts_by.values() for bp in parts if bp.id is not None
]
floor_dims_by_part = self._floor_dims_by_part(part_ids)
rir_by_part = self._rir_by_part(part_ids)
surfaces_by_rir = self._rir_surfaces_by_rir(
[r.id for r in rir_by_part.values() if r.id is not None]
)
result: dict[int, EpcPropertyData] = {}
for key, parent in parents_by_key.items():
@ -589,6 +672,8 @@ class EpcPostgresRepository(EpcRepository):
heating_rows=heating_by.get(epc_id, []),
part_rows=parts_by.get(epc_id, []),
floor_dims_by_part=floor_dims_by_part,
rir_by_part=rir_by_part,
surfaces_by_rir=surfaces_by_rir,
window_rows=windows_by.get(epc_id, []),
pv_array_rows=pv_arrays_by.get(epc_id, []),
roof_window_rows=roof_windows_by.get(epc_id, []),
@ -612,6 +697,33 @@ class EpcPostgresRepository(EpcRepository):
grouped.setdefault(row.epc_building_part_id, []).append(row)
return grouped
@private
def _rir_by_part(self, part_ids: list[int]) -> dict[int, EpcRoomInRoofModel]:
if not part_ids:
return {}
rows = self._session.exec(
select(EpcRoomInRoofModel).where(
col(EpcRoomInRoofModel.epc_building_part_id).in_(part_ids)
)
).all()
return {row.epc_building_part_id: row for row in rows}
@private
def _rir_surfaces_by_rir(
self, rir_ids: list[int]
) -> dict[int, list[EpcRoomInRoofSurfaceModel]]:
if not rir_ids:
return {}
rows = self._session.exec(
select(EpcRoomInRoofSurfaceModel)
.where(col(EpcRoomInRoofSurfaceModel.epc_room_in_roof_id).in_(rir_ids))
.order_by(EpcRoomInRoofSurfaceModel.surface_index) # type: ignore[arg-type]
).all()
grouped: dict[int, list[EpcRoomInRoofSurfaceModel]] = {}
for row in rows:
grouped.setdefault(row.epc_room_in_roof_id, []).append(row)
return grouped
def get(self, epc_property_id: int) -> EpcPropertyData:
p = self._session.get(EpcPropertyModel, epc_property_id)
if p is None:
@ -655,8 +767,11 @@ class EpcPostgresRepository(EpcRepository):
window_rows = self._windows(epc_property_id)
pv_array_rows = self._pv_arrays(epc_property_id)
roof_window_rows = self._roof_windows(epc_property_id)
floor_dims_by_part = self._floor_dims_by_part(
[bp.id for bp in part_rows if bp.id is not None]
part_ids = [bp.id for bp in part_rows if bp.id is not None]
floor_dims_by_part = self._floor_dims_by_part(part_ids)
rir_by_part = self._rir_by_part(part_ids)
surfaces_by_rir = self._rir_surfaces_by_rir(
[r.id for r in rir_by_part.values() if r.id is not None]
)
return self._compose(
p=p,
@ -665,6 +780,8 @@ class EpcPostgresRepository(EpcRepository):
heating_rows=heating_rows,
part_rows=part_rows,
floor_dims_by_part=floor_dims_by_part,
rir_by_part=rir_by_part,
surfaces_by_rir=surfaces_by_rir,
window_rows=window_rows,
pv_array_rows=pv_array_rows,
roof_window_rows=roof_window_rows,
@ -681,6 +798,8 @@ class EpcPostgresRepository(EpcRepository):
heating_rows: list[EpcMainHeatingDetailModel],
part_rows: list[EpcBuildingPartModel],
floor_dims_by_part: dict[int, list[EpcFloorDimensionModel]],
rir_by_part: dict[int, EpcRoomInRoofModel],
surfaces_by_rir: dict[int, list[EpcRoomInRoofSurfaceModel]],
window_rows: list[EpcWindowModel],
pv_array_rows: list[EpcPhotovoltaicArrayModel],
roof_window_rows: list[EpcRoofWindowModel],
@ -721,7 +840,10 @@ class EpcPostgresRepository(EpcRepository):
sap_energy_source=self._to_energy_source(p, pv_array_rows),
sap_building_parts=[
self._to_building_part(
bp, floor_dims_by_part.get(bp.id, []) if bp.id is not None else []
bp,
floor_dims_by_part.get(bp.id, []) if bp.id is not None else [],
rir_by_part.get(bp.id) if bp.id is not None else None,
surfaces_by_rir,
)
for bp in part_rows
],
@ -1028,7 +1150,11 @@ class EpcPostgresRepository(EpcRepository):
@private
def _to_building_part(
self, bp: EpcBuildingPartModel, floor_rows: list[EpcFloorDimensionModel]
self,
bp: EpcBuildingPartModel,
floor_rows: list[EpcFloorDimensionModel],
rir_model: Optional[EpcRoomInRoofModel],
surfaces_by_rir: dict[int, list[EpcRoomInRoofSurfaceModel]],
) -> SapBuildingPart:
return SapBuildingPart(
identifier=BuildingPartIdentifier(bp.identifier),
@ -1059,19 +1185,49 @@ class EpcPostgresRepository(EpcRepository):
roof_insulation_thickness=bp.roof_insulation_thickness,
rafter_insulation_thickness=bp.rafter_insulation_thickness,
wall_is_basement=bp.wall_is_basement,
sap_room_in_roof=(
SapRoomInRoof(
floor_area=bp.room_in_roof_floor_area,
construction_age_band=_require(
bp.room_in_roof_construction_age_band,
"room_in_roof_construction_age_band",
),
)
if bp.room_in_roof_floor_area is not None
else None
# Reconstruct from the epc_room_in_roof child table (full geometry
# incl. detailed_surfaces); the flat room_in_roof_* columns on
# epc_building_part are superseded and dropped in a follow-up (#1664).
sap_room_in_roof=self._to_room_in_roof(rir_model, surfaces_by_rir),
)
@private
def _to_room_in_roof(
self,
rir_model: Optional[EpcRoomInRoofModel],
surfaces_by_rir: dict[int, list[EpcRoomInRoofSurfaceModel]],
) -> Optional[SapRoomInRoof]:
if rir_model is None:
return None
surfaces = (
surfaces_by_rir.get(rir_model.id, []) if rir_model.id is not None else []
)
return SapRoomInRoof(
floor_area=rir_model.floor_area,
construction_age_band=rir_model.construction_age_band,
common_wall_length_m=rir_model.common_wall_length_m,
common_wall_height_m=rir_model.common_wall_height_m,
gable_1_length_m=rir_model.gable_1_length_m,
gable_1_height_m=rir_model.gable_1_height_m,
gable_2_length_m=rir_model.gable_2_length_m,
gable_2_height_m=rir_model.gable_2_height_m,
detailed_surfaces=(
[self._to_rir_surface(s) for s in surfaces] if surfaces else None
),
)
@private
def _to_rir_surface(
self, s: EpcRoomInRoofSurfaceModel
) -> SapRoomInRoofSurface:
return SapRoomInRoofSurface(
kind=s.kind,
area_m2=s.area_m2,
insulation_thickness_mm=s.insulation_thickness_mm,
insulation_type=s.insulation_type,
u_value=s.u_value,
)
@private
def _to_alt_wall(
self, bp: EpcBuildingPartModel, n: int

11
tests/repositories/epc/test_epc_persistence_field_coverage.py
View file

@ -43,17 +43,6 @@ _UNPERSISTED_ALLOWLIST: dict[str, str] = {
"EpcPropertyData.solar_hw_collector_orientation": "FE column pending — tracked round-trip gap",
"EpcPropertyData.solar_hw_collector_pitch_deg": "FE column pending — tracked round-trip gap",
"EpcPropertyData.solar_hw_overshading": "FE column pending — tracked round-trip gap",
# Room-in-roof detail: only floor_area + age_band reconstruct from the
# inlined columns; the geometry / surface detail awaits an FE child table
# (tracked round-trip gap, docs/migrations §2).
"SapRoomInRoof.common_wall_height_m": "FE child table pending — room-in-roof detail",
"SapRoomInRoof.common_wall_length_m": "FE child table pending — room-in-roof detail",
"SapRoomInRoof.detailed_surfaces": "FE child table pending — room-in-roof detail",
"SapRoomInRoof.gable_1_height_m": "FE child table pending — room-in-roof detail",
"SapRoomInRoof.gable_1_length_m": "FE child table pending — room-in-roof detail",
"SapRoomInRoof.gable_2_height_m": "FE child table pending — room-in-roof detail",
"SapRoomInRoof.gable_2_length_m": "FE child table pending — room-in-roof detail",
"SapRoomInRoofSurface": "FE child table pending — room-in-roof surface detail",
# --- DELIBERATELY not persisted (#1656). These four ARE read by the
# calculator (`heat_transmission.py:994` wall U, `:1068` roof U, `:1118`
# floor U, `:955` basement wall) — do NOT re-label them "not read by the