Persist roof windows via the epc_roof_window child table 🟩

EpcPropertyData.sap_roof_windows (List[SapRoofWindow]) had no table, so 55
corpus certs lost their rooflights on the DB round-trip and were re-scored
without them (§3 (27a) heat transmission + §6 solar gain). Add EpcRoofWindowModel
(mirrors EpcPhotovoltaicArrayModel) with a roof_window_index order column and the
resolved SapRoofWindow fields; wire save (batch insert), delete (both delete
paths), both read paths (get + get_many) and _compose reconstruction; drop the
two _UNPERSISTED_ALLOWLIST entries.

u_value_raw / g_perpendicular / frame_factor are stored resolved, not re-derived:
the Table 24 lookup is keyed on glazing_type AND glazing_gap, and glazing_gap is
not retained on SapRoofWindow, so they can't be reconstructed from the persisted
fields. window_location is JSONB (int on API / str on site-notes). Verified: a
roof-window cert round-trips sap_roof_windows to deep equality.

Companion FE migration: assessment-model#443 (epc_roof_window). Deploy gate.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

infrastructure/postgres/epc_property_table.py

@@ -15,6 +15,7 @@ from datatypes.epc.domain.epc_property_data import (
     SapBuildingPart,
     SapFloorDimension,
     SapFlatDetails,
+    SapRoofWindow,
     SapWindow,
 )
 
@@ -896,6 +897,49 @@ class EpcPhotovoltaicArrayModel(SQLModel, table=True):
         )
 
 
+class EpcRoofWindowModel(SQLModel, table=True):
+    __tablename__: ClassVar[str] = (
+        "epc_roof_window"  # pyright: ignore[reportIncompatibleVariableOverride]
+    )
+
+    id: Optional[int] = Field(default=None, primary_key=True)
+    epc_property_id: int = Field(foreign_key="epc_property.id", nullable=False)
+
+    # List position on `sap_roof_windows`, persisted so the order is recoverable
+    # for round-trip equality (#1665). Fields mirror SapRoofWindow: area/U feed
+    # §3 (27a) heat transmission; orientation/pitch/g/frame_factor feed §6 solar.
+    roof_window_index: int
+    area_m2: float
+    u_value_raw: float
+    orientation: int
+    pitch_deg: float
+    g_perpendicular: float
+    frame_factor: float
+    glazing_type: int
+    # SAP building-part index — int on the API path, str on site-notes; JSONB
+    # preserves the int-vs-str distinction (mirrors SapWindow.window_location).
+    window_location: Union[int, str] = Field(
+        sa_column=Column(JSONB, nullable=False)
+    )
+
+    @classmethod
+    def from_domain(
+        cls, rw: SapRoofWindow, roof_window_index: int, epc_property_id: int
+    ) -> EpcRoofWindowModel:
+        return cls(
+            epc_property_id=epc_property_id,
+            roof_window_index=roof_window_index,
+            area_m2=rw.area_m2,
+            u_value_raw=rw.u_value_raw,
+            orientation=rw.orientation,
+            pitch_deg=rw.pitch_deg,
+            g_perpendicular=rw.g_perpendicular,
+            frame_factor=rw.frame_factor,
+            glazing_type=rw.glazing_type,
+            window_location=rw.window_location,
+        )
+
+
 class EpcEnergyElementModel(SQLModel, table=True):
     __tablename__: ClassVar[str] = (
         "epc_energy_element"  # pyright: ignore[reportIncompatibleVariableOverride]

repositories/epc/epc_postgres_repository.py

@@ -29,6 +29,7 @@ from datatypes.epc.domain.epc_property_data import (
     SapFlatDetails,
     SapFloorDimension,
     SapHeating,
+    SapRoofWindow,
     SapRoomInRoof,
     SapVentilation,
     SapWindow,
@@ -48,6 +49,7 @@ from infrastructure.postgres.epc_property_table import (
     EpcPropertyEnergyPerformanceModel,
     EpcPropertyModel,
     EpcRenewableHeatIncentiveModel,
+    EpcRoofWindowModel,
     EpcWindowModel,
 )
 from repositories.epc.epc_repository import (
@@ -195,6 +197,7 @@ class EpcPostgresRepository(EpcRepository):
         parts_ordered: list[tuple[Any, int]] = []  # (SapBuildingPart, epc_property_id)
         window_rows: list[dict[str, Any]] = []
         pv_rows: list[dict[str, Any]] = []
+        roof_window_rows: list[dict[str, Any]] = []
         element_rows: list[dict[str, Any]] = []
         flat_rows: list[dict[str, Any]] = []
         rhi_rows: list[dict[str, Any]] = []
@@ -231,6 +234,13 @@ class EpcPostgresRepository(EpcRepository):
                         frozenset({"id"}),
                     )
                 )
+            for idx, roof_window in enumerate(d.sap_roof_windows or []):
+                roof_window_rows.append(
+                    _col_values(
+                        EpcRoofWindowModel.from_domain(roof_window, idx, epc_pid),
+                        frozenset({"id"}),
+                    )
+                )
             for etype, els in (
                 ("roof", d.roofs),
                 ("wall", d.walls),
@@ -284,6 +294,8 @@ class EpcPostgresRepository(EpcRepository):
             self._session.execute(_sa_insert(EpcWindowModel), window_rows)  # type: ignore[deprecated]
         if pv_rows:
             self._session.execute(_sa_insert(EpcPhotovoltaicArrayModel), pv_rows)  # type: ignore[deprecated]
+        if roof_window_rows:
+            self._session.execute(_sa_insert(EpcRoofWindowModel), roof_window_rows)  # type: ignore[deprecated]
         if element_rows:
             self._session.execute(_sa_insert(EpcEnergyElementModel), element_rows)  # type: ignore[deprecated]
         if flat_rows:
@@ -359,6 +371,7 @@ class EpcPostgresRepository(EpcRepository):
             EpcBuildingPartModel,
             EpcWindowModel,
             EpcPhotovoltaicArrayModel,
+            EpcRoofWindowModel,
             EpcFlatDetailsModel,
             EpcRenewableHeatIncentiveModel,
         ):
@@ -406,6 +419,7 @@ class EpcPostgresRepository(EpcRepository):
             EpcBuildingPartModel,
             EpcWindowModel,
             EpcPhotovoltaicArrayModel,
+            EpcRoofWindowModel,
             EpcFlatDetailsModel,
             EpcRenewableHeatIncentiveModel,
         ):
@@ -553,6 +567,13 @@ class EpcPostgresRepository(EpcRepository):
                 .order_by(EpcPhotovoltaicArrayModel.array_index)  # type: ignore[arg-type]
             ).all()
         )
+        roof_windows_by = _group_by_epc(
+            self._session.exec(
+                select(EpcRoofWindowModel)
+                .where(col(EpcRoofWindowModel.epc_property_id).in_(epc_ids))
+                .order_by(EpcRoofWindowModel.roof_window_index)  # type: ignore[arg-type]
+            ).all()
+        )
         part_ids = [
             bp.id for parts in parts_by.values() for bp in parts if bp.id is not None
         ]
@@ -570,6 +591,7 @@ class EpcPostgresRepository(EpcRepository):
                 floor_dims_by_part=floor_dims_by_part,
                 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, []),
                 flat_row=flat_by.get(epc_id),
                 rhi_row=rhi_by.get(epc_id),
             )
@@ -632,6 +654,7 @@ class EpcPostgresRepository(EpcRepository):
         ).first()
         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]
         )
@@ -644,6 +667,7 @@ class EpcPostgresRepository(EpcRepository):
             floor_dims_by_part=floor_dims_by_part,
             window_rows=window_rows,
             pv_array_rows=pv_array_rows,
+            roof_window_rows=roof_window_rows,
             flat_row=flat_row,
             rhi_row=rhi_row,
         )
@@ -659,6 +683,7 @@ class EpcPostgresRepository(EpcRepository):
         floor_dims_by_part: dict[int, list[EpcFloorDimensionModel]],
         window_rows: list[EpcWindowModel],
         pv_array_rows: list[EpcPhotovoltaicArrayModel],
+        roof_window_rows: list[EpcRoofWindowModel],
         flat_row: Optional[EpcFlatDetailsModel],
         rhi_row: Optional[EpcRenewableHeatIncentiveModel],
     ) -> EpcPropertyData:
@@ -688,6 +713,11 @@ class EpcPostgresRepository(EpcRepository):
             door_count=p.door_count,
             sap_heating=self._to_sap_heating(p, heating_rows),
             sap_windows=[self._to_window(w) for w in window_rows],
+            sap_roof_windows=(
+                [self._to_roof_window(r) for r in roof_window_rows]
+                if roof_window_rows
+                else None
+            ),
             sap_energy_source=self._to_energy_source(p, pv_array_rows),
             sap_building_parts=[
                 self._to_building_part(
@@ -869,6 +899,29 @@ class EpcPostgresRepository(EpcRepository):
             ).all()
         )
 
+    @private
+    def _roof_windows(self, epc_property_id: int) -> list[EpcRoofWindowModel]:
+        return list(
+            self._session.exec(
+                select(EpcRoofWindowModel)
+                .where(EpcRoofWindowModel.epc_property_id == epc_property_id)
+                .order_by(EpcRoofWindowModel.roof_window_index)  # type: ignore[arg-type]
+            ).all()
+        )
+
+    @private
+    def _to_roof_window(self, r: EpcRoofWindowModel) -> SapRoofWindow:
+        return SapRoofWindow(
+            area_m2=r.area_m2,
+            u_value_raw=r.u_value_raw,
+            orientation=r.orientation,
+            pitch_deg=r.pitch_deg,
+            g_perpendicular=r.g_perpendicular,
+            frame_factor=r.frame_factor,
+            glazing_type=r.glazing_type,
+            window_location=r.window_location,
+        )
+
     @private
     def _to_energy_element(self, e: EpcEnergyElementModel) -> EnergyElement:
         return EnergyElement(

tests/repositories/epc/test_epc_persistence_field_coverage.py

@@ -40,8 +40,6 @@ _UNPERSISTED_ALLOWLIST: dict[str, str] = {
     "EpcPropertyData.lzc_energy_sources": "dormant — not read by the calculator; no FE column",
     # Scoring-relevant round-trip gaps awaiting FE columns / child tables
     # (tracked follow-ups; see docs/migrations/epc-property-round-trip-fidelity.md).
-    "EpcPropertyData.sap_roof_windows": "FE child table pending — tracked round-trip gap",
-    "SapRoofWindow": "FE child table pending — tracked round-trip gap (sap_roof_windows)",
     "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",