from __future__ import annotations from collections.abc import Sequence from dataclasses import dataclass, field from datetime import date, datetime from typing import Any, Optional, Protocol, TypeVar from sqlalchemy import insert as _sa_insert from sqlmodel import Session, SQLModel, col, delete, select from datatypes.epc.domain.epc import Epc from datatypes.epc.domain.epc_property_data import ( Addendum, BuildingPartIdentifier, EnergyElement, EpcPropertyData, InstantaneousWwhrs, MainHeatingDetail, PhotovoltaicArray, PhotovoltaicSupply, PhotovoltaicSupplyNoneOrNoDetails, PvBatteries, PvBattery, RenewableHeatIncentive, SapAlternativeWall, SapBuildingPart, SapConservatory, SapEnergySource, SapFlatDetails, SapFloorDimension, SapHeating, SapRoomInRoof, SapVentilation, SapWindow, ShowerOutlet, ShowerOutlets, WindowsTransmissionDetails, WindowTransmissionDetails, WindTurbineDetails, ) from infrastructure.postgres.epc_property_table import ( EpcBuildingPartModel, EpcEnergyElementModel, EpcFlatDetailsModel, EpcFloorDimensionModel, EpcMainHeatingDetailModel, EpcPhotovoltaicArrayModel, EpcPropertyEnergyPerformanceModel, EpcPropertyModel, EpcRenewableHeatIncentiveModel, EpcWindowModel, ) from repositories.epc.epc_repository import EpcRepository, EpcSource from utilities.private import private _T = TypeVar("_T") @dataclass(frozen=True) class EpcSaveRequest: data: EpcPropertyData property_id: Optional[int] = None portfolio_id: Optional[int] = None source: EpcSource = field(default="lodged") def _col_values(model: SQLModel, exclude: frozenset[str] = frozenset()) -> dict[str, Any]: """Extract column-keyed values from a SQLModel instance for Core INSERT.""" return { c.name: getattr(model, c.name) # type: ignore[union-attr] for c in model.__table__.c # type: ignore[attr-defined] if c.name not in exclude # type: ignore[union-attr] } def _require(value: Optional[_T], field: str) -> _T: if value is None: raise ValueError(f"epc_property row is missing required field {field!r}") 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 _RowT = TypeVar("_RowT", bound=_HasEpcPropertyId) def _group_by_epc(rows: Sequence[_RowT]) -> dict[int, list[_RowT]]: grouped: dict[int, list[_RowT]] = {} for row in rows: grouped.setdefault(row.epc_property_id, []).append(row) return grouped class EpcPostgresRepository(EpcRepository): """Maps EpcPropertyData to/from the epc_property parent row + child tables. Round-trip fidelity over the persisted projection is pinned by the Slice-1 round-trip test (Hestia-Homes/Model#1129). Fields the schema does not yet store (see docs/migrations/epc-property-round-trip-fidelity.md §2) reconstruct as their dataclass defaults — tracked as follow-up migrations. """ def __init__(self, session: Session) -> None: self._session = session def save( self, data: EpcPropertyData, property_id: Optional[int] = None, portfolio_id: Optional[int] = None, source: EpcSource = "lodged", ) -> int: return self.save_batch([EpcSaveRequest(data, property_id, portfolio_id, source)])[0] def save_batch(self, requests: list[EpcSaveRequest]) -> list[int]: """Insert all EPCs in `requests` in one pass per table, returning one epc_property_id per request in the same order as the input. Deletes are batched first (one IN-query per child table per source), then all parent rows are inserted with a single RETURNING statement so positional ordering maps each returned id to its request. Building-part ids are captured the same way so floor-dimension FKs are resolved without any per-property flush round-trips (ADR-0012). """ if not requests: return [] # Batch-delete existing rows grouped by source so the lodged and predicted # slots remain independent (ADR-0031). pids_by_source: dict[EpcSource, list[int]] = {} for r in requests: if r.property_id is not None: pids_by_source.setdefault(r.source, []).append(r.property_id) for src, pids in pids_by_source.items(): self._delete_for_properties(pids, src) # Insert all parent (epc_property) rows; capture returned ids positionally. parent_rows = [ _col_values( EpcPropertyModel.from_epc_property_data( r.data, property_id=r.property_id, portfolio_id=r.portfolio_id, source=r.source ), exclude=frozenset({"id"}), ) for r in requests ] returned_parents = self._session.execute( # type: ignore[deprecated] _sa_insert(EpcPropertyModel).returning(EpcPropertyModel.__table__.c["id"]), # type: ignore[attr-defined] parent_rows, ).all() epc_property_ids = [row[0] for row in returned_parents] # Collect child rows, accumulating building parts in an ordered list so # the positional RETURNING trick can map part objects to their new ids. perf_rows: list[dict[str, Any]] = [] heating_rows: list[dict[str, Any]] = [] parts_ordered: list[tuple[Any, int]] = [] # (SapBuildingPart, epc_property_id) window_rows: list[dict[str, Any]] = [] pv_rows: list[dict[str, Any]] = [] element_rows: list[dict[str, Any]] = [] flat_rows: list[dict[str, Any]] = [] rhi_rows: list[dict[str, Any]] = [] for r, epc_pid in zip(requests, epc_property_ids): d = r.data perf_rows.append( _col_values( EpcPropertyEnergyPerformanceModel.from_epc_property_data(d, epc_pid), exclude=frozenset({"id"}), ) ) for detail in d.sap_heating.main_heating_details: heating_rows.append( _col_values(EpcMainHeatingDetailModel.from_domain(detail, epc_pid), frozenset({"id"})) ) for part in d.sap_building_parts: parts_ordered.append((part, epc_pid)) for window in d.sap_windows: window_rows.append( _col_values(EpcWindowModel.from_domain(window, epc_pid), frozenset({"id"})) ) for idx, array in enumerate(d.sap_energy_source.photovoltaic_arrays or []): pv_rows.append( _col_values(EpcPhotovoltaicArrayModel.from_domain(array, idx, epc_pid), frozenset({"id"})) ) for etype, els in ( ("roof", d.roofs), ("wall", d.walls), ("floor", d.floors), ("main_heating", d.main_heating), ): for el in els: element_rows.append( _col_values(EpcEnergyElementModel.from_domain(el, etype, epc_pid), frozenset({"id"})) ) for el, etype in ( (d.window, "window"), (d.lighting, "lighting"), (d.hot_water, "hot_water"), (d.secondary_heating, "secondary_heating"), (d.main_heating_controls, "main_heating_controls"), ): if el is not None: element_rows.append( _col_values(EpcEnergyElementModel.from_domain(el, etype, epc_pid), frozenset({"id"})) ) if d.sap_flat_details is not None: flat_rows.append( _col_values(EpcFlatDetailsModel.from_domain(d.sap_flat_details, epc_pid), frozenset({"id"})) ) if d.renewable_heat_incentive is not None: rhi_rows.append( _col_values(EpcRenewableHeatIncentiveModel.from_domain(d.renewable_heat_incentive, epc_pid), frozenset({"id"})) ) # Bulk-insert all simple child tables (no downstream FK dependency). if perf_rows: self._session.execute(_sa_insert(EpcPropertyEnergyPerformanceModel), perf_rows) # type: ignore[deprecated] if heating_rows: self._session.execute(_sa_insert(EpcMainHeatingDetailModel), heating_rows) # type: ignore[deprecated] if window_rows: 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 element_rows: self._session.execute(_sa_insert(EpcEnergyElementModel), element_rows) # type: ignore[deprecated] if flat_rows: self._session.execute(_sa_insert(EpcFlatDetailsModel), flat_rows) # type: ignore[deprecated] if rhi_rows: self._session.execute(_sa_insert(EpcRenewableHeatIncentiveModel), rhi_rows) # type: ignore[deprecated] # Building parts: insert with RETURNING and zip positionally to resolve # floor-dimension FKs. Do NOT key by id(part) — the same EpcPropertyData # object can appear in multiple requests (same epc, different property_ids), # giving identical object ids that collapse the dict and mis-wire FKs. # Positional zip is safe because PostgreSQL preserves VALUES order in RETURNING. if parts_ordered: bp_rows = [ _col_values(EpcBuildingPartModel.from_domain(part, epc_pid), frozenset({"id"})) for part, epc_pid in parts_ordered ] returned_bps = self._session.execute( # type: ignore[deprecated] _sa_insert(EpcBuildingPartModel).returning(EpcBuildingPartModel.__table__.c["id"]), # type: ignore[attr-defined] bp_rows, ).all() floor_rows: list[dict[str, Any]] = [ _col_values(EpcFloorDimensionModel.from_domain(dim, bp_row[0]), frozenset({"id"})) for (part, _), bp_row in zip(parts_ordered, returned_bps) for dim in part.sap_floor_dimensions ] if floor_rows: self._session.execute(_sa_insert(EpcFloorDimensionModel), floor_rows) # type: ignore[deprecated] return epc_property_ids def _delete_for_properties(self, property_ids: list[int], source: EpcSource) -> None: """Batch-delete every EPC graph for the given property_ids and source in one pass per child table (IN queries), replacing the per-property flush loop that drove RDS CPU to saturation during bulk modelling runs.""" epc_ids = [ i for i in self._session.exec( select(EpcPropertyModel.id) .where(col(EpcPropertyModel.property_id).in_(property_ids)) .where(EpcPropertyModel.source == source) ).all() if i is not None ] if not epc_ids: return part_ids = [ i for i in self._session.exec( select(EpcBuildingPartModel.id).where( col(EpcBuildingPartModel.epc_property_id).in_(epc_ids) ) ).all() if i is not None ] if part_ids: self._session.exec( # type: ignore[call-overload] delete(EpcFloorDimensionModel).where( col(EpcFloorDimensionModel.epc_building_part_id).in_(part_ids) ) ) for child in ( EpcPropertyEnergyPerformanceModel, EpcEnergyElementModel, EpcMainHeatingDetailModel, EpcBuildingPartModel, EpcWindowModel, EpcPhotovoltaicArrayModel, EpcFlatDetailsModel, EpcRenewableHeatIncentiveModel, ): self._session.exec( # type: ignore[call-overload] delete(child).where(col(child.epc_property_id).in_(epc_ids)) ) self._session.exec( # type: ignore[call-overload] delete(EpcPropertyModel).where(col(EpcPropertyModel.id).in_(epc_ids)) ) def _delete_for_property(self, property_id: int, source: EpcSource) -> None: """Remove the property's existing EPC graph for `source` (parent + child tables) so a re-save replaces rather than duplicates (ADR-0012), without disturbing the other source's slot (ADR-0031).""" epc_ids = [ i for i in self._session.exec( select(EpcPropertyModel.id) .where(EpcPropertyModel.property_id == property_id) .where(EpcPropertyModel.source == source) ).all() if i is not None ] if not epc_ids: return part_ids = [ i for i in self._session.exec( select(EpcBuildingPartModel.id).where( col(EpcBuildingPartModel.epc_property_id).in_(epc_ids) ) ).all() if i is not None ] if part_ids: self._session.exec( # type: ignore[call-overload] delete(EpcFloorDimensionModel).where( col(EpcFloorDimensionModel.epc_building_part_id).in_(part_ids) ) ) for child in ( EpcPropertyEnergyPerformanceModel, EpcEnergyElementModel, EpcMainHeatingDetailModel, EpcBuildingPartModel, EpcWindowModel, EpcPhotovoltaicArrayModel, EpcFlatDetailsModel, EpcRenewableHeatIncentiveModel, ): self._session.exec( # type: ignore[call-overload] delete(child).where(col(child.epc_property_id).in_(epc_ids)) ) self._session.exec( # type: ignore[call-overload] delete(EpcPropertyModel).where(col(EpcPropertyModel.id).in_(epc_ids)) ) def get_for_property(self, property_id: int) -> Optional[EpcPropertyData]: return self._get_for_property(property_id, source="lodged") def get_predicted_for_property( self, property_id: int ) -> Optional[EpcPropertyData]: return self._get_for_property(property_id, source="predicted") def _get_for_property( self, property_id: int, source: EpcSource ) -> Optional[EpcPropertyData]: row = self._session.exec( select(EpcPropertyModel) .where(EpcPropertyModel.property_id == property_id) .where(EpcPropertyModel.source == source) .order_by(EpcPropertyModel.id) # type: ignore[arg-type] ).first() if row is None or row.id is None: return None return self.get(row.id) def get_for_properties( self, property_ids: list[int] ) -> dict[int, EpcPropertyData]: """Bulk-hydrate a batch's LODGED EPCs, keyed by property_id.""" return self._for_properties(property_ids, source="lodged") def get_predicted_for_properties( self, property_ids: list[int] ) -> dict[int, EpcPropertyData]: """Bulk-hydrate a batch's PREDICTED EPCs (ADR-0031), keyed by property_id.""" return self._for_properties(property_ids, source="predicted") def _for_properties( self, property_ids: list[int], source: EpcSource ) -> dict[int, EpcPropertyData]: """Bulk-hydrate a batch's EPCs of one `source` in a handful of per-table IN queries (ADR-0012), not N x per-property. Load-whole per ADR-0002.""" if not property_ids: return {} parents = self._session.exec( select(EpcPropertyModel) .where(col(EpcPropertyModel.property_id).in_(property_ids)) .where(EpcPropertyModel.source == source) .order_by(EpcPropertyModel.id) # type: ignore[arg-type] ).all() parent_by_property: dict[int, EpcPropertyModel] = {} for parent in parents: if parent.property_id is not None and parent.id is not None: parent_by_property.setdefault(parent.property_id, parent) epc_ids = [p.id for p in parent_by_property.values() if p.id is not None] if not epc_ids: return {} perf_by = { r.epc_property_id: r for r in self._session.exec( select(EpcPropertyEnergyPerformanceModel).where( col(EpcPropertyEnergyPerformanceModel.epc_property_id).in_(epc_ids) ) ).all() } flat_by = { r.epc_property_id: r for r in self._session.exec( select(EpcFlatDetailsModel).where( col(EpcFlatDetailsModel.epc_property_id).in_(epc_ids) ) ).all() } rhi_by = { r.epc_property_id: r for r in self._session.exec( select(EpcRenewableHeatIncentiveModel).where( col(EpcRenewableHeatIncentiveModel.epc_property_id).in_(epc_ids) ) ).all() } elements_by = _group_by_epc( self._session.exec( select(EpcEnergyElementModel) .where(col(EpcEnergyElementModel.epc_property_id).in_(epc_ids)) .order_by(EpcEnergyElementModel.id) # type: ignore[arg-type] ).all() ) heating_by = _group_by_epc( self._session.exec( select(EpcMainHeatingDetailModel) .where(col(EpcMainHeatingDetailModel.epc_property_id).in_(epc_ids)) .order_by(EpcMainHeatingDetailModel.id) # type: ignore[arg-type] ).all() ) parts_by = _group_by_epc( self._session.exec( select(EpcBuildingPartModel) .where(col(EpcBuildingPartModel.epc_property_id).in_(epc_ids)) .order_by(EpcBuildingPartModel.id) # type: ignore[arg-type] ).all() ) windows_by = _group_by_epc( self._session.exec( select(EpcWindowModel) .where(col(EpcWindowModel.epc_property_id).in_(epc_ids)) .order_by(EpcWindowModel.id) # type: ignore[arg-type] ).all() ) pv_arrays_by = _group_by_epc( self._session.exec( select(EpcPhotovoltaicArrayModel) .where(col(EpcPhotovoltaicArrayModel.epc_property_id).in_(epc_ids)) .order_by(EpcPhotovoltaicArrayModel.array_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 ] floor_dims_by_part = self._floor_dims_by_part(part_ids) result: dict[int, EpcPropertyData] = {} for property_id, parent in parent_by_property.items(): epc_id = _require(parent.id, "id") result[property_id] = self._compose( p=parent, perf=perf_by.get(epc_id), elements=elements_by.get(epc_id, []), heating_rows=heating_by.get(epc_id, []), part_rows=parts_by.get(epc_id, []), floor_dims_by_part=floor_dims_by_part, window_rows=windows_by.get(epc_id, []), pv_array_rows=pv_arrays_by.get(epc_id, []), flat_row=flat_by.get(epc_id), rhi_row=rhi_by.get(epc_id), ) return result def _floor_dims_by_part( self, part_ids: list[int] ) -> dict[int, list[EpcFloorDimensionModel]]: if not part_ids: return {} rows = self._session.exec( select(EpcFloorDimensionModel) .where(col(EpcFloorDimensionModel.epc_building_part_id).in_(part_ids)) .order_by(EpcFloorDimensionModel.id) # type: ignore[arg-type] ).all() grouped: dict[int, list[EpcFloorDimensionModel]] = {} for row in rows: grouped.setdefault(row.epc_building_part_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: raise ValueError(f"epc_property {epc_property_id} not found") perf = self._session.exec( select(EpcPropertyEnergyPerformanceModel).where( EpcPropertyEnergyPerformanceModel.epc_property_id == epc_property_id ) ).first() elements = list( self._session.exec( select(EpcEnergyElementModel) .where(EpcEnergyElementModel.epc_property_id == epc_property_id) .order_by(EpcEnergyElementModel.id) # type: ignore[arg-type] ).all() ) heating_rows = list( self._session.exec( select(EpcMainHeatingDetailModel) .where(EpcMainHeatingDetailModel.epc_property_id == epc_property_id) .order_by(EpcMainHeatingDetailModel.id) # type: ignore[arg-type] ).all() ) part_rows = list( self._session.exec( select(EpcBuildingPartModel) .where(EpcBuildingPartModel.epc_property_id == epc_property_id) .order_by(EpcBuildingPartModel.id) # type: ignore[arg-type] ).all() ) flat_row = self._session.exec( select(EpcFlatDetailsModel).where( EpcFlatDetailsModel.epc_property_id == epc_property_id ) ).first() rhi_row = self._session.exec( select(EpcRenewableHeatIncentiveModel).where( EpcRenewableHeatIncentiveModel.epc_property_id == epc_property_id ) ).first() window_rows = self._windows(epc_property_id) pv_array_rows = self._pv_arrays(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] ) return self._compose( p=p, perf=perf, elements=elements, heating_rows=heating_rows, part_rows=part_rows, floor_dims_by_part=floor_dims_by_part, window_rows=window_rows, pv_array_rows=pv_array_rows, flat_row=flat_row, rhi_row=rhi_row, ) def _compose( self, *, p: EpcPropertyModel, perf: Optional[EpcPropertyEnergyPerformanceModel], elements: list[EpcEnergyElementModel], heating_rows: list[EpcMainHeatingDetailModel], part_rows: list[EpcBuildingPartModel], floor_dims_by_part: dict[int, list[EpcFloorDimensionModel]], window_rows: list[EpcWindowModel], pv_array_rows: list[EpcPhotovoltaicArrayModel], flat_row: Optional[EpcFlatDetailsModel], rhi_row: Optional[EpcRenewableHeatIncentiveModel], ) -> EpcPropertyData: def _elements(element_type: str) -> list[EnergyElement]: return [self._to_energy_element(e) for e in elements if e.element_type == element_type] def _single(element_type: str) -> Optional[EnergyElement]: found = _elements(element_type) return found[0] if found else None return EpcPropertyData( dwelling_type=p.dwelling_type, 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"), postcode=_require(p.postcode, "postcode"), post_town=_require(p.post_town, "post_town"), roofs=_elements("roof"), walls=_elements("wall"), floors=_elements("floor"), main_heating=_elements("main_heating"), 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_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 [] ) for bp in part_rows ], solar_water_heating=p.solar_water_heating, has_hot_water_cylinder=p.has_hot_water_cylinder, has_fixed_air_conditioning=p.has_fixed_air_conditioning, wet_rooms_count=p.wet_rooms_count, extensions_count=p.extensions_count, heated_rooms_count=p.heated_rooms_count, open_chimneys_count=p.open_chimneys_count, habitable_rooms_count=p.habitable_rooms_count, insulated_door_count=p.insulated_door_count, cfl_fixed_lighting_bulbs_count=p.cfl_fixed_lighting_bulbs_count, led_fixed_lighting_bulbs_count=p.led_fixed_lighting_bulbs_count, incandescent_fixed_lighting_bulbs_count=p.incandescent_fixed_lighting_bulbs_count, total_floor_area_m2=p.total_floor_area_m2, assessment_type=p.assessment_type, sap_version=p.sap_version, uprn=p.uprn, status=p.status, window=_single("window"), lighting=_single("lighting"), hot_water=_single("hot_water"), secondary_heating=_single("secondary_heating"), main_heating_controls=_single("main_heating_controls"), schema_type=p.schema_type, schema_versions_original=p.schema_versions_original, report_type=p.report_type, report_reference=p.report_reference, uprn_source=p.uprn_source, address_line_2=p.address_line_2, region_code=p.region_code, country_code=p.country_code, built_form=p.built_form, property_type=p.property_type, pressure_test=p.pressure_test, language_code=p.language_code, completion_date=( _as_date(p.completion_date) if p.completion_date else None ), registration_date=( _as_date(p.registration_date) if p.registration_date else None ), measurement_type=p.measurement_type, conservatory_type=p.conservatory_type, has_conservatory=p.has_conservatory, has_heated_separate_conservatory=p.has_heated_separate_conservatory, sap_conservatory=self._to_conservatory(p), blocked_chimneys_count=p.blocked_chimneys_count, energy_rating_average=p.energy_rating_average, current_energy_efficiency_band=( Epc(perf.current_energy_efficiency_band) if perf and perf.current_energy_efficiency_band else None ), environmental_impact_current=( perf.environmental_impact_current if perf else None ), heating_cost_current=perf.heating_cost_current if perf else None, co2_emissions_current=perf.co2_emissions_current if perf else None, energy_consumption_current=( perf.energy_consumption_current if perf else None ), energy_rating_current=perf.energy_rating_current if perf else None, lighting_cost_current=perf.lighting_cost_current if perf else None, hot_water_cost_current=perf.hot_water_cost_current if perf else None, insulated_door_u_value=p.insulated_door_u_value, mechanical_ventilation=p.mechanical_ventilation, percent_draughtproofed=p.percent_draughtproofed, heating_cost_potential=perf.heating_cost_potential if perf else None, co2_emissions_potential=perf.co2_emissions_potential if perf else None, energy_consumption_potential=( perf.energy_consumption_potential if perf else None ), energy_rating_potential=perf.energy_rating_potential if perf else None, lighting_cost_potential=perf.lighting_cost_potential if perf else None, hot_water_cost_potential=perf.hot_water_cost_potential if perf else None, environmental_impact_potential=( perf.environmental_impact_potential if perf else None ), potential_energy_efficiency_band=( Epc(perf.potential_energy_efficiency_band) if perf and perf.potential_energy_efficiency_band else None ), draughtproofed_door_count=p.draughtproofed_door_count, mechanical_vent_duct_type=p.mechanical_vent_duct_type, windows_transmission_details=( WindowsTransmissionDetails( u_value=p.windows_transmission_u_value, data_source=_require( p.windows_transmission_data_source, "windows_transmission_data_source", ), solar_transmittance=_require( p.windows_transmission_solar_transmittance, "windows_transmission_solar_transmittance", ), ) if p.windows_transmission_u_value is not None else None ), multiple_glazed_proportion=p.multiple_glazed_proportion, # Top-level mirror of sap_ventilation.extract_fans_count (the mapper # sets both from the same source). The deterministic calculator reads # the sap_ventilation copy, not this one; reconstruct it from the same # column anyway so EpcPropertyData round-trips complete, with no # allow-list exception for a field that trivially mirrors a persisted one. extract_fans_count=p.ventilation_extract_fans_count, calculation_software_version=p.calculation_software_version, mechanical_vent_duct_placement=p.mechanical_vent_duct_placement, mechanical_vent_duct_insulation=p.mechanical_vent_duct_insulation, pressure_test_certificate_number=p.pressure_test_certificate_number, mechanical_ventilation_index_number=p.mechanical_ventilation_index_number, mechanical_vent_measured_installation=p.mechanical_vent_measured_installation, co2_emissions_current_per_floor_area=( perf.co2_emissions_current_per_floor_area if perf else None ), low_energy_fixed_lighting_bulbs_count=p.low_energy_fixed_lighting_bulbs_count, sap_flat_details=( self._to_flat_details(flat_row) if flat_row is not None else None ), fixed_lighting_outlets_count=p.fixed_lighting_outlets_count, low_energy_fixed_lighting_outlets_count=p.low_energy_fixed_lighting_outlets_count, sap_ventilation=self._to_ventilation(p), number_of_storeys=p.number_of_storeys, any_unheated_rooms=p.any_unheated_rooms, waste_water_heat_recovery=p.waste_water_heat_recovery, hydro=p.hydro, photovoltaic_array=p.photovoltaic_array, renewable_heat_incentive=( RenewableHeatIncentive( space_heating_kwh=rhi_row.space_heating_kwh, water_heating_kwh=rhi_row.water_heating_kwh, impact_of_loft_insulation_kwh=rhi_row.impact_of_loft_insulation_kwh, impact_of_cavity_insulation_kwh=rhi_row.impact_of_cavity_insulation_kwh, impact_of_solid_wall_insulation_kwh=rhi_row.impact_of_solid_wall_insulation_kwh, ) if rhi_row is not None else None ), mechanical_vent_duct_insulation_level=p.mechanical_vent_duct_insulation_level, addendum=( Addendum( stone_walls=p.addendum_stone_walls, system_build=p.addendum_system_build, addendum_numbers=p.addendum_numbers, ) if ( p.addendum_stone_walls is not None or p.addendum_system_build is not None or p.addendum_numbers is not None ) else None ), ) @private def _windows(self, epc_property_id: int) -> list[EpcWindowModel]: return list( self._session.exec( select(EpcWindowModel) .where(EpcWindowModel.epc_property_id == epc_property_id) .order_by(EpcWindowModel.id) # type: ignore[arg-type] ).all() ) @private def _pv_arrays( self, epc_property_id: int ) -> list[EpcPhotovoltaicArrayModel]: return list( self._session.exec( select(EpcPhotovoltaicArrayModel) .where(EpcPhotovoltaicArrayModel.epc_property_id == epc_property_id) .order_by(EpcPhotovoltaicArrayModel.array_index) # type: ignore[arg-type] ).all() ) @private def _to_energy_element(self, e: EpcEnergyElementModel) -> EnergyElement: return EnergyElement( description=e.description, energy_efficiency_rating=e.energy_efficiency_rating, environmental_efficiency_rating=e.environmental_efficiency_rating, ) @private def _to_sap_heating( self, p: EpcPropertyModel, heating_rows: list[EpcMainHeatingDetailModel] ) -> SapHeating: shower_outlets = ( ShowerOutlets( shower_outlet=ShowerOutlet( shower_outlet_type=p.heating_shower_outlet_type, shower_wwhrs=p.heating_shower_wwhrs, ) ) if p.heating_shower_outlet_type is not None else None ) return SapHeating( instantaneous_wwhrs=InstantaneousWwhrs( wwhrs_index_number1=p.heating_wwhrs_index_number_1, wwhrs_index_number2=p.heating_wwhrs_index_number_2, ), main_heating_details=[self._to_main_heating(m) for m in heating_rows], has_fixed_air_conditioning=p.has_fixed_air_conditioning, cylinder_size=p.heating_cylinder_size, water_heating_code=p.heating_water_heating_code, water_heating_fuel=p.heating_water_heating_fuel, immersion_heating_type=p.heating_immersion_heating_type, shower_outlets=shower_outlets, cylinder_insulation_type=p.heating_cylinder_insulation_type, cylinder_thermostat=p.heating_cylinder_thermostat, secondary_fuel_type=p.heating_secondary_fuel_type, secondary_heating_type=p.heating_secondary_heating_type, cylinder_insulation_thickness_mm=p.heating_cylinder_insulation_thickness_mm, cylinder_volume_measured_l=p.heating_cylinder_volume_measured_l, number_baths=p.heating_number_baths, number_baths_wwhrs=p.heating_number_baths_wwhrs, electric_shower_count=p.heating_electric_shower_count, mixer_shower_count=p.heating_mixer_shower_count, ) @private def _to_main_heating(self, m: EpcMainHeatingDetailModel) -> MainHeatingDetail: return MainHeatingDetail( has_fghrs=m.has_fghrs, main_fuel_type=m.main_fuel_type, heat_emitter_type=m.heat_emitter_type, emitter_temperature=m.emitter_temperature, main_heating_control=m.main_heating_control, fan_flue_present=m.fan_flue_present, boiler_flue_type=m.boiler_flue_type, boiler_ignition_type=m.boiler_ignition_type, central_heating_pump_age=m.central_heating_pump_age, central_heating_pump_age_str=m.central_heating_pump_age_str, main_heating_index_number=m.main_heating_index_number, sap_main_heating_code=m.sap_main_heating_code, main_heating_number=m.main_heating_number, main_heating_category=m.main_heating_category, main_heating_fraction=m.main_heating_fraction, main_heating_data_source=m.main_heating_data_source, condensing=m.condensing, weather_compensator=m.weather_compensator, community_heating_boiler_fuel_type=m.community_heating_boiler_fuel_type, community_heating_chp_fraction=m.community_heating_chp_fraction, ) @private def _to_window(self, w: EpcWindowModel) -> SapWindow: return SapWindow( frame_material=w.frame_material, glazing_gap=w.glazing_gap, orientation=w.orientation, window_type=w.window_type, glazing_type=w.glazing_type, window_width=w.window_width, window_height=w.window_height, draught_proofed=w.draught_proofed, window_location=w.window_location, window_wall_type=w.window_wall_type, permanent_shutters_present=w.permanent_shutters_present, frame_factor=w.frame_factor, window_transmission_details=( WindowTransmissionDetails( u_value=w.transmission_u_value, data_source=_require( w.transmission_data_source, "window.transmission_data_source" ), solar_transmittance=_require( w.transmission_solar_transmittance, "window.transmission_solar_transmittance", ), ) if w.transmission_u_value is not None else None ), permanent_shutters_insulated=w.permanent_shutters_insulated, ) @private def _to_building_part( self, bp: EpcBuildingPartModel, floor_rows: list[EpcFloorDimensionModel] ) -> SapBuildingPart: return SapBuildingPart( identifier=BuildingPartIdentifier(bp.identifier), construction_age_band=bp.construction_age_band, wall_construction=bp.wall_construction, wall_insulation_type=bp.wall_insulation_type, wall_thickness_measured=bp.wall_thickness_measured, party_wall_construction=bp.party_wall_construction, sap_floor_dimensions=[self._to_floor_dimension(f) for f in floor_rows], building_part_number=bp.building_part_number, wall_dry_lined=bp.wall_dry_lined, wall_thickness_mm=bp.wall_thickness_mm, wall_insulation_thickness=bp.wall_insulation_thickness, wall_insulation_thermal_conductivity=bp.wall_insulation_thermal_conductivity, sap_alternative_wall_1=self._to_alt_wall(bp, 1), sap_alternative_wall_2=self._to_alt_wall(bp, 2), floor_heat_loss=bp.floor_heat_loss, floor_insulation_thickness=bp.floor_insulation_thickness, flat_roof_insulation_thickness=bp.flat_roof_insulation_thickness, floor_type=bp.floor_type, floor_construction_type=bp.floor_construction_type, floor_insulation_type_str=bp.floor_insulation_type_str, floor_u_value_known=bp.floor_u_value_known, roof_construction=bp.roof_construction, roof_construction_type=bp.roof_construction_type, curtain_wall_age=bp.curtain_wall_age, roof_insulation_location=bp.roof_insulation_location, roof_insulation_thickness=bp.roof_insulation_thickness, 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 ), ) @private def _to_alt_wall( self, bp: EpcBuildingPartModel, n: int ) -> Optional[SapAlternativeWall]: area = bp.alt_wall_1_area if n == 1 else bp.alt_wall_2_area if area is None: return None dry_lined = bp.alt_wall_1_dry_lined if n == 1 else bp.alt_wall_2_dry_lined construction = ( bp.alt_wall_1_construction if n == 1 else bp.alt_wall_2_construction ) insulation_type = ( bp.alt_wall_1_insulation_type if n == 1 else bp.alt_wall_2_insulation_type ) thickness_measured = ( bp.alt_wall_1_thickness_measured if n == 1 else bp.alt_wall_2_thickness_measured ) insulation_thickness = ( bp.alt_wall_1_insulation_thickness if n == 1 else bp.alt_wall_2_insulation_thickness ) sheltered = bp.alt_wall_1_is_sheltered if n == 1 else bp.alt_wall_2_is_sheltered return SapAlternativeWall( wall_area=area, wall_dry_lined=_require(dry_lined, f"alt_wall_{n}_dry_lined"), wall_construction=_require(construction, f"alt_wall_{n}_construction"), wall_insulation_type=_require( insulation_type, f"alt_wall_{n}_insulation_type" ), wall_thickness_measured=_require( thickness_measured, f"alt_wall_{n}_thickness_measured" ), wall_insulation_thickness=insulation_thickness, # Nullable column (added later than the alt wall itself); a row from # before the column existed reads None → the domain default False. is_sheltered=sheltered if sheltered is not None else False, ) @private def _to_floor_dimension(self, f: EpcFloorDimensionModel) -> SapFloorDimension: return SapFloorDimension( room_height_m=f.room_height_m, total_floor_area_m2=f.total_floor_area_m2, party_wall_length_m=f.party_wall_length_m, heat_loss_perimeter_m=f.heat_loss_perimeter_m, floor=f.floor, floor_insulation=f.floor_insulation, floor_construction=f.floor_construction, is_exposed_floor=f.is_exposed_floor, is_above_partially_heated_space=f.is_above_partially_heated_space, ) @private def _to_energy_source( self, p: EpcPropertyModel, pv_array_rows: list[EpcPhotovoltaicArrayModel] ) -> SapEnergySource: return SapEnergySource( photovoltaic_arrays=( [ PhotovoltaicArray( peak_power=a.peak_power, pitch=a.pitch, overshading=a.overshading, orientation=a.orientation, ) for a in sorted(pv_array_rows, key=lambda a: a.array_index) ] if pv_array_rows else None ), pv_diverter_present=p.energy_pv_diverter_present, mains_gas=p.energy_mains_gas, meter_type=p.energy_meter_type, pv_battery_count=p.energy_pv_battery_count, wind_turbines_count=p.energy_wind_turbines_count, gas_smart_meter_present=p.energy_gas_smart_meter_present, is_dwelling_export_capable=p.energy_is_dwelling_export_capable, wind_turbines_terrain_type=p.energy_wind_turbines_terrain_type, electricity_smart_meter_present=p.energy_electricity_smart_meter_present, pv_connection=p.energy_pv_connection, photovoltaic_supply=( PhotovoltaicSupply( none_or_no_details=PhotovoltaicSupplyNoneOrNoDetails( percent_roof_area=p.energy_pv_percent_roof_area ) ) if p.energy_pv_percent_roof_area is not None else None ), wind_turbine_details=( WindTurbineDetails( hub_height=p.energy_wind_turbine_hub_height, rotor_diameter=_require( p.energy_wind_turbine_rotor_diameter, "energy_wind_turbine_rotor_diameter", ), ) if p.energy_wind_turbine_hub_height is not None else None ), pv_batteries=( PvBatteries( pv_battery=PvBattery(battery_capacity=p.energy_pv_battery_capacity) ) if p.energy_pv_battery_capacity is not None else None ), ) @private def _to_conservatory(self, p: EpcPropertyModel) -> Optional[SapConservatory]: # RdSAP 10 §6.1 — rebuild the non-separated conservatory the mapper # split out of the fabric parts. Presence is signalled by the floor # area (the §6.1 fold always sets all five geometry columns together); # all None when there is no non-separated conservatory (ADR-0036). if p.conservatory_floor_area_m2 is None: return None return SapConservatory( floor_area_m2=p.conservatory_floor_area_m2, glazed_perimeter_m=_require( p.conservatory_glazed_perimeter_m, "conservatory_glazed_perimeter_m" ), double_glazed=_require( p.conservatory_double_glazed, "conservatory_double_glazed" ), thermally_separated=_require( p.conservatory_thermally_separated, "conservatory_thermally_separated", ), room_height_storeys=_require( p.conservatory_room_height_storeys, "conservatory_room_height_storeys", ), ) @private def _to_ventilation(self, p: EpcPropertyModel) -> Optional[SapVentilation]: if not p.ventilation_present: return None return SapVentilation( ventilation_type=p.ventilation_type, draught_lobby=p.ventilation_draught_lobby, pressure_test=p.ventilation_pressure_test, open_flues_count=p.ventilation_open_flues_count, closed_flues_count=p.ventilation_closed_flues_count, boiler_flues_count=p.ventilation_boiler_flues_count, other_flues_count=p.ventilation_other_flues_count, extract_fans_count=p.ventilation_extract_fans_count, passive_vents_count=p.ventilation_passive_vents_count, flueless_gas_fires_count=p.ventilation_flueless_gas_fires_count, ventilation_in_pcdf_database=p.ventilation_in_pcdf_database, sheltered_sides=p.ventilation_sheltered_sides, has_suspended_timber_floor=p.ventilation_has_suspended_timber_floor, suspended_timber_floor_sealed=p.ventilation_suspended_timber_floor_sealed, has_draught_lobby=p.ventilation_has_draught_lobby, air_permeability_ap4_m3_h_m2=p.ventilation_air_permeability_ap4_m3_h_m2, air_permeability_ap50_m3_h_m2=p.ventilation_air_permeability_ap50_m3_h_m2, mechanical_ventilation_kind=p.ventilation_mechanical_ventilation_kind, ) @private def _to_flat_details(self, f: EpcFlatDetailsModel) -> SapFlatDetails: return SapFlatDetails( level=f.level, top_storey=f.top_storey, flat_location=f.flat_location, heat_loss_corridor=f.heat_loss_corridor, storey_count=f.storey_count, unheated_corridor_length_m=f.unheated_corridor_length_m, )