Model/infrastructure/postgres/epc_property_table.py
Khalim Conn-Kowlessar 50493f07c6 Persist the lodged wall_u_value override 🟩
SapBuildingPart.wall_u_value (the gov-EPC lodged wall U, authoritative over the
derived default) had no column and dropped on save, so the reloaded dwelling fell
back to the age-band default U — -2.32 SAP on the 2 affected RdSAP corpus certs.
Surfaced by the new DB-round-trip gate. The prior _UNPERSISTED_ALLOWLIST
justification ("deliberate — full-SAP only; RdSAP re-model uses default U-values")
was stale: the cascade DOES honour the lodged U on the API path.

Add wall_u_value to EpcBuildingPartModel, write in from_domain, reconstruct in
_to_building_part, drop the allowlist entry. Verified: certs 39036600 / 38137278
round-trip 72.43 / 69.11 exactly. Needs an added FE column
(epc_building_part.wall_u_value, double precision NULL) on assessment-model#443.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-23 09:23:26 +00:00

1077 lines
52 KiB
Python

from __future__ import annotations
from typing import ClassVar, Optional, Union
from sqlalchemy import BigInteger, Column
from sqlalchemy import Enum as SAEnum
from sqlalchemy.dialects.postgresql import JSONB
from sqlmodel import SQLModel, Field
from datatypes.epc.domain.epc_property_data import (
EpcPropertyData,
EnergyElement,
MainHeatingDetail,
PhotovoltaicArray,
RenewableHeatIncentive,
SapBuildingPart,
SapFloorDimension,
SapFlatDetails,
SapRoofWindow,
SapRoomInRoof,
SapRoomInRoofSurface,
SapWindow,
)
class EpcPropertyModel(SQLModel, table=True):
__tablename__: ClassVar[str] = (
"epc_property" # pyright: ignore[reportIncompatibleVariableOverride]
)
id: Optional[int] = Field(default=None, primary_key=True)
property_id: Optional[int] = Field(default=None)
portfolio_id: Optional[int] = Field(default=None)
uploaded_file_id: Optional[int] = Field(default=None)
# Provenance of this EPC picture: "lodged" (a real public/landlord EPC) or
# "predicted" (EPC Prediction gap-fill, ADR-0031). A property may hold one of
# each, so reads filter on it. Defaults to "lodged" — every existing row is a
# real EPC. (Requires a matching `source` column in the Drizzle schema — see
# docs/handover; the SQLModel mirror is what the test DB builds from.)
source: str = Field(default="lodged")
# Identity / admin
# BIGINT: real 12-digit UPRNs overflow int32 (the FE `property.uprn` is
# already bigint). SQLModel maps `int` -> INTEGER by default, so pin the
# column type or every real-UPRN insert hard-fails NumericValueOutOfRange.
uprn: Optional[int] = Field(
default=None, sa_column=Column(BigInteger, nullable=True)
)
uprn_source: Optional[str] = Field(default=None)
report_reference: Optional[str] = Field(default=None)
report_type: Optional[str] = Field(default=None)
assessment_type: Optional[str] = Field(default=None)
sap_version: Optional[float] = Field(default=None)
schema_type: Optional[str] = Field(default=None)
schema_versions_original: Optional[str] = Field(default=None)
status: Optional[str] = Field(default=None)
calculation_software_version: Optional[str] = Field(default=None)
# Address
address_line_1: Optional[str] = Field(default=None)
address_line_2: Optional[str] = Field(default=None)
post_town: Optional[str] = Field(default=None)
postcode: Optional[str] = Field(default=None)
region_code: Optional[str] = Field(default=None)
country_code: Optional[str] = Field(default=None)
language_code: Optional[str] = Field(default=None)
# Property description
dwelling_type: str
property_type: Optional[str] = Field(default=None)
built_form: Optional[str] = Field(default=None)
tenure: str
transaction_type: str
inspection_date: str # store as ISO string; cast on read if needed
completion_date: Optional[str] = Field(default=None)
registration_date: Optional[str] = Field(default=None)
total_floor_area_m2: float
measurement_type: Optional[int] = Field(default=None)
# Flags
solar_water_heating: bool
has_hot_water_cylinder: bool
has_fixed_air_conditioning: bool
has_conservatory: Optional[bool] = Field(default=None)
has_heated_separate_conservatory: Optional[bool] = Field(default=None)
conservatory_type: Optional[int] = Field(default=None)
# RdSAP 10 §6.1 — geometry of a NON-SEPARATED conservatory folded into the
# dwelling (`EpcPropertyData.sap_conservatory`). 1:1 with the dwelling, so
# flat nullable columns rather than a child table; all None when there is no
# non-separated conservatory. Must round-trip — scoring reads the reloaded
# picture, and a dropped conservatory silently disregards it (ADR-0036).
conservatory_floor_area_m2: Optional[float] = Field(default=None)
conservatory_glazed_perimeter_m: Optional[float] = Field(default=None)
conservatory_double_glazed: Optional[bool] = Field(default=None)
conservatory_thermally_separated: Optional[bool] = Field(default=None)
conservatory_room_height_storeys: Optional[float] = Field(default=None)
# Counts
door_count: int
wet_rooms_count: int
extensions_count: int
heated_rooms_count: int
open_chimneys_count: int
habitable_rooms_count: int
insulated_door_count: int
cfl_fixed_lighting_bulbs_count: int
led_fixed_lighting_bulbs_count: int
incandescent_fixed_lighting_bulbs_count: int
blocked_chimneys_count: Optional[int] = Field(default=None)
draughtproofed_door_count: Optional[int] = Field(default=None)
energy_rating_average: Optional[int] = Field(default=None)
low_energy_fixed_lighting_bulbs_count: Optional[int] = Field(default=None)
fixed_lighting_outlets_count: Optional[int] = Field(default=None)
low_energy_fixed_lighting_outlets_count: Optional[int] = Field(default=None)
number_of_storeys: Optional[int] = Field(default=None)
any_unheated_rooms: Optional[bool] = Field(default=None)
mechanical_vent_duct_insulation_level: Optional[int] = Field(default=None)
# Addendum (cert-level construction flags)
addendum_stone_walls: Optional[bool] = Field(default=None)
addendum_system_build: Optional[bool] = Field(default=None)
addendum_numbers: Optional[list[int]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
# Misc
hydro: Optional[bool] = Field(default=None)
photovoltaic_array: Optional[bool] = Field(default=None)
waste_water_heat_recovery: Optional[str] = Field(default=None)
pressure_test: Optional[int] = Field(default=None)
pressure_test_certificate_number: Optional[int] = Field(default=None)
percent_draughtproofed: Optional[int] = Field(default=None)
insulated_door_u_value: Optional[float] = Field(default=None)
multiple_glazed_proportion: Optional[int] = Field(default=None)
windows_transmission_u_value: Optional[float] = Field(default=None)
windows_transmission_data_source: Optional[int] = Field(default=None)
windows_transmission_solar_transmittance: Optional[float] = Field(default=None)
# Energy source
energy_gas_connection_available: bool
energy_meter_type: str
energy_pv_battery_count: int
energy_wind_turbines_count: int
energy_gas_smart_meter_present: bool
energy_is_dwelling_export_capable: bool
energy_wind_turbines_terrain_type: str
energy_electricity_smart_meter_present: bool
energy_pv_connection: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
energy_pv_percent_roof_area: Optional[int] = Field(default=None)
energy_pv_battery_capacity: Optional[float] = Field(default=None)
# An EV/PV diverter present on the dwelling (SAP §M; the calculator reads it
# via cert_to_inputs). Non-optional bool defaulting False, matching the domain.
energy_pv_diverter_present: bool = Field(default=False)
energy_wind_turbine_hub_height: Optional[float] = Field(default=None)
energy_wind_turbine_rotor_diameter: Optional[float] = Field(default=None)
# Heating config
# Union[int, str] code fields stored as JSONB to preserve the int (API) vs
# str (Site Notes) distinction on round-trip (see docs/migrations/epc-property-round-trip-fidelity.md §1).
heating_cylinder_size: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
heating_water_heating_code: Optional[int] = Field(default=None)
heating_water_heating_fuel: Optional[int] = Field(default=None)
heating_immersion_heating_type: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
heating_cylinder_insulation_type: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
heating_cylinder_thermostat: Optional[str] = Field(default=None)
heating_secondary_fuel_type: Optional[int] = Field(default=None)
heating_secondary_heating_type: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
heating_cylinder_insulation_thickness_mm: Optional[int] = Field(default=None)
heating_cylinder_volume_measured_l: Optional[int] = Field(default=None)
# #1665: read by cert_to_inputs (§4 cylinder loss); dropping it billed the
# cylinder at the Table 2b age-band default (worst case -15.53 SAP).
heating_cylinder_heat_loss: Optional[float] = Field(default=None)
heating_wwhrs_index_number_1: Optional[int] = Field(default=None)
heating_wwhrs_index_number_2: Optional[int] = Field(default=None)
heating_shower_outlet_type: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
heating_shower_wwhrs: Optional[int] = Field(default=None)
heating_number_baths: Optional[int] = Field(default=None)
heating_number_baths_wwhrs: Optional[int] = Field(default=None)
heating_electric_shower_count: Optional[int] = Field(default=None)
heating_mixer_shower_count: Optional[int] = Field(default=None)
# Ventilation
ventilation_type: Optional[str] = Field(default=None)
ventilation_draught_lobby: Optional[bool] = Field(default=None)
ventilation_pressure_test: Optional[str] = Field(default=None)
ventilation_open_flues_count: Optional[int] = Field(default=None)
ventilation_closed_flues_count: Optional[int] = Field(default=None)
ventilation_boiler_flues_count: Optional[int] = Field(default=None)
ventilation_other_flues_count: Optional[int] = Field(default=None)
ventilation_extract_fans_count: Optional[int] = Field(default=None)
ventilation_passive_vents_count: Optional[int] = Field(default=None)
ventilation_flueless_gas_fires_count: Optional[int] = Field(default=None)
ventilation_in_pcdf_database: Optional[bool] = Field(default=None)
# SAP 10.2 §2 lodgements + a presence flag so an all-None SapVentilation
# round-trips as present (not collapsed to None).
ventilation_present: bool = Field(default=False)
ventilation_sheltered_sides: Optional[int] = Field(default=None)
ventilation_has_suspended_timber_floor: Optional[bool] = Field(default=None)
ventilation_suspended_timber_floor_sealed: Optional[bool] = Field(default=None)
ventilation_has_draught_lobby: Optional[bool] = Field(default=None)
ventilation_air_permeability_ap4_m3_h_m2: Optional[float] = Field(default=None)
ventilation_air_permeability_ap50_m3_h_m2: Optional[float] = Field(default=None)
ventilation_mechanical_ventilation_kind: Optional[str] = Field(default=None)
mechanical_ventilation: Optional[int] = Field(default=None)
mechanical_vent_duct_type: Optional[int] = Field(default=None)
mechanical_vent_duct_placement: Optional[int] = Field(default=None)
mechanical_vent_duct_insulation: Optional[int] = Field(default=None)
mechanical_ventilation_index_number: Optional[int] = Field(default=None)
mechanical_vent_measured_installation: Optional[str] = Field(default=None)
@classmethod
def from_epc_property_data(
cls,
data: EpcPropertyData,
property_id: Optional[int] = None,
portfolio_id: Optional[int] = None,
source: str = "lodged",
) -> EpcPropertyModel:
es = data.sap_energy_source
h = data.sap_heating
v = data.sap_ventilation
shower = h.shower_outlets.shower_outlet if h.shower_outlets else None
pv = es.photovoltaic_supply
wt = es.wind_turbine_details
pvb = es.pv_batteries
return cls(
property_id=property_id,
portfolio_id=portfolio_id,
source=source,
uprn=data.uprn,
uprn_source=data.uprn_source,
report_reference=data.report_reference,
report_type=data.report_type,
assessment_type=data.assessment_type,
sap_version=data.sap_version,
schema_type=data.schema_type,
schema_versions_original=data.schema_versions_original,
status=data.status,
calculation_software_version=data.calculation_software_version,
address_line_1=data.address_line_1,
address_line_2=data.address_line_2,
post_town=data.post_town,
postcode=data.postcode,
region_code=data.region_code,
country_code=data.country_code,
language_code=data.language_code,
dwelling_type=data.dwelling_type,
property_type=data.property_type,
built_form=data.built_form,
tenure=data.tenure,
transaction_type=data.transaction_type,
inspection_date=data.inspection_date.isoformat(),
completion_date=(
data.completion_date.isoformat() if data.completion_date else None
),
registration_date=(
data.registration_date.isoformat() if data.registration_date else None
),
total_floor_area_m2=data.total_floor_area_m2,
measurement_type=data.measurement_type,
solar_water_heating=data.solar_water_heating,
has_hot_water_cylinder=data.has_hot_water_cylinder,
has_fixed_air_conditioning=data.has_fixed_air_conditioning,
has_conservatory=data.has_conservatory,
has_heated_separate_conservatory=data.has_heated_separate_conservatory,
conservatory_type=data.conservatory_type,
conservatory_floor_area_m2=(
data.sap_conservatory.floor_area_m2 if data.sap_conservatory else None
),
conservatory_glazed_perimeter_m=(
data.sap_conservatory.glazed_perimeter_m
if data.sap_conservatory
else None
),
conservatory_double_glazed=(
data.sap_conservatory.double_glazed if data.sap_conservatory else None
),
conservatory_thermally_separated=(
data.sap_conservatory.thermally_separated
if data.sap_conservatory
else None
),
conservatory_room_height_storeys=(
data.sap_conservatory.room_height_storeys
if data.sap_conservatory
else None
),
door_count=data.door_count,
wet_rooms_count=data.wet_rooms_count,
extensions_count=data.extensions_count,
heated_rooms_count=data.heated_rooms_count,
open_chimneys_count=data.open_chimneys_count,
habitable_rooms_count=data.habitable_rooms_count,
insulated_door_count=data.insulated_door_count,
cfl_fixed_lighting_bulbs_count=data.cfl_fixed_lighting_bulbs_count,
led_fixed_lighting_bulbs_count=data.led_fixed_lighting_bulbs_count,
incandescent_fixed_lighting_bulbs_count=data.incandescent_fixed_lighting_bulbs_count,
blocked_chimneys_count=data.blocked_chimneys_count,
draughtproofed_door_count=data.draughtproofed_door_count,
energy_rating_average=data.energy_rating_average,
low_energy_fixed_lighting_bulbs_count=data.low_energy_fixed_lighting_bulbs_count,
fixed_lighting_outlets_count=data.fixed_lighting_outlets_count,
low_energy_fixed_lighting_outlets_count=data.low_energy_fixed_lighting_outlets_count,
number_of_storeys=data.number_of_storeys,
any_unheated_rooms=data.any_unheated_rooms,
mechanical_vent_duct_insulation_level=data.mechanical_vent_duct_insulation_level,
addendum_stone_walls=data.addendum.stone_walls if data.addendum else None,
addendum_system_build=(
data.addendum.system_build if data.addendum else None
),
addendum_numbers=data.addendum.addendum_numbers if data.addendum else None,
hydro=data.hydro,
photovoltaic_array=data.photovoltaic_array,
waste_water_heat_recovery=data.waste_water_heat_recovery,
pressure_test=data.pressure_test,
pressure_test_certificate_number=data.pressure_test_certificate_number,
percent_draughtproofed=data.percent_draughtproofed,
insulated_door_u_value=data.insulated_door_u_value,
multiple_glazed_proportion=data.multiple_glazed_proportion,
windows_transmission_u_value=(
data.windows_transmission_details.u_value
if data.windows_transmission_details
else None
),
windows_transmission_data_source=(
data.windows_transmission_details.data_source
if data.windows_transmission_details
else None
),
windows_transmission_solar_transmittance=(
data.windows_transmission_details.solar_transmittance
if data.windows_transmission_details
else None
),
energy_gas_connection_available=es.gas_connection_available,
energy_meter_type=str(es.meter_type),
energy_pv_battery_count=es.pv_battery_count,
energy_wind_turbines_count=es.wind_turbines_count,
energy_gas_smart_meter_present=es.gas_smart_meter_present,
energy_is_dwelling_export_capable=es.is_dwelling_export_capable,
energy_wind_turbines_terrain_type=str(es.wind_turbines_terrain_type),
energy_electricity_smart_meter_present=es.electricity_smart_meter_present,
energy_pv_connection=es.pv_connection,
energy_pv_percent_roof_area=(
pv.none_or_no_details.percent_roof_area if pv else None
),
energy_pv_battery_capacity=pvb.pv_battery.battery_capacity if pvb else None,
energy_pv_diverter_present=es.pv_diverter_present,
energy_wind_turbine_hub_height=wt.hub_height if wt else None,
energy_wind_turbine_rotor_diameter=wt.rotor_diameter if wt else None,
heating_cylinder_size=h.cylinder_size,
heating_water_heating_code=h.water_heating_code,
heating_water_heating_fuel=h.water_heating_fuel,
heating_immersion_heating_type=h.immersion_heating_type,
heating_cylinder_insulation_type=h.cylinder_insulation_type,
heating_cylinder_thermostat=h.cylinder_thermostat,
heating_secondary_fuel_type=h.secondary_fuel_type,
heating_secondary_heating_type=h.secondary_heating_type,
heating_cylinder_insulation_thickness_mm=h.cylinder_insulation_thickness_mm,
heating_cylinder_volume_measured_l=h.cylinder_volume_measured_l,
heating_cylinder_heat_loss=h.cylinder_heat_loss,
heating_wwhrs_index_number_1=h.instantaneous_wwhrs.wwhrs_index_number1,
heating_wwhrs_index_number_2=h.instantaneous_wwhrs.wwhrs_index_number2,
heating_shower_outlet_type=shower.shower_outlet_type if shower else None,
heating_shower_wwhrs=shower.shower_wwhrs if shower else None,
heating_number_baths=h.number_baths,
heating_number_baths_wwhrs=h.number_baths_wwhrs,
heating_electric_shower_count=h.electric_shower_count,
heating_mixer_shower_count=h.mixer_shower_count,
ventilation_type=v.ventilation_type if v else None,
ventilation_draught_lobby=v.draught_lobby if v else None,
ventilation_pressure_test=v.pressure_test if v else None,
ventilation_open_flues_count=v.open_flues_count if v else None,
ventilation_closed_flues_count=v.closed_flues_count if v else None,
ventilation_boiler_flues_count=v.boiler_flues_count if v else None,
ventilation_other_flues_count=v.other_flues_count if v else None,
ventilation_extract_fans_count=v.extract_fans_count if v else None,
ventilation_passive_vents_count=v.passive_vents_count if v else None,
ventilation_flueless_gas_fires_count=(
v.flueless_gas_fires_count if v else None
),
ventilation_in_pcdf_database=v.ventilation_in_pcdf_database if v else None,
ventilation_present=v is not None,
ventilation_sheltered_sides=v.sheltered_sides if v else None,
ventilation_has_suspended_timber_floor=(
v.has_suspended_timber_floor if v else None
),
ventilation_suspended_timber_floor_sealed=(
v.suspended_timber_floor_sealed if v else None
),
ventilation_has_draught_lobby=v.has_draught_lobby if v else None,
ventilation_air_permeability_ap4_m3_h_m2=(
v.air_permeability_ap4_m3_h_m2 if v else None
),
ventilation_air_permeability_ap50_m3_h_m2=(
v.air_permeability_ap50_m3_h_m2 if v else None
),
ventilation_mechanical_ventilation_kind=(
v.mechanical_ventilation_kind if v else None
),
mechanical_ventilation=data.mechanical_ventilation,
mechanical_vent_duct_type=data.mechanical_vent_duct_type,
mechanical_vent_duct_placement=data.mechanical_vent_duct_placement,
mechanical_vent_duct_insulation=data.mechanical_vent_duct_insulation,
mechanical_ventilation_index_number=data.mechanical_ventilation_index_number,
mechanical_vent_measured_installation=data.mechanical_vent_measured_installation,
)
class EpcPropertyEnergyPerformanceModel(SQLModel, table=True):
__tablename__: ClassVar[str] = (
"epc_property_energy_performance" # pyright: ignore[reportIncompatibleVariableOverride]
)
id: Optional[int] = Field(default=None, primary_key=True)
epc_property_id: int = Field(
foreign_key="epc_property.id", nullable=False, unique=True
)
energy_rating_current: Optional[int] = Field(default=None)
energy_consumption_current: Optional[int] = Field(default=None)
environmental_impact_current: Optional[int] = Field(default=None)
heating_cost_current: Optional[float] = Field(default=None)
lighting_cost_current: Optional[float] = Field(default=None)
hot_water_cost_current: Optional[float] = Field(default=None)
co2_emissions_current: Optional[float] = Field(default=None)
co2_emissions_current_per_floor_area: Optional[int] = Field(default=None)
current_energy_efficiency_band: Optional[str] = Field(default=None)
energy_rating_potential: Optional[float] = Field(default=None)
energy_consumption_potential: Optional[int] = Field(default=None)
environmental_impact_potential: Optional[int] = Field(default=None)
heating_cost_potential: Optional[float] = Field(default=None)
lighting_cost_potential: Optional[float] = Field(default=None)
hot_water_cost_potential: Optional[float] = Field(default=None)
co2_emissions_potential: Optional[float] = Field(default=None)
potential_energy_efficiency_band: Optional[str] = Field(default=None)
@classmethod
def from_epc_property_data(
cls, data: EpcPropertyData, epc_property_id: int
) -> EpcPropertyEnergyPerformanceModel:
return cls(
epc_property_id=epc_property_id,
energy_rating_current=data.energy_rating_current,
energy_consumption_current=data.energy_consumption_current,
environmental_impact_current=data.environmental_impact_current,
heating_cost_current=data.heating_cost_current,
lighting_cost_current=data.lighting_cost_current,
hot_water_cost_current=data.hot_water_cost_current,
co2_emissions_current=data.co2_emissions_current,
co2_emissions_current_per_floor_area=data.co2_emissions_current_per_floor_area,
current_energy_efficiency_band=(
data.current_energy_efficiency_band.value
if data.current_energy_efficiency_band
else None
),
energy_rating_potential=data.energy_rating_potential,
energy_consumption_potential=data.energy_consumption_potential,
environmental_impact_potential=data.environmental_impact_potential,
heating_cost_potential=data.heating_cost_potential,
lighting_cost_potential=data.lighting_cost_potential,
hot_water_cost_potential=data.hot_water_cost_potential,
co2_emissions_potential=data.co2_emissions_potential,
potential_energy_efficiency_band=(
data.potential_energy_efficiency_band.value
if data.potential_energy_efficiency_band
else None
),
)
class EpcRenewableHeatIncentiveModel(SQLModel, table=True):
__tablename__: ClassVar[str] = (
"epc_renewable_heat_incentive" # pyright: ignore[reportIncompatibleVariableOverride]
)
id: Optional[int] = Field(default=None, primary_key=True)
epc_property_id: int = Field(
foreign_key="epc_property.id", nullable=False, unique=True
)
space_heating_kwh: float
water_heating_kwh: float
impact_of_loft_insulation_kwh: Optional[float] = Field(default=None)
impact_of_cavity_insulation_kwh: Optional[float] = Field(default=None)
impact_of_solid_wall_insulation_kwh: Optional[float] = Field(default=None)
@classmethod
def from_domain(
cls, rhi: RenewableHeatIncentive, epc_property_id: int
) -> EpcRenewableHeatIncentiveModel:
return cls(
epc_property_id=epc_property_id,
space_heating_kwh=rhi.space_heating_kwh,
water_heating_kwh=rhi.water_heating_kwh,
impact_of_loft_insulation_kwh=rhi.impact_of_loft_insulation_kwh,
impact_of_cavity_insulation_kwh=rhi.impact_of_cavity_insulation_kwh,
impact_of_solid_wall_insulation_kwh=rhi.impact_of_solid_wall_insulation_kwh,
)
class EpcFlatDetailsModel(SQLModel, table=True):
__tablename__: ClassVar[str] = (
"epc_flat_details" # pyright: ignore[reportIncompatibleVariableOverride]
)
id: Optional[int] = Field(default=None, primary_key=True)
epc_property_id: int = Field(
foreign_key="epc_property.id", nullable=False, unique=True
)
level: int
top_storey: str
flat_location: int
heat_loss_corridor: int
storey_count: Optional[int] = Field(default=None)
unheated_corridor_length_m: Optional[int] = Field(default=None)
@classmethod
def from_domain(
cls, flat: SapFlatDetails, epc_property_id: int
) -> EpcFlatDetailsModel:
return cls(
epc_property_id=epc_property_id,
level=flat.level,
top_storey=flat.top_storey,
flat_location=flat.flat_location,
heat_loss_corridor=flat.heat_loss_corridor,
storey_count=flat.storey_count,
unheated_corridor_length_m=flat.unheated_corridor_length_m,
)
class EpcMainHeatingDetailModel(SQLModel, table=True):
__tablename__: ClassVar[str] = (
"epc_main_heating_detail" # pyright: ignore[reportIncompatibleVariableOverride]
)
id: Optional[int] = Field(default=None, primary_key=True)
epc_property_id: int = Field(foreign_key="epc_property.id", nullable=False)
has_fghrs: bool
# Union[int, str] code fields — JSONB to preserve int/str on round-trip.
main_fuel_type: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
heat_emitter_type: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
emitter_temperature: Union[int, str] = Field(
sa_column=Column(JSONB, nullable=False)
)
main_heating_control: Union[int, str] = Field(
sa_column=Column(JSONB, nullable=False)
)
fan_flue_present: Optional[bool] = Field(default=None)
boiler_flue_type: Optional[int] = Field(default=None)
boiler_ignition_type: Optional[int] = Field(default=None)
central_heating_pump_age: Optional[int] = Field(default=None)
central_heating_pump_age_str: Optional[str] = Field(default=None)
main_heating_index_number: Optional[int] = Field(default=None)
sap_main_heating_code: Optional[int] = Field(default=None)
main_heating_number: Optional[int] = Field(default=None)
main_heating_category: Optional[int] = Field(default=None)
main_heating_fraction: Optional[int] = Field(default=None)
main_heating_data_source: Optional[int] = Field(default=None)
condensing: Optional[bool] = Field(default=None)
weather_compensator: Optional[bool] = Field(default=None)
# Community-heating fields (SAP Table 4d; the calculator reads them via
# cert_to_inputs for a community-heated dwelling).
community_heating_boiler_fuel_type: Optional[int] = Field(default=None)
community_heating_chp_fraction: Optional[float] = Field(default=None)
@classmethod
def from_domain(
cls, detail: MainHeatingDetail, epc_property_id: int
) -> EpcMainHeatingDetailModel:
return cls(
epc_property_id=epc_property_id,
has_fghrs=detail.has_fghrs,
main_fuel_type=detail.main_fuel_type,
heat_emitter_type=detail.heat_emitter_type,
emitter_temperature=detail.emitter_temperature,
main_heating_control=detail.main_heating_control,
fan_flue_present=detail.fan_flue_present,
boiler_flue_type=detail.boiler_flue_type,
boiler_ignition_type=detail.boiler_ignition_type,
central_heating_pump_age=detail.central_heating_pump_age,
central_heating_pump_age_str=detail.central_heating_pump_age_str,
main_heating_index_number=detail.main_heating_index_number,
sap_main_heating_code=detail.sap_main_heating_code,
main_heating_number=detail.main_heating_number,
main_heating_category=detail.main_heating_category,
main_heating_fraction=detail.main_heating_fraction,
main_heating_data_source=detail.main_heating_data_source,
condensing=detail.condensing,
weather_compensator=detail.weather_compensator,
community_heating_boiler_fuel_type=detail.community_heating_boiler_fuel_type,
community_heating_chp_fraction=detail.community_heating_chp_fraction,
)
class EpcBuildingPartModel(SQLModel, table=True):
__tablename__: ClassVar[str] = (
"epc_building_part" # pyright: ignore[reportIncompatibleVariableOverride]
)
id: Optional[int] = Field(default=None, primary_key=True)
epc_property_id: int = Field(foreign_key="epc_property.id", nullable=False)
identifier: str
construction_age_band: str
# Union[int, str] code fields — JSONB to preserve int/str on round-trip.
wall_construction: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
wall_insulation_type: Union[int, str] = Field(
sa_column=Column(JSONB, nullable=False)
)
wall_thickness_measured: bool
party_wall_construction: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
building_part_number: Optional[int] = Field(default=None)
wall_dry_lined: Optional[bool] = Field(default=None)
wall_thickness_mm: Optional[int] = Field(default=None)
# Union[str, int] — int mm when the API lodges
# `wall_insulation_thickness == "measured"` (resolved by
# `_api_resolve_wall_insulation_thickness`), else the lodged string
# ("NI", a numeric string, ...). JSONB to preserve int vs str on
# round-trip, exactly like the sibling `roof_insulation_thickness` /
# `flat_roof_insulation_thickness`.
wall_insulation_thickness: Optional[Union[str, int]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
# Union[int, str] SAP code (int from the API, str from Site Notes) — JSONB to
# preserve the type on round-trip, like the insulation-thickness siblings.
wall_insulation_thermal_conductivity: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
floor_heat_loss: Optional[int] = Field(default=None)
floor_insulation_thickness: Optional[str] = Field(default=None)
flat_roof_insulation_thickness: Optional[Union[str, int]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
floor_type: Optional[str] = Field(default=None)
floor_construction_type: Optional[str] = Field(default=None)
floor_insulation_type_str: Optional[str] = Field(default=None)
floor_u_value_known: Optional[bool] = Field(default=None)
roof_construction: Optional[int] = Field(default=None)
roof_construction_type: Optional[str] = Field(default=None)
curtain_wall_age: Optional[str] = Field(default=None)
roof_insulation_location: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
roof_insulation_thickness: Optional[Union[str, int]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
# RdSAP 10 §5.11.2 — a roof insulated AT RAFTERS lodges its thickness here,
# NOT in `roof_insulation_thickness` (which stays null for rafter roofs).
# `heat_transmission.py:922` prefers it when `roof_insulation_location == 1`
# so the measured Table 16 column (2) row applies; without it the roof falls
# to the Table 18 column (2) unknown-thickness default (2.30 for bands A-D),
# a 10x U-value error. Union[str, int] ("250mm" from the API) → JSONB, like
# the `roof_insulation_thickness` sibling.
rafter_insulation_thickness: Optional[Union[str, int]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
# RdSAP 10 §5.17 / Table 23 — selects the basement-wall U-value column for
# the part's primary wall (`heat_transmission.py:955` via
# `SapBuildingPart.main_wall_is_basement`). NOT a lodged U-value: it picks
# WHICH RdSAP default applies, so unlike wall/roof/floor_u_value it must be
# persisted. Nullable is load-bearing — `None` means "not stated, fall back
# to the gov-API code-6 heuristic", `False` means "explicitly system-built,
# do NOT apply the heuristic". Collapsing False to None inverts the answer.
wall_is_basement: Optional[bool] = Field(default=None)
# #1665: lodged wall U (gov-EPC `wall_u_value`), authoritative over the
# derived default when present; dropped it fell back to the age-band default
# (-2.32 SAP). The prior allowlist "RdSAP uses defaults" was stale — the
# cascade does honour the lodged U on the API path.
wall_u_value: Optional[float] = Field(default=None)
room_in_roof_floor_area: Optional[float] = Field(default=None)
room_in_roof_construction_age_band: Optional[str] = Field(default=None)
alt_wall_1_area: Optional[float] = Field(default=None)
alt_wall_1_dry_lined: Optional[str] = Field(default=None)
alt_wall_1_construction: Optional[int] = Field(default=None)
alt_wall_1_insulation_type: Optional[int] = Field(default=None)
alt_wall_1_thickness_measured: Optional[str] = Field(default=None)
alt_wall_1_insulation_thickness: Optional[str] = Field(default=None)
alt_wall_1_is_sheltered: Optional[bool] = Field(default=None)
# #1665: calculator-read alt-wall fields (heat_transmission.py) with no column.
alt_wall_1_u_value: Optional[float] = Field(default=None)
alt_wall_1_thickness_mm: Optional[int] = Field(default=None)
# Nullable is load-bearing: None ("not stated") != False ("not a basement").
# A NOT NULL DEFAULT false re-enables the code-6 basement heuristic (cf #1661).
alt_wall_1_is_basement: Optional[bool] = Field(default=None)
alt_wall_2_area: Optional[float] = Field(default=None)
alt_wall_2_dry_lined: Optional[str] = Field(default=None)
alt_wall_2_construction: Optional[int] = Field(default=None)
alt_wall_2_insulation_type: Optional[int] = Field(default=None)
alt_wall_2_thickness_measured: Optional[str] = Field(default=None)
alt_wall_2_insulation_thickness: Optional[str] = Field(default=None)
alt_wall_2_is_sheltered: Optional[bool] = Field(default=None)
alt_wall_2_u_value: Optional[float] = Field(default=None)
alt_wall_2_thickness_mm: Optional[int] = Field(default=None)
alt_wall_2_is_basement: Optional[bool] = Field(default=None)
@classmethod
def from_domain(
cls, part: SapBuildingPart, epc_property_id: int
) -> EpcBuildingPartModel:
rir = part.sap_room_in_roof
aw1 = part.sap_alternative_wall_1
aw2 = part.sap_alternative_wall_2
return cls(
epc_property_id=epc_property_id,
identifier=part.identifier.value,
construction_age_band=part.construction_age_band,
wall_construction=part.wall_construction,
wall_insulation_type=part.wall_insulation_type,
wall_thickness_measured=part.wall_thickness_measured,
party_wall_construction=part.party_wall_construction,
building_part_number=part.building_part_number,
wall_dry_lined=part.wall_dry_lined,
wall_thickness_mm=part.wall_thickness_mm,
wall_insulation_thickness=part.wall_insulation_thickness,
wall_insulation_thermal_conductivity=part.wall_insulation_thermal_conductivity,
floor_heat_loss=part.floor_heat_loss,
floor_insulation_thickness=part.floor_insulation_thickness,
flat_roof_insulation_thickness=part.flat_roof_insulation_thickness,
floor_type=part.floor_type,
floor_construction_type=part.floor_construction_type,
floor_insulation_type_str=part.floor_insulation_type_str,
floor_u_value_known=part.floor_u_value_known,
roof_construction=part.roof_construction,
roof_construction_type=part.roof_construction_type,
curtain_wall_age=part.curtain_wall_age,
roof_insulation_location=part.roof_insulation_location,
roof_insulation_thickness=part.roof_insulation_thickness,
rafter_insulation_thickness=part.rafter_insulation_thickness,
wall_is_basement=part.wall_is_basement,
wall_u_value=part.wall_u_value,
room_in_roof_floor_area=float(rir.floor_area) if rir else None,
room_in_roof_construction_age_band=(
rir.construction_age_band if rir else None
),
alt_wall_1_area=aw1.wall_area if aw1 else None,
alt_wall_1_dry_lined=aw1.wall_dry_lined if aw1 else None,
alt_wall_1_construction=aw1.wall_construction if aw1 else None,
alt_wall_1_insulation_type=aw1.wall_insulation_type if aw1 else None,
alt_wall_1_thickness_measured=aw1.wall_thickness_measured if aw1 else None,
alt_wall_1_insulation_thickness=(
aw1.wall_insulation_thickness if aw1 else None
),
alt_wall_1_is_sheltered=aw1.is_sheltered if aw1 else None,
alt_wall_1_u_value=aw1.u_value if aw1 else None,
alt_wall_1_thickness_mm=aw1.wall_thickness_mm if aw1 else None,
alt_wall_1_is_basement=aw1.is_basement if aw1 else None,
alt_wall_2_area=aw2.wall_area if aw2 else None,
alt_wall_2_dry_lined=aw2.wall_dry_lined if aw2 else None,
alt_wall_2_construction=aw2.wall_construction if aw2 else None,
alt_wall_2_insulation_type=aw2.wall_insulation_type if aw2 else None,
alt_wall_2_thickness_measured=aw2.wall_thickness_measured if aw2 else None,
alt_wall_2_insulation_thickness=(
aw2.wall_insulation_thickness if aw2 else None
),
alt_wall_2_is_sheltered=aw2.is_sheltered if aw2 else None,
alt_wall_2_u_value=aw2.u_value if aw2 else None,
alt_wall_2_thickness_mm=aw2.wall_thickness_mm if aw2 else None,
alt_wall_2_is_basement=aw2.is_basement if aw2 else None,
)
class EpcFloorDimensionModel(SQLModel, table=True):
__tablename__: ClassVar[str] = (
"epc_floor_dimension" # pyright: ignore[reportIncompatibleVariableOverride]
)
id: Optional[int] = Field(default=None, primary_key=True)
epc_building_part_id: int = Field(
foreign_key="epc_building_part.id", nullable=False
)
floor: Optional[int] = Field(default=None)
room_height_m: float
total_floor_area_m2: float
party_wall_length_m: float
heat_loss_perimeter_m: float
floor_insulation: Optional[int] = Field(default=None)
floor_construction: Optional[int] = Field(default=None)
# Heat-loss flags (SAP 10.2 §3.3): a floor over outside air / over a
# partially-heated space takes a different U-value path. Default False
# matches the domain default for an ordinary ground floor.
is_exposed_floor: bool = Field(default=False)
is_above_partially_heated_space: bool = Field(default=False)
@classmethod
def from_domain(
cls, dim: SapFloorDimension, epc_building_part_id: int
) -> EpcFloorDimensionModel:
return cls(
epc_building_part_id=epc_building_part_id,
floor=dim.floor,
room_height_m=dim.room_height_m,
total_floor_area_m2=dim.total_floor_area_m2,
party_wall_length_m=dim.party_wall_length_m,
heat_loss_perimeter_m=dim.heat_loss_perimeter_m,
floor_insulation=dim.floor_insulation,
floor_construction=dim.floor_construction,
is_exposed_floor=dim.is_exposed_floor,
is_above_partially_heated_space=dim.is_above_partially_heated_space,
)
class EpcWindowModel(SQLModel, table=True):
__tablename__: ClassVar[str] = (
"epc_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)
frame_material: Optional[str] = Field(default=None)
# Union[int, str] / Union[bool, str] code fields — JSONB to preserve type on round-trip.
glazing_gap: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
orientation: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
window_type: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
glazing_type: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
window_width: float
window_height: float
draught_proofed: Union[bool, str] = Field(sa_column=Column(JSONB, nullable=False))
window_location: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
window_wall_type: Union[int, str] = Field(sa_column=Column(JSONB, nullable=False))
permanent_shutters_present: Union[bool, str] = Field(
sa_column=Column(JSONB, nullable=False)
)
frame_factor: Optional[float] = Field(default=None)
permanent_shutters_insulated: Optional[str] = Field(default=None)
transmission_u_value: Optional[float] = Field(default=None)
transmission_data_source: Optional[Union[int, str]] = Field(
default=None, sa_column=Column(JSONB, nullable=True)
)
transmission_solar_transmittance: Optional[float] = Field(default=None)
@classmethod
def from_domain(cls, window: SapWindow, epc_property_id: int) -> EpcWindowModel:
td = window.window_transmission_details
return cls(
epc_property_id=epc_property_id,
frame_material=window.frame_material,
glazing_gap=window.glazing_gap,
orientation=window.orientation,
window_type=window.window_type,
glazing_type=window.glazing_type,
window_width=window.window_width,
window_height=window.window_height,
draught_proofed=window.draught_proofed,
window_location=window.window_location,
window_wall_type=window.window_wall_type,
permanent_shutters_present=window.permanent_shutters_present,
frame_factor=window.frame_factor,
permanent_shutters_insulated=window.permanent_shutters_insulated,
transmission_u_value=td.u_value if td else None,
transmission_data_source=td.data_source if td else None,
transmission_solar_transmittance=td.solar_transmittance if td else None,
)
class EpcPhotovoltaicArrayModel(SQLModel, table=True):
__tablename__: ClassVar[str] = (
"epc_photovoltaic_array" # 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_energy_source.photovoltaic_arrays`. Persisted so the
# array order is recoverable for round-trip equality (a dwelling can carry
# several arrays at different pitch/orientation).
array_index: int
# kWp as a float (e.g. 3.24). pitch / overshading / orientation are SAP
# enumeration CODES (small categorical ints), not physical degrees.
peak_power: float
pitch: int
overshading: int
orientation: Optional[int] = Field(default=None)
@classmethod
def from_domain(
cls, array: PhotovoltaicArray, array_index: int, epc_property_id: int
) -> EpcPhotovoltaicArrayModel:
return cls(
epc_property_id=epc_property_id,
array_index=array_index,
peak_power=array.peak_power,
pitch=array.pitch,
overshading=array.overshading,
orientation=array.orientation,
)
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 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]
)
id: Optional[int] = Field(default=None, primary_key=True)
epc_property_id: int = Field(foreign_key="epc_property.id", nullable=False)
# The live column is the ``energy_element_type`` Postgres enum (owned by the
# Drizzle schema). Binding it as a bare ``str`` makes SQLAlchemy cast the
# value to VARCHAR, which Postgres won't implicitly coerce to the enum on
# INSERT — so declare the native enum explicitly (``create_type=False``: the
# type already exists). The values stay plain strings on the domain side.
element_type: str = Field(
sa_column=Column(
SAEnum(
"roof",
"wall",
"floor",
"main_heating",
"window",
"lighting",
"hot_water",
"secondary_heating",
"main_heating_controls",
name="energy_element_type",
create_type=False,
),
nullable=False,
)
)
description: str
energy_efficiency_rating: int
environmental_efficiency_rating: int
@classmethod
def from_domain(
cls, element: EnergyElement, element_type: str, epc_property_id: int
) -> EpcEnergyElementModel:
return cls(
epc_property_id=epc_property_id,
element_type=element_type,
description=element.description,
energy_efficiency_rating=element.energy_efficiency_rating,
environmental_efficiency_rating=element.environmental_efficiency_rating,
)