Merge branch 'main' into feature/pashub-to-ara

This commit is contained in:
Daniel Roth 2026-04-16 15:15:49 +00:00
commit ead01979dc
35 changed files with 5258 additions and 205 deletions

View file

@ -16,6 +16,7 @@ from sqlalchemy import (
from backend.app.db.base import Base
from backend.app.db.models.users import UserModel # noqa
from backend.app.db.models.materials import MaterialType
from datatypes.epc.domain.epc import Epc
class PortfolioStatus(enum.Enum):
@ -100,16 +101,6 @@ class PropertyCreationStatus(enum.Enum):
ERROR = "ERROR"
class Epc(enum.Enum): # TODO: Move to domain?
A = "A"
B = "B"
C = "C"
D = "D"
E = "E"
F = "F"
G = "G"
class PropertyModel(Base):
__tablename__ = "property"
id = Column(Integer, primary_key=True, autoincrement=True)

View file

@ -17,8 +17,8 @@ from datetime import datetime
from backend.app.db.base import Base
from backend.app.db.models.portfolio import Portfolio, PortfolioGoal, PropertyModel
from backend.app.db.models.materials import Material
from backend.app.db.models.portfolio import Epc
from datatypes.enums import QuantityUnits
from datatypes.epc.domain.epc import Epc
def portfolio_goal_values(enum_cls: Type[PortfolioGoal]) -> List[str]:
@ -54,9 +54,7 @@ class Recommendation(Base):
class RecommendationMaterials(Base):
__tablename__ = "recommendation_materials"
id: Mapped[int] = mapped_column(
BigInteger, primary_key=True, autoincrement=True
)
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
recommendation_id: Mapped[int] = mapped_column(
BigInteger,

View file

@ -16,6 +16,7 @@ class FileTypeEnum(enum.Enum):
PAS_2023_OCCUPANCY = "pas_2023_occupancy"
ECMK_SITE_NOTE = "ecmk_site_note"
ECMK_RD_SAP_SITE_NOTE = "ecmk_rd_sap_site_note"
ECMK_SURVEY_XML = "ecmk_survey_xml"
class FileSourceEnum(enum.Enum):

View file

@ -2,8 +2,8 @@ from dataclasses import dataclass
from datetime import datetime
from typing import Optional
from backend.app.db.models.portfolio import Epc
from backend.app.db.models.recommendations import PlanTypeEnum
from datatypes.epc.domain.epc import Epc
@dataclass(frozen=True)

View file

@ -6,7 +6,8 @@ from backend.app.domain.classes.plan import Plan
from backend.app.domain.classes.scenario import Scenario
from backend.app.domain.records.plan_record import PlanRecord
from backend.app.domain.records.scenario_record import ScenarioRecord
from backend.app.db.models.portfolio import Epc, PortfolioGoal
from backend.app.db.models.portfolio import PortfolioGoal
from datatypes.epc.domain.epc import Epc
@pytest.fixture

View file

@ -6,8 +6,9 @@ from backend.app.domain.classes.plan import Plan
from backend.app.domain.classes.scenario import Scenario
from backend.app.domain.records.plan_record import PlanRecord
from backend.app.domain.records.scenario_record import ScenarioRecord
from backend.app.db.models.portfolio import Epc, PortfolioGoal
from backend.app.db.models.portfolio import PortfolioGoal
from backend.categorisation.processor import choose_cheapest_relevant_plan
from datatypes.epc.domain.epc import Epc
@pytest.fixture

View file

@ -0,0 +1,53 @@
import os
from typing import Any
from openpyxl import Workbook, load_workbook
from openpyxl.worksheet.worksheet import Worksheet
def write_row(file_path: str, row_data: dict[str, Any]) -> None:
new_keys = list(row_data.keys())
if not os.path.exists(file_path):
wb = Workbook()
ws: Worksheet = wb.active # type: ignore[assignment]
ws.append(new_keys)
ws.append(list(row_data.values()))
wb.save(file_path)
return
wb = load_workbook(file_path)
ws = wb.active # type: ignore[assignment]
# Build a mutable header list and insert new columns using insert_cols so
# that existing row data shifts along with the headers.
# Filter out None to guard against blank columns in the source file.
headers: list[str] = [cell.value for cell in ws[1] if cell.value is not None] # type: ignore[misc]
for key in new_keys:
if key in headers:
continue
# Find the first key that comes after this one in new_keys that already
# exists in headers — insert before it to keep columns logically grouped.
insert_before: str | None = None
found = False
for k in new_keys:
if k == key:
found = True
continue
if found and k in headers:
insert_before = k
break
if insert_before is not None:
col_idx = headers.index(insert_before) + 1 # 1-based
ws.insert_cols(col_idx)
ws.cell(row=1, column=col_idx, value=key)
headers.insert(col_idx - 1, key)
else:
headers.append(key)
ws.cell(row=1, column=len(headers), value=key)
ws.append([row_data.get(col) for col in headers])
wb.save(file_path)

View file

@ -26,13 +26,17 @@ from backend.ecmk_fetcher.browser import (
)
from backend.ecmk_fetcher.reports import (
REPORT_TYPES,
FileDownloadButtonType,
build_property_id,
map_report_type_to_db_file_type,
)
from backend.ecmk_fetcher.excel_writer import write_row
from backend.ecmk_fetcher.upload import (
upload_excel_to_sharepoint,
upload_file_to_s3_and_update_db,
upload_file_to_sharepoint,
)
from backend.ecmk_fetcher.xml_processor import flatten_sap_property, parse_rdsap
from utils.logger import setup_logger
from utils.sharepoint.domna_sharepoint_client import DomnaSharepointClient
from utils.sharepoint.domna_sites import DomnaSites
@ -59,6 +63,15 @@ def run_job() -> None:
)
sharepoint_base_path: str = "/Projects/Southern Housing/SH-SURV-26-001/Assessments"
sharepoint_excel_path: str = "/Projects/Southern Housing/SH-SURV-26-001/Modelling"
DIMENSIONS_FILENAME: str = "Dimensions.xlsx"
local_dimensions_path: str = os.path.join(BASE_DIR, DIMENSIONS_FILENAME)
sharepoint_client.download_file(
sharepoint_path=f"{sharepoint_excel_path}/{DIMENSIONS_FILENAME}",
local_path=local_dimensions_path,
)
s3_bucket: str = "retrofit-energy-assessments-dev"
@ -141,15 +154,30 @@ def run_job() -> None:
)
try:
upload_file_to_sharepoint(
client=sharepoint_client,
file_path=file_path,
base_path=sharepoint_base_path,
subpath=sharepoint_address,
)
logger.info(
f"Successfully loaded {os.path.basename(file_path)} to sharepoint for {address}"
)
if report_type == FileDownloadButtonType.RAW_XML.value:
with open(file_path, "r", encoding="utf-8") as f:
xml_string = f.read()
details = parse_rdsap(xml_string)
row_data = flatten_sap_property(details)
write_row(local_dimensions_path, row_data)
upload_excel_to_sharepoint(
client=sharepoint_client,
file_path=local_dimensions_path,
sharepoint_path=sharepoint_excel_path,
)
logger.info(
f"Written dimensions row and uploaded Dimensions.xlsx for {address}"
)
else:
upload_file_to_sharepoint(
client=sharepoint_client,
file_path=file_path,
base_path=sharepoint_base_path,
subpath=sharepoint_address,
)
logger.info(
f"Successfully loaded {os.path.basename(file_path)} to sharepoint for {address}"
)
# Upload to s3 and update db
upload_file_to_s3_and_update_db(

View file

@ -14,6 +14,7 @@ class FileDownloadButtonType(Enum):
REPORT_TYPES = [
FileDownloadButtonType.ASSESSOR_HUB_SITENOTE_REPORT.value,
FileDownloadButtonType.SITENOTE_REPORT.value,
FileDownloadButtonType.RAW_XML.value,
]
@ -23,6 +24,8 @@ def map_report_type_to_db_file_type(report_type: int) -> FileTypeEnum:
return FileTypeEnum.ECMK_SITE_NOTE
case FileDownloadButtonType.SITENOTE_REPORT.value:
return FileTypeEnum.ECMK_RD_SAP_SITE_NOTE
case FileDownloadButtonType.RAW_XML.value:
return FileTypeEnum.ECMK_SURVEY_XML
case _:
raise ValueError("Unknown report type")

View file

@ -0,0 +1,123 @@
import os
import pathlib
import pytest
from openpyxl import load_workbook
from openpyxl.worksheet.worksheet import Worksheet
from backend.ecmk_fetcher.excel_writer import write_row
@pytest.fixture
def xlsx_path(tmp_path: pathlib.Path) -> str:
return str(tmp_path / "output.xlsx")
def _active_sheet(file_path: str) -> Worksheet:
ws = load_workbook(file_path).active
assert isinstance(ws, Worksheet)
return ws
def test_write_row_creates_file(xlsx_path: str):
# arrange
row = {
"address": "1 Fake Avenue, AB24 5CD",
"property_type": "House",
"main_dwelling_floor_1_area_m2": 43.61,
}
# act
write_row(xlsx_path, row)
# assert
assert os.path.exists(xlsx_path)
ws = _active_sheet(xlsx_path)
assert [c.value for c in ws[1]] == list(row.keys())
assert [c.value for c in ws[2]] == list(row.values())
def test_write_row_appends_to_existing(xlsx_path: str):
# arrange
row_a = {
"address": "1 Fake Avenue, AB24 5CD",
"property_type": "House",
"main_dwelling_floor_1_area_m2": 43.61,
}
row_b = {
"address": "2 Other Street, XY1 2AB",
"property_type": "Flat",
"main_dwelling_floor_1_area_m2": 30.0,
}
# act
write_row(xlsx_path, row_a)
write_row(xlsx_path, row_b)
# assert
ws = _active_sheet(xlsx_path)
assert ws.max_row == 3 # 1 header + 2 data rows
assert [c.value for c in ws[1]] == list(row_a.keys())
assert [c.value for c in ws[2]] == list(row_a.values())
assert [c.value for c in ws[3]] == list(row_b.values())
def test_write_row_inserts_new_columns_at_logical_positions(xlsx_path: str):
# arrange
# First row: main_dwelling floor 1 + roof
# Second row: also has main_dwelling floor 2 — should be inserted between floor 1 and roof,
# not appended to the end
row_a = {
"address": "1 Fake Avenue, AB24 5CD",
"property_type": "House",
"main_dwelling_floor_1_area_m2": 43.61,
"main_dwelling_floor_1_height_m": 2.46,
"main_dwelling_roof_construction": 4,
}
row_b = {
"address": "2 Other Street, XY1 2AB",
"property_type": "House",
"main_dwelling_floor_1_area_m2": 50.0,
"main_dwelling_floor_1_height_m": 2.5,
"main_dwelling_floor_2_area_m2": 48.0,
"main_dwelling_floor_2_height_m": 2.4,
"main_dwelling_roof_construction": 4,
}
# act
write_row(xlsx_path, row_a)
write_row(xlsx_path, row_b)
# assert
ws = _active_sheet(xlsx_path)
assert [c.value for c in ws[1]] == [
"address",
"property_type",
"main_dwelling_floor_1_area_m2",
"main_dwelling_floor_1_height_m",
"main_dwelling_floor_2_area_m2", # inserted before roof, not at end
"main_dwelling_floor_2_height_m",
"main_dwelling_roof_construction",
]
# row_a had no floor_2 data — those cells should be empty
assert [c.value for c in ws[2]] == [
"1 Fake Avenue, AB24 5CD",
"House",
43.61,
2.46,
None, # main_dwelling_floor_2_area_m2
None, # main_dwelling_floor_2_height_m
4,
]
# row_b should be fully populated
assert [c.value for c in ws[3]] == [
"2 Other Street, XY1 2AB",
"House",
50.0,
2.5,
48.0,
2.4,
4,
]

View file

@ -0,0 +1,329 @@
from backend.ecmk_fetcher.xml_processor import (
SapPropertyDetails,
flatten_sap_property,
parse_rdsap,
)
SAMPLE_XML = """<RdSAP-Report xmlns="https://epbr.digital.communities.gov.uk/xsd/rdsap">
<Report-Header>
<Property>
<Address>
<Address-Line-1>1</Address-Line-1>
<Address-Line-2>Fake Avenue</Address-Line-2>
<Post-Town>Random</Post-Town>
<Postcode>AB24 5CD</Postcode>
</Address>
</Property>
</Report-Header>
<SAP-Data>
<SAP-Property-Details>
<Property-Type>0</Property-Type>
<SAP-Building-Parts>
<SAP-Building-Part>
<Building-Part-Number>1</Building-Part-Number>
<Identifier>Main Dwelling</Identifier>
<Construction-Age-Band>C</Construction-Age-Band>
<Floor-Heat-Loss>7</Floor-Heat-Loss>
<Roof-Construction>4</Roof-Construction>
<Roof-Insulation-Location>2</Roof-Insulation-Location>
<Roof-Insulation-Thickness>100mm</Roof-Insulation-Thickness>
<Wall-Construction>4</Wall-Construction>
<Wall-Insulation-Type>4</Wall-Insulation-Type>
<SAP-Floor-Dimensions>
<SAP-Floor-Dimension>
<Heat-Loss-Perimeter quantity="metres">25.31</Heat-Loss-Perimeter>
<Room-Height quantity="metres">2.46</Room-Height>
<Total-Floor-Area quantity="square metres">43.61</Total-Floor-Area>
<Floor>0</Floor>
<Party-Wall-Length>0</Party-Wall-Length>
</SAP-Floor-Dimension>
<SAP-Floor-Dimension>
<Heat-Loss-Perimeter quantity="metres">26.16</Heat-Loss-Perimeter>
<Room-Height quantity="metres">2.44</Room-Height>
<Total-Floor-Area quantity="square metres">42.33</Total-Floor-Area>
<Floor>1</Floor>
<Party-Wall-Length>0</Party-Wall-Length>
</SAP-Floor-Dimension>
</SAP-Floor-Dimensions>
</SAP-Building-Part>
<SAP-Building-Part>
<Building-Part-Number>2</Building-Part-Number>
<Identifier>Extension</Identifier>
<Construction-Age-Band>C</Construction-Age-Band>
<Roof-Construction>8</Roof-Construction>
<Roof-Insulation-Location>7</Roof-Insulation-Location>
<Sloping-Ceiling-Insulation-Thickness>AB</Sloping-Ceiling-Insulation-Thickness>
<Wall-Construction>3</Wall-Construction>
<Wall-Insulation-Type>4</Wall-Insulation-Type>
<SAP-Floor-Dimensions>
<SAP-Floor-Dimension>
<Heat-Loss-Perimeter quantity="metres">6.85</Heat-Loss-Perimeter>
<Room-Height quantity="metres">2.24</Room-Height>
<Total-Floor-Area quantity="square metres">4.46</Total-Floor-Area>
<Floor>0</Floor>
<Party-Wall-Length>0</Party-Wall-Length>
</SAP-Floor-Dimension>
</SAP-Floor-Dimensions>
</SAP-Building-Part>
</SAP-Building-Parts>
</SAP-Property-Details>
</SAP-Data>
</RdSAP-Report>
"""
NO_ROOF_XML = """<RdSAP-Report xmlns="https://epbr.digital.communities.gov.uk/xsd/rdsap">
<Report-Header>
<Property>
<Address>
<Address-Line-1>5</Address-Line-1>
<Post-Town>Somewhere</Post-Town>
<Postcode>XY1 2AB</Postcode>
</Address>
</Property>
</Report-Header>
<SAP-Data>
<SAP-Property-Details>
<Property-Type>0</Property-Type>
<SAP-Building-Parts>
<SAP-Building-Part>
<Identifier>Main Dwelling</Identifier>
<SAP-Floor-Dimensions>
<SAP-Floor-Dimension>
<Heat-Loss-Perimeter quantity="metres">10.0</Heat-Loss-Perimeter>
<Room-Height quantity="metres">2.5</Room-Height>
<Total-Floor-Area quantity="square metres">50.0</Total-Floor-Area>
<Floor>0</Floor>
<Party-Wall-Length>3.0</Party-Wall-Length>
</SAP-Floor-Dimension>
</SAP-Floor-Dimensions>
</SAP-Building-Part>
</SAP-Building-Parts>
</SAP-Property-Details>
</SAP-Data>
</RdSAP-Report>
"""
def test_parse_rdsap_contract():
# arrange + act
result: SapPropertyDetails = parse_rdsap(SAMPLE_XML)
# assert
assert result == {
"reference": "1AB245CD",
"address": "1, Fake Avenue, Random, AB24 5CD",
"property_type": "House",
"building_parts": [
{
"identifier": "Main Dwelling",
"floors": [
{
"area_m2": 43.61,
"height_m": 2.46,
"heat_loss_perimeter_m": 25.31,
"party_wall_length_m": 0.0,
},
{
"area_m2": 42.33,
"height_m": 2.44,
"heat_loss_perimeter_m": 26.16,
"party_wall_length_m": 0.0,
},
],
"roof": {
"construction": 4,
"insulation_location": 2,
"insulation_thickness_mm": 100.0,
},
},
{
"identifier": "Extension",
"floors": [
{
"area_m2": 4.46,
"height_m": 2.24,
"heat_loss_perimeter_m": 6.85,
"party_wall_length_m": 0.0,
}
],
"roof": {
"construction": 8,
"insulation_location": 7,
},
},
],
}
ND_THICKNESS_XML = """<RdSAP-Report xmlns="https://epbr.digital.communities.gov.uk/xsd/rdsap">
<Report-Header>
<Property>
<Address>
<Address-Line-1>1</Address-Line-1>
<Post-Town>Somewhere</Post-Town>
<Postcode>AB1 2CD</Postcode>
</Address>
</Property>
</Report-Header>
<SAP-Data>
<SAP-Property-Details>
<Property-Type>0</Property-Type>
<SAP-Building-Parts>
<SAP-Building-Part>
<Identifier>Main Dwelling</Identifier>
<Roof-Construction>4</Roof-Construction>
<Roof-Insulation-Location>2</Roof-Insulation-Location>
<Roof-Insulation-Thickness>ND</Roof-Insulation-Thickness>
<SAP-Floor-Dimensions>
<SAP-Floor-Dimension>
<Heat-Loss-Perimeter quantity="metres">10.0</Heat-Loss-Perimeter>
<Room-Height quantity="metres">2.5</Room-Height>
<Total-Floor-Area quantity="square metres">50.0</Total-Floor-Area>
<Floor>0</Floor>
<Party-Wall-Length>0</Party-Wall-Length>
</SAP-Floor-Dimension>
</SAP-Floor-Dimensions>
</SAP-Building-Part>
</SAP-Building-Parts>
</SAP-Property-Details>
</SAP-Data>
</RdSAP-Report>
"""
ND_INSULATION_LOCATION_XML = """<RdSAP-Report xmlns="https://epbr.digital.communities.gov.uk/xsd/rdsap">
<Report-Header>
<Property>
<Address>
<Address-Line-1>1</Address-Line-1>
<Post-Town>Somewhere</Post-Town>
<Postcode>AB1 2CD</Postcode>
</Address>
</Property>
</Report-Header>
<SAP-Data>
<SAP-Property-Details>
<Property-Type>0</Property-Type>
<SAP-Building-Parts>
<SAP-Building-Part>
<Identifier>Main Dwelling</Identifier>
<Roof-Construction>4</Roof-Construction>
<Roof-Insulation-Location>ND</Roof-Insulation-Location>
<Roof-Insulation-Thickness>250</Roof-Insulation-Thickness>
<SAP-Floor-Dimensions>
<SAP-Floor-Dimension>
<Heat-Loss-Perimeter quantity="metres">10.0</Heat-Loss-Perimeter>
<Room-Height quantity="metres">2.5</Room-Height>
<Total-Floor-Area quantity="square metres">50.0</Total-Floor-Area>
<Floor>0</Floor>
<Party-Wall-Length>0</Party-Wall-Length>
</SAP-Floor-Dimension>
</SAP-Floor-Dimensions>
</SAP-Building-Part>
</SAP-Building-Parts>
</SAP-Property-Details>
</SAP-Data>
</RdSAP-Report>
"""
def test_parse_rdsap_nd_thickness():
# 'ND' (not determined) is a valid value in the wild for Roof-Insulation-Thickness
# — it should be retained as-is rather than raising
# arrange + act
result: SapPropertyDetails = parse_rdsap(ND_THICKNESS_XML)
# assert
assert result["building_parts"][0]["roof"] == {
"construction": 4,
"insulation_location": 2,
"insulation_thickness_mm": "ND",
}
def test_parse_rdsap_nd_location():
# 'ND' (not determined) is a valid value in the wild for Roof-Insulation-Location
# — it should be retained as-is rather than raising
# arrange + act
result: SapPropertyDetails = parse_rdsap(ND_INSULATION_LOCATION_XML)
# assert
assert result["building_parts"][0]["roof"] == {
"construction": 4,
"insulation_location": "ND",
"insulation_thickness_mm": 250,
}
def test_flatten_full():
# Two building parts; Main Dwelling has two floors + full roof,
# Extension has one floor + partial roof (no thickness)
# arrange
details: SapPropertyDetails = parse_rdsap(SAMPLE_XML)
# act
result = flatten_sap_property(details)
# assert
assert result == {
"reference": "1AB245CD",
"address": "1, Fake Avenue, Random, AB24 5CD",
"property_type": "House",
"main_dwelling_floor_1_area_m2": 43.61,
"main_dwelling_floor_1_height_m": 2.46,
"main_dwelling_floor_1_heat_loss_perimeter_m": 25.31,
"main_dwelling_floor_1_party_wall_length_m": 0.0,
"main_dwelling_floor_2_area_m2": 42.33,
"main_dwelling_floor_2_height_m": 2.44,
"main_dwelling_floor_2_heat_loss_perimeter_m": 26.16,
"main_dwelling_floor_2_party_wall_length_m": 0.0,
"main_dwelling_roof_construction": 4,
"main_dwelling_roof_insulation_location": 2,
"main_dwelling_roof_insulation_thickness_mm": 100.0,
"extension_floor_1_area_m2": 4.46,
"extension_floor_1_height_m": 2.24,
"extension_floor_1_heat_loss_perimeter_m": 6.85,
"extension_floor_1_party_wall_length_m": 0.0,
"extension_roof_construction": 8,
"extension_roof_insulation_location": 7,
}
def test_flatten_no_roof():
# Single building part with no roof — roof keys must be absent entirely
# arrange
details: SapPropertyDetails = parse_rdsap(NO_ROOF_XML)
# act
result = flatten_sap_property(details)
# assert
assert result == {
"reference": "5XY12AB",
"address": "5, Somewhere, XY1 2AB",
"property_type": "House",
"main_dwelling_floor_1_area_m2": 50.0,
"main_dwelling_floor_1_height_m": 2.5,
"main_dwelling_floor_1_heat_loss_perimeter_m": 10.0,
"main_dwelling_floor_1_party_wall_length_m": 3.0,
}

View file

@ -28,6 +28,18 @@ def upload_file_to_sharepoint(
)
def upload_excel_to_sharepoint(
client: DomnaSharepointClient,
file_path: str,
sharepoint_path: str,
) -> None:
client.upload_file(
file_path=file_path,
sharepoint_path=sharepoint_path,
file_name=os.path.basename(file_path),
)
def upload_file_to_s3_and_update_db(
bucket: str, file_path: str, hubspot_listing_id: str, file_type: FileTypeEnum
) -> None:

View file

@ -0,0 +1,226 @@
import xml.etree.ElementTree as ET
from typing import Any, List, Optional, TypedDict
from backend.ecmk_fetcher.reports import build_property_id
from datatypes.epc.domain.field_mappings import PROPERTY_TYPE_LOOKUP
# This file should ultimately live somewhere different, probably
class Floor(TypedDict):
area_m2: float
height_m: float
heat_loss_perimeter_m: float
party_wall_length_m: float
class Roof(TypedDict, total=False):
construction: int # TODO: map to str
insulation_location: int | str # TODO: map to str
insulation_thickness_mm: float | str
class BuildingPart(TypedDict):
identifier: str # e.g. "Main Dwelling", "Extension"
floors: List[Floor]
roof: Optional[Roof]
class SapPropertyDetails(TypedDict):
reference: str
address: str
property_type: str
building_parts: List[BuildingPart]
def _get_namespace(tag: str) -> str:
return tag.split("}")[0].strip("{")
def _require_text(value: Optional[str], field: str) -> str:
if value is None:
raise ValueError(f"Missing required field: {field}")
return value
def _parse_float(value: Optional[str], field: str) -> float:
if value is None:
raise ValueError(f"Missing float field: {field}")
return float(value)
def _parse_int(value: Optional[str], field: str) -> int:
if value is None:
raise ValueError(f"Missing int field: {field}")
return int(value)
def _parse_thickness_mm(value: Optional[str]) -> Optional[float | str]:
if value is None:
return None
stripped = value.replace("mm", "").strip()
try:
return float(stripped)
except ValueError:
return stripped
def parse_rdsap(xml_string: str) -> SapPropertyDetails:
root = ET.fromstring(xml_string)
ns_uri: str = _get_namespace(root.tag)
ns: dict[str, str] = {"r": ns_uri}
# --- Address ---
addr_elem = root.find(".//r:Address", ns)
if addr_elem is None:
raise ValueError("Address element not found")
address_line_1: str = addr_elem.findtext(
"r:Address-Line-1", default="", namespaces=ns
)
postcode: str = addr_elem.findtext("r:Postcode", default="", namespaces=ns)
address_parts: List[str] = [
address_line_1,
addr_elem.findtext("r:Address-Line-2", default="", namespaces=ns),
addr_elem.findtext("r:Post-Town", default="", namespaces=ns),
postcode,
]
address: str = ", ".join(part for part in address_parts if part)
reference: str = build_property_id(address_line_1, postcode)
# --- Property Type ---
prop_type_text = root.findtext(".//r:Property-Type", namespaces=ns)
prop_type_code: int = _parse_int(prop_type_text, "Property-Type")
property_type: str = PROPERTY_TYPE_LOOKUP[prop_type_code]
# --- Building Parts ---
building_parts: List[BuildingPart] = []
for bp in root.findall(".//r:SAP-Building-Part", ns):
identifier_text = bp.findtext("r:Identifier", namespaces=ns)
identifier: str = _require_text(identifier_text, "Identifier")
# Floors
floors: List[Floor] = []
for f in bp.findall(".//r:SAP-Floor-Dimension", ns):
area = _parse_float(
f.findtext("r:Total-Floor-Area", namespaces=ns),
"Total-Floor-Area",
)
height = _parse_float(
f.findtext("r:Room-Height", namespaces=ns),
"Room-Height",
)
heat_loss = _parse_float(
f.findtext("r:Heat-Loss-Perimeter", namespaces=ns),
"Heat-Loss-Perimeter",
)
party_wall = _parse_float(
f.findtext("r:Party-Wall-Length", namespaces=ns),
"Party-Wall-Length",
)
floor: Floor = {
"area_m2": area,
"height_m": height,
"heat_loss_perimeter_m": heat_loss,
"party_wall_length_m": party_wall,
}
floors.append(floor)
# Roof (optional)
roof: Optional[Roof] = None
roof_construction_text = bp.findtext("r:Roof-Construction", namespaces=ns)
roof_ins_loc_text = bp.findtext("r:Roof-Insulation-Location", namespaces=ns)
roof_thickness_text = bp.findtext("r:Roof-Insulation-Thickness", namespaces=ns)
if (
roof_construction_text is not None
or roof_ins_loc_text is not None
or roof_thickness_text is not None
):
roof_dict: Roof = {}
if roof_construction_text is not None:
roof_dict["construction"] = _parse_int(
roof_construction_text, "Roof-Construction"
)
if roof_ins_loc_text is not None:
try:
roof_dict["insulation_location"] = _parse_int(
roof_ins_loc_text, "Roof-Insulation-Location"
)
except ValueError:
roof_dict["insulation_location"] = roof_ins_loc_text
thickness = _parse_thickness_mm(roof_thickness_text)
if thickness is not None:
roof_dict["insulation_thickness_mm"] = thickness
roof = roof_dict
building_part: BuildingPart = {
"identifier": identifier,
"floors": floors,
"roof": roof,
}
building_parts.append(building_part)
result: SapPropertyDetails = {
"reference": reference,
"address": address,
"property_type": property_type,
"building_parts": building_parts,
}
return result
def _normalise_identifier(identifier: str) -> str:
return identifier.lower().replace(" ", "_").replace("-", "_")
def flatten_sap_property(details: SapPropertyDetails) -> dict[str, Any]:
row: dict[str, Any] = {}
row["reference"] = details["reference"]
row["address"] = details["address"]
row["property_type"] = details["property_type"]
for bp in details["building_parts"]:
prefix = _normalise_identifier(bp["identifier"])
for i, floor in enumerate(bp["floors"], start=1):
floor_prefix = f"{prefix}_floor_{i}"
row[f"{floor_prefix}_area_m2"] = floor["area_m2"]
row[f"{floor_prefix}_height_m"] = floor["height_m"]
row[f"{floor_prefix}_heat_loss_perimeter_m"] = floor[
"heat_loss_perimeter_m"
]
row[f"{floor_prefix}_party_wall_length_m"] = floor["party_wall_length_m"]
roof = bp.get("roof")
if roof:
if "construction" in roof:
row[f"{prefix}_roof_construction"] = roof["construction"]
if "insulation_location" in roof:
row[f"{prefix}_roof_insulation_location"] = roof["insulation_location"]
if "insulation_thickness_mm" in roof:
row[f"{prefix}_roof_insulation_thickness_mm"] = roof[
"insulation_thickness_mm"
]
return row

View file

@ -5,11 +5,22 @@ import time
from backend.export.property_scenarios.main import process_export
from backend.export.property_scenarios.input_schema import ExportRequest
from backend.app.db.models.portfolio import PropertyModel, Epc, Portfolio, PortfolioStatus, PortfolioGoal, \
PropertyCreationStatus, PropertyDetailsEpcModel
from backend.app.db.models.recommendations import PlanModel, Recommendation, PlanRecommendations, \
RecommendationMaterials
from backend.app.db.models.portfolio import (
PropertyModel,
Portfolio,
PortfolioStatus,
PortfolioGoal,
PropertyCreationStatus,
PropertyDetailsEpcModel,
)
from backend.app.db.models.recommendations import (
PlanModel,
Recommendation,
PlanRecommendations,
RecommendationMaterials,
)
from backend.app.db.models.materials import Material
from datatypes.epc.domain.epc import Epc
from utils.logger import setup_logger
FIXTURE_PATH = Path("backend/export/tests/fixtures")
@ -78,11 +89,13 @@ def test_default_export_integration(db_session):
else None
)
prop = PropertyModel(**{
col: row_dict[col]
for col in PropertyModel.__table__.columns.keys()
if col in row_dict
})
prop = PropertyModel(
**{
col: row_dict[col]
for col in PropertyModel.__table__.columns.keys()
if col in row_dict
}
)
prop.creation_status = PropertyCreationStatus[
row_dict["creation_status"].split(".")[-1]
@ -90,9 +103,7 @@ def test_default_export_integration(db_session):
prop.status = PortfolioStatus[row_dict["status"].split(".")[-1]]
if row_dict.get("current_epc_rating"):
prop.current_epc_rating = Epc[
row_dict["current_epc_rating"].split(".")[-1]
]
prop.current_epc_rating = Epc[row_dict["current_epc_rating"].split(".")[-1]]
properties.append(prop)
@ -112,7 +123,8 @@ def test_default_export_integration(db_session):
epc_data = {
col.name: row_dict[col.name]
for col in PropertyDetailsEpcModel.__table__.columns.values()
if col.name in row_dict and col.name not in ["id", "property_id", "portfolio_id"]
if col.name in row_dict
and col.name not in ["id", "property_id", "portfolio_id"]
}
epc = PropertyDetailsEpcModel(
@ -142,11 +154,13 @@ def test_default_export_integration(db_session):
row_dict["scenario_id"] = None
plan = PlanModel(**{
col: row_dict[col]
for col in PlanModel.__table__.columns.keys()
if col in row_dict
})
plan = PlanModel(
**{
col: row_dict[col]
for col in PlanModel.__table__.columns.keys()
if col in row_dict
}
)
plans.append(plan)
@ -158,11 +172,13 @@ def test_default_export_integration(db_session):
# ----------------------------------------
recs = [
Recommendation(**{
col: row[col]
for col in Recommendation.__table__.columns.keys()
if col in row
})
Recommendation(
**{
col: row[col]
for col in Recommendation.__table__.columns.keys()
if col in row
}
)
for _, row in recommendations_df.iterrows()
]
@ -203,28 +219,19 @@ def test_default_export_integration(db_session):
# ----------------------------------------
logger.info(
"Recommendation count in DB: %s",
db_session.query(Recommendation).count()
"Recommendation count in DB: %s", db_session.query(Recommendation).count()
)
logger.info(
"Property count in DB: %s",
db_session.query(PropertyModel).count()
)
logger.info("Property count in DB: %s", db_session.query(PropertyModel).count())
logger.info(
"Property EPC in DB: %s",
db_session.query(PropertyDetailsEpcModel).count()
"Property EPC in DB: %s", db_session.query(PropertyDetailsEpcModel).count()
)
logger.info(
"Plan count in DB: %s",
db_session.query(PlanModel).count()
)
logger.info("Plan count in DB: %s", db_session.query(PlanModel).count())
logger.info(
"PlanRecommendatons count in DB: %s",
db_session.query(PlanModel).count()
"PlanRecommendatons count in DB: %s", db_session.query(PlanModel).count()
)
logger.info("Starting process_export")
@ -232,17 +239,23 @@ def test_default_export_integration(db_session):
result = process_export(payload, session=db_session)
logger.info("process_export finished in %.2f seconds", time.perf_counter() - process_t0)
logger.info(
"process_export finished in %.2f seconds", time.perf_counter() - process_t0
)
# ----------------------------------------
# 8) Assertions
# ----------------------------------------
assert "default_plans" in result, "Expected 'default_plans' in export result, got {}".format(result.keys())
assert (
"default_plans" in result
), "Expected 'default_plans' in export result, got {}".format(result.keys())
df = result["default_plans"]
assert df.shape[0] == 10, "Expected 10 properties in the export, got {}".format(df.shape[0])
assert df.shape[0] == 10, "Expected 10 properties in the export, got {}".format(
df.shape[0]
)
failed = df[df["predicted_post_works_sap"] < 69]
failed_property_types = failed["property_type"].value_counts().to_dict()
@ -251,19 +264,28 @@ def test_default_export_integration(db_session):
assert failed.shape[0]
assert df["total_retrofit_cost"].sum() == 41706.585999999996, (
"Expected total retrofit cost to be 10000, got {}".format(df["total_retrofit_cost"].sum())
assert (
df["total_retrofit_cost"].sum() == 41706.585999999996
), "Expected total retrofit cost to be 10000, got {}".format(
df["total_retrofit_cost"].sum()
)
assert df["predicted_post_works_sap"].sum() == 698.1, (
"Expected total predicted post works SAP to be 698.1, got {}".format(df["predicted_post_works_sap"].sum())
assert (
df["predicted_post_works_sap"].sum() == 698.1
), "Expected total predicted post works SAP to be 698.1, got {}".format(
df["predicted_post_works_sap"].sum()
)
assert df["sap_points"].sum() == 100.10000000000001, (
"Expected total SAP points increase to be 100.10000000000001, got {}".format(df["sap_points"].sum())
assert (
df["sap_points"].sum() == 100.10000000000001
), "Expected total SAP points increase to be 100.10000000000001, got {}".format(
df["sap_points"].sum()
)
assert df.shape == (10, 100), "Expected dataframe shape to be (10, 100), got {}".format(df.shape)
assert df.shape == (
10,
100,
), "Expected dataframe shape to be (10, 100), got {}".format(df.shape)
def test_solar_with_battery_example(db_session):
@ -271,116 +293,251 @@ def test_solar_with_battery_example(db_session):
test_property_id = 1
portfolio_df = pd.DataFrame(
[{'id': test_portfolio_id, 'name': 'Example', 'budget': None,
'status': 'PortfolioStatus.SCOPING', 'goal': 'PortfolioGoal.NONE', 'cost': None, 'number_of_properties': None,
'co2_equivalent_savings': None, 'energy_savings': None, 'energy_cost_savings': None,
'property_valuation_increase': None, 'rental_yield_increase': None, 'total_work_hours': None,
'labour_days': None, 'created_at': '2026-02-12 21:23:37.862000+00:00',
'updated_at': '2026-02-12 21:23:37.862000+00:00', 'epc_breakdown_pre_retrofit': None,
'epc_breakdown_post_retrofit': None, 'n_units_to_retrofit': None, 'co2_per_unit_pre_retrofit': None,
'co2_per_unit_post_retrofit': None, 'energy_bill_per_unit_pre_retrofit': None,
'energy_bill_per_unit_post_retrofit': None, 'energy_consumption_per_unit_pre_retrofit': None,
'energy_consumption_per_unit_post_retrofit': None, 'valuation_improvement_per_unit': None,
'cost_per_unit': None, 'cost_per_co2_saved': None, 'cost_per_sap_point': None,
'valuation_return_on_investment': None}]
[
{
"id": test_portfolio_id,
"name": "Example",
"budget": None,
"status": "PortfolioStatus.SCOPING",
"goal": "PortfolioGoal.NONE",
"cost": None,
"number_of_properties": None,
"co2_equivalent_savings": None,
"energy_savings": None,
"energy_cost_savings": None,
"property_valuation_increase": None,
"rental_yield_increase": None,
"total_work_hours": None,
"labour_days": None,
"created_at": "2026-02-12 21:23:37.862000+00:00",
"updated_at": "2026-02-12 21:23:37.862000+00:00",
"epc_breakdown_pre_retrofit": None,
"epc_breakdown_post_retrofit": None,
"n_units_to_retrofit": None,
"co2_per_unit_pre_retrofit": None,
"co2_per_unit_post_retrofit": None,
"energy_bill_per_unit_pre_retrofit": None,
"energy_bill_per_unit_post_retrofit": None,
"energy_consumption_per_unit_pre_retrofit": None,
"energy_consumption_per_unit_post_retrofit": None,
"valuation_improvement_per_unit": None,
"cost_per_unit": None,
"cost_per_co2_saved": None,
"cost_per_sap_point": None,
"valuation_return_on_investment": None,
}
]
)
properties_df = pd.DataFrame(
[{'id': test_property_id, 'portfolio_id': test_portfolio_id, 'creation_status': 'PropertyCreationStatus.READY',
'uprn': 100090438731, 'landlord_property_id': 'BARR052', 'building_reference_number': 3460742868.0,
'status': 'PortfolioStatus.ASSESSMENT', 'address': '52, Barrack Street', 'postcode': 'CO1 2LR',
'has_pre_condition_report': True, 'has_recommendations': True, 'created_at': '2026-02-12 21:59:02.744427',
'updated_at': '2026-02-19 16:18:57.941443', 'property_type': 'House', 'built_form': 'End-Terrace',
'local_authority': 'Colchester', 'constituency': 'Colchester', 'number_of_rooms': 4.0, 'year_built': 1900.0,
'tenure': 'rental (private)', 'current_epc_rating': 'Epc.E', 'current_sap_points': 53.0,
'current_valuation': 0.0, 'installed_measures_sap_point_adjustment': 0.0,
'is_sap_points_adjusted_for_installed_measures': False, 'original_sap_points': 53.0}]
[
{
"id": test_property_id,
"portfolio_id": test_portfolio_id,
"creation_status": "PropertyCreationStatus.READY",
"uprn": 100090438731,
"landlord_property_id": "BARR052",
"building_reference_number": 3460742868.0,
"status": "PortfolioStatus.ASSESSMENT",
"address": "52, Barrack Street",
"postcode": "CO1 2LR",
"has_pre_condition_report": True,
"has_recommendations": True,
"created_at": "2026-02-12 21:59:02.744427",
"updated_at": "2026-02-19 16:18:57.941443",
"property_type": "House",
"built_form": "End-Terrace",
"local_authority": "Colchester",
"constituency": "Colchester",
"number_of_rooms": 4.0,
"year_built": 1900.0,
"tenure": "rental (private)",
"current_epc_rating": "Epc.E",
"current_sap_points": 53.0,
"current_valuation": 0.0,
"installed_measures_sap_point_adjustment": 0.0,
"is_sap_points_adjusted_for_installed_measures": False,
"original_sap_points": 53.0,
}
]
)
property_details_epc_df = pd.DataFrame(
[
{'id': 1534934, 'property_id': test_property_id, 'portfolio_id': test_portfolio_id,
'full_address': '48, Medcalf Road', 'lodgement_date': '2018-09-05', 'is_expired': False,
'total_floor_area': 68.0, 'walls': 'Solid brick, as built, no insulation', 'walls_rating': 1,
'roof': 'Pitched, no insulation', 'roof_rating': 1.0, 'floor': 'Solid, no insulation',
'floor_rating': None,
'windows': 'Fully double glazed', 'windows_rating': 4, 'heating': 'Boiler and radiators, mains gas',
'heating_rating': 4, 'heating_controls': 'Programmer, room thermostat and trvs',
'heating_controls_rating': 4,
'hot_water': 'From main system', 'hot_water_rating': 4,
'lighting': 'Low energy lighting in all fixed outlets', 'lighting_rating': 5,
'mainfuel': 'Mains gas not community', 'ventilation': 'natural', 'solar_pv': 0.0, 'solar_hot_water': False,
'wind_turbine': 0.0, 'floor_height': 2.55, 'number_heated_rooms': None, 'heat_loss_corridor': False,
'unheated_corridor_length': None, 'number_of_open_fireplaces': 0, 'number_of_extensions': 0,
'number_of_storeys': None, 'mains_gas': True, 'energy_tariff': 'Single',
'primary_energy_consumption': 278.0,
'co2_emissions': 3.81, 'current_energy_demand': 14643.366,
'current_energy_demand_heating_hotwater': 12185.6,
'estimated': False, 'sap_05_overwritten': False, 'sap_05_score': None, 'sap_05_epc_rating': None,
'heating_cost_current': 711.0628, 'hot_water_cost_current': 139.06198, 'lighting_cost_current': 70.770935,
'appliances_cost_current': 609.7844, 'gas_standing_charge': 128.0785,
'electricity_standing_charge': 199.8375,
'original_co2_emissions': 3.81, 'original_primary_energy_consumption': 278.0,
'original_current_energy_demand': 14643.366, 'original_current_energy_demand_heating_hotwater': 12185.6,
'installed_measures_co2_adjustment': 0.0, 'installed_measures_energy_demand_adjustment': 0.0,
'installed_measures_total_energy_bill_adjustment': 0.0, 'installed_measures_heat_demand_adjustment': 0.0,
'is_epc_adjusted_for_installed_measures': False}
{
"id": 1534934,
"property_id": test_property_id,
"portfolio_id": test_portfolio_id,
"full_address": "48, Medcalf Road",
"lodgement_date": "2018-09-05",
"is_expired": False,
"total_floor_area": 68.0,
"walls": "Solid brick, as built, no insulation",
"walls_rating": 1,
"roof": "Pitched, no insulation",
"roof_rating": 1.0,
"floor": "Solid, no insulation",
"floor_rating": None,
"windows": "Fully double glazed",
"windows_rating": 4,
"heating": "Boiler and radiators, mains gas",
"heating_rating": 4,
"heating_controls": "Programmer, room thermostat and trvs",
"heating_controls_rating": 4,
"hot_water": "From main system",
"hot_water_rating": 4,
"lighting": "Low energy lighting in all fixed outlets",
"lighting_rating": 5,
"mainfuel": "Mains gas not community",
"ventilation": "natural",
"solar_pv": 0.0,
"solar_hot_water": False,
"wind_turbine": 0.0,
"floor_height": 2.55,
"number_heated_rooms": None,
"heat_loss_corridor": False,
"unheated_corridor_length": None,
"number_of_open_fireplaces": 0,
"number_of_extensions": 0,
"number_of_storeys": None,
"mains_gas": True,
"energy_tariff": "Single",
"primary_energy_consumption": 278.0,
"co2_emissions": 3.81,
"current_energy_demand": 14643.366,
"current_energy_demand_heating_hotwater": 12185.6,
"estimated": False,
"sap_05_overwritten": False,
"sap_05_score": None,
"sap_05_epc_rating": None,
"heating_cost_current": 711.0628,
"hot_water_cost_current": 139.06198,
"lighting_cost_current": 70.770935,
"appliances_cost_current": 609.7844,
"gas_standing_charge": 128.0785,
"electricity_standing_charge": 199.8375,
"original_co2_emissions": 3.81,
"original_primary_energy_consumption": 278.0,
"original_current_energy_demand": 14643.366,
"original_current_energy_demand_heating_hotwater": 12185.6,
"installed_measures_co2_adjustment": 0.0,
"installed_measures_energy_demand_adjustment": 0.0,
"installed_measures_total_energy_bill_adjustment": 0.0,
"installed_measures_heat_demand_adjustment": 0.0,
"is_epc_adjusted_for_installed_measures": False,
}
]
)
plans_df = pd.DataFrame(
[
{'id': 0, 'name': None, 'portfolio_id': test_portfolio_id, 'property_id': test_property_id,
'scenario_id': 1060, 'created_at': '2026-02-19 16:14:45.560816', 'is_default': True,
'valuation_increase_lower_bound': 0.0302,
'valuation_increase_upper_bound': 0.07, 'valuation_increase_average': 0.048226666, 'plan_type': None,
'post_sap_points': 71.5, 'post_epc_rating': 'Epc.C', 'post_co2_emissions': 4.1813498,
'co2_savings': 0.71865046, 'post_energy_bill': 1447.5204, 'energy_bill_savings': 691.6662,
'post_energy_consumption': 15303.688, 'energy_consumption_savings': 3276.7622,
'valuation_post_retrofit': None, 'valuation_increase': None, 'cost_of_works': 6984.568,
'contingency_cost': 1003.9568}
{
"id": 0,
"name": None,
"portfolio_id": test_portfolio_id,
"property_id": test_property_id,
"scenario_id": 1060,
"created_at": "2026-02-19 16:14:45.560816",
"is_default": True,
"valuation_increase_lower_bound": 0.0302,
"valuation_increase_upper_bound": 0.07,
"valuation_increase_average": 0.048226666,
"plan_type": None,
"post_sap_points": 71.5,
"post_epc_rating": "Epc.C",
"post_co2_emissions": 4.1813498,
"co2_savings": 0.71865046,
"post_energy_bill": 1447.5204,
"energy_bill_savings": 691.6662,
"post_energy_consumption": 15303.688,
"energy_consumption_savings": 3276.7622,
"valuation_post_retrofit": None,
"valuation_increase": None,
"cost_of_works": 6984.568,
"contingency_cost": 1003.9568,
}
]
)
plan_recs_df = pd.DataFrame(
[{'id': 0, 'plan_id': 0, 'recommendation_id': 0}]
)
plan_recs_df = pd.DataFrame([{"id": 0, "plan_id": 0, "recommendation_id": 0}])
recommendations_df = pd.DataFrame(
[{'id': 0, 'property_id': test_property_id, 'created_at': '2026-02-19 16:14:45.560816',
'type': 'solar_pv', 'measure_type': 'solar_pv',
'description': 'Fit solar',
'estimated_cost': 10000, 'default': True, 'starting_u_value': None, 'new_u_value': None, 'sap_points': 1.5,
'heat_demand': 14.9, 'kwh_savings': 1041.2, 'co2_equivalent_savings': 0.2, 'energy_savings': 14.9,
'energy_cost_savings': 72.639015, 'property_valuation_increase': None, 'rental_yield_increase': None,
'total_work_hours': 4.16, 'labour_days': 1.0, 'already_installed': False, 'plan_name': 'whatever'}
]
[
{
"id": 0,
"property_id": test_property_id,
"created_at": "2026-02-19 16:14:45.560816",
"type": "solar_pv",
"measure_type": "solar_pv",
"description": "Fit solar",
"estimated_cost": 10000,
"default": True,
"starting_u_value": None,
"new_u_value": None,
"sap_points": 1.5,
"heat_demand": 14.9,
"kwh_savings": 1041.2,
"co2_equivalent_savings": 0.2,
"energy_savings": 14.9,
"energy_cost_savings": 72.639015,
"property_valuation_increase": None,
"rental_yield_increase": None,
"total_work_hours": 4.16,
"labour_days": 1.0,
"already_installed": False,
"plan_name": "whatever",
}
]
)
recommendations_materials_df = pd.DataFrame(
[
{
"id": 0, "recommendation_id": 0, "material_id": 0, "depth": None, "quantity": 1.0,
"id": 0,
"recommendation_id": 0,
"material_id": 0,
"depth": None,
"quantity": 1.0,
"quantity_unit": "part",
"estimated_cost": 10000, "created_at": '2026-02-19 16:14:45.560816',
"updated_at": '2026-02-19 16:14:45.560816',
"estimated_cost": 10000,
"created_at": "2026-02-19 16:14:45.560816",
"updated_at": "2026-02-19 16:14:45.560816",
}
]
)
materials_df = pd.DataFrame(
[
{'id': 0, 'type': 'solar_pv', 'description': 'Some solar product',
'depth': 75.0,
'depth_unit': 'mm', 'cost': None, 'cost_unit': 'gbp_per_m2', 'r_value_per_mm': 0.030303031,
'r_value_unit': 'square_meter_kelvin_per_watt', 'thermal_conductivity': 0.033,
'thermal_conductivity_unit': 'watt_per_meter_kelvin', 'link': 'Test',
'created_at': "'2026-02-19 16:14:45.560816", 'is_active': True,
'prime_material_cost': None,
'material_cost': 0.0, 'labour_cost': 0.0, 'labour_hours_per_unit': 0.0, 'plant_cost': 0.0,
'total_cost': 10000,
'notes': None, 'is_installer_quote': True, 'innovation_rate': 0.25, 'size': None, 'size_unit': None,
'includes_scaffolding': True, 'includes_battery': True, 'battery_size': 5.8}
{
"id": 0,
"type": "solar_pv",
"description": "Some solar product",
"depth": 75.0,
"depth_unit": "mm",
"cost": None,
"cost_unit": "gbp_per_m2",
"r_value_per_mm": 0.030303031,
"r_value_unit": "square_meter_kelvin_per_watt",
"thermal_conductivity": 0.033,
"thermal_conductivity_unit": "watt_per_meter_kelvin",
"link": "Test",
"created_at": "'2026-02-19 16:14:45.560816",
"is_active": True,
"prime_material_cost": None,
"material_cost": 0.0,
"labour_cost": 0.0,
"labour_hours_per_unit": 0.0,
"plant_cost": 0.0,
"total_cost": 10000,
"notes": None,
"is_installer_quote": True,
"innovation_rate": 0.25,
"size": None,
"size_unit": None,
"includes_scaffolding": True,
"includes_battery": True,
"battery_size": 5.8,
}
]
)
@ -463,7 +620,7 @@ def test_solar_with_battery_example(db_session):
already_installed=row.already_installed,
sap_points=row.sap_points,
type=row.type,
description=row.description
description=row.description,
)
db_session.add(rec)
db_session.flush()
@ -515,13 +672,15 @@ def test_solar_with_battery_example(db_session):
db_session.commit()
payload = ExportRequest.model_validate({
"task_id": "test",
"subtask_id": "test",
"portfolio_id": test_portfolio_id,
"scenario_ids": [],
"default_plans_only": True,
})
payload = ExportRequest.model_validate(
{
"task_id": "test",
"subtask_id": "test",
"portfolio_id": test_portfolio_id,
"scenario_ids": [],
"default_plans_only": True,
}
)
result = process_export(payload, session=db_session)
@ -534,7 +693,9 @@ def test_solar_with_battery_example(db_session):
# solar_pv should NOT exist
assert "solar_pv" not in df.columns
assert df.shape[0] == 1, "Expected 1 property in the export, got {}".format(df.shape[0])
assert df.shape[0] == 1, "Expected 1 property in the export, got {}".format(
df.shape[0]
)
# Cost should land in correct column
assert df["solar_pv_with_battery"].iloc[0] == 10000

View file

View file

@ -0,0 +1,11 @@
from enum import Enum
class Epc(Enum):
A = "A"
B = "B"
C = "C"
D = "D"
E = "E"
F = "F"
G = "G"

View file

@ -0,0 +1,348 @@
from dataclasses import dataclass
from datetime import date
from typing import List, Optional, Union
from datatypes.epc.domain.epc import Epc
@dataclass
class EnergyElement:
description: str
energy_efficiency_rating: int
environmental_efficiency_rating: int
@dataclass
class InstantaneousWwhrs:
wwhrs_index_number1: Optional[int] = None
wwhrs_index_number2: Optional[int] = None
@dataclass
class MainHeatingDetail:
has_fghrs: bool
main_fuel_type: Union[int, str] # int from API, str from site notes
heat_emitter_type: Union[int, str] # int from API, str from site notes
emitter_temperature: Union[int, str]
main_heating_control: Union[int, str] # int from API, str from site notes
fan_flue_present: Optional[bool] = None
boiler_flue_type: Optional[int] = None # TODO: make enum?
boiler_ignition_type: Optional[int] = None # TODO: make enum?
central_heating_pump_age: Optional[int] = None
main_heating_index_number: Optional[int] = None
sap_main_heating_code: Optional[int] = None # TODO: make enum?
main_heating_number: Optional[int] = None
main_heating_category: Optional[int] = None
main_heating_fraction: Optional[int] = None
main_heating_data_source: Optional[int] = None
@dataclass
class ShowerOutlet:
shower_wwhrs: int
shower_outlet_type: int
@dataclass
class ShowerOutlets:
# TODO: consolidate ShowerOutlet and ShowerOutlets
shower_outlet: ShowerOutlet
@dataclass
class SapHeating:
instantaneous_wwhrs: InstantaneousWwhrs
main_heating_details: List[MainHeatingDetail]
has_fixed_air_conditioning: bool
cylinder_size: Optional[int] = (
None # int code from API; not directly available from site notes
)
water_heating_code: Optional[int] = None # TODO: make enum?
water_heating_fuel: Optional[int] = None # TODO: make enum?
immersion_heating_type: Optional[Union[int, str]] = None # TODO: make enum?
shower_outlets: Optional[ShowerOutlets] = None
cylinder_insulation_type: Optional[int] = None
cylinder_thermostat: Optional[str] = None
secondary_fuel_type: Optional[int] = None
secondary_heating_type: Optional[int] = None
cylinder_insulation_thickness: Optional[int] = None
@dataclass
class WindowTransmissionDetails:
u_value: float
data_source: int
solar_transmittance: float
@dataclass
class SapWindow:
pvc_frame: str
glazing_gap: Union[int, str]
orientation: Union[int, str]
window_type: Union[int, str]
glazing_type: Union[int, str]
window_width: float
window_height: float
draught_proofed: Union[bool, str] # TODO: make enum/mapping?
window_location: Union[int, str] # TODO: make enum/mapping
window_wall_type: Union[int, str] # TODO: make enum/mapping
permanent_shutters_present: Union[bool, str] # TODO: make enum/mapping
frame_factor: Optional[float] = None
window_transmission_details: Optional[WindowTransmissionDetails] = None
permanent_shutters_insulated: Optional[str] = None
@dataclass
class PvBattery:
battery_capacity: float
@dataclass
class PvBatteries:
pv_battery: PvBattery
@dataclass
class WindTurbineDetails:
hub_height: float
rotor_diameter: float
@dataclass
class PhotovoltaicSupplyNoneOrNoDetails:
percent_roof_area: int
@dataclass
class PhotovoltaicSupply:
none_or_no_details: PhotovoltaicSupplyNoneOrNoDetails
@dataclass
class SapEnergySource:
mains_gas: bool
meter_type: str # int in API, str (e.g. "Single") in site notes
pv_battery_count: int
wind_turbines_count: int
gas_smart_meter_present: bool
is_dwelling_export_capable: bool
wind_turbines_terrain_type: str # int in API, str (e.g. "Suburban") in site notes
electricity_smart_meter_present: bool
pv_connection: Optional[int] = None
photovoltaic_supply: Optional[PhotovoltaicSupply] = None
wind_turbine_details: Optional[WindTurbineDetails] = None
pv_batteries: Optional[PvBatteries] = None
@dataclass
class SapFloorDimension:
room_height_m: float
total_floor_area_m2: float
party_wall_length_m: float
heat_loss_perimeter_m: float
floor: Optional[int] = None
floor_insulation: Optional[int] = None
floor_construction: Optional[int] = None
@dataclass
class SapRoomInRoof:
floor_area: Union[int, float]
construction_age_band: str
@dataclass
class SapAlternativeWall:
wall_area: float
wall_dry_lined: str
wall_construction: int
wall_insulation_type: int
wall_thickness_measured: str
wall_insulation_thickness: Optional[str] = None
@dataclass
class SapBuildingPart:
# General
identifier: str # e.g. "main", "roof"
construction_age_band: str
# Wall
wall_construction: Union[
int, str
] # int from API, str from site notes TODO: make enum/mapping?
wall_insulation_type: Union[
int, str
] # int from API, str from site notes TODO: make enum/mapping?
wall_thickness_measured: bool
party_wall_construction: Union[int, str] # TODO: make enum/mapping?
# Floor
sap_floor_dimensions: List[
SapFloorDimension
] # Not included in site notes; should this be optional?
# Optional
building_part_number: Optional[int] = (
None # Not sure how we get this from site notes
)
wall_dry_lined: Optional[bool] = None # Don't think we have this in site notes
wall_thickness_mm: Optional[int] = None
wall_insulation_thickness: Optional[str] = None
sap_alternative_wall_1: Optional[SapAlternativeWall] = None
sap_alternative_wall_2: Optional[SapAlternativeWall] = None
floor_heat_loss: Optional[int] = None
floor_insulation_thickness: Optional[str] = None
flat_roof_insulation_thickness: Optional[Union[str, int]] = (
None # TODO: make enum/mapping?
)
roof_construction: Optional[int] = None
roof_insulation_location: Optional[Union[int, str]] = (
None # TODO: make enum/mapping?
)
roof_insulation_thickness: Optional[Union[str, int]] = (
None # TODO: make enum/mapping?
)
sap_room_in_roof: Optional[SapRoomInRoof] = None
@dataclass
class WindowsTransmissionDetails:
u_value: float
data_source: int
solar_transmittance: float
@dataclass
class SapFlatDetails:
level: int
top_storey: str
flat_location: int
heat_loss_corridor: int
storey_count: Optional[int] = None
unheated_corridor_length_m: Optional[int] = None
@dataclass
class EpcPropertyData:
# General
dwelling_type: str # TODO: make enum?
inspection_date: date
tenure: str # str in site notes; stringified int (e.g. "1") from API
transaction_type: str # str in site notes; stringified int from API
# Elements
roofs: List[EnergyElement]
walls: List[EnergyElement]
floors: List[EnergyElement]
main_heating: List[EnergyElement]
door_count: int
sap_heating: SapHeating
sap_windows: List[SapWindow]
sap_energy_source: SapEnergySource
sap_building_parts: List[SapBuildingPart]
solar_water_heating: bool
has_hot_water_cylinder: bool # must be inferred when mapping from site notes
has_fixed_air_conditioning: bool
# Counts
wet_rooms_count: int # If this isn't provided, should it be 0 or None?
extensions_count: int # If this isn't provided, should it be 0 or None?
heated_rooms_count: int # If this isn't provided, should it be 0 or None?
open_chimneys_count: int
habitable_rooms_count: int
insulated_door_count: (
int # Called "number_of_insulated_external_doors" in site notes; same thing?
)
cfl_fixed_lighting_bulbs_count: int
led_fixed_lighting_bulbs_count: int
incandescent_fixed_lighting_bulbs_count: int
# Measurements
total_floor_area_m2: float
# Optional fields
assessment_type: Optional[str] = None # not available from site notes
sap_version: Optional[float] = None # not available from site notes
uprn: Optional[int] = None # not available from site notes
address_line_1: Optional[str] = None # not available from site notes
postcode: Optional[str] = None # not available from site notes
post_town: Optional[str] = None # not available from site notes
status: Optional[str] = None # not available from site notes
window: Optional[EnergyElement] = None # not available from site notes
lighting: Optional[EnergyElement] = None # not available from site notes
hot_water: Optional[EnergyElement] = None # not available from site notes
schema_type: Optional[str] = None
schema_versions_original: Optional[str] = None
report_type: Optional[str] = None # TODO: make enum?
uprn_source: Optional[str] = None
address_line_2: Optional[str] = None
region_code: Optional[str] = None # TODO: make enum?
country_code: Optional[str] = None
built_form: Optional[str] = None # TODO: make enum?
property_type: Optional[str] = None
pressure_test: Optional[int] = None
language_code: Optional[str] = None
completion_date: Optional[date] = None
registration_date: Optional[date] = None
measurement_type: Optional[int] = None # What is this?
conservatory_type: Optional[int] = (
None # What is this? site notes have "has_conservatory" flag
)
has_conservatory: Optional[bool] = None # mapped directly from site notes
has_heated_separate_conservatory: Optional[bool] = None
secondary_heating: Optional[EnergyElement] = (
None # For site notes, secondary_fuel maps to sap_heating.secondary_fuel_type
)
blocked_chimneys_count: Optional[int] = None
energy_rating_average: Optional[int] = None
main_heating_controls: Optional[EnergyElement] = (
None # site notes has heating_and_hot_water.main_heating.controls: str - doesn't map to EnergyElement
)
current_energy_efficiency_band: Optional[Epc] = None # not available in site notes?
environmental_impact_current: Optional[int] = None
heating_cost_current: Optional[float] = None
co2_emissions_current: Optional[float] = None
energy_consumption_current: Optional[int] = None
energy_rating_current: Optional[int] = None
lighting_cost_current: Optional[float] = None
hot_water_cost_current: Optional[float] = None
insulated_door_u_value: Optional[float] = None # Not available in site notes
mechanical_ventilation: Optional[int] = (
None # ventilation details present in site notes, but I'm not sure they correspond directly to the integers returned by the API here
)
percent_draughtproofed: Optional[int] = (
None # Site notes have draught_proofed: bool field for each window, can we use that to infer percentage?
)
heating_cost_potential: Optional[float] = None
co2_emissions_potential: Optional[float] = None
energy_consumption_potential: Optional[int] = None
energy_rating_potential: Optional[float] = None
lighting_cost_potential: Optional[float] = None
hot_water_cost_potential: Optional[float] = None
environmental_impact_potential: Optional[int] = None
potential_energy_efficiency_band: Optional[Epc] = (
None # not available in site notes
)
# renewable_heat_incentive: Optional[Any] = None # Not sure what this is, skip for now
draughtproofed_door_count: Optional[int] = None
mechanical_vent_duct_type: Optional[int] = None
windows_transmission_details: Optional[WindowsTransmissionDetails] = None
multiple_glazed_propertion: Optional[int] = None
calculation_software_version: Optional[str] = None # Do we care about this?
mechanical_vent_duct_placement: Optional[int] = None
mechanical_vent_duct_insulation: Optional[int] = None
pressure_test_certificate_number: Optional[int] = None
mechanical_ventilation_index_number: Optional[int] = None
mechanical_vent_measured_installation: Optional[str] = None
co2_emissions_current_per_floor_area: Optional[int] = None
low_energy_fixed_lighting_bulbs_count: Optional[int] = None
sap_flat_details: Optional[SapFlatDetails] = None
# survey_addendum: Optional[Any] = None # not sure how to handle, skip for now
fixed_lighting_outlets_count: Optional[int] = None
low_energy_fixed_lighting_outlets_count: Optional[int] = None

View file

@ -0,0 +1,3 @@
PROPERTY_TYPE_LOOKUP = {0: "House", 1: "Bungalow", 2: "Flat", 3: "Maisonette"}
ROOF_CONSTRUCTION_LOOKUP = {}
ROOF_INSULATION_LOCATION_LOOKUP = {}

File diff suppressed because it is too large Load diff

View file

View file

@ -0,0 +1,530 @@
import json
import os
from datetime import date
from typing import Any, Dict
import pytest
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from datatypes.epc.domain.mapper import EpcPropertyDataMapper
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
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.schema.tests.helpers import from_dict
FIXTURES = os.path.join(os.path.dirname(__file__), "../../schema/tests/fixtures")
def load(filename: str) -> Dict[str, Any]:
with open(os.path.join(FIXTURES, filename)) as f:
return json.load(f) # type: ignore[no-any-return]
# ---------------------------------------------------------------------------
# Schema 17.0
# ---------------------------------------------------------------------------
class TestFromRdSapSchema17_0:
@pytest.fixture
def result(self) -> EpcPropertyData:
schema = from_dict(RdSapSchema17_0, load("17_0.json"))
return EpcPropertyDataMapper.from_rdsap_schema_17_0(schema)
def test_uprn(self, result: EpcPropertyData) -> None:
assert result.uprn == 12457
def test_assessment_type(self, result: EpcPropertyData) -> None:
assert result.assessment_type == "RdSAP"
def test_sap_version(self, result: EpcPropertyData) -> None:
assert result.sap_version == 9.92
def test_dwelling_type(self, result: EpcPropertyData) -> None:
# dwelling_type is a localised object in 17.0; mapper extracts the string value
assert result.dwelling_type == "Mid-floor flat"
def test_tenure(self, result: EpcPropertyData) -> None:
# tenure: 2 — stored as stringified int
assert result.tenure == "2"
def test_door_count(self, result: EpcPropertyData) -> None:
assert result.door_count == 2
def test_built_form(self, result: EpcPropertyData) -> None:
assert result.built_form == "2"
def test_property_type(self, result: EpcPropertyData) -> None:
assert result.property_type == "2"
# ---------------------------------------------------------------------------
# Schema 17.1
# ---------------------------------------------------------------------------
class TestFromRdSapSchema17_1:
@pytest.fixture
def result(self) -> EpcPropertyData:
schema = from_dict(RdSapSchema17_1, load("17_1.json"))
return EpcPropertyDataMapper.from_rdsap_schema_17_1(schema)
def test_uprn(self, result: EpcPropertyData) -> None:
assert result.uprn == 12457
def test_assessment_type(self, result: EpcPropertyData) -> None:
assert result.assessment_type == "RdSAP"
def test_sap_version(self, result: EpcPropertyData) -> None:
assert result.sap_version == 9.92
def test_dwelling_type(self, result: EpcPropertyData) -> None:
# dwelling_type is a localised object in 17.1; mapper extracts the string value
assert result.dwelling_type == "Detached house"
def test_tenure(self, result: EpcPropertyData) -> None:
# tenure: 1
assert result.tenure == "1"
def test_door_count(self, result: EpcPropertyData) -> None:
assert result.door_count == 4
def test_built_form(self, result: EpcPropertyData) -> None:
assert result.built_form == "1"
def test_property_type(self, result: EpcPropertyData) -> None:
assert result.property_type == "0"
# ---------------------------------------------------------------------------
# Schema 18.0
# ---------------------------------------------------------------------------
class TestFromRdSapSchema18_0:
@pytest.fixture
def result(self) -> EpcPropertyData:
schema = from_dict(RdSapSchema18_0, load("18_0.json"))
return EpcPropertyDataMapper.from_rdsap_schema_18_0(schema)
def test_uprn(self, result: EpcPropertyData) -> None:
assert result.uprn == 12457
def test_assessment_type(self, result: EpcPropertyData) -> None:
assert result.assessment_type == "RdSAP"
def test_sap_version(self, result: EpcPropertyData) -> None:
assert result.sap_version == 9.92
def test_dwelling_type(self, result: EpcPropertyData) -> None:
# dwelling_type is a localised object in 18.0; mapper extracts the string value
assert result.dwelling_type == "Mid-terrace house"
def test_tenure(self, result: EpcPropertyData) -> None:
assert result.tenure == "1"
def test_door_count(self, result: EpcPropertyData) -> None:
assert result.door_count == 2
def test_built_form(self, result: EpcPropertyData) -> None:
assert result.built_form == "4"
def test_property_type(self, result: EpcPropertyData) -> None:
assert result.property_type == "0"
# ---------------------------------------------------------------------------
# Schema 19.0
# ---------------------------------------------------------------------------
class TestFromRdSapSchema19_0:
@pytest.fixture
def result(self) -> EpcPropertyData:
schema = from_dict(RdSapSchema19_0, load("19_0.json"))
return EpcPropertyDataMapper.from_rdsap_schema_19_0(schema)
def test_uprn(self, result: EpcPropertyData) -> None:
assert result.uprn == 12457
def test_assessment_type(self, result: EpcPropertyData) -> None:
assert result.assessment_type == "RdSAP"
def test_sap_version(self, result: EpcPropertyData) -> None:
assert result.sap_version == 9.94
def test_dwelling_type(self, result: EpcPropertyData) -> None:
# dwelling_type is a localised object in 19.0; mapper extracts the string value
assert result.dwelling_type == "Semi-detached house"
def test_tenure(self, result: EpcPropertyData) -> None:
# tenure: 3
assert result.tenure == "3"
def test_door_count(self, result: EpcPropertyData) -> None:
assert result.door_count == 1
def test_built_form(self, result: EpcPropertyData) -> None:
assert result.built_form == "2"
def test_property_type(self, result: EpcPropertyData) -> None:
assert result.property_type == "0"
# ---------------------------------------------------------------------------
# Schema 20.0.0
# ---------------------------------------------------------------------------
class TestFromRdSapSchema20_0_0:
@pytest.fixture
def result(self) -> EpcPropertyData:
schema = from_dict(RdSapSchema20_0_0, load("20_0_0.json"))
return EpcPropertyDataMapper.from_rdsap_schema_20_0_0(schema)
def test_uprn(self, result: EpcPropertyData) -> None:
assert result.uprn == 12457
def test_assessment_type(self, result: EpcPropertyData) -> None:
assert result.assessment_type == "RdSAP"
def test_sap_version(self, result: EpcPropertyData) -> None:
assert result.sap_version == 9.8
def test_dwelling_type(self, result: EpcPropertyData) -> None:
# dwelling_type is a plain string from 20.0.0 onwards
assert result.dwelling_type == "Mid-terrace house"
def test_tenure(self, result: EpcPropertyData) -> None:
assert result.tenure == "1"
def test_door_count(self, result: EpcPropertyData) -> None:
assert result.door_count == 2
def test_built_form(self, result: EpcPropertyData) -> None:
assert result.built_form == "2"
def test_property_type(self, result: EpcPropertyData) -> None:
assert result.property_type == "0"
# ---------------------------------------------------------------------------
# Schema 21.0.0
# ---------------------------------------------------------------------------
class TestFromRdSapSchema21_0_0:
@pytest.fixture
def result(self) -> EpcPropertyData:
schema = from_dict(RdSapSchema21_0_0, load("21_0_0.json"))
return EpcPropertyDataMapper.from_rdsap_schema_21_0_0(schema)
def test_uprn(self, result: EpcPropertyData) -> None:
assert result.uprn == 12457
def test_assessment_type(self, result: EpcPropertyData) -> None:
assert result.assessment_type == "RdSAP"
def test_sap_version(self, result: EpcPropertyData) -> None:
assert result.sap_version == 10.2
def test_dwelling_type(self, result: EpcPropertyData) -> None:
assert result.dwelling_type == "Mid-terrace house"
def test_tenure(self, result: EpcPropertyData) -> None:
assert result.tenure == "1"
def test_door_count(self, result: EpcPropertyData) -> None:
assert result.door_count == 3
def test_built_form(self, result: EpcPropertyData) -> None:
assert result.built_form == "2"
def test_property_type(self, result: EpcPropertyData) -> None:
assert result.property_type == "0"
# ---------------------------------------------------------------------------
# Schema 21.0.1 (most comprehensive — full field coverage)
# ---------------------------------------------------------------------------
class TestFromRdSapSchema21_0_1:
@pytest.fixture
def schema(self) -> RdSapSchema21_0_1:
return from_dict(RdSapSchema21_0_1, load("21_0_1.json"))
@pytest.fixture
def result(self, schema: RdSapSchema21_0_1) -> EpcPropertyData:
return EpcPropertyDataMapper.from_rdsap_schema_21_0_1(schema)
# --- general ---
def test_uprn(self, result: EpcPropertyData) -> None:
assert result.uprn == 12457
def test_assessment_type(self, result: EpcPropertyData) -> None:
assert result.assessment_type == "RdSAP"
def test_sap_version(self, result: EpcPropertyData) -> None:
assert result.sap_version == 10.2
def test_dwelling_type(self, result: EpcPropertyData) -> None:
assert result.dwelling_type == "Mid-terrace house"
def test_property_type(self, result: EpcPropertyData) -> None:
assert result.property_type == "0"
def test_built_form(self, result: EpcPropertyData) -> None:
assert result.built_form == "2"
def test_address_line_1(self, result: EpcPropertyData) -> None:
assert result.address_line_1 == "1 Some Street"
def test_postcode(self, result: EpcPropertyData) -> None:
assert result.postcode == "A0 0AA"
def test_post_town(self, result: EpcPropertyData) -> None:
assert result.post_town == "Whitbury"
def test_status(self, result: EpcPropertyData) -> None:
assert result.status == "entered"
def test_tenure(self, result: EpcPropertyData) -> None:
# tenure: 1 — stored as stringified int
assert result.tenure == "1"
def test_transaction_type(self, result: EpcPropertyData) -> None:
# transaction_type: 16 — stored as stringified int
assert result.transaction_type == "16"
def test_inspection_date(self, result: EpcPropertyData) -> None:
assert result.inspection_date == date(2025, 4, 4)
def test_total_floor_area(self, result: EpcPropertyData) -> None:
assert result.total_floor_area_m2 == 55.0
# --- property flags ---
def test_solar_water_heating(self, result: EpcPropertyData) -> None:
# solar_water_heating: "N"
assert result.solar_water_heating is False
def test_has_hot_water_cylinder(self, result: EpcPropertyData) -> None:
# has_hot_water_cylinder: "true"
assert result.has_hot_water_cylinder is True
def test_has_fixed_air_conditioning(self, result: EpcPropertyData) -> None:
# has_fixed_air_conditioning: "false"
assert result.has_fixed_air_conditioning is False
def test_no_conservatory(self, result: EpcPropertyData) -> None:
# conservatory_type: 1 → no conservatory
assert result.has_conservatory is False
# --- counts ---
def test_door_count(self, result: EpcPropertyData) -> None:
assert result.door_count == 3
def test_habitable_rooms(self, result: EpcPropertyData) -> None:
assert result.habitable_rooms_count == 5
def test_heated_rooms(self, result: EpcPropertyData) -> None:
assert result.heated_rooms_count == 5
def test_wet_rooms(self, result: EpcPropertyData) -> None:
assert result.wet_rooms_count == 0
def test_extensions_count(self, result: EpcPropertyData) -> None:
assert result.extensions_count == 0
def test_open_chimneys(self, result: EpcPropertyData) -> None:
assert result.open_chimneys_count == 1
def test_insulated_doors(self, result: EpcPropertyData) -> None:
assert result.insulated_door_count == 2
def test_draughtproofed_doors(self, result: EpcPropertyData) -> None:
assert result.draughtproofed_door_count == 1
# --- lighting ---
def test_led_bulbs(self, result: EpcPropertyData) -> None:
assert result.led_fixed_lighting_bulbs_count == 10
def test_cfl_bulbs(self, result: EpcPropertyData) -> None:
assert result.cfl_fixed_lighting_bulbs_count == 5
def test_incandescent_bulbs(self, result: EpcPropertyData) -> None:
assert result.incandescent_fixed_lighting_bulbs_count == 0
# --- energy elements ---
def test_roof_count(self, result: EpcPropertyData) -> None:
assert len(result.roofs) == 2
def test_roof_description(self, result: EpcPropertyData) -> None:
assert result.roofs[0].description == "Pitched, 25 mm loft insulation"
def test_roof_energy_efficiency_rating(self, result: EpcPropertyData) -> None:
assert result.roofs[0].energy_efficiency_rating == 2
def test_wall_count(self, result: EpcPropertyData) -> None:
assert len(result.walls) == 2
def test_window_element_description(self, result: EpcPropertyData) -> None:
assert result.window is not None
assert result.window.description == "Fully double glazed"
def test_window_element_rating(self, result: EpcPropertyData) -> None:
assert result.window is not None
assert result.window.energy_efficiency_rating == 3
def test_lighting_element_description(self, result: EpcPropertyData) -> None:
assert result.lighting is not None
assert result.lighting.description == "Low energy lighting in 50% of fixed outlets"
def test_hot_water_element_description(self, result: EpcPropertyData) -> None:
assert result.hot_water is not None
assert result.hot_water.description == "From main system"
def test_secondary_heating_element(self, result: EpcPropertyData) -> None:
assert result.secondary_heating is not None
assert result.secondary_heating.description == "Room heaters, electric"
def test_main_heating_element_count(self, result: EpcPropertyData) -> None:
assert len(result.main_heating) == 2
def test_main_heating_element_description(self, result: EpcPropertyData) -> None:
assert result.main_heating[0].description == "Boiler and radiators, anthracite"
# --- sap energy source ---
def test_mains_gas(self, result: EpcPropertyData) -> None:
# mains_gas: "Y"
assert result.sap_energy_source.mains_gas is True
def test_electricity_smart_meter(self, result: EpcPropertyData) -> None:
# electricity_smart_meter_present: "true"
assert result.sap_energy_source.electricity_smart_meter_present is True
def test_gas_smart_meter(self, result: EpcPropertyData) -> None:
# gas_smart_meter_present: "false"
assert result.sap_energy_source.gas_smart_meter_present is False
def test_pv_battery_count(self, result: EpcPropertyData) -> None:
assert result.sap_energy_source.pv_battery_count == 1
def test_wind_turbines_count(self, result: EpcPropertyData) -> None:
assert result.sap_energy_source.wind_turbines_count == 0
# --- sap heating ---
def test_cylinder_size(self, result: EpcPropertyData) -> None:
assert result.sap_heating.cylinder_size == 1
def test_water_heating_code(self, result: EpcPropertyData) -> None:
assert result.sap_heating.water_heating_code == 901
def test_water_heating_fuel(self, result: EpcPropertyData) -> None:
assert result.sap_heating.water_heating_fuel == 26
def test_secondary_fuel_type(self, result: EpcPropertyData) -> None:
# secondary_fuel_type: 25
assert result.sap_heating.secondary_fuel_type == 25
def test_main_heating_no_fghrs(self, result: EpcPropertyData) -> None:
# has_fghrs: "N"
assert result.sap_heating.main_heating_details[0].has_fghrs is False
def test_main_heating_fuel_type(self, result: EpcPropertyData) -> None:
# main_fuel_type: 26
assert result.sap_heating.main_heating_details[0].main_fuel_type == 26
def test_main_heating_fan_flue(self, result: EpcPropertyData) -> None:
# fan_flue_present: "N"
assert result.sap_heating.main_heating_details[0].fan_flue_present is False
def test_main_heating_control(self, result: EpcPropertyData) -> None:
assert result.sap_heating.main_heating_details[0].main_heating_control == 2106
def test_main_heating_category(self, result: EpcPropertyData) -> None:
assert result.sap_heating.main_heating_details[0].main_heating_category == 2
def test_main_heating_number(self, result: EpcPropertyData) -> None:
assert result.sap_heating.main_heating_details[0].main_heating_number == 1
# --- sap windows ---
def test_window_count(self, result: EpcPropertyData) -> None:
assert len(result.sap_windows) == 1
def test_window_height(self, result: EpcPropertyData) -> None:
assert result.sap_windows[0].window_height == 2.0
def test_window_width(self, result: EpcPropertyData) -> None:
assert result.sap_windows[0].window_width == 1.2
def test_window_draught_proofed(self, result: EpcPropertyData) -> None:
# draught_proofed: "true"
assert result.sap_windows[0].draught_proofed is True
# --- sap building parts ---
def test_building_part_count(self, result: EpcPropertyData) -> None:
assert len(result.sap_building_parts) == 1
def test_construction_age_band(self, result: EpcPropertyData) -> None:
assert result.sap_building_parts[0].construction_age_band == "M"
def test_wall_construction(self, result: EpcPropertyData) -> None:
# wall_construction: 4 (int preserved from API)
assert result.sap_building_parts[0].wall_construction == 4
def test_wall_insulation_type(self, result: EpcPropertyData) -> None:
# wall_insulation_type: 2 (int preserved from API)
assert result.sap_building_parts[0].wall_insulation_type == 2
def test_wall_thickness_not_measured(self, result: EpcPropertyData) -> None:
# wall_thickness_measured: "N"
assert result.sap_building_parts[0].wall_thickness_measured is False
def test_wall_thickness_mm_absent(self, result: EpcPropertyData) -> None:
assert result.sap_building_parts[0].wall_thickness_mm is None
def test_roof_insulation_thickness(self, result: EpcPropertyData) -> None:
# roof_insulation_thickness: "200mm" — preserved as-is from schema
assert result.sap_building_parts[0].roof_insulation_thickness == "200mm"
def test_room_in_roof_present(self, result: EpcPropertyData) -> None:
# sap_room_in_roof is present in the fixture
assert result.sap_building_parts[0].sap_room_in_roof is not None
# --- floor dimensions ---
def test_floor_count(self, result: EpcPropertyData) -> None:
assert len(result.sap_building_parts[0].sap_floor_dimensions) == 1
def test_floor_area(self, result: EpcPropertyData) -> None:
assert result.sap_building_parts[0].sap_floor_dimensions[0].total_floor_area_m2 == 45.82
def test_floor_height(self, result: EpcPropertyData) -> None:
assert result.sap_building_parts[0].sap_floor_dimensions[0].room_height_m == 2.45
def test_heat_loss_perimeter(self, result: EpcPropertyData) -> None:
assert result.sap_building_parts[0].sap_floor_dimensions[0].heat_loss_perimeter_m == 19.5
def test_party_wall_length(self, result: EpcPropertyData) -> None:
assert result.sap_building_parts[0].sap_floor_dimensions[0].party_wall_length_m == 7.9

View file

@ -0,0 +1,499 @@
import json
import os
from datetime import date
from typing import Any, Dict
import pytest
from datatypes.epc.domain.epc_property_data import (
EpcPropertyData,
InstantaneousWwhrs,
MainHeatingDetail,
SapBuildingPart,
SapEnergySource,
SapFloorDimension,
SapHeating,
SapWindow,
)
from datatypes.epc.domain.mapper import EpcPropertyDataMapper
from datatypes.epc.schema.tests.helpers import from_dict
from datatypes.epc.surveys.pashub_rdsap_site_notes import PasHubRdSapSiteNotes
FIXTURES = os.path.join(
os.path.dirname(__file__),
"../../surveys/tests/fixtures",
)
def load(filename: str) -> Dict[str, Any]:
with open(os.path.join(FIXTURES, filename)) as f:
return json.load(f) # type: ignore[no-any-return]
class TestFromSiteNotesExample1:
"""
Fixture: pashub_rdsap_site_notes_example1.json
No extensions, regular boiler with cylinder, natural ventilation.
"""
@pytest.fixture
def survey(self) -> PasHubRdSapSiteNotes:
return from_dict(
PasHubRdSapSiteNotes, load("pashub_rdsap_site_notes_example1.json")
)
@pytest.fixture
def result(self, survey: PasHubRdSapSiteNotes) -> EpcPropertyData:
return EpcPropertyDataMapper.from_site_notes(survey)
# --- property details ---
def test_dwelling_type(self, result: EpcPropertyData) -> None:
# general.property_type + general.detachment_type → "Mid-terrace house"
assert result.dwelling_type == "Mid-terrace house"
def test_tenure(self, result: EpcPropertyData) -> None:
# general.tenure: "Rented Social"
assert result.tenure == "Rented Social"
def test_transaction_type(self, result: EpcPropertyData) -> None:
# general.transaction_type: "None of the Above"
assert result.transaction_type == "None of the Above"
def test_inspection_date(self, result: EpcPropertyData) -> None:
# general.inspection_date: "2026-03-31"
assert result.inspection_date == date(2026, 3, 31)
def test_built_form(self, result: EpcPropertyData) -> None:
# general.detachment_type: "Mid-terrace"
assert result.built_form == "Mid-terrace"
def test_property_type(self, result: EpcPropertyData) -> None:
# general.property_type: "House"
assert result.property_type == "House"
# --- energy elements are not available from site notes ---
def test_roofs_empty(self, result: EpcPropertyData) -> None:
assert result.roofs == []
def test_walls_empty(self, result: EpcPropertyData) -> None:
assert result.walls == []
def test_floors_empty(self, result: EpcPropertyData) -> None:
assert result.floors == []
def test_main_heating_elements_empty(self, result: EpcPropertyData) -> None:
assert result.main_heating == []
def test_window_element_absent(self, result: EpcPropertyData) -> None:
assert result.window is None
def test_lighting_element_absent(self, result: EpcPropertyData) -> None:
assert result.lighting is None
def test_hot_water_element_absent(self, result: EpcPropertyData) -> None:
assert result.hot_water is None
# --- energy source ---
def test_mains_gas_available(self, result: EpcPropertyData) -> None:
# general.mains_gas_available: true
assert result.sap_energy_source.mains_gas is True
def test_electricity_smart_meter(self, result: EpcPropertyData) -> None:
# general.electricity_smart_meter: true
assert result.sap_energy_source.electricity_smart_meter_present is True
def test_gas_smart_meter(self, result: EpcPropertyData) -> None:
# general.gas_smart_meter: true
assert result.sap_energy_source.gas_smart_meter_present is True
def test_meter_type(self, result: EpcPropertyData) -> None:
# general.electric_meter_type: "Single"
assert result.sap_energy_source.meter_type == "Single"
def test_dwelling_export_capable(self, result: EpcPropertyData) -> None:
# general.dwelling_export_capable: true
assert result.sap_energy_source.is_dwelling_export_capable is True
def test_wind_turbines_terrain_type(self, result: EpcPropertyData) -> None:
# general.terrain_type: "Suburban"
assert result.sap_energy_source.wind_turbines_terrain_type == "Suburban"
def test_no_wind_turbines(self, result: EpcPropertyData) -> None:
# renewables.wind_turbines: false → count 0
assert result.sap_energy_source.wind_turbines_count == 0
def test_no_pv_batteries(self, result: EpcPropertyData) -> None:
# renewables.number_of_pv_batteries: 0
assert result.sap_energy_source.pv_battery_count == 0
# --- renewables ---
def test_no_solar_hot_water(self, result: EpcPropertyData) -> None:
# renewables.solar_hot_water: false
assert result.solar_water_heating is False
# --- ventilation ---
def test_no_fixed_air_conditioning(self, result: EpcPropertyData) -> None:
# ventilation.has_fixed_air_conditioning: false
assert result.has_fixed_air_conditioning is False
# --- conservatory ---
def test_no_conservatory(self, result: EpcPropertyData) -> None:
# conservatories.has_conservatory: false
assert result.has_conservatory is False
# --- hot water / cylinder ---
def test_has_hot_water_cylinder(self, result: EpcPropertyData) -> None:
# water_heating.cylinder_size is present → cylinder exists
assert result.has_hot_water_cylinder is True
# --- main heating ---
def test_main_heating_fuel(self, result: EpcPropertyData) -> None:
# heating_and_hot_water.main_heating.fuel: "Mains gas"
assert result.sap_heating.main_heating_details[0].main_fuel_type == "Mains gas"
def test_main_heating_emitter(self, result: EpcPropertyData) -> None:
# heating_and_hot_water.main_heating.emitter: "Radiators"
assert (
result.sap_heating.main_heating_details[0].heat_emitter_type == "Radiators"
)
def test_main_heating_no_fghrs(self, result: EpcPropertyData) -> None:
# heating_and_hot_water.main_heating.flue_gas_heat_recovery_system: false
assert result.sap_heating.main_heating_details[0].has_fghrs is False
def test_main_heating_fan_flue_present(self, result: EpcPropertyData) -> None:
# heating_and_hot_water.main_heating.fan_assist: true
assert result.sap_heating.main_heating_details[0].fan_flue_present is True
def test_no_secondary_heating(self, result: EpcPropertyData) -> None:
# secondary_heating.secondary_fuel: "No Secondary Heating" → no secondary fuel type
assert result.sap_heating.secondary_fuel_type is None
# --- windows ---
def test_window_count(self, result: EpcPropertyData) -> None:
# 4 windows in fixture
assert len(result.sap_windows) == 4
def test_window_height(self, result: EpcPropertyData) -> None:
assert result.sap_windows[0].window_height == 1.36
def test_window_width(self, result: EpcPropertyData) -> None:
assert result.sap_windows[0].window_width == 1.0
def test_window_draught_proofed(self, result: EpcPropertyData) -> None:
# windows[0].draught_proofed: true
assert result.sap_windows[0].draught_proofed is True
def test_window_orientation(self, result: EpcPropertyData) -> None:
assert result.sap_windows[0].orientation == "South East"
def test_window_glazing_type(self, result: EpcPropertyData) -> None:
assert (
result.sap_windows[0].glazing_type == "Double glazing, Unknown install date"
)
# --- building parts ---
def test_building_parts_count(self, result: EpcPropertyData) -> None:
# no extensions → one building part for main building
assert len(result.sap_building_parts) == 1
def test_building_part_identifier(self, result: EpcPropertyData) -> None:
assert result.sap_building_parts[0].identifier == "main"
def test_construction_age_band(self, result: EpcPropertyData) -> None:
# main_building.age_range: "I: 1996 - 2002" → letter "I"
assert result.sap_building_parts[0].construction_age_band == "I"
def test_wall_construction(self, result: EpcPropertyData) -> None:
# main_building.walls_construction_type: "Cavity"
assert result.sap_building_parts[0].wall_construction == "Cavity"
def test_wall_insulation_type(self, result: EpcPropertyData) -> None:
# main_building.walls_insulation_type: "As built"
assert result.sap_building_parts[0].wall_insulation_type == "As built"
def test_wall_thickness_measured(self, result: EpcPropertyData) -> None:
# main_building.wall_thickness_mm: 280 → thickness was measured
assert result.sap_building_parts[0].wall_thickness_measured is True
def test_wall_thickness_mm(self, result: EpcPropertyData) -> None:
assert result.sap_building_parts[0].wall_thickness_mm == 280
# --- floor dimensions ---
def test_floor_count(self, result: EpcPropertyData) -> None:
# 2 floors in main building
assert len(result.sap_building_parts[0].sap_floor_dimensions) == 2
def test_floor_area(self, result: EpcPropertyData) -> None:
# building_measurements.main_building.floors[0].area_m2: 24.78
assert (
result.sap_building_parts[0].sap_floor_dimensions[0].total_floor_area_m2
== 24.78
)
def test_floor_height(self, result: EpcPropertyData) -> None:
# floors[0].height_m: 2.37
assert (
result.sap_building_parts[0].sap_floor_dimensions[0].room_height_m == 2.37
)
def test_heat_loss_perimeter(self, result: EpcPropertyData) -> None:
# floors[0].heat_loss_perimeter_m: 14.21
assert (
result.sap_building_parts[0].sap_floor_dimensions[0].heat_loss_perimeter_m
== 14.21
)
def test_party_wall_length(self, result: EpcPropertyData) -> None:
# floors[0].pwl_m: 6.15
assert (
result.sap_building_parts[0].sap_floor_dimensions[0].party_wall_length_m
== 6.15
)
def test_total_floor_area(self, result: EpcPropertyData) -> None:
# sum of all floor areas: 24.78 + 24.78 = 49.56
assert result.total_floor_area_m2 == 49.56
# --- room counts ---
def test_habitable_rooms(self, result: EpcPropertyData) -> None:
# room_count_elements.number_of_habitable_rooms: 2
assert result.habitable_rooms_count == 2
def test_external_doors(self, result: EpcPropertyData) -> None:
# room_count_elements.number_of_external_doors: 2
assert result.door_count == 2
def test_open_chimneys(self, result: EpcPropertyData) -> None:
# room_count_elements.number_of_open_chimneys: 0
assert result.open_chimneys_count == 0
def test_blocked_chimneys(self, result: EpcPropertyData) -> None:
# room_count_elements.number_of_blocked_chimneys: 0
assert result.blocked_chimneys_count == 0
def test_insulated_doors(self, result: EpcPropertyData) -> None:
# room_count_elements.number_of_insulated_external_doors: 0
assert result.insulated_door_count == 0
def test_draughtproofed_doors(self, result: EpcPropertyData) -> None:
# room_count_elements.number_of_draughtproofed_external_doors: 2
assert result.draughtproofed_door_count == 2
def test_extensions_count(self, result: EpcPropertyData) -> None:
# general.number_of_extensions: 0
assert result.extensions_count == 0
def test_heated_rooms_count(self, result: EpcPropertyData) -> None:
# room_count_elements.number_of_heated_rooms: 0
# no equivalent in site notes defaults to 0
assert result.heated_rooms_count == 0
def test_wet_rooms_count_defaults_to_zero(self, result: EpcPropertyData) -> None:
# no equivalent in site notes; mapper must default to 0
assert result.wet_rooms_count == 0
# --- lighting ---
def test_led_bulbs(self, result: EpcPropertyData) -> None:
# room_count_elements.number_of_fixed_led_bulbs: 5
assert result.led_fixed_lighting_bulbs_count == 5
def test_cfl_bulbs(self, result: EpcPropertyData) -> None:
# room_count_elements.number_of_fixed_cfl_bulbs: 4
assert result.cfl_fixed_lighting_bulbs_count == 4
def test_incandescent_bulbs(self, result: EpcPropertyData) -> None:
# room_count_elements.number_of_fixed_incandescent_bulbs: 0
assert result.incandescent_fixed_lighting_bulbs_count == 0
# --- api-only fields absent ---
def test_assessment_type_absent(self, result: EpcPropertyData) -> None:
assert result.assessment_type is None
def test_sap_version_absent(self, result: EpcPropertyData) -> None:
assert result.sap_version is None
def test_uprn_absent(self, result: EpcPropertyData) -> None:
assert result.uprn is None
def test_address_absent(self, result: EpcPropertyData) -> None:
assert result.address_line_1 is None
def test_postcode_absent(self, result: EpcPropertyData) -> None:
assert result.postcode is None
def test_post_town_absent(self, result: EpcPropertyData) -> None:
assert result.post_town is None
def test_status_absent(self, result: EpcPropertyData) -> None:
assert result.status is None
# --- full object equality ---
def test_full_mapping(self, survey: PasHubRdSapSiteNotes) -> None:
result = EpcPropertyDataMapper.from_site_notes(survey)
expected = EpcPropertyData(
# General
assessment_type=None,
sap_version=None,
dwelling_type="Mid-terrace house",
uprn=None,
address_line_1=None,
postcode=None,
post_town=None,
inspection_date=date(2026, 3, 31),
status=None,
tenure="Rented Social",
transaction_type="None of the Above",
# Elements (not available from site notes)
roofs=[],
walls=[],
floors=[],
main_heating=[],
window=None,
lighting=None,
hot_water=None,
door_count=2,
# Heating
sap_heating=SapHeating(
instantaneous_wwhrs=InstantaneousWwhrs(),
main_heating_details=[
MainHeatingDetail(
has_fghrs=False,
main_fuel_type="Mains gas",
heat_emitter_type="Radiators",
emitter_temperature="Unknown",
fan_flue_present=True,
main_heating_control="Programmer, room thermostat and TRVs",
)
],
has_fixed_air_conditioning=False,
),
# Windows
sap_windows=[
SapWindow(
pvc_frame="Wooden or PVC",
glazing_gap="16 mm or more",
orientation="South East",
window_type="Window",
glazing_type="Double glazing, Unknown install date",
window_width=1.0,
window_height=1.36,
draught_proofed=True,
window_location="Main Building",
window_wall_type="External wall",
permanent_shutters_present=False,
),
SapWindow(
pvc_frame="Wooden or PVC",
glazing_gap="16 mm or more",
orientation="South East",
window_type="Window",
glazing_type="Double glazing, Unknown install date",
window_width=0.96,
window_height=1.33,
draught_proofed=True,
window_location="Main Building",
window_wall_type="External wall",
permanent_shutters_present=False,
),
SapWindow(
pvc_frame="Wooden or PVC",
glazing_gap="16 mm or more",
orientation="North West",
window_type="Window",
glazing_type="Double glazing, Unknown install date",
window_width=0.96,
window_height=1.04,
draught_proofed=True,
window_location="Main Building",
window_wall_type="External wall",
permanent_shutters_present=False,
),
SapWindow(
pvc_frame="Wooden or PVC",
glazing_gap="16 mm or more",
orientation="North West",
window_type="Window",
glazing_type="Double glazing, Unknown install date",
window_width=0.97,
window_height=1.02,
draught_proofed=True,
window_location="Main Building",
window_wall_type="External wall",
permanent_shutters_present=False,
),
],
# Energy source
sap_energy_source=SapEnergySource(
mains_gas=True,
meter_type="Single",
pv_battery_count=0,
wind_turbines_count=0,
gas_smart_meter_present=True,
is_dwelling_export_capable=True,
wind_turbines_terrain_type="Suburban",
electricity_smart_meter_present=True,
),
# Building parts
sap_building_parts=[
SapBuildingPart(
identifier="main",
construction_age_band="I",
wall_construction="Cavity",
wall_insulation_type="As built",
wall_thickness_measured=True,
party_wall_construction="Cavity Masonry, Unfilled",
sap_floor_dimensions=[
SapFloorDimension(
room_height_m=2.37,
total_floor_area_m2=24.78,
party_wall_length_m=6.15,
heat_loss_perimeter_m=14.21,
),
SapFloorDimension(
room_height_m=2.35,
total_floor_area_m2=24.78,
party_wall_length_m=6.15,
heat_loss_perimeter_m=14.21,
),
],
wall_thickness_mm=280,
)
],
solar_water_heating=False,
has_hot_water_cylinder=True,
has_fixed_air_conditioning=False,
# Counts
wet_rooms_count=0, # no equivalent in site notes
extensions_count=0,
heated_rooms_count=0,
open_chimneys_count=0,
habitable_rooms_count=2,
insulated_door_count=0,
cfl_fixed_lighting_bulbs_count=4,
led_fixed_lighting_bulbs_count=5,
incandescent_fixed_lighting_bulbs_count=0,
total_floor_area_m2=49.56,
# Optional fields populated from site notes
built_form="Mid-terrace",
property_type="House",
has_conservatory=False,
blocked_chimneys_count=0,
draughtproofed_door_count=2,
)
assert result == expected

View file

@ -33,13 +33,14 @@ class ShowerOutlets:
@dataclass
class InstantaneousWwhrs:
"""References WWHRS product index numbers (introduced in 21.0.0)."""
wwhrs_index_number1: Optional[int] = None
wwhrs_index_number2: Optional[int] = None
@dataclass
class MainHeatingDetail:
has_fghrs: str
has_fghrs: str # TODO: make bool
main_fuel_type: int
heat_emitter_type: int
emitter_temperature: Union[int, str]
@ -49,7 +50,7 @@ class MainHeatingDetail:
main_heating_fraction: int
main_heating_data_source: int
boiler_flue_type: Optional[int] = None
fan_flue_present: Optional[str] = None
fan_flue_present: Optional[str] = None # TODO: make bool
boiler_ignition_type: Optional[int] = None
central_heating_pump_age: Optional[int] = None
main_heating_index_number: Optional[int] = None
@ -132,10 +133,10 @@ class SapWindow:
glazing_type: int
window_width: float
window_height: float
draught_proofed: str
draught_proofed: str # TODO: make bool
window_location: int
window_wall_type: int
permanent_shutters_present: str
permanent_shutters_present: str # TODO: make bool
window_transmission_details: WindowTransmissionDetails
permanent_shutters_insulated: str

View file

@ -0,0 +1,3 @@
from .pashub_rdsap_site_notes import PasHubRdSapSiteNotes
__all__ = ["PasHubRdSapSiteNotes"]

View file

@ -0,0 +1,291 @@
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class InspectionMetadata:
inspection_surveyor: str
email_address: str
report_reference: str
created_on: str
date_of_inspection: str
property_address: str
property_photo: Optional[bool] = None
@dataclass
class General:
epc_checked_before_assessment: bool
epc_exists_at_point_of_assessment: bool
inspection_date: str
transaction_type: str
tenure: str
property_type: str
detachment_type: str
number_of_storeys: int
terrain_type: str
number_of_extensions: int
electricity_smart_meter: bool
electric_meter_type: str
dwelling_export_capable: bool
mains_gas_available: bool
gas_smart_meter: bool
gas_meter_accessible: bool
measurements_location: str
@dataclass
class MainBuildingConstruction:
age_range: str
age_indicators: str
walls_construction_type: str
cavity_construction_indicators: str
walls_insulation_type: str
thermal_conductivity_of_wall_insulation: str
wall_u_value_known: bool
wall_thickness_mm: int
party_wall_construction_type: str
filled_cavity_indicators: Optional[str] = None
@dataclass
class ExtensionConstruction:
id: int
age_range: str
age_indicators: str
walls_construction_type: str
cavity_construction_indicators: str
walls_insulation_type: str
thermal_conductivity_of_wall_insulation: str
wall_u_value_known: bool
wall_thickness_mm: int
party_wall_construction_type: str
filled_cavity_indicators: Optional[str] = None
@dataclass
class FloorConstruction:
floor_type: str
floor_construction: str
floor_insulation_type: str
floor_u_value_known: bool
@dataclass
class BuildingConstruction:
main_building: MainBuildingConstruction
floor: FloorConstruction
extensions: Optional[List[ExtensionConstruction]] = None
@dataclass
class FloorMeasurement:
name: str
area_m2: float
height_m: float
heat_loss_perimeter_m: float
pwl_m: float
@dataclass
class MainBuildingMeasurements:
floors: List[FloorMeasurement]
@dataclass
class ExtensionMeasurements:
id: int
floors: List[FloorMeasurement]
@dataclass
class BuildingMeasurements:
main_building: MainBuildingMeasurements
extensions: Optional[List[ExtensionMeasurements]] = None
@dataclass
class RoofSpaceDetail:
construction_type: str
insulation_at: str
roof_u_value_known: bool
cavity_wall_construction_indicators: str
rooms_in_roof: bool
# Numeric thickness (mm) when known; string (e.g. "As built") when not measured
insulation_thickness_mm: Optional[int] = None
insulation_thickness: Optional[str] = None
@dataclass
class ExtensionRoofSpace:
id: int
construction_type: str
insulation_at: str
roof_u_value_known: bool
cavity_wall_construction_indicators: str
rooms_in_roof: bool
insulation_thickness_mm: Optional[int] = None
insulation_thickness: Optional[str] = None
@dataclass
class RoofSpace:
main_building: RoofSpaceDetail
extensions: Optional[List[ExtensionRoofSpace]] = None
@dataclass
class Window:
id: int
location: str
wall_type: str
glazing_type: str
window_type: str
frame_type: str
glazing_gap: str
draught_proofed: bool
permanent_shutters: bool
height_m: float
width_m: float
orientation: str
@dataclass
class MainHeating:
selection_method: str
system_type: str
product_id: int
manufacturer: str
model: str
orig_manufacturer: str
fuel: str
summer_efficiency: float
type: str
condensing: bool
year: str
mount: str
open_flue: str
fan_assist: bool
status: str
central_heating_pump_age: str
controls: str
flue_gas_heat_recovery_system: bool
weather_compensator: bool
emitter: str
emitter_temperature: str
@dataclass
class SecondaryHeating:
secondary_fuel: str
@dataclass
class WaterHeating:
type: str
system: str
cylinder_size: str
cylinder_measured_heat_loss: Optional[str] = None
insulation_type: Optional[str] = None
insulation_thickness_mm: Optional[int] = None
has_thermostat: Optional[bool] = None
@dataclass
class HeatingAndHotWater:
main_heating: MainHeating
secondary_heating: SecondaryHeating
water_heating: WaterHeating
@dataclass
class Ventilation:
ventilation_type: str
has_fixed_air_conditioning: bool
number_of_open_flues: int
number_of_closed_flues: int
number_of_boiler_flues: int
number_of_other_flues: int
number_of_extract_fans: int
number_of_passive_vents: int
number_of_flueless_gas_fires: int
pressure_test: str
draught_lobby: bool
ventilation_in_pcdf_database: Optional[bool] = None
@dataclass
class Conservatories:
has_conservatory: bool
@dataclass
class Renewables:
wind_turbines: bool
solar_hot_water: bool
photovoltaic_array: bool
number_of_pv_batteries: int
hydro: bool
@dataclass
class RoomCountElements:
number_of_habitable_rooms: int
any_unheated_rooms: bool
number_of_external_doors: int
number_of_insulated_external_doors: int
number_of_draughtproofed_external_doors: int
number_of_open_chimneys: int
number_of_blocked_chimneys: int
number_of_fixed_incandescent_bulbs: int
exact_led_cfl_count_known: bool
number_of_fixed_led_bulbs: int
number_of_fixed_cfl_bulbs: int
waste_water_heat_recovery: str
number_of_heated_rooms: Optional[int] = None
@dataclass
class Shower:
id: int
outlet_type: str
@dataclass
class WaterUse:
number_of_baths: int
number_of_special_features: int
showers: List[Shower]
@dataclass
class CustomerResponse:
customer_present: bool
willing_to_answer_satisfaction_survey: bool
@dataclass
class SurveyAddendum:
addendum: str
related_party_disclosure: str
hard_to_treat_cavity_access_issues: bool
hard_to_treat_cavity_high_exposure: bool
hard_to_treat_cavity_narrow_cavities: bool
@dataclass
class PasHubRdSapSiteNotes:
inspection_metadata: InspectionMetadata
general: General
building_construction: BuildingConstruction
building_measurements: BuildingMeasurements
roof_space: RoofSpace
windows: List[Window]
heating_and_hot_water: HeatingAndHotWater
ventilation: Ventilation
conservatories: Conservatories
renewables: Renewables
room_count_elements: RoomCountElements
water_use: WaterUse
customer_response: CustomerResponse
addendum: SurveyAddendum

View file

View file

@ -0,0 +1,232 @@
{
"inspection_metadata": {
"inspection_surveyor": "test",
"email_address": "test@test.com",
"report_reference": "49D422A9-0779-44DD-9665-464D35DFF1A8",
"created_on": "2026-03-31",
"date_of_inspection": "2026-03-31",
"property_address": "test"
},
"general": {
"epc_checked_before_assessment": true,
"epc_exists_at_point_of_assessment": false,
"inspection_date": "2026-03-31",
"transaction_type": "None of the Above",
"tenure": "Rented Social",
"property_type": "House",
"detachment_type": "Mid-terrace",
"number_of_storeys": 2,
"terrain_type": "Suburban",
"number_of_extensions": 0,
"electricity_smart_meter": true,
"electric_meter_type": "Single",
"dwelling_export_capable": true,
"mains_gas_available": true,
"gas_smart_meter": true,
"gas_meter_accessible": true,
"measurements_location": "Internal"
},
"building_construction": {
"main_building": {
"age_range": "I: 1996 - 2002",
"age_indicators": "local knowledge",
"walls_construction_type": "Cavity",
"cavity_construction_indicators": "stretcher bond",
"walls_insulation_type": "As built",
"thermal_conductivity_of_wall_insulation": "Unknown",
"wall_u_value_known": false,
"wall_thickness_mm": 280,
"party_wall_construction_type": "Cavity Masonry, Unfilled"
},
"floor": {
"floor_type": "Ground Floor",
"floor_construction": "Suspended, not timber",
"floor_insulation_type": "As Built",
"floor_u_value_known": false
}
},
"building_measurements": {
"main_building": {
"floors": [
{
"name": "Floor 1",
"area_m2": 24.78,
"height_m": 2.37,
"heat_loss_perimeter_m": 14.21,
"pwl_m": 6.15
},
{
"name": "Floor 0",
"area_m2": 24.78,
"height_m": 2.35,
"heat_loss_perimeter_m": 14.21,
"pwl_m": 6.15
}
]
}
},
"roof_space": {
"main_building": {
"construction_type": "Pitched roof (Slates or tiles), Access to loft",
"insulation_at": "Joists",
"roof_u_value_known": false,
"insulation_thickness_mm": 100,
"cavity_wall_construction_indicators": "No indicator of construction visible",
"rooms_in_roof": false
}
},
"windows": [
{
"id": 1,
"location": "Main Building",
"wall_type": "External wall",
"glazing_type": "Double glazing, Unknown install date",
"window_type": "Window",
"frame_type": "Wooden or PVC",
"glazing_gap": "16 mm or more",
"draught_proofed": true,
"permanent_shutters": false,
"height_m": 1.36,
"width_m": 1.0,
"orientation": "South East"
},
{
"id": 2,
"location": "Main Building",
"wall_type": "External wall",
"glazing_type": "Double glazing, Unknown install date",
"window_type": "Window",
"frame_type": "Wooden or PVC",
"glazing_gap": "16 mm or more",
"draught_proofed": true,
"permanent_shutters": false,
"height_m": 1.33,
"width_m": 0.96,
"orientation": "South East"
},
{
"id": 3,
"location": "Main Building",
"wall_type": "External wall",
"glazing_type": "Double glazing, Unknown install date",
"window_type": "Window",
"frame_type": "Wooden or PVC",
"glazing_gap": "16 mm or more",
"draught_proofed": true,
"permanent_shutters": false,
"height_m": 1.04,
"width_m": 0.96,
"orientation": "North West"
},
{
"id": 4,
"location": "Main Building",
"wall_type": "External wall",
"glazing_type": "Double glazing, Unknown install date",
"window_type": "Window",
"frame_type": "Wooden or PVC",
"glazing_gap": "16 mm or more",
"draught_proofed": true,
"permanent_shutters": false,
"height_m": 1.02,
"width_m": 0.97,
"orientation": "North West"
}
],
"heating_and_hot_water": {
"main_heating": {
"selection_method": "PCDF Search",
"system_type": "Boiler with radiators or underfloor heating",
"product_id": 18400,
"manufacturer": "Vaillant",
"model": "ecoFIT sustain 415",
"orig_manufacturer": "Vaillant",
"fuel": "Mains gas",
"summer_efficiency": 0,
"type": "Regular",
"condensing": true,
"year": "2018 - current",
"mount": "Wall",
"open_flue": "Room-sealed",
"fan_assist": true,
"status": "Normal status for an actual product",
"central_heating_pump_age": "Unknown",
"controls": "Programmer, room thermostat and TRVs",
"flue_gas_heat_recovery_system": false,
"weather_compensator": false,
"emitter": "Radiators",
"emitter_temperature": "Unknown"
},
"secondary_heating": {
"secondary_fuel": "No Secondary Heating"
},
"water_heating": {
"type": "Regular",
"system": "From main heating 1",
"cylinder_size": "Normal (90-130 litres)",
"cylinder_measured_heat_loss": "Not known",
"insulation_type": "Factory fitted",
"insulation_thickness_mm": 12,
"has_thermostat": true
}
},
"ventilation": {
"ventilation_type": "Natural",
"has_fixed_air_conditioning": false,
"number_of_open_flues": 0,
"number_of_closed_flues": 0,
"number_of_boiler_flues": 0,
"number_of_other_flues": 0,
"number_of_extract_fans": 2,
"number_of_passive_vents": 0,
"number_of_flueless_gas_fires": 0,
"pressure_test": "No test",
"draught_lobby": false
},
"conservatories": {
"has_conservatory": false
},
"renewables": {
"wind_turbines": false,
"solar_hot_water": false,
"photovoltaic_array": false,
"number_of_pv_batteries": 0,
"hydro": false
},
"room_count_elements": {
"number_of_habitable_rooms": 2,
"any_unheated_rooms": true,
"number_of_heated_rooms": 0,
"number_of_external_doors": 2,
"number_of_insulated_external_doors": 0,
"number_of_draughtproofed_external_doors": 2,
"number_of_open_chimneys": 0,
"number_of_blocked_chimneys": 0,
"number_of_fixed_incandescent_bulbs": 0,
"exact_led_cfl_count_known": true,
"number_of_fixed_led_bulbs": 5,
"number_of_fixed_cfl_bulbs": 4,
"waste_water_heat_recovery": "None"
},
"water_use": {
"number_of_baths": 1,
"number_of_special_features": 0,
"showers": [
{
"id": 1,
"outlet_type": "Non-Electric Shower"
}
]
},
"customer_response": {
"customer_present": true,
"willing_to_answer_satisfaction_survey": false
},
"addendum": {
"addendum": "PV Recommended",
"related_party_disclosure": "No related party",
"hard_to_treat_cavity_access_issues": false,
"hard_to_treat_cavity_high_exposure": false,
"hard_to_treat_cavity_narrow_cavities": false
}
}

View file

@ -0,0 +1,330 @@
{
"inspection_metadata": {
"inspection_surveyor": "test",
"email_address": "test@test.com",
"report_reference": "6EA2A86D-94CE-4792-8D49-AB495C744EDD",
"created_on": "2025-11-10",
"date_of_inspection": "2025-09-25",
"property_address": "test",
"property_photo": true
},
"general": {
"epc_checked_before_assessment": true,
"epc_exists_at_point_of_assessment": false,
"inspection_date": "2025-09-25",
"transaction_type": "Grant-Scheme (ECO, RHI, etc.)",
"tenure": "Rented Social",
"property_type": "House",
"detachment_type": "Mid-terrace",
"number_of_storeys": 2,
"terrain_type": "Suburban",
"number_of_extensions": 1,
"electricity_smart_meter": true,
"electric_meter_type": "Single",
"dwelling_export_capable": true,
"mains_gas_available": true,
"gas_smart_meter": true,
"gas_meter_accessible": true,
"measurements_location": "Internal"
},
"building_construction": {
"main_building": {
"age_range": "1950-1966",
"age_indicators": "local knowledge, enquiries of owner",
"walls_construction_type": "Cavity",
"cavity_construction_indicators": "wall thickness over 270 mm",
"walls_insulation_type": "Filled Cavity",
"filled_cavity_indicators": "evidence of cavity fill drill holes",
"thermal_conductivity_of_wall_insulation": "Unknown",
"wall_u_value_known": false,
"wall_thickness_mm": 310,
"party_wall_construction_type": "Cavity Masonry, Filled"
},
"extensions": [
{
"id": 1,
"age_range": "2003-2006",
"age_indicators": "local knowledge, enquiries of owner",
"walls_construction_type": "Cavity",
"cavity_construction_indicators": "wall thickness over 270 mm",
"walls_insulation_type": "As built",
"thermal_conductivity_of_wall_insulation": "Unknown",
"wall_u_value_known": false,
"wall_thickness_mm": 310,
"party_wall_construction_type": "Cavity Masonry, Filled"
}
],
"floor": {
"floor_type": "Ground Floor",
"floor_construction": "Solid",
"floor_insulation_type": "As Built",
"floor_u_value_known": false
}
},
"building_measurements": {
"main_building": {
"floors": [
{
"name": "Floor 1",
"area_m2": 35.68,
"height_m": 2.19,
"heat_loss_perimeter_m": 13.44,
"pwl_m": 10.62
},
{
"name": "Floor 0",
"area_m2": 35.68,
"height_m": 2.17,
"heat_loss_perimeter_m": 11.0,
"pwl_m": 10.62
}
]
},
"extensions": [
{
"id": 1,
"floors": [
{
"name": "Floor 0",
"area_m2": 3.8,
"height_m": 2.0,
"heat_loss_perimeter_m": 5.7,
"pwl_m": 0.0
}
]
}
]
},
"roof_space": {
"main_building": {
"construction_type": "Pitched roof (Slates or tiles), Access to loft",
"insulation_at": "Joists",
"roof_u_value_known": false,
"insulation_thickness_mm": 100,
"cavity_wall_construction_indicators": "cavity visible in roof space",
"rooms_in_roof": false
},
"extensions": [
{
"id": 1,
"construction_type": "Pitched roof, Sloping ceiling",
"insulation_at": "Sloping ceiling insulation",
"roof_u_value_known": false,
"insulation_thickness": "As built",
"cavity_wall_construction_indicators": "No indicator of construction visible",
"rooms_in_roof": false
}
]
},
"windows": [
{
"id": 1,
"location": "Main Building",
"wall_type": "External wall",
"glazing_type": "Double glazing, Unknown install date",
"window_type": "Window",
"frame_type": "Wooden or PVC",
"glazing_gap": "16 mm or more",
"draught_proofed": true,
"permanent_shutters": false,
"height_m": 1.2,
"width_m": 2.3,
"orientation": "North West"
},
{
"id": 2,
"location": "Main Building",
"wall_type": "External wall",
"glazing_type": "Double glazing, Unknown install date",
"window_type": "Window",
"frame_type": "Wooden or PVC",
"glazing_gap": "16 mm or more",
"draught_proofed": true,
"permanent_shutters": false,
"height_m": 1.2,
"width_m": 1.0,
"orientation": "North West"
},
{
"id": 3,
"location": "Main Building",
"wall_type": "External wall",
"glazing_type": "Double glazing, Unknown install date",
"window_type": "Window",
"frame_type": "Wooden or PVC",
"glazing_gap": "16 mm or more",
"draught_proofed": true,
"permanent_shutters": false,
"height_m": 0.9,
"width_m": 1.0,
"orientation": "North East"
},
{
"id": 4,
"location": "Extension 1",
"wall_type": "External wall",
"glazing_type": "Double glazing, Unknown install date",
"window_type": "Window",
"frame_type": "Wooden or PVC",
"glazing_gap": "16 mm or more",
"draught_proofed": true,
"permanent_shutters": false,
"height_m": 0.9,
"width_m": 1.0,
"orientation": "North"
},
{
"id": 5,
"location": "Extension 1",
"wall_type": "External wall",
"glazing_type": "Double glazing, Unknown install date",
"window_type": "Window",
"frame_type": "Wooden or PVC",
"glazing_gap": "16 mm or more",
"draught_proofed": true,
"permanent_shutters": false,
"height_m": 0.9,
"width_m": 1.7,
"orientation": "North East"
},
{
"id": 6,
"location": "Extension 1",
"wall_type": "External wall",
"glazing_type": "Double glazing, Unknown install date",
"window_type": "Window",
"frame_type": "Wooden or PVC",
"glazing_gap": "16 mm or more",
"draught_proofed": true,
"permanent_shutters": false,
"height_m": 0.9,
"width_m": 2.3,
"orientation": "North West"
},
{
"id": 7,
"location": "Extension 1",
"wall_type": "External wall",
"glazing_type": "Double glazing, Unknown install date",
"window_type": "Window",
"frame_type": "Wooden or PVC",
"glazing_gap": "16 mm or more",
"draught_proofed": true,
"permanent_shutters": false,
"height_m": 1.0,
"width_m": 1.2,
"orientation": "North West"
},
{
"id": 8,
"location": "Extension 1",
"wall_type": "External wall",
"glazing_type": "Double glazing, Unknown install date",
"window_type": "Window",
"frame_type": "Wooden or PVC",
"glazing_gap": "16 mm or more",
"draught_proofed": true,
"permanent_shutters": false,
"height_m": 0.9,
"width_m": 1.0,
"orientation": "North East"
}
],
"heating_and_hot_water": {
"main_heating": {
"selection_method": "PCDF Search",
"system_type": "Boiler with radiators or underfloor heating",
"product_id": 16839,
"manufacturer": "Vaillant",
"model": "ecoTEC pro 28",
"orig_manufacturer": "Vaillant",
"fuel": "Mains gas",
"summer_efficiency": 0,
"type": "Combi",
"condensing": true,
"year": "2005 - 2015",
"mount": "Wall",
"open_flue": "Room-sealed",
"fan_assist": true,
"status": "Normal status for an actual product",
"central_heating_pump_age": "Unknown",
"controls": "Programmer, room thermostat and TRVs",
"flue_gas_heat_recovery_system": false,
"weather_compensator": false,
"emitter": "Radiators",
"emitter_temperature": "Unknown"
},
"secondary_heating": {
"secondary_fuel": "No Secondary Heating"
},
"water_heating": {
"type": "Regular",
"system": "From main heating 1",
"cylinder_size": "No Cylinder",
"cylinder_measured_heat_loss": null,
"insulation_type": null,
"insulation_thickness_mm": null,
"has_thermostat": null
}
},
"ventilation": {
"ventilation_type": "Mechanical Extract - Decentralised",
"ventilation_in_pcdf_database": false,
"has_fixed_air_conditioning": false,
"number_of_open_flues": 0,
"number_of_closed_flues": 0,
"number_of_boiler_flues": 0,
"number_of_other_flues": 0,
"number_of_extract_fans": 0,
"number_of_passive_vents": 0,
"number_of_flueless_gas_fires": 0,
"pressure_test": "No test",
"draught_lobby": false
},
"conservatories": {
"has_conservatory": false
},
"renewables": {
"wind_turbines": false,
"solar_hot_water": false,
"photovoltaic_array": false,
"number_of_pv_batteries": 0,
"hydro": false
},
"room_count_elements": {
"number_of_habitable_rooms": 3,
"any_unheated_rooms": false,
"number_of_heated_rooms": null,
"number_of_external_doors": 2,
"number_of_insulated_external_doors": 0,
"number_of_draughtproofed_external_doors": 2,
"number_of_open_chimneys": 0,
"number_of_blocked_chimneys": 0,
"number_of_fixed_incandescent_bulbs": 4,
"exact_led_cfl_count_known": true,
"number_of_fixed_led_bulbs": 0,
"number_of_fixed_cfl_bulbs": 1,
"waste_water_heat_recovery": "None"
},
"water_use": {
"number_of_baths": 1,
"number_of_special_features": 0,
"showers": [
{
"id": 1,
"outlet_type": "Non-Electric Shower"
}
]
},
"customer_response": {
"customer_present": true,
"willing_to_answer_satisfaction_survey": false
},
"addendum": {
"addendum": "None",
"related_party_disclosure": "No related party",
"hard_to_treat_cavity_access_issues": false,
"hard_to_treat_cavity_high_exposure": false,
"hard_to_treat_cavity_narrow_cavities": false
}
}

View file

@ -0,0 +1,366 @@
import json
import os
from typing import Any, Dict
import pytest
from datatypes.epc.schema.tests.helpers import from_dict
from datatypes.epc.surveys.pashub_rdsap_site_notes import (
ExtensionConstruction,
ExtensionMeasurements,
ExtensionRoofSpace,
PasHubRdSapSiteNotes,
)
FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures")
def load(filename: str) -> Dict[str, Any]:
with open(os.path.join(FIXTURES, filename)) as f:
return json.load(f) # type: ignore[no-any-return]
class TestExample1:
"""No extensions; regular boiler with hot water cylinder; natural ventilation."""
@pytest.fixture
def survey(self) -> PasHubRdSapSiteNotes:
return from_dict(
PasHubRdSapSiteNotes, load("pashub_rdsap_site_notes_example1.json")
)
# --- inspection_metadata ---
def test_report_reference(self, survey: PasHubRdSapSiteNotes) -> None:
assert (
survey.inspection_metadata.report_reference
== "49D422A9-0779-44DD-9665-464D35DFF1A8"
)
def test_created_on(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.inspection_metadata.created_on == "2026-03-31"
def test_property_photo_absent(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.inspection_metadata.property_photo is None
# --- general ---
def test_transaction_type(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.general.transaction_type == "None of the Above"
def test_number_of_extensions(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.general.number_of_extensions == 0
def test_smart_meters(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.general.electricity_smart_meter is True
assert survey.general.gas_smart_meter is True
# --- building_construction ---
def test_main_building_wall_thickness(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.building_construction.main_building.wall_thickness_mm == 280
def test_main_building_walls_insulation_type(
self, survey: PasHubRdSapSiteNotes
) -> None:
assert (
survey.building_construction.main_building.walls_insulation_type
== "As built"
)
def test_filled_cavity_indicators_absent(
self, survey: PasHubRdSapSiteNotes
) -> None:
assert (
survey.building_construction.main_building.filled_cavity_indicators is None
)
def test_no_extensions_in_construction(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.building_construction.extensions is None
def test_floor_construction(self, survey: PasHubRdSapSiteNotes) -> None:
assert (
survey.building_construction.floor.floor_construction
== "Suspended, not timber"
)
# --- building_measurements ---
def test_main_building_floor_count(self, survey: PasHubRdSapSiteNotes) -> None:
assert len(survey.building_measurements.main_building.floors) == 2
def test_floor_area(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.building_measurements.main_building.floors[0].area_m2 == 24.78
def test_no_extension_measurements(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.building_measurements.extensions is None
# --- roof_space ---
def test_roof_insulation_thickness_mm(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.roof_space.main_building.insulation_thickness_mm == 100
def test_roof_insulation_thickness_string_absent(
self, survey: PasHubRdSapSiteNotes
) -> None:
assert survey.roof_space.main_building.insulation_thickness is None
def test_no_extension_roof_spaces(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.roof_space.extensions is None
def test_rooms_in_roof_false(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.roof_space.main_building.rooms_in_roof is False
# --- windows ---
def test_window_count(self, survey: PasHubRdSapSiteNotes) -> None:
assert len(survey.windows) == 4
def test_window_dimensions(self, survey: PasHubRdSapSiteNotes) -> None:
w = survey.windows[0]
assert w.height_m == 1.36
assert w.width_m == 1.0
def test_window_orientation(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.windows[0].orientation == "South East"
assert survey.windows[2].orientation == "North West"
def test_window_glazing_gap(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.windows[0].glazing_gap == "16 mm or more"
# --- heating_and_hot_water ---
def test_main_heating_manufacturer(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.heating_and_hot_water.main_heating.manufacturer == "Vaillant"
def test_main_heating_model(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.heating_and_hot_water.main_heating.model == "ecoFIT sustain 415"
def test_main_heating_product_id(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.heating_and_hot_water.main_heating.product_id == 18400
def test_main_heating_type_regular(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.heating_and_hot_water.main_heating.type == "Regular"
def test_water_heating_cylinder_present(self, survey: PasHubRdSapSiteNotes) -> None:
wh = survey.heating_and_hot_water.water_heating
assert wh.cylinder_size == "Normal (90-130 litres)"
assert wh.insulation_type == "Factory fitted"
assert wh.insulation_thickness_mm == 12
assert wh.has_thermostat is True
def test_secondary_heating_none(self, survey: PasHubRdSapSiteNotes) -> None:
assert (
survey.heating_and_hot_water.secondary_heating.secondary_fuel
== "No Secondary Heating"
)
# --- ventilation ---
def test_ventilation_type(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.ventilation.ventilation_type == "Natural"
def test_ventilation_pcdf_absent(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.ventilation.ventilation_in_pcdf_database is None
def test_extract_fans(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.ventilation.number_of_extract_fans == 2
# --- room_count_elements ---
def test_habitable_rooms(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.room_count_elements.number_of_habitable_rooms == 2
def test_heated_rooms_zero(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.room_count_elements.number_of_heated_rooms == 0
def test_led_bulbs(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.room_count_elements.number_of_fixed_led_bulbs == 5
def test_cfl_bulbs(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.room_count_elements.number_of_fixed_cfl_bulbs == 4
# --- water_use ---
def test_shower_outlet_type(self, survey: PasHubRdSapSiteNotes) -> None:
assert len(survey.water_use.showers) == 1
assert survey.water_use.showers[0].outlet_type == "Non-Electric Shower"
# --- addendum ---
def test_addendum_value(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.addendum.addendum == "PV Recommended"
def test_related_party_disclosure(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.addendum.related_party_disclosure == "No related party"
def test_hard_to_treat_flags_false(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.addendum.hard_to_treat_cavity_access_issues is False
assert survey.addendum.hard_to_treat_cavity_high_exposure is False
assert survey.addendum.hard_to_treat_cavity_narrow_cavities is False
class TestExample2:
"""With extensions; combi boiler (no cylinder); mechanical extract ventilation."""
@pytest.fixture
def survey(self) -> PasHubRdSapSiteNotes:
return from_dict(
PasHubRdSapSiteNotes, load("pashub_rdsap_site_notes_example2.json")
)
# --- inspection_metadata ---
def test_report_reference(self, survey: PasHubRdSapSiteNotes) -> None:
assert (
survey.inspection_metadata.report_reference
== "6EA2A86D-94CE-4792-8D49-AB495C744EDD"
)
def test_property_photo_present(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.inspection_metadata.property_photo is True
def test_created_on_differs_from_inspection_date(
self, survey: PasHubRdSapSiteNotes
) -> None:
assert survey.inspection_metadata.created_on == "2025-11-10"
assert survey.inspection_metadata.date_of_inspection == "2025-09-25"
# --- general ---
def test_number_of_extensions(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.general.number_of_extensions == 1
def test_transaction_type(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.general.transaction_type == "Grant-Scheme (ECO, RHI, etc.)"
# --- building_construction ---
def test_main_building_filled_cavity_indicators(
self, survey: PasHubRdSapSiteNotes
) -> None:
assert (
survey.building_construction.main_building.filled_cavity_indicators
== "evidence of cavity fill drill holes"
)
def test_extension_construction_present(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.building_construction.extensions is not None
assert len(survey.building_construction.extensions) == 1
def test_extension_construction_type(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.building_construction.extensions is not None
ext: ExtensionConstruction = survey.building_construction.extensions[0]
assert ext.id == 1
assert ext.walls_insulation_type == "As built"
assert ext.wall_thickness_mm == 310
def test_extension_no_filled_cavity_indicators(
self, survey: PasHubRdSapSiteNotes
) -> None:
assert survey.building_construction.extensions is not None
assert (
survey.building_construction.extensions[0].filled_cavity_indicators is None
)
# --- building_measurements ---
def test_main_building_floor_count(self, survey: PasHubRdSapSiteNotes) -> None:
assert len(survey.building_measurements.main_building.floors) == 2
def test_extension_measurements_present(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.building_measurements.extensions is not None
assert len(survey.building_measurements.extensions) == 1
def test_extension_floor_area(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.building_measurements.extensions is not None
ext: ExtensionMeasurements = survey.building_measurements.extensions[0]
assert ext.id == 1
assert len(ext.floors) == 1
assert ext.floors[0].area_m2 == 3.8
# --- roof_space ---
def test_main_roof_insulation_thickness_mm(
self, survey: PasHubRdSapSiteNotes
) -> None:
assert survey.roof_space.main_building.insulation_thickness_mm == 100
def test_extension_roof_spaces_present(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.roof_space.extensions is not None
assert len(survey.roof_space.extensions) == 1
def test_extension_roof_uses_string_thickness(
self, survey: PasHubRdSapSiteNotes
) -> None:
assert survey.roof_space.extensions is not None
ext: ExtensionRoofSpace = survey.roof_space.extensions[0]
assert ext.insulation_thickness == "As built"
assert ext.insulation_thickness_mm is None
def test_extension_roof_construction_type(
self, survey: PasHubRdSapSiteNotes
) -> None:
assert survey.roof_space.extensions is not None
assert (
survey.roof_space.extensions[0].construction_type
== "Pitched roof, Sloping ceiling"
)
# --- windows ---
def test_window_count(self, survey: PasHubRdSapSiteNotes) -> None:
assert len(survey.windows) == 8
def test_extension_windows(self, survey: PasHubRdSapSiteNotes) -> None:
extension_windows = [w for w in survey.windows if w.location == "Extension 1"]
assert len(extension_windows) == 5
def test_window_ids_sequential(self, survey: PasHubRdSapSiteNotes) -> None:
ids = [w.id for w in survey.windows]
assert ids == list(range(1, 9))
# --- heating_and_hot_water ---
def test_main_heating_type_combi(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.heating_and_hot_water.main_heating.type == "Combi"
def test_main_heating_model(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.heating_and_hot_water.main_heating.model == "ecoTEC pro 28"
def test_water_heating_no_cylinder(self, survey: PasHubRdSapSiteNotes) -> None:
wh = survey.heating_and_hot_water.water_heating
assert wh.cylinder_size == "No Cylinder"
assert wh.cylinder_measured_heat_loss is None
assert wh.insulation_type is None
assert wh.insulation_thickness_mm is None
assert wh.has_thermostat is None
# --- ventilation ---
def test_ventilation_type(self, survey: PasHubRdSapSiteNotes) -> None:
assert (
survey.ventilation.ventilation_type == "Mechanical Extract - Decentralised"
)
def test_ventilation_in_pcdf_database(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.ventilation.ventilation_in_pcdf_database is False
def test_no_extract_fans_for_mev(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.ventilation.number_of_extract_fans == 0
# --- room_count_elements ---
def test_habitable_rooms(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.room_count_elements.number_of_habitable_rooms == 3
def test_heated_rooms_null(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.room_count_elements.number_of_heated_rooms is None
def test_incandescent_bulbs(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.room_count_elements.number_of_fixed_incandescent_bulbs == 4
# --- addendum ---
def test_addendum_none(self, survey: PasHubRdSapSiteNotes) -> None:
assert survey.addendum.addendum == "None"

View file

@ -77,14 +77,8 @@ class KwhData:
'Cheapest tariff (Large legacy suppliers)', 'Cheapest tariff (All suppliers)',
'Cheapest tariff (Basket)', 'Default tariff cap level']
# Extract data rows
data_rows = []
for row in data[1:]:
date = row['\ufeff"']
values = row[None]
data_rows.append([date] + values)
self.retail_price_comparison = pd.DataFrame(data_rows, columns=header)
self.retail_price_comparison = pd.DataFrame(data)
self.retail_price_comparison.columns = header
self.retail_price_comparison['Date'] = pd.to_datetime(self.retail_price_comparison['Date'], errors='coerce')
@staticmethod

View file

@ -12,8 +12,8 @@ from backend.app.db.models.recommendations import (
from backend.app.db.models.portfolio import PropertyModel, PropertyDetailsEpcModel
from backend.app.utils import sap_to_epc
from typing import Dict, List, Set
from datatypes.epc.domain.epc import Epc
from recommendations.Costs import Costs
from backend.app.db.models.portfolio import Epc
pd.set_option("display.max_rows", 500)
pd.set_option("display.max_columns", 500)

View file

@ -3,6 +3,6 @@ pythonpath = .
log_cli = true
log_cli_level = INFO
addopts = --cov-report term-missing --cov=etl/epc --cov=recommendations --cov=backend --cov=etl/epc_clean --cov=etl/spatial
testpaths = recommendations/tests backend/tests etl/epc/tests etl/epc_clean/tests etl/spatial/tests backend/condition/tests backend/address2UPRN/tests backend/onboarders/tests backend/categorisation/tests backend/export/tests etl/hubspot/tests backend/hubspot_trigger_orchestrator/tests datatypes/epc/schema/tests
testpaths = recommendations/tests backend/tests etl/epc/tests etl/epc_clean/tests etl/spatial/tests backend/condition/tests backend/address2UPRN/tests backend/onboarders/tests backend/categorisation/tests backend/export/tests etl/hubspot/tests backend/hubspot_trigger_orchestrator/tests datatypes/epc/schema/tests datatypes/epc/surveys/tests datatypes/epc/domain/tests backend/ecmk_fetcher/tests/
markers =
integration: mark a test as an integration test

View file

@ -230,7 +230,7 @@ for scenario_id in SCENARIOS:
# Get recs for this scenario
recommended_measures_df = recommendations_df[
recommendations_df["scenario_id"] == scenario_id
][["property_id", "measure_type", "estimated_cost", "default"]]
][["property_id", "measure_type", "estimated_cost", "default"]]
recommended_measures_df = recommended_measures_df[
recommended_measures_df["default"]
]
@ -238,7 +238,7 @@ for scenario_id in SCENARIOS:
post_install_sap = recommendations_df[
recommendations_df["scenario_id"] == scenario_id
][["property_id", "default", "sap_points"]]
][["property_id", "default", "sap_points"]]
post_install_sap = post_install_sap[post_install_sap["default"]]
# Sum up the sap points by property id
post_install_sap = (
@ -282,6 +282,7 @@ for scenario_id in SCENARIOS:
"windows",
"current_epc_rating",
"current_sap_points",
"original_sap_points",
"total_floor_area",
"number_of_rooms",
"lodgement_date",
@ -303,31 +304,6 @@ for scenario_id in SCENARIOS:
)
df["uprn"] = df["uprn"].astype(str)
relevant_plans = plans_df[plans_df["scenario_id"] == scenario_id]
df2 = df.merge(
relevant_plans[["property_id", "post_sap_points", "post_epc_rating"]],
how="left",
on="property_id",
suffixes=("", "_plan"),
)
print(df2["predicted_post_works_epc"].value_counts())
print(df2["post_epc_rating"].value_counts())
z = df2[
(df2["predicted_post_works_epc"] != "D")
& (df2["post_epc_rating"].astype(str) == "Epc.D")
]
df2["predicted_post_works_epc"].value_counts()
df2["post_epc_rating"].astype(str).value_counts()
df2[df2["total_retrofit_cost"] > 0].shape
getting_works = df[df["total_retrofit_cost"] > 0]
getting_works["predicted_post_works_epc"].value_counts()
df[df["predicted_post_works_sap"] == ""]
# Expected columns list
expected_columns = [
"suspended_floor_insulation",

View file

@ -90,6 +90,41 @@ class DomnaSharepointClient:
file_name, get_file_stream(file_path), sharepoint_path
)
def download_file(self, sharepoint_path: str, local_path: str) -> bool:
"""
Download a file from SharePoint to a local path.
Returns True if the file was downloaded, False if it does not exist yet.
Raises on any other error.
"""
sharepoint_client = SharePointClient(
tenant_id=self.sharepoint_tenant_id,
client_id=self.sharepoint_client_id,
client_secret=self.sharepoint_client_secret,
site_id=self.sharepoint_drive.value,
)
try:
metadata: Dict[str, Any] = sharepoint_client.get_file_metadata(sharepoint_path)
except ValueError:
return False
download_url: Optional[str] = metadata.get("@microsoft.graph.downloadUrl")
if not download_url:
return False
content: BytesIO = SharePointClient.download_sharepoint_file(download_url)
parent_dir = os.path.dirname(local_path)
if parent_dir:
os.makedirs(parent_dir, exist_ok=True)
with open(local_path, "wb") as f:
f.write(content.getvalue())
self.logger.debug(f"Downloaded SharePoint file to: {local_path}")
return True
def create_temp_file(self, content: BytesIO, path: str):
# Ensure the path is under /tmp/
new_path = os.path.join("/tmp/sharepoint", path)

View file

@ -278,6 +278,17 @@ class SharePointClient:
# logger.debug(f"Listing folder contents from URL: {url}")
return "GET", url, None
@api_call_decorator
def get_file_metadata(self, file_path: str) -> Dict[str, Any]:
"""
GET /drives/{drive-id}/root:/{file_path}
Returns file metadata, including '@microsoft.graph.downloadUrl'.
Raises ValueError if the file does not exist (404).
"""
url = f"https://graph.microsoft.com/v1.0/drives/{self.document_drive_id}/root:/{file_path}"
return "GET", url, None
@api_call_decorator
def create_folder(self, file_name: str, folder_path: str) -> Dict[str, Any]:
"""
@ -325,7 +336,7 @@ class SharePointClient:
return self.upload_file(file_name, sharepoint_parent_id, file_stream)
@staticmethod
def download_sharepoint_file(download_url):
def download_sharepoint_file(download_url: str) -> BytesIO:
"""
Downloads a file from the given URL and returns its content.