Map to domain from site notes objects 🟩

This commit is contained in:
Daniel Roth 2026-04-14 09:31:07 +00:00
parent 0fbcacb8cd
commit 419892410b

View file

@ -1,5 +1,16 @@
from typing import Union
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from datetime import date
from typing import List, Union
from datatypes.epc.domain.epc_property_data import (
EpcPropertyData,
InstantaneousWwhrs,
MainHeatingDetail,
SapBuildingPart,
SapEnergySource,
SapFloorDimension,
SapHeating,
SapWindow,
)
from datatypes.epc.schema.rdsap_schema_17_0 import RdSapSchema17_0
from datatypes.epc.schema.rdsap_schema_17_1 import RdSapSchema17_1
from datatypes.epc.schema.rdsap_schema_18_0 import RdSapSchema18_0
@ -7,7 +18,17 @@ from datatypes.epc.schema.rdsap_schema_19_0 import RdSapSchema19_0
from datatypes.epc.schema.rdsap_schema_20_0_0 import RdSapSchema20_0_0
from datatypes.epc.schema.rdsap_schema_21_0_0 import RdSapSchema21_0_0
from datatypes.epc.schema.rdsap_schema_21_0_1 import RdSapSchema21_0_1
from datatypes.epc.surveys.pashub_rdsap_site_notes import PasHubRdSapSiteNotes
from datatypes.epc.surveys.pashub_rdsap_site_notes import (
BuildingConstruction,
BuildingMeasurements,
ExtensionConstruction,
ExtensionMeasurements,
FloorMeasurement,
HeatingAndHotWater,
PasHubRdSapSiteNotes,
Ventilation,
Window,
)
AnyRdSapSchema = Union[
RdSapSchema17_0,
@ -24,8 +45,170 @@ class EpcPropertyDataMapper:
@staticmethod
def from_site_notes(survey: PasHubRdSapSiteNotes) -> EpcPropertyData:
raise NotImplementedError
general = survey.general
construction = survey.building_construction
measurements = survey.building_measurements
heating = survey.heating_and_hot_water
ventilation = survey.ventilation
renewables = survey.renewables
room_counts = survey.room_count_elements
sap_building_parts = [_map_main_building_part(construction, measurements)]
if construction.extensions and measurements.extensions:
for ext_c in construction.extensions:
matching = [m for m in measurements.extensions if m.id == ext_c.id]
if matching:
sap_building_parts.append(_map_extension_building_part(ext_c, matching[0]))
total_floor_area = round(
sum(
floor.total_floor_area_m2
for part in sap_building_parts
for floor in part.sap_floor_dimensions
),
2,
)
return EpcPropertyData(
dwelling_type=f"{general.detachment_type} {general.property_type.lower()}",
inspection_date=date.fromisoformat(general.inspection_date),
tenure=general.tenure,
transaction_type=general.transaction_type,
roofs=[],
walls=[],
floors=[],
main_heating=[],
door_count=room_counts.number_of_external_doors,
sap_heating=_map_sap_heating(heating, ventilation),
sap_windows=[_map_sap_window(w) for w in survey.windows],
sap_energy_source=SapEnergySource(
mains_gas=general.mains_gas_available,
meter_type=general.electric_meter_type,
pv_battery_count=renewables.number_of_pv_batteries,
wind_turbines_count=0 if not renewables.wind_turbines else 1,
gas_smart_meter_present=general.gas_smart_meter,
is_dwelling_export_capable=general.dwelling_export_capable,
wind_turbines_terrain_type=general.terrain_type,
electricity_smart_meter_present=general.electricity_smart_meter,
),
sap_building_parts=sap_building_parts,
solar_water_heating=renewables.solar_hot_water,
has_hot_water_cylinder=heating.water_heating.cylinder_size != "No Cylinder",
has_fixed_air_conditioning=ventilation.has_fixed_air_conditioning,
wet_rooms_count=0, # no equivalent in site notes
extensions_count=general.number_of_extensions,
heated_rooms_count=room_counts.number_of_heated_rooms or 0, # absent in site notes → 0
open_chimneys_count=room_counts.number_of_open_chimneys,
habitable_rooms_count=room_counts.number_of_habitable_rooms,
insulated_door_count=room_counts.number_of_insulated_external_doors,
cfl_fixed_lighting_bulbs_count=room_counts.number_of_fixed_cfl_bulbs,
led_fixed_lighting_bulbs_count=room_counts.number_of_fixed_led_bulbs,
incandescent_fixed_lighting_bulbs_count=room_counts.number_of_fixed_incandescent_bulbs,
total_floor_area_m2=total_floor_area,
built_form=general.detachment_type,
property_type=general.property_type,
has_conservatory=survey.conservatories.has_conservatory,
blocked_chimneys_count=room_counts.number_of_blocked_chimneys,
draughtproofed_door_count=room_counts.number_of_draughtproofed_external_doors,
)
@staticmethod
def from_rdsap_schema(_schema: AnyRdSapSchema) -> EpcPropertyData:
raise NotImplementedError
# ---------------------------------------------------------------------------
# Private helpers
# ---------------------------------------------------------------------------
def _extract_age_band(age_range: str) -> str:
"""Return the letter code from a site-notes age range, e.g. 'I: 1996 - 2002''I'."""
return age_range.split(":")[0].strip()
def _map_floor_dimensions(floors: List[FloorMeasurement]) -> List[SapFloorDimension]:
return [
SapFloorDimension(
room_height_m=floor.height_m,
total_floor_area_m2=floor.area_m2,
party_wall_length_m=floor.pwl_m,
heat_loss_perimeter_m=floor.heat_loss_perimeter_m,
)
for floor in floors
]
def _map_main_building_part(
construction: BuildingConstruction,
measurements: BuildingMeasurements,
) -> SapBuildingPart:
main = construction.main_building
return SapBuildingPart(
identifier="main",
construction_age_band=_extract_age_band(main.age_range),
wall_construction=main.walls_construction_type,
wall_insulation_type=main.walls_insulation_type,
wall_thickness_measured=main.wall_thickness_mm > 0,
party_wall_construction=main.party_wall_construction_type,
sap_floor_dimensions=_map_floor_dimensions(measurements.main_building.floors),
wall_thickness_mm=main.wall_thickness_mm,
)
def _map_extension_building_part(
ext_c: ExtensionConstruction,
ext_m: ExtensionMeasurements,
) -> SapBuildingPart:
return SapBuildingPart(
identifier=f"extension_{ext_c.id}",
construction_age_band=_extract_age_band(ext_c.age_range),
wall_construction=ext_c.walls_construction_type,
wall_insulation_type=ext_c.walls_insulation_type,
wall_thickness_measured=ext_c.wall_thickness_mm > 0,
party_wall_construction=ext_c.party_wall_construction_type,
sap_floor_dimensions=_map_floor_dimensions(ext_m.floors),
wall_thickness_mm=ext_c.wall_thickness_mm,
)
def _map_sap_window(window: Window) -> SapWindow:
return SapWindow(
pvc_frame=window.frame_type,
glazing_gap=window.glazing_gap,
orientation=window.orientation,
window_type=window.window_type,
glazing_type=window.glazing_type,
window_width=window.width_m,
window_height=window.height_m,
draught_proofed=window.draught_proofed,
window_location=window.location,
window_wall_type=window.wall_type,
permanent_shutters_present=window.permanent_shutters,
)
def _map_sap_heating(heating: HeatingAndHotWater, ventilation: Ventilation) -> SapHeating:
main = heating.main_heating
secondary = heating.secondary_heating
# secondary_fuel_type is an int code in the domain model; we can't map a
# site-notes string directly, so leave it None unless there is secondary heating.
# The string fuel type is preserved via sap_heating when needed.
secondary_fuel_type = None if secondary.secondary_fuel == "No Secondary Heating" else None
return SapHeating(
instantaneous_wwhrs=InstantaneousWwhrs(),
main_heating_details=[
MainHeatingDetail(
has_fghrs=main.flue_gas_heat_recovery_system,
main_fuel_type=main.fuel,
heat_emitter_type=main.emitter,
emitter_temperature=main.emitter_temperature,
fan_flue_present=main.fan_assist,
main_heating_control=main.controls,
)
],
has_fixed_air_conditioning=str(ventilation.has_fixed_air_conditioning).lower(),
secondary_fuel_type=secondary_fuel_type,
)