Model/infrastructure/postgres/epc_property_table.py
Khalim Conn-Kowlessar 77f90e144e review: store epc_building_part.wall_insulation_thickness as JSONB
PR feedback (dancafc): the SQLModel column was Optional[str], but the
domain `SapBuildingPart.wall_insulation_thickness` is Optional[Union[str,
int]] — `_api_resolve_wall_insulation_thickness` returns an int mm when the
API lodges `wall_insulation_thickness == "measured"` (SAP 10.2 §5.7 /
Table 8). The plain str column round-trips that int back as the string
"100", corrupting the Table 8 insulated-wall U-value lookup.

This column was missed in the round-trip-fidelity §1 JSONB sweep
(#1129) — its `Union[str, int]` sibling `roof_insulation_thickness` was
converted, but `wall_insulation_thickness` was not, and no 21.0.0/21.0.1
fixture lodges "measured" so the gap stayed latent. Convert to JSONB
(matching `roof_insulation_thickness` / `flat_roof_insulation_thickness`),
align the column type to Optional[Union[str, int]] (also removes a pyright
type-mismatch), record it in the migration doc §1, and add a round-trip
guard test asserting an int survives as an int (fails as '100' == 100 on
the old str column).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-05 10:07:24 +00:00

753 lines
37 KiB
Python

from __future__ import annotations
from typing import ClassVar, Optional, Union
from sqlalchemy import Column
from sqlalchemy.dialects.postgresql import JSONB
from sqlmodel import SQLModel, Field
from datatypes.epc.domain.epc_property_data import (
EpcPropertyData,
EnergyElement,
MainHeatingDetail,
RenewableHeatIncentive,
SapBuildingPart,
SapFloorDimension,
SapFlatDetails,
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)
# Identity / admin
uprn: Optional[int] = Field(default=None)
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)
# 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_mains_gas: 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)
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_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_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,
) -> 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,
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,
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_mains_gas=es.mains_gas,
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_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_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_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)
@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,
)
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)
)
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)
)
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_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)
@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,
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,
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_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
),
)
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)
@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,
)
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 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)
element_type: str # roof | wall | floor | main_heating | window | lighting | hot_water | secondary_heating | main_heating_controls
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,
)