Merge branch 'main' of github.com:Hestia-Homes/Model into etl-michael

This commit is contained in:
Michael Duong 2025-11-01 15:34:20 +00:00
commit 0e0a473afe
34 changed files with 2623 additions and 711 deletions

View file

@ -309,6 +309,17 @@ class AssetList:
'NAME OF SURVEYOR'
]
# Solar non-intrusive fields
NON_INTRUSIVES_SOLAR_COLNAMES = [
'PV, ACCESS ISSUE, SEE NOTES', 'ROOF ORIENTATION',
'AREA (m²) OF ROOF WHERE PV WILL BE SITUATED ', 'SHADING',
'Roof Tiles - CONCRETE/SLATE/ROSEMARY',
'NO. OF PANELS (Typical size of 420W panel is 1mx1.7m and need 30cm all the way around panels)',
'SCAFFOLD REQUIRED? IF YES, ARE THERE ANY SURROUNDING ACCESS ISSUES - PLEASE DESCRIBE',
'IF PANELS ARE GOING ON REAR PLEASE CHECK FOR SPACE FOR SCAFFOLDING - DESCRIBE ANY ISSUES BELOW',
'DATE', 'NAME OF SURVEYOR'
]
NON_INTRUSIVES_ELIGIBILITY_COLUMN = "Eligibility (Red/Yellow/Green)"
OLD_FORMAT_NON_INTRUSIVE_COLNAMES = ['WFT Findings', 'ECO Eligibility']
@ -461,6 +472,8 @@ class AssetList:
self.new_format_non_insturives_present_v2 = 'TILE HUNG' in self.raw_asset_list.columns
self.solar_non_intrusives_present = "AREA (m²) OF ROOF WHERE PV WILL BE SITUATED" in self.raw_asset_list.columns
# Names of columns
self.landlord_property_id = landlord_property_id
self.address1_colname = address1_colname
@ -774,6 +787,9 @@ class AssetList:
if self.new_format_non_insturives_present_v2:
non_intrusive_columns += self.NON_INTRUSIVES_NEW_FORMAT_COLNAMES_V2
if self.solar_non_intrusives_present:
non_intrusive_columns += self.NON_INTRUSIVES_SOLAR_COLNAMES
if self.old_format_non_intrusives_present:
# We check if we have the ECO Eligibility column, which we might not have
non_intrusive_columns = [
@ -946,7 +962,7 @@ class AssetList:
if self.phase:
# We filter on just the properties that have had an inspection
if self.new_format_non_insturives_present_v2:
if self.new_format_non_insturives_present_v2 or self.solar_non_intrusives_present:
self.standardised_asset_list = self.standardised_asset_list[
~self.standardised_asset_list['NAME OF SURVEYOR'].isin(
["YET TO BE SURVEYED", "", None]
@ -1341,10 +1357,10 @@ class AssetList:
# for identifying cavity jobs
if self.non_intrusives_present and not self.old_format_non_intrusives_present:
if self.new_format_non_insturives_present_v2:
if self.new_format_non_insturives_present_v2 or self.solar_non_intrusives_present:
existing_solar_non_intrusives_check = (
self.standardised_asset_list["non-intrusives: ROOF ORIENTATION"].str.strip().isin(
["ALREADY HAS SOLAR PV"]
["ALREADY HAS SOLAR PV", "ALREADY HAS PV"]
)
)
else:
@ -1783,9 +1799,16 @@ class AssetList:
)
)
not_a_flat = (
self.standardised_asset_list[self.STANDARD_PROPERTY_TYPE] != "flat"
)
# Determine if the client gave us property type in the first place
if all(self.standardised_asset_list[self.STANDARD_PROPERTY_TYPE] == "unknown"):
# Use EPC
not_a_flat = (
self.standardised_asset_list[self.EPC_API_DATA_NAMES["property-type"]] != "Flat"
)
else:
not_a_flat = (
self.standardised_asset_list[self.STANDARD_PROPERTY_TYPE] != "flat"
)
solar_roof_meets_criteria = (
self.standardised_asset_list["solar_epc_roof_insulated"] |
@ -3452,7 +3475,13 @@ class AssetList:
raise ValueError("No installer column found in master data")
measure_mix_col = "MEASURE COMBO"
town_colname = "TOWN" if "TOWN" in master_data.columns else 'Town/Area'
if "TOWN" in master_data.columns:
town_colname = "TOWN"
elif 'Town/Area' in master_data.columns:
town_colname = 'Town/Area'
else:
town_colname = "Town/City"
logger.info("Matching master data to asset list")
matched = []

View file

@ -59,6 +59,278 @@ def app():
Property UPRN
"""
# Stonewater Solar
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Stonewater/October 2025 Solar"
data_filename = "Copy of AP Stonewater Ammended address list - PV AM Amended - Khalim initial review.xlsx"
sheet_name = "Proposed Sheet"
postcode_column = 'Postcode'
address1_column = None
address1_method = "house_number_extraction"
fulladdress_column = "Address"
address_cols_to_concat = []
missing_postcodes_method = None
landlord_year_built = None
landlord_os_uprn = None
landlord_property_type = "Property Type"
landlord_built_form = "Property Type"
landlord_wall_construction = "Walls"
landlord_roof_construction = "Roofs"
landlord_heating_system = "Heating"
landlord_existing_pv = None
landlord_property_id = "Asset Id"
landlord_sap = "SAP"
outcomes_filename = None
outcomes_sheetname = None
outcomes_postcode = None
outcomes_houseno = None
outcomes_id = None
outcomes_address = None
master_filepaths = []
master_id_colnames = []
master_to_asset_list_filepath = None
phase = False
ecosurv_landlords = None
asset_list_header = 0
landlord_block_reference = None
#
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Cambridge/"
data_filename = "22.10_Cambridge_west addresses.xlsx"
sheet_name = "Asset List"
postcode_column = 'Postcode'
address1_column = None
address1_method = "house_number_extraction"
fulladdress_column = "Full Address"
address_cols_to_concat = []
missing_postcodes_method = None
landlord_year_built = None
landlord_os_uprn = None
landlord_property_type = None
landlord_built_form = None
landlord_wall_construction = None
landlord_roof_construction = None
landlord_heating_system = None
landlord_existing_pv = None
landlord_property_id = "id"
landlord_sap = None
outcomes_filename = None
outcomes_sheetname = None
outcomes_postcode = None
outcomes_houseno = None
outcomes_id = None
outcomes_address = None
master_filepaths = []
master_id_colnames = []
master_to_asset_list_filepath = None
phase = False
ecosurv_landlords = None
asset_list_header = 0
landlord_block_reference = None
# Property Box
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/NRLA/Property Box"
data_filename = "Property Box Finance Portfolio.xlsx"
sheet_name = "Sheet1"
postcode_column = 'Postcode'
address1_column = None
address1_method = "house_number_extraction"
fulladdress_column = "Address 1"
address_cols_to_concat = []
missing_postcodes_method = None
landlord_year_built = None
landlord_os_uprn = None
landlord_property_type = None
landlord_built_form = None
landlord_wall_construction = None
landlord_roof_construction = None
landlord_heating_system = None
landlord_existing_pv = None
landlord_property_id = "row_id"
landlord_sap = None
outcomes_filename = None
outcomes_sheetname = None
outcomes_postcode = None
outcomes_houseno = None
outcomes_id = None
outcomes_address = None
master_filepaths = []
master_id_colnames = []
master_to_asset_list_filepath = None
phase = False
ecosurv_landlords = None
asset_list_header = 0
landlord_block_reference = "block_id"
# CDS - able-to-pay
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/CDS/Able to pay"
data_filename = "CDS_ASSET LIST_(2314).xlsx"
sheet_name = "Sheet1"
postcode_column = 'Property Address - Postcode'
address1_column = "Property Address - Line 1"
address1_method = None
fulladdress_column = "Property Address - Line 1"
address_cols_to_concat = []
missing_postcodes_method = None
landlord_year_built = None
landlord_os_uprn = None
landlord_property_type = None
landlord_built_form = None
landlord_wall_construction = None
landlord_roof_construction = None
landlord_heating_system = None
landlord_existing_pv = None
landlord_property_id = "row_id"
landlord_sap = None
outcomes_filename = None
outcomes_sheetname = None
outcomes_postcode = None
outcomes_houseno = None
outcomes_id = None
outcomes_address = None
master_filepaths = []
master_id_colnames = []
master_to_asset_list_filepath = None
phase = False
ecosurv_landlords = None
asset_list_header = 0
landlord_block_reference = None
# Hyde - solar
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Hyde/Solar"
data_filename = "Domna Property Analysis HYDE (Chichester Removed)V2-Completed.xlsx"
sheet_name = "Electric Property Inspections"
postcode_column = 'Postcode'
address1_column = None # Is only patchily populated so we create it
address1_method = 'house_number_extraction'
fulladdress_column = "Address"
address_cols_to_concat = []
missing_postcodes_method = None
landlord_year_built = None
landlord_os_uprn = None
landlord_property_type = "Property Type"
landlord_built_form = "Property Type"
landlord_wall_construction = "Walls "
landlord_roof_construction = "Roofs"
landlord_heating_system = "Heating"
landlord_existing_pv = None
landlord_property_id = "Address ID"
landlord_sap = "SAP"
outcomes_filename = None
outcomes_sheetname = None
outcomes_postcode = None
outcomes_houseno = None
outcomes_id = None
outcomes_address = None
master_filepaths = []
master_id_colnames = []
master_to_asset_list_filepath = None
phase = False
ecosurv_landlords = None
asset_list_header = 0
landlord_block_reference = None
# Hyde cavity
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Hyde/Cavity"
data_filename = "Domna Property Analysis HYDE (Chichester Removed)V2-Completed.xlsx"
sheet_name = "Cavity Inspections"
postcode_column = 'Postcode'
address1_column = None # Is only patchily populated so we create it
address1_method = 'house_number_extraction'
fulladdress_column = "Address"
address_cols_to_concat = []
missing_postcodes_method = None
landlord_year_built = None
landlord_os_uprn = None
landlord_property_type = "Property Type"
landlord_built_form = "Property Type"
landlord_wall_construction = "Walls "
landlord_roof_construction = "Roofs"
landlord_heating_system = "Heating"
landlord_existing_pv = None
landlord_property_id = "Address ID"
landlord_sap = "SAP"
outcomes_filename = None
outcomes_sheetname = None
outcomes_postcode = None
outcomes_houseno = None
outcomes_id = None
outcomes_address = None
master_filepaths = []
master_id_colnames = []
master_to_asset_list_filepath = None
phase = False
ecosurv_landlords = None
asset_list_header = 0
landlord_block_reference = None
# CDS - Sept 2025
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/CDS/September 2025 Programme"
data_filename = "Founder Estates CDS.xlsx"
sheet_name = "Combined List"
postcode_column = 'Postcode'
address1_column = None # Is only patchily populated so we create it
address1_method = 'house_number_extraction'
fulladdress_column = "Address"
address_cols_to_concat = []
missing_postcodes_method = None
landlord_year_built = None
landlord_os_uprn = None
landlord_property_type = "Property Type"
landlord_built_form = None
landlord_wall_construction = None
landlord_roof_construction = None
landlord_heating_system = "Heating Type"
landlord_existing_pv = None
landlord_property_id = "(Do Not Modify) Property"
landlord_sap = None
outcomes_filename = None
outcomes_sheetname = None
outcomes_postcode = None
outcomes_houseno = None
outcomes_id = None
outcomes_address = None
master_filepaths = []
master_id_colnames = []
master_to_asset_list_filepath = None
phase = False
ecosurv_landlords = None
asset_list_header = 0
landlord_block_reference = None
# Project from Nick
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/sfr/Sep2025 Project"
data_filename = "AL Test.xlsx"
sheet_name = "Sheet1"
postcode_column = 'postcode'
address1_column = None
address1_method = 'house_number_extraction'
fulladdress_column = "address"
address_cols_to_concat = []
missing_postcodes_method = None
landlord_year_built = None
landlord_os_uprn = None
landlord_property_type = None
landlord_built_form = None
landlord_wall_construction = None
landlord_roof_construction = None
landlord_heating_system = None
landlord_existing_pv = None
landlord_property_id = "row_id"
landlord_sap = None
outcomes_filename = None
outcomes_sheetname = None
outcomes_postcode = None
outcomes_houseno = None
outcomes_id = None
outcomes_address = None
master_filepaths = []
master_id_colnames = []
master_to_asset_list_filepath = None
phase = False
ecosurv_landlords = None
asset_list_header = 0
landlord_block_reference = None
# Lambeth
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Lambeth"
data_filename = "LAMBETH Asset List ( Incomplete).xlsx"
@ -1307,6 +1579,26 @@ def app():
filename = os.path.join(data_folder, ".".join(data_filename.split(".")[:-1])) + " - Standardised.xlsx"
# Store the data in two tabs. One for the asset list with the EPC data and the second with the flat data
# Determine inspections priority
# solar_jobs = asset_list.standardised_asset_list[~pd.isnull(asset_list.standardised_asset_list["solar_reason"])][
# "domna_postcode"].unique()
# asset_list.standardised_asset_list["in_solar_area"] = asset_list.standardised_asset_list["domna_postcode"].isin(
# solar_jobs
# )
# # Same for cav
# cavity_jobs = asset_list.standardised_asset_list[
# ~pd.isnull(asset_list.standardised_asset_list["cavity_reason"])
# ]["domna_postcode"].unique()
# asset_list.standardised_asset_list["in_cavity_area"] = asset_list.standardised_asset_list["domna_postcode"].isin(
# cavity_jobs
# )
# # We prioritise properties that are in solar areas and cavity areas
# import numpy as np
# asset_list.standardised_asset_list["inspection_priority"] = np.where(
# asset_list.standardised_asset_list["in_solar_area"] | asset_list.standardised_asset_list["in_cavity_area"],
# 1, 2
# )
with pd.ExcelWriter(filename) as writer:
asset_list.standardised_asset_list.to_excel(writer, sheet_name="Standardised Asset List", index=False)
if asset_list.block_analysis_df is not None:

View file

@ -438,6 +438,26 @@ BUILT_FORM_MAPPINGS = {
'Maisonette - Mid Terrace': 'mid-terrace',
'Chalet - Wheelchair': 'unknown',
'Studio Flat': 'unknown',
'Bungalow - Attached': 'semi-detached'
'Bungalow - Attached': 'semi-detached',
'ND': 'unknown',
'Maisonette: Mid Terrace: Mid Floor': 'mid-floor',
'Maisonette: Semi Detached: Ground Floor': 'semi-detached',
'Maisonette: Enclosed Mid Terrace: Ground Floor': 'enclosed mid-terrace',
'Maisonette: Enclosed End Terrace: Ground Floor': 'end-terrace',
'Maisonette: Mid Terrace: Ground Floor': 'mid-terrace',
'Flat: Semi Detached: Basement': 'semi-detached',
'Maisonette: Semi Detached: Top Floor': 'semi-detached',
'Maisonette: Enclosed Mid Terrace: Mid Floor': 'enclosed mid-terrace',
'Flat: Detached: Basement': 'detached',
'Maisonette: Enclosed Mid Terrace: Top Floor': 'enclosed mid-terrace',
'Maisonette: End Terrace: Top Floor': 'top-floor',
'House: Mid Terrace: Ground Floor': 'ground floor',
'Maisonette: Semi Detached: Mid Floor': 'detached',
'Maisonette: Detached: Mid Floor': 'detached',
'Bungalow: EnclosedMidTerrace': 'enclosed mid-terrace',
'House: EnclosedMidTerrace': 'enclosed mid-terrace'
}

View file

@ -473,5 +473,27 @@ HEATING_MAPPINGS = {
'Boiler and radiators, oil': 'oil boiler',
'Boiler and radiators, electric': 'electric boiler',
'No system present: electric heaters assumed': 'electric radiators',
'Boiler and radiators, anthracite': 'solid fuel'
'Boiler and radiators, anthracite': 'solid fuel',
'Heat networks Heat networks (mains gas)': 'communal heating',
'ND Oil': 'oil fuel',
'Boiler Biofuel': 'boiler - other fuel',
'Electric (direct acting) room heaters: Water- or oil-filled radiators': 'room heaters',
'Other: Electric ceiling heating': 'electric ceiling',
'Heat Pump: Electric Heat pumps: Air source heat pump with flow temperature <= 35°C': 'air source heat pump',
'Oil room heaters: Room heater, 2000 or later': 'room heaters',
'Electric Underfloor Heating: In screed above insulation (standard or off peak)': 'electric underfloor',
'Heat Pump: Electric Heat pumps: Air source heat pump in other cases': 'air source heat pump',
'Electric Storage Systems: Old (large volume) storage heaters': 'electric storage heaters',
'Gas (including LPG) room heaters: Condensing gas fire': 'room heaters',
'Solid fuel room heaters: Open fire in grate': 'solid fuel',
'Solid fuel room heaters: Open fire with back boiler (no radiators)': 'solid fuel',
'Community Heating Systems: Community heat pump (RdSAP)': 'communal heating',
'Gas (including LPG) room heaters: Gas fire, open flue, 1980 or later (open fronted), sitting proud of, '
'and sealed to, fireplace opening': 'room heaters',
'Boiler: A rated Regular Boiler, System 2: Boiler: C rated Regular Boiler': 'boiler - other fuel',
'Boiler: G rated Combi': 'gas condensing combi'
}

View file

@ -343,5 +343,25 @@ PROPERTY_MAPPING = {
'bungalow': 'bungalow',
'flat': 'flat',
'FLA': 'flat',
'HOU': 'house'
'HOU': 'house',
'Maisonette: Mid Terrace: Mid Floor': 'maisonette',
'Maisonette: Semi Detached: Ground Floor': 'maisonette',
'Maisonette: Enclosed Mid Terrace: Ground Floor': 'maisonette',
'Maisonette: Enclosed End Terrace: Ground Floor': 'maisonette',
'Maisonette: Mid Terrace: Ground Floor': 'maisonette',
'Flat: Semi Detached: Basement': 'flat',
'Maisonette: Semi Detached: Top Floor': 'maisonette',
'Maisonette: Enclosed Mid Terrace: Mid Floor': 'maisonette',
'Flat: Detached: Basement': 'flat',
'Maisonette: Enclosed Mid Terrace: Top Floor': 'maisonette',
'Maisonette: End Terrace: Top Floor': 'maisonette',
'House: Mid Terrace: Ground Floor': 'house',
'Bungalow: EnclosedMidTerrace': 'bungalow',
'Maisonette: Semi Detached: Mid Floor': 'maisonette',
'Maisonette: Detached: Mid Floor': 'maisonette',
'House: EnclosedMidTerrace': 'house'
}

View file

@ -246,4 +246,59 @@ ROOF_CONSTRUCTION_MAPPINGS = {
'Pitched, 150 mm loft insulation': 'pitched insulated',
'Flat, limited insulation (assumed)': 'flat uninsulated',
'Pitched (no access to loft) 350mm': 'pitched insulated',
'Pitched (no access to loft) 200mm': 'pitched insulated',
'Pitched (access to loft) 200mm': 'pitched insulated',
'Pitched (no access to loft) 250mm': 'pitched insulated',
'Pitched (access to loft) 100mm': 'pitched insulated',
'Another dwelling above ND (inferred)': 'another dwelling above',
'Pitched (no access to loft) N/A': 'pitched no access to loft',
'Pitched (no access to loft) ND (inferred)': 'pitched no access to loft',
'Pitched (no access to loft) 150mm': 'pitched insulated',
'Pitched (access to loft) 400mm+': 'pitched insulated',
'Pitched (no access to loft) 300mm': 'pitched insulated',
'Pitched (access to loft) <25mm': 'pitched less than 100mm insulation',
'Pitched (access to loft) None': 'pitched less than 100mm insulation',
'Pitched (access to loft) 300mm': 'pitched insulated',
'Pitched (access to loft) 50mm': 'pitched less than 100mm insulation',
'Pitched (access to loft) 270mm': 'pitched insulated',
'Pitched (access to loft) Non-joist': 'pitched access to loft',
'Pitched (access to loft) 250mm': 'pitched insulated',
'Another dwelling above N/A': 'another dwelling above',
'Pitched (access to loft) 150mm': 'pitched insulated',
'Pitched (access to loft) ND (inferred)': 'pitched access to loft',
'Pitched (access to loft) 350mm': 'pitched insulated',
'Pitched (access to loft) NR': 'pitched unknown insulation',
'Pitched (access to loft) 75mm': 'pitched less than 100mm insulation',
'Pitched (access to loft) N/A': 'pitched access to loft',
'ND (inferred) 250mm': 'unknown insulated',
'Pitched (vaulted ceiling) Non-joist': 'pitched unknown insulation',
'ND (inferred) ND (inferred)': 'unknown',
'Flat Non-joist': 'flat insulated',
'Same dwelling above N/A': 'another dwelling above',
'Flat: As Built, PitchedNormalLoftAccess: Unknown': 'flat unknown insulation',
'PitchedNormalLoftAccess: Unknown, PitchedNormalNoLoftAccess: Unknown': 'pitched unknown insulation',
'PitchedNormalLoftAccess: 400mm+': 'pitched insulated',
'AnotherDwellingAbove: 150mm': 'another dwelling above',
'Flat: 150mm': 'flat insulated',
'AnotherDwellingAbove: 50mm': 'another dwelling above',
'PitchedNormalNoLoftAccess: As Built': 'pitched no access to loft',
'PitchedNormalLoftAccess: 250mm, PitchedWithSlopingCeiling: As Built': 'pitched insulated',
'PitchedNormalLoftAccess: 200mm, PitchedWithSlopingCeiling: As Built': 'pitched insulated',
'PitchedNormalLoftAccess: 350mm': 'pitched insulated',
'PitchedNormalNoLoftAccess: 270mm': 'pitched no access to loft',
'AnotherDwellingAbove: 100mm': 'another dwelling above',
'PitchedWithSlopingCeiling: Unknown': 'piched unknown insulation',
'AnotherDwellingAbove: Unknown, Flat: As Built': 'another dwelling above',
'Flat: Unknown, PitchedNormalLoftAccess: 25mm': 'flat unknown insulation',
'SameDwellingAbove: Unknown': 'another dwelling above',
'Flat: Unknown': 'flat unknown insulation',
'Flat: 50mm, PitchedNormalLoftAccess: 100mm': 'flat insulated',
'Flat: As Built, PitchedNormalLoftAccess: 250mm, PitchedWithSlopingCeiling: As Built': 'flat unknown insulation',
'Flat: As Built, PitchedNormalLoftAccess: 400mm+': 'flat unknown insulation',
'PitchedWithSlopingCeiling: As Built': 'pitched insulated',
'PitchedNormalLoftAccess: As Built': 'pitched unknown insulation',
}

View file

@ -342,5 +342,18 @@ WALL_CONSTRUCTION_MAPPINGS = {
'Solid brick, as built, partial insulation (assumed)': 'insulated solid brick',
'Sandstone, as built, no insulation (assumed)': 'uninsulated sandstone or limestone',
'System built, as built, partial insulation (assumed)': 'system built unknown insulation',
'Timber frame, with external insulation': 'insulated timber frame'
'Timber frame, with external insulation': 'insulated timber frame',
'Cob As-built': 'cob',
'System built Unknown insulation': 'system built unknown insulation',
'Solid brick Unknown insulation': 'solid brick unknown insulation',
'Timber frame Internal': 'insulated timber frame',
'System built External': 'insulated system built',
'Stone As-built': 'uninsulated sandstone or limestone',
'System built As-built': "uninsulated system built",
'System built Internal': 'insulated system built',
'Cavity: AsBuilt (1976-1982), TimberFrame: AsBuilt': 'cavity unknown insulation',
'Cavity: FilledCavityPlusExternal': 'filled cavity'
}

View file

@ -1,11 +1,14 @@
from enum import Enum
from typing import List
import pandas as pd
from utils.logger import setup_logger
from etl.epc_clean.epc_attributes.MainheatAttributes import MainHeatAttributes
from backend.app.plan.schemas import VALID_HOUSING_TYPES, WALL_INSULATION_MEASURES, ROOF_INSULATION_MEASURES, \
MEASURE_MAP
logger = setup_logger(__name__)
class EligibilityCaveats(Enum):
EPC_RATING = "epc_rating" # EPC requirements not met
@ -578,6 +581,11 @@ class Funding:
return pps.squeeze()["Cost Savings"]
if measure_type == "flat_roof_insulation":
# Not funding for properties starting at C or above
if self.starting_sap_band in ["Low_C", "High_C", "Low_B", "High_B", "Low_A", "High_A"]:
return 0
pps = filtered_pps_matrix[filtered_pps_matrix["Measure_Type"] == "FRI"]
if pps.shape[0] != 1:
raise ValueError("Invalid FRI category")
@ -632,13 +640,25 @@ class Funding:
if self.starting_sap_band in ["Low_C", "High_C", "Low_B", "High_B", "Low_A", "High_A"]:
return 0
pps = filtered_pps_matrix[
(filtered_pps_matrix["Pre_Main_Heating_Source"] == pre_heating_system) &
(filtered_pps_matrix["Post_Main_Heating_Source"] == "Air to Water ASHP") &
(filtered_pps_matrix["Measure_Type"] == "B_Upgrade_nopreHCs")
pps_data = filtered_pps_matrix[
filtered_pps_matrix["Post_Main_Heating_Source"] == "Air to Water ASHP"
]
if pre_heating_system not in pps_data["Pre_Main_Heating_Source"].values:
logger.info(
f"No PPS data for ASHP upgrade from {pre_heating_system}, returning 0"
)
return 0
pps = pps_data[
(pps_data["Pre_Main_Heating_Source"] == pre_heating_system) &
(pps_data["Measure_Type"] == "B_Upgrade_nopreHCs")
# We assume we'll be making a heating system upgrade
]
# Not every pre heating system will result in PPS, e.g. a ground source heat pump to ASHP upgrade
# won't have a PPS.
if pps.shape[0] != 1:
raise ValueError("something went wrong, more than one pps for ashp")
return pps.squeeze()["Cost Savings"]

View file

@ -347,7 +347,8 @@ class SearchEpc:
# We update the data with the correct uprn
if self.uprn:
for x in api_response["response"]["rows"]:
x["uprn"] = self.uprn
if pd.isnull(x["uprn"]):
x["uprn"] = self.uprn
data["rows"].extend(api_response["response"]["rows"])
@ -357,6 +358,8 @@ class SearchEpc:
row for row in data["rows"]
if row["lmk-key"] not in seen and not seen.add(row["lmk-key"])
]
# Overwrite the data
self.data = data
if data["rows"]:
api_response["msg"] = self.SUCCESS
@ -415,7 +418,20 @@ class SearchEpc:
address, [", ".join([r["address"]]) for r in rows], score_cutoff=0
)
# Pick the largest score
if best_match1[1] >= best_match2[1]:
if best_match1[1] == best_match2[1]:
# if thery're the same, we'll work under the assumption that the addresses are the same and we'll
# take whichever has the newest EPC
rows_filtered = [
r for r in rows
if (", ".join([r["address"], r["posttown"]]) == best_match1[0]) or
(r["address"] == best_match2[0])
]
rows_filtered = [
r for r in rows_filtered
if r["lodgement-datetime"] == max([x["lodgement-datetime"] for x in rows_filtered])
]
elif best_match1[1] > best_match2[1]:
# Get all of the scores
rows_filtered = [r for r in rows if ", ".join([r["address"], r["posttown"]]) == best_match1[0]]
else:

View file

@ -332,7 +332,6 @@ class GoogleSolarApi:
)
if solar_product is None:
logger.info("No suitable solar product found for the configuration with %d panels.", total_panels)
continue
total_cost = Costs.solar_pv(
@ -855,18 +854,21 @@ class GoogleSolarApi:
):
continue
solar_api_client = cls(api_key=google_solar_api_key, solar_materials=solar_materials)
if unit["longitude"] is None or unit["latitude"] is None:
# At this point, we've checked that solar PV is valid, and so we provide some defaults
property_instance.set_solar_panel_configuration(
solar_panel_configuration={
"insights_data": None,
"panel_performance": cls.default_panel_performance(property_instance=property_instance),
"panel_performance": solar_api_client.default_panel_performance(
property_instance=property_instance
),
"unit_share_of_energy": 1
},
)
continue
solar_api_client = cls(api_key=google_solar_api_key, solar_materials=solar_materials)
solar_api_client.get(
longitude=unit["longitude"],
latitude=unit["latitude"],

View file

@ -0,0 +1,163 @@
import enum
import pytz
import datetime
from sqlalchemy import (
Column,
BigInteger,
Text,
DateTime,
Enum,
ForeignKey,
)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
# -------------------------------------------------------------------
# ENUM DEFINITIONS (equivalent to drizzle pgEnum calls)
# -------------------------------------------------------------------
class InspectionArchetype(enum.Enum):
BUNGALOW = "Bungalow"
FLAT = "Flat"
MAISONETTE = "Maisonette"
HOUSE = "House"
NON_DOMESTIC = "non-domestic"
class InspectionArchetype2(enum.Enum):
DETACHED = "detached"
MID_TERRACE = "mid-terrace"
ENCLOSED_MID_TERRACE = "enclosed mid-terrace"
END_TERRACE = "end-terrace"
ENCLOSED_END_TERRACE = "enclosed end-terrace"
SEMI_DETACHED = "semi-detached"
class InspectionsWallConstruction(enum.Enum):
CAVITY = "cavity"
SOLID = "solid"
SYSTEM_BUILT = "system built"
TIMBER_FRAMED = "timber framed"
STEEL_FRAMED = "steel framed"
RE_WALLED_CAVITY = "re-walled cavity"
MANSARD_PRE_FAB = "mansard pre-fab"
MANSARD_EWI = "mansard ewi"
MANSARD_RE_WALLED = "mansard re-walled"
class InspectionsWallInsulation(enum.Enum):
EMPTY_CAVITY = "empty cavity"
FILLED_AT_BUILD = "filled at build"
PARTIAL = "partial"
RETRO_DRILLED = "retro drilled"
EWI = "ewi"
IWI = "iwi"
SOLID_NON_CAVITY = "solid non-cavity"
SYSTEM_BUILT = "system built"
TIMBER_FRAMED = "timber framed"
STEEL_FRAMED = "steel framed"
class InspectionsInsulationMaterial(enum.Enum):
EMPTY_50_90 = "empty 50-90"
EMPTY_100_PLUS = "empty 100+"
EMPTY_30_40 = "empty 30-40"
EMPTY_LESS_THAN_30 = "empty less than 30"
LOOSE_FIBRE_WOOL = "loose fibre/wool"
EPS_CELO_KING = "eps/celo/king"
FIBRE_BATTS_WITH_CAVITY = "fibre batts - with cavity"
FIBRE_BATTS_NO_CAVITY = "fibre batts - no cavity"
LOOSE_BEAD = "loose bead"
GLUED_BEAD = "glued bead"
FORMALDEHYDE = "formaldehyde"
BUBBLE_WRAP = "bubble wrap"
POLY_CHUNKS = "poly chunks"
class InspectionBorescoped(enum.Enum):
YES = "yes"
NO = "no"
REFUSED = "refused"
class InspectionsRoofOrientation(enum.Enum):
NORTH = "north"
EAST = "east"
SOUTH = "south"
WEST = "west"
NORTH_EAST = "north-east"
NORTH_WEST = "north-west"
SOUTH_EAST = "south-east"
SOUTH_WEST = "south-west"
N_S_SPLIT = "n/s split"
E_W_SPLIT = "e/w split"
NE_SW_SPLIT = "ne/sw split"
NW_SE_SPLIT = "nw/se split"
FLAT_ROOF = "flat roof"
NO_ROOF = "no roof"
ROOF_TOO_SMALL = "roof too small"
ALREADY_HAS_SOLAR_PV = "already has solar pv"
class InspectionsTileHung(enum.Enum):
YES = "yes"
NO = "no"
FIRST_FLOOR_FLATS_TILE_HUNG = "first floor flats are tile hung"
class InspectionsRendered(enum.Enum):
NO_RENDER = "no render"
INSUFFICIENT_DPC_SPACE = "rendered with “insufficient” space between dpc and render"
SUFFICIENT_DPC_SPACE = "rendered with “sufficient” space between dpc and render"
class InspectionsCladding(enum.Enum):
NONE = "none"
SUFFICIENT_SPACE = "cladded with “sufficient space to fill the wall”"
INSUFFICIENT_SPACE = "cladded with “insufficient space to fill the wall”"
class InspectionsAccessIssues(enum.Enum):
SEE_NOTES = "see notes"
DAMP_ISSUES = "damp issues"
FOLIAGE_ON_WALLS = "foliage on walls"
BUSHES_AGAINST_WALL = "bushes against wall"
TREES_AROUND_ABOVE = "trees around/anove property"
HIGH_RISE = "high rise block flats/maisonettes"
CONSERVATORY = "conservatory"
LEAN_TO = "lean-to"
GARAGE = "garage"
EXTENSION = "extension"
DECKING = "decking"
SHED_AGAINST_WALL = "shed against wall"
class InspectionModel(Base):
__tablename__ = "inspections"
id = Column(BigInteger, primary_key=True, autoincrement=True)
property_id = Column(BigInteger, ForeignKey("property.id"), nullable=False)
archetype = Column(Enum(InspectionArchetype), nullable=True)
archetype_2 = Column(Enum(InspectionArchetype2), nullable=True)
wall_construction = Column(Enum(InspectionsWallConstruction), nullable=True)
insulation = Column(Enum(InspectionsWallInsulation), nullable=True)
insulation_material = Column(Enum(InspectionsInsulationMaterial), nullable=True)
borescoped = Column(Enum(InspectionBorescoped), nullable=True)
roof_orientation = Column(Enum(InspectionsRoofOrientation), nullable=True)
tile_hung = Column(Enum(InspectionsTileHung), nullable=True)
rendered = Column(Enum(InspectionsRendered), nullable=True)
cladding = Column(Enum(InspectionsCladding), nullable=True)
access_issues = Column(Enum(InspectionsAccessIssues), nullable=True)
notes = Column(Text)
surveyor_name = Column(Text)
created_at = Column(
DateTime, nullable=False, default=datetime.datetime.now(pytz.utc)
)
uploaded_at = Column(
DateTime, nullable=False, default=datetime.datetime.now(pytz.utc)
)

View file

@ -19,6 +19,7 @@ class MaterialType(enum.Enum):
flat_roof_insulation = "flat_roof_insulation"
room_roof_insulation = "room_roof_insulation"
windows_glazing = "windows_glazing"
secondary_glazing = "secondary_glazing"
cavity_wall_extraction = "cavity_wall_extraction"
iwi_wall_demolition = "iwi_wall_demolition"
@ -45,6 +46,8 @@ class MaterialType(enum.Enum):
scaffolding = "scaffolding"
high_heat_retention_storage_heaters = "high_heat_retention_storage_heaters"
sealing_fireplace = "sealing_fireplace"
roomstat_programmer_trvs = "roomstat_programmer_trvs"
time_temperature_zone_control = "time_temperature_zone_control"
class DepthUnit(enum.Enum):

View file

@ -145,14 +145,17 @@ def extract_portfolio_aggregation_data(
cost = sum([r["total"] for r in default_recommendations])
sap_point_improvement = sum([r["sap_points"] for r in default_recommendations])
lower_bound_valuation_uplift = (
property_value_increase_ranges[p.id]["lower_bound_increased_value"] -
property_value_increase_ranges[p.id]["current_value"]
)
upper_bound_valuation_uplift = (
property_value_increase_ranges[p.id]["upper_bound_increased_value"] -
property_value_increase_ranges[p.id]["current_value"]
)
if not pd.isnull(property_value_increase_ranges[p.id]["current_value"]):
lower_bound_valuation_uplift = (
property_value_increase_ranges[p.id]["lower_bound_increased_value"] -
property_value_increase_ranges[p.id]["current_value"]
)
upper_bound_valuation_uplift = (
property_value_increase_ranges[p.id]["upper_bound_increased_value"] -
property_value_increase_ranges[p.id]["current_value"]
)
else:
lower_bound_valuation_uplift, upper_bound_valuation_uplift = 0, 0
agg_data.append({
"pre_retrofit_epc": p.data["current-energy-rating"],
@ -484,12 +487,19 @@ async def model_engine(body: PlanTriggerRequest):
plan_input["uprn"] = np.where(plan_input["estimated"].isin([1, True]), None, plan_input["uprn"])
# We handle the landlord property type and built form
plan_input["property_type"] = plan_input["landlord_property_type"].copy()
plan_input["built_form"] = plan_input["landlord_built_form"].copy()
if "landlord_built_form" in plan_input.columns:
plan_input["built_form"] = plan_input["landlord_built_form"].copy()
else:
plan_input["built_form"] = None
plan_input["property_type"] = np.where(
plan_input["property_type"] == "unknown",
plan_input["epc_property_type"],
plan_input["property_type"]
)
if "epc_archetype" not in plan_input.columns:
plan_input["epc_archetype"] = None
plan_input["built_form"] = np.where(
plan_input["built_form"] == "unknown", plan_input["epc_archetype"], plan_input["built_form"]
)
@ -516,6 +526,7 @@ async def model_engine(body: PlanTriggerRequest):
plan_input["built_form"] = plan_input["built_form"].map(built_form_map)
plan_input = plan_input.to_dict("records")
else:
raise ValueError("Other formats not yet supported")
@ -534,11 +545,21 @@ async def model_engine(body: PlanTriggerRequest):
if input_uprns:
# Check for dupes
if len(input_uprns) != len(set(input_uprns)):
raise ValueError("Duplicate UPRNs in the input data")
# Find the duplicate UPRNs
duplicates = set([x for x in input_uprns if input_uprns.count(x) > 1])
# de-dupe input_uprns
raise ValueError(f"Duplicate UPRNs in the input data: {duplicates}")
# If we have patches or overrides, we should read them in here
patches, already_installed, non_invasive_recommendations, valuation_data = get_request_property_data(body)
if body.file_type == "xlsx" and body.file_format == "domna_asset_list":
# We check if we have valution data
if not valuation_data and body.valuation_file_path in [None, ""]:
# We check plan_input
if "domna_valuation" in plan_input[0]:
valuation_data = [{"uprn": x["uprn"], "valuation": x["domna_valuation"]} for x in plan_input]
cleaning_data = read_dataframe_from_s3_parquet(
bucket_name=get_settings().DATA_BUCKET, file_key="sap_change_model/cleaning_dataset.parquet",
)
@ -553,12 +574,22 @@ async def model_engine(body: PlanTriggerRequest):
if uprn:
uprn = int(float(uprn))
address1 = config.get("address", None)
# Handle domna address list format
if pd.isnull(address1) and body.file_format == "domna_asset_list":
address1 = config.get("domna_full_address", None)
address1 = str(int(address1)) if isinstance(address1, float) else str(address1)
full_address = config["domna_full_address"] if body.file_format == "domna_asset_list" else None
epc_searcher = SearchEpc(
address1=str(config["address"]),
address1=address1,
postcode=config["postcode"],
uprn=uprn,
auth_token=get_settings().EPC_AUTH_TOKEN,
os_api_key="",
full_address=full_address
)
epc_searcher.ordnance_survey_client.built_form = config.get("built_form", None)
epc_searcher.ordnance_survey_client.property_type = config.get("property_type", None)
@ -900,7 +931,7 @@ async def model_engine(body: PlanTriggerRequest):
r["uplift_project_score"]
) = funding.get_innovation_uplift(
measure=r,
starting_sap=p.data["current-energy-efficiency"],
starting_sap=int(p.data["current-energy-efficiency"]),
floor_area=p.floor_area,
is_cavity=p.walls["is_cavity_wall"],
current_wall_uvalue=current_wall_u_value,
@ -928,15 +959,19 @@ async def model_engine(body: PlanTriggerRequest):
)
# Given the solutions we select the optimal one
# 1) If the scheme is ECO4, the full project funding and uplift are deducted from the cost
# 2) If the sheme is GBIS, the partial project funding and uplift are deducted from the cost
# 3) Otherwise, no funding is deducted from the cost
solutions["cost_less_full_project_funding"] = np.where(
solutions["scheme"] == "eco4",
solutions["total_cost"] - solutions["full_project_funding"] - solutions["total_uplift"],
solutions["total_cost"] - solutions["partial_project_funding"] - solutions["total_uplift"]
solutions["scheme"] == "none",
solutions["total_cost"],
np.where(
solutions["scheme"] == "eco4",
solutions["total_cost"] - solutions["full_project_funding"] - solutions["total_uplift"],
solutions["total_cost"] - solutions["partial_project_funding"] - solutions["total_uplift"]
)
)
solutions["cost_less_full_project_funding"] = (
solutions["total_cost"] - solutions["full_project_funding"] - solutions["total_uplift"]
)
solutions = solutions.sort_values("cost_less_full_project_funding", ascending=True)
if solutions["meets_upgrade_target"].any():
@ -1166,9 +1201,10 @@ async def model_engine(body: PlanTriggerRequest):
upload_funding(session, p, new_plan_id, recommendations_to_upload)
property_valuation_increases.append(
valuations["average_increased_value"] - valuations["current_value"]
)
if valuations["current_value"] > 0:
property_valuation_increases.append(
valuations["average_increased_value"] - valuations["current_value"]
)
# Commit the session after each batch
session.commit()

View file

@ -219,12 +219,19 @@ class PropertyValuation:
current_epc = property_instance.data["current-energy-rating"]
if not current_value:
# In this case, we return a % improvement rather than an absolute
relative_improvement = cls.estimate_valuation_improvement(
current_value=1,
current_epc=current_epc,
target_epc=target_epc,
total_cost=1
)
return {
"current_value": 0,
"lower_bound_increased_value": 0,
"upper_bound_increased_value": 0,
"average_increased_value": 0,
"average_increase": 0
"lower_bound_increased_value": relative_improvement["lower_bound_increased_value"] - 1,
"upper_bound_increased_value": relative_improvement["upper_bound_increased_value"] - 1,
"average_increased_value": relative_improvement["average_increased_value"] - 1,
"average_increase": relative_improvement["average_increase"]
}
return cls.estimate_valuation_improvement(current_value, current_epc, target_epc, total_cost)

View file

@ -4,7 +4,7 @@ innovation_scenarios = [
# 1) Innovation PV, non-eligible heating system in place, EPC D - not eligible
{
"description": "Innovation PV, non-eligible heating system in place, EPC D",
"measures": [{"type": "solar_pv", "is_innovation": True, "uplift": 0.45}],
"measures": [{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45}],
"starting_sap": 60,
"mainheat_description": "Electric storage heaters",
"heating_control_description": "Manual charge control",
@ -16,7 +16,7 @@ innovation_scenarios = [
# 2) Innovation PV, eligible heating system in place, EPC D - eligible
{
"description": "Innovation PV, eligible heating system in place, EPC D",
"measures": [{"type": "solar_pv", "is_innovation": True, "uplift": 0.45}],
"measures": [{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45}],
"starting_sap": 60,
"mainheat_description": "Air source heat pump, radiators",
"heating_control_description": "Programmer, room thermostat and TRVs",
@ -29,8 +29,8 @@ innovation_scenarios = [
{
"description": "Innovation PV + HHRSH upgrade, EPC E",
"measures": [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "high_heat_retention_storage_heater", "is_innovation": True, "uplift": 0.1}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "high_heat_retention_storage_heater", "is_innovation": False, "innovation_uplift": 0}
],
"starting_sap": 50,
"mainheat_description": "Electric storage heaters",
@ -44,8 +44,8 @@ innovation_scenarios = [
{
"description": "Innovation PV + HHRSH upgrade, EPC E",
"measures": [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "high_heat_retention_storage_heater", "is_innovation": True, "uplift": 0.1}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "high_heat_retention_storage_heater", "is_innovation": False, "innovation_uplift": 0}
],
"starting_sap": 50,
"mainheat_description": "Electric storage heaters",
@ -58,7 +58,7 @@ innovation_scenarios = [
# 5) Innovation PV, needs wall insulation, no wall insulation measure - not eligible
{
"description": "Innovation PV, wall insulation recommended, but not installed",
"measures": [{"type": "solar_pv", "is_innovation": True, "uplift": 0.45}],
"measures": [{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45}],
"starting_sap": 60,
"mainheat_description": "Air source heat pump, radiators",
"heating_control_description": "Programmer, room thermostat and TRVs",
@ -71,8 +71,8 @@ innovation_scenarios = [
{
"description": "Innovation PV, wall insulation recommended and installed",
"measures": [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0.25}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0.25}
],
"starting_sap": 60,
"mainheat_description": "Air source heat pump, radiators",
@ -85,7 +85,7 @@ innovation_scenarios = [
# 7) Innovation PV, needs roof insulation, no roof insulation measure - not eligible
{
"description": "Innovation PV, roof insulation recommended, not installed",
"measures": [{"type": "solar_pv", "is_innovation": True, "uplift": 0.45}],
"measures": [{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45}],
"starting_sap": 60,
"mainheat_description": "Air source heat pump, radiators",
"heating_control_description": "Programmer, room thermostat and TRVs",
@ -98,8 +98,8 @@ innovation_scenarios = [
{
"description": "Innovation PV, roof insulation recommended and installed",
"measures": [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "loft_insulation", "is_innovation": False, "uplift": 0}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "loft_insulation", "is_innovation": False, "innovation_uplift": 0}
],
"starting_sap": 60,
"mainheat_description": "Air source heat pump, radiators",
@ -112,7 +112,7 @@ innovation_scenarios = [
# 9) Innovation PV, needs both roof + wall insulation, no insulation - not eligible
{
"description": "Innovation PV, both insulations recommended, none installed",
"measures": [{"type": "solar_pv", "is_innovation": True, "uplift": 0.45}],
"measures": [{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45}],
"starting_sap": 60,
"mainheat_description": "Air source heat pump, radiators",
"heating_control_description": "Programmer, room thermostat and TRVs",
@ -125,8 +125,8 @@ innovation_scenarios = [
{
"description": "Innovation PV, both insulations recommended, only wall done",
"measures": [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0.25}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0.25}
],
"starting_sap": 60,
"mainheat_description": "Air source heat pump, radiators",
@ -140,8 +140,8 @@ innovation_scenarios = [
{
"description": "Innovation PV, both insulations recommended, only roof done",
"measures": [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "loft_insulation", "is_innovation": False, "uplift": 0}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "loft_insulation", "is_innovation": False, "innovation_uplift": 0}
],
"starting_sap": 60,
"mainheat_description": "Air source heat pump, radiators",
@ -155,9 +155,9 @@ innovation_scenarios = [
{
"description": "Innovation PV, both insulations recommended and installed",
"measures": [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0.25},
{"type": "loft_insulation", "is_innovation": False, "uplift": 0}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0.25},
{"type": "loft_insulation", "is_innovation": False, "innovation_uplift": 0}
],
"starting_sap": 60,
"mainheat_description": "Air source heat pump, radiators",

View file

@ -120,7 +120,7 @@ def test_eco4_prs_eligible_with_swi(
# 3) is getting a solid was measure
# so it's eligible for ECO4
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0}]
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0}]
funding.check_funding(
measures=measures,
starting_sap=50, # EPC E
@ -162,7 +162,7 @@ def test_eco4_prs_not_eligible_high_epc(
tenure="Private",
)
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0}]
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0}]
funding.check_funding(
measures=measures,
starting_sap=72, # EPC C (too high)
@ -203,7 +203,7 @@ def test_gbis_prs_general_eligibility(
tenure="Private",
)
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0}]
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0}]
funding.check_funding(
measures=measures,
starting_sap=65, # EPC D
@ -244,7 +244,7 @@ def test_gbis_prs_low_income_caveat(
tenure="Private",
)
measures = [{"type": "cavity_wall_insulation", "is_innovation": False, "uplift": 0}]
measures = [{"type": "cavity_wall_insulation", "is_innovation": False, "innovation_uplift": 0}]
funding.check_funding(
measures=measures,
starting_sap=60, # EPC D
@ -290,7 +290,7 @@ def test_eco4_sh_epc_e_eligible(
tenure="Social",
)
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0}]
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0}]
funding.check_funding(
measures=measures,
starting_sap=50, # EPC E
@ -330,7 +330,7 @@ def test_eco4_sh_epc_d_requires_innovation(
tenure="Social",
)
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0}]
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0}]
funding.check_funding(
measures=measures,
starting_sap=60, # EPC D
@ -365,7 +365,7 @@ def test_eco4_sh_epc_d_requires_innovation(
gbis_private_solid_abs_rate=28,
tenure="Social",
)
measures2 = [{"type": "internal_wall_insulation", "is_innovation": True, "uplift": 0.25}]
measures2 = [{"type": "internal_wall_insulation", "is_innovation": True, "innovation_uplift": 0.25}]
funding2.check_funding(
measures=measures2,
starting_sap=60, # EPC D
@ -403,7 +403,7 @@ def test_eco4_sh_epc_d_requires_innovation(
gbis_private_solid_abs_rate=28,
tenure="Social",
)
measures3 = [{"type": "solar_pv", "is_innovation": True, "uplift": 0.45}]
measures3 = [{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45}]
funding3.check_funding(
measures=measures3,
starting_sap=60, # EPC D
@ -439,7 +439,7 @@ def test_eco4_sh_epc_d_requires_innovation(
tenure="Social",
)
measures4 = [{"type": "solar_pv", "is_innovation": True, "uplift": 0.45}, ]
measures4 = [{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45}, ]
funding4.check_funding(
measures=measures4,
starting_sap=60, # EPC D
@ -476,8 +476,8 @@ def test_eco4_sh_epc_d_requires_innovation(
)
measures5 = [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "high_heat_retention_storage_heater", "is_innovation": False, "uplift": 0}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "high_heat_retention_storage_heater", "is_innovation": False, "innovation_uplift": 0}
]
funding5.check_funding(
measures=measures5,
@ -516,7 +516,7 @@ def test_eco4_sh_epc_d_requires_innovation(
)
measures6 = [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
]
funding6.check_funding(
measures=measures6,
@ -556,9 +556,9 @@ def test_eco4_sh_epc_d_requires_innovation(
tenure="Social",
)
measures7 = [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "cavity_wall_insulation", "is_innovation": False, "uplift": 0.25},
{"type": "loft_insulation", "is_innovation": False, "uplift": 0}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "cavity_wall_insulation", "is_innovation": False, "innovation_uplift": 0.25},
{"type": "loft_insulation", "is_innovation": False, "innovation_uplift": 0}
]
funding7.check_funding(
measures=measures7,
@ -599,7 +599,7 @@ def test_eco4_sh_solar_pv_requires_heating(
tenure="Social",
)
measures = [{"type": "solar_pv", "is_innovation": True, "uplift": 0.45}]
measures = [{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45}]
funding.check_funding(
measures=measures,
starting_sap=60, # EPC D
@ -641,8 +641,8 @@ def test_eco4_sh_solar_pv_with_heating_is_ok(
)
measures = [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "air_source_heat_pump", "is_innovation": False, "uplift": 0}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "air_source_heat_pump", "is_innovation": False, "innovation_uplift": 0}
]
funding.check_funding(
measures=measures,
@ -684,7 +684,7 @@ def test_eco4_upgrade_requirement_e_to_c_pass(
tenure="Private",
)
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0}]
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0}]
# E (SAP 50) → C (SAP 70) meets upgrade rule
funding.check_funding(
@ -727,7 +727,7 @@ def test_eco4_upgrade_requirement_e_to_d_fail(
tenure="Private",
)
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0}]
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0}]
# E (SAP 50) → D (SAP 65) does NOT meet ECO4 upgrade rule
funding.check_funding(
@ -770,7 +770,7 @@ def test_eco4_upgrade_requirement_f_to_d_pass(
tenure="Private",
)
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0}]
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0}]
# F (SAP 35) → D (SAP 60) is OK for ECO4
funding.check_funding(
@ -813,7 +813,7 @@ def test_eco4_upgrade_requirement_f_to_e_fail(
tenure="Private",
)
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0}]
measures = [{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0}]
# F (SAP 35) → E (SAP 50) does NOT meet ECO4 rule
funding.check_funding(
@ -859,7 +859,7 @@ def test_epc_d_social_no_innovation_no_heating(
)
measures = [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45}
]
funding.check_funding(
@ -905,10 +905,10 @@ def test_epc_d_social_with_heating_and_insulation(
# Should NOT be eligible as the ASHP is not an innovation measure
measures = [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0},
{"type": "loft_insulation", "is_innovation": False, "uplift": 0},
{"type": "air_source_heat_pump", "is_innovation": False, "uplift": 0}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0},
{"type": "loft_insulation", "is_innovation": False, "innovation_uplift": 0},
{"type": "air_source_heat_pump", "is_innovation": False, "innovation_uplift": 0}
]
funding.check_funding(
@ -954,9 +954,9 @@ def test_epc_d_social_solar_with_only_minimum_insulation_should_fail(
# Solar PV innovation with insulation, but no heating system upgrade => not eligible
measures = [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0},
{"type": "loft_insulation", "is_innovation": False, "uplift": 0}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0},
{"type": "loft_insulation", "is_innovation": False, "innovation_uplift": 0}
]
funding.check_funding(
@ -1002,8 +1002,8 @@ def test_epc_d_social_solar_with_ashp_and_no_insulation_should_fail(
# Solar PV innovation with heating, but no insulation when insulation is recommended => not eligible
measures = [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "air_source_heat_pump", "is_innovation": False, "uplift": 0}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "air_source_heat_pump", "is_innovation": False, "innovation_uplift": 0}
]
funding.check_funding(
@ -1050,10 +1050,10 @@ def test_epc_d_social_solar_with_heating_and_minimum_insulation_should_pass(
# Innovation solar + insulation measures + eligible heating upgrade = not valid because the heat pump isn;t
# an innovation measure
measures = [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0},
{"type": "loft_insulation", "is_innovation": False, "uplift": 0},
{"type": "air_source_heat_pump", "is_innovation": False, "uplift": 0}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0},
{"type": "loft_insulation", "is_innovation": False, "innovation_uplift": 0},
{"type": "air_source_heat_pump", "is_innovation": False, "innovation_uplift": 0}
]
funding.check_funding(
@ -1095,10 +1095,10 @@ def test_epc_d_social_solar_with_heating_and_minimum_insulation_should_pass(
# Innovation solar + insulation measures + eligible heating upgrade = should be valid because the
# heat pump is an innovation measure
measures2 = [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0},
{"type": "loft_insulation", "is_innovation": False, "uplift": 0},
{"type": "air_source_heat_pump", "is_innovation": True, "uplift": 0.25}
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0},
{"type": "loft_insulation", "is_innovation": False, "innovation_uplift": 0},
{"type": "air_source_heat_pump", "is_innovation": True, "innovation_uplift": 0.25}
]
funding2.check_funding(
@ -1203,11 +1203,11 @@ def test_uplift(
# # TODO: Add a scenario with multiple measures, where some are innovation, some are not and we have
# TODO: Make sure private works too
measures = [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "uplift": 0},
{"type": "loft_insulation", "is_innovation": False, "uplift": 0},
{"type": "air_source_heat_pump", "is_innovation": False, "uplift": 0},
{"type": "cavity_wall_insulation", "is_innovation": False, "uplift": 0.25},
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "internal_wall_insulation", "is_innovation": False, "innovation_uplift": 0},
{"type": "loft_insulation", "is_innovation": False, "innovation_uplift": 0},
{"type": "air_source_heat_pump", "is_innovation": False, "innovation_uplift": 0},
{"type": "cavity_wall_insulation", "is_innovation": False, "innovation_uplift": 0.25},
]
funding.check_funding(
@ -1229,7 +1229,7 @@ def test_uplift(
)
assert funding.eco4_funding == 5302.3949999999995
assert funding.full_project_abs == 392.77 # is 280 + the 112.77 innovation uplift
assert funding.full_project_abs == 280 # Doesn't include the eco4 uplift
assert funding.eco4_uplift == 112.77
@ -1311,7 +1311,7 @@ def test_private_epc_e_solar_needs_heating(
tenure="Private",
)
measures = [{"type": "solar_pv", "is_innovation": True, "uplift": 0.45}]
measures = [{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45}]
funding.check_funding(
measures=measures,
starting_sap=54, # EPC E - eligible for private on EPC
@ -1360,10 +1360,10 @@ def test_private_epc_e_solar_with_heating_and_minimum_insulation_produces_uplift
)
measures = [
{"type": "solar_pv", "is_innovation": True, "uplift": 0.45},
{"type": "air_source_heat_pump", "is_innovation": False, "uplift": 0},
{"type": "cavity_wall_insulation", "is_innovation": False, "uplift": 0},
{"type": "loft_insulation", "is_innovation": False, "uplift": 0},
{"type": "solar_pv", "is_innovation": True, "innovation_uplift": 0.45},
{"type": "air_source_heat_pump", "is_innovation": False, "innovation_uplift": 0},
{"type": "cavity_wall_insulation", "is_innovation": False, "innovation_uplift": 0},
{"type": "loft_insulation", "is_innovation": False, "innovation_uplift": 0},
]
funding.check_funding(
@ -1393,3 +1393,85 @@ def test_private_epc_e_solar_with_heating_and_minimum_insulation_produces_uplift
assert funding.eco4_uplift and funding.eco4_uplift > 0
# And total funding should include that uplift
assert funding.eco4_funding and funding.eco4_funding > 0
def test_existing_gshp_to_ashp():
r = {'phase': 3, 'parts': [], 'type': 'heating', 'measure_type': 'air_source_heat_pump',
'description': 'Install a 5KW air source heat pump, and upgrade heating controls to Smart Thermostats, '
'room sensors and smart radiator valves (time & temperature zone control). Ensure you have a '
'single tariff',
'starting_u_value': None, 'new_u_value': None, 'sap_points': 7.7, 'already_installed': False,
'simulation_config': {'mainheat_energy_eff_ending': 'Good', 'hot_water_energy_eff_ending': 'Average',
'has_air_source_heat_pump_ending': True, 'has_ground_source_heat_pump_ending': False,
'extra_features_ending': None,
'thermostatic_control_ending': 'time and temperature zone control',
'switch_system_ending': None, 'multiple_room_thermostats_ending': False,
'mainheatc_energy_eff_ending': 'Very Good'},
'description_simulation': {'mainheat-description': 'Air source heat pump, radiators, electric',
'mainheat-energy-eff': 'Good', 'hot-water-energy-eff': 'Average',
'hotwater-description': 'From main system',
'mainheatcont-description': 'Time and temperature zone control',
'mainheatc-energy-eff': 'Very Good'}, 'total': 13188.996000000001,
'contingency': 3145.8150000000005, 'contingency_rate': 0.35, 'vat': 2080.666, 'labour_hours': 44.7,
'labour_days': 6.0, 'innovation_rate': 0, 'recommendation_id': '6_phase=3',
'efficiency': 13188.996000000001, 'co2_equivalent_savings': 0.4999999999999998,
'heat_demand': 53.20000000000002, 'kwh_savings': 801.5000000000005,
'energy_cost_savings': 327.31316785714296
}
funding = Funding(
project_scores_matrix=mock_project_scores_matrix,
partial_project_scores_matrix=mock_partial_scores_matrix,
whlg_eligible_postcodes=mock_whlg_postcodes,
eco4_social_cavity_abs_rate=13.5,
eco4_social_solid_abs_rate=17,
eco4_private_cavity_abs_rate=13.5,
eco4_private_solid_abs_rate=17,
gbis_social_cavity_abs_rate=21,
gbis_social_solid_abs_rate=25,
gbis_private_cavity_abs_rate=22,
gbis_private_solid_abs_rate=28,
tenure="Private",
)
(
pps, ppf, iu, ups
) = funding.get_innovation_uplift(
measure=r,
starting_sap=62,
floor_area=69,
is_cavity=True,
current_wall_uvalue=0.7,
is_partial=False,
existing_li_thickness=200,
mainheating={
'original_description': 'Ground source heat pump, radiators, electric',
'clean_description': 'Ground source heat pump, radiators, electric', 'has_radiators': True,
'has_fan_coil_units': False, 'has_pipes_in_screed_above_insulation': False,
'has_pipes_in_insulated_timber_floor': False, 'has_pipes_in_concrete_slab': False, 'has_boiler': False,
'has_air_source_heat_pump': False, 'has_room_heaters': False, 'has_electric_storage_heaters': False,
'has_warm_air': False, 'has_electric_underfloor_heating': False, 'has_electric_ceiling_heating': False,
'has_community_scheme': False, 'has_ground_source_heat_pump': True, 'has_no_system_present': False,
'has_portable_electric_heaters': False, 'has_water_source_heat_pump': False,
'has_electric_heat_pump': False, 'has_micro-cogeneration': False, 'has_solar_assisted_heat_pump': False,
'has_exhaust_source_heat_pump': False, 'has_community_heat_pump': False, 'has_hot-water-only': False,
'has_electric': True, 'has_mains_gas': False, 'has_wood_logs': False, 'has_coal': False, 'has_oil': False,
'has_wood_pellets': False, 'has_anthracite': False, 'has_dual_fuel_mineral_and_wood': False,
'has_smokeless_fuel': False, 'has_lpg': False, 'has_b30k': False, 'has_mineral_and_wood': False,
'has_dual_fuel_appliance': False, 'has_assumed': False, 'has_electricaire': False,
'has_assumed_for_most_rooms': False, 'has_underfloor_heating': False
},
main_fuel={
'original_description': 'electricity (not community)',
'clean_description': 'Electricity not community', 'fuel_type': 'electricity', 'tariff_type': None,
'is_community': False, 'no_individual_heating_or_community_network': False,
'complex_fuel_type': None
},
mainheat_energy_eff="Poor",
)
# All should be zero
assert pps == 0
assert ppf == 0
assert iu == 0
assert ups == 0

File diff suppressed because it is too large Load diff

View file

@ -26,7 +26,7 @@ class TestSearchEpcIntegration:
# Test case 2: Another valid address and postcode
# In this case, the newest EPC, does not have a uprn associated to it. If we did a search by
# uprn, we would get an old EPC
("Flat 8, Hainton House", "DN32 9AQ", 10090082018, True,
("Flat 8, Hainton House", "DN32 9AQ", "", True,
"bd1149a20a73397184f07a9955f872424826e70f4870c058d71be887766ee1f8", 2),
# Test case 3: When we make a request to the API for this property, we get back results for
# flats 1, 2 and 3. We have some logic to handle the response so that we get back flat 1
@ -56,7 +56,6 @@ class TestSearchEpcIntegration:
# We check that we have the correct epc
assert epc_searcher.newest_epc["lmk-key"] == lmk_key
assert epc_searcher.newest_epc["uprn"] == uprn
assert len(epc_searcher.older_epcs) == n_old_epcs
def test_search_housenumber(self):

View file

@ -310,7 +310,7 @@ class KwhData:
False: "N",
None: "N",
"Y": "Y",
"N": "N"
"N": "N",
}
for v in bools_to_remap:
epc[v] = bool_map[epc[v]]

View file

@ -0,0 +1,744 @@
import json
import os
import pandas as pd
from datetime import datetime
def years_between(d1, d2):
# precise year difference (accounts for months/days)
return (d1.year - d2.year) - ((d1.month, d1.day) < (d2.month, d2.day))
def get_element(elements, label):
"""Safely get an element dict by display label (your JSON keys)."""
return elements.get(label)
def append_result(decent_homes_meta, criteria, variable, sub_variable, result, install_date=None, expiry_date=None):
decent_homes_meta.append({
"criteria": criteria,
"variable": variable,
"sub_variable": sub_variable,
"result": result,
"hhsrs_rank": None,
"hhsrs_score": None,
"install_date": install_date,
"expiry_date": expiry_date,
})
# Read in static json, which is transformed by Jun-te's script
folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Waltham Forest/Decent Homes Pilot"
filenames = ["flat 1.json", "house 1.json"]
# Standardised variables which will form the enums in the db
HHSRS_VARIABLES = [
"damp_and_mould_growth",
"excess_cold",
"excess_heat",
"asbestos_and_mm_fibres",
"biocides",
"carbon_monoxide_and_fuel_combustion_products",
"lead",
"radiation",
"uncombusted_fuel_gas",
"volatile_organic_compounds",
"crowding_and_space",
"entry_by_intruders",
"lighting",
"noise",
"domestic_hygiene_pests_and_refuse",
"food_safety",
"personal_hygiene_sanitation_and_drainage",
"water_supply",
"falls_associated_with_baths",
"falls_on_level_surfaces",
"falls_on_stairs_and_steps",
"falls_between_levels",
"electrical_hazards",
"fire",
"flames_hot_surfaces_and_materials",
"collision_and_entrapment",
"explosions",
"ergonomics",
"structural_collapse_and_falling_elements"
]
ELEMENT_CODE_TO_DESCRIPTION = {
# One-to-one
"HHSRSDAMP": "damp_and_mould_growth",
"HHSRSCOLD": "excess_cold",
"HHSRSHEAT": "excess_heat",
"HHSRSASB": "asbestos_and_mm_fibres",
"HHSRSBIOC": "biocides",
"HHSRSLEAD": "lead",
"HHSRSRADIA": "radiation",
"HHSRSFUEL": "uncombusted_fuel_gas",
"HHSRSORGAN": "volatile_organic_compounds",
"HHSRSCROWD": "crowding_and_space",
"HHSRSENTRY": "entry_by_intruders",
"HHSRSLIGHT": "lighting",
"HHSRSNOISE": "noise",
"HHSRSDOMES": "domestic_hygiene_pests_and_refuse",
"HHSRSFOOD": "food_safety",
"HHSRSPERS": "personal_hygiene_sanitation_and_drainage",
"HHSRSWATER": "water_supply",
"HHSRSFBATH": "falls_associated_with_baths",
"HHSRSFLEVE": "falls_on_level_surfaces",
"HHSRSFSTAI": "falls_on_stairs_and_steps",
"HHSRSFBETW": "falls_between_levels",
"HHSRSELEC": "electrical_hazards",
"HHSRSFIRE": "fire",
"HHSRSFLAME": "flames_hot_surfaces_and_materials",
"HHSRSEXPLO": "explosions",
"HHSRSPOSI": "ergonomics",
"HHSRSSTRUC": "structural_collapse_and_falling_elements",
# One-to-many expansions
"HHSRSCO": "carbon_monoxide",
"HHSRSSO2": "sulphur_dioxide_and_smoke",
"HHSRSNO2": "nitrogen_dioxide",
"HHSRSENTRP": "collision_and_entrapment",
"HHSRSCLOW": "collision_hazards_and_low_headroom",
}
CRITERION_B_VARIABLES = [
"external_walls_structure", "lintels", "brickwork_spalling", "wall_finish", "roof_structure", "roof_finish",
"chimneys", "windows", "external_doors", "kitchens", "bathrooms", "central_heating_boiler",
"central_heating_distribution_system", "heating_other", "electrical_systems",
]
CRITERION_C_VARIABLES = [
"kitchen_less_than_20_years_old", "kitchen_adequate_space_and_layout", "bathroom_less_than_30_years_old",
"bathroom_wc_appropriately_located", "adequate_external_noise_insulation", "adequate_common_entrance_areas",
]
# Criterion C explicit age limits (different from component lifespans used elsewhere)
CRITERION_C_AGE_LIMITS = {
"kitchen_years_max": 20,
"bathroom_years_max": 30,
}
# Field labels as they appear in your JSON (based on your code)
LABEL_KITCHEN = "Adequacy of Kitchen and Type in Property"
LABEL_BATHROOM = "Adequacy of Bathroom Location in Property"
LABEL_NOISE = "Adequacy of Noise Insulation in Property"
LABEL_COMMON_CIRC = "Circulation Space in Common Area" # flats only
STANDARD_HHSRS_MAPPING = {"pass": "TYPRISK", "fail": "MODRISK", "no_data": "TOBEASSESS"}
# Criterion A - mapping of HHSRS variables to Waltham forest element codes
HHSRS_MAPPING = {
"damp_and_mould_growth": {"HHSRSDAMP": STANDARD_HHSRS_MAPPING},
"excess_cold": {"HHSRSCOLD": STANDARD_HHSRS_MAPPING},
"excess_heat": {"HHSRSHEAT": STANDARD_HHSRS_MAPPING},
"asbestos_and_mm_fibres": {"HHSRSASB": STANDARD_HHSRS_MAPPING},
"biocides": {"HHSRSBIOC": STANDARD_HHSRS_MAPPING},
"carbon_monoxide_and_fuel_combustion_products": {
"HHSRSCO": STANDARD_HHSRS_MAPPING,
"HHSRSSO2": STANDARD_HHSRS_MAPPING,
"HHSRSNO2": STANDARD_HHSRS_MAPPING
},
"lead": {"HHSRSLEAD": STANDARD_HHSRS_MAPPING},
"radiation": {"HHSRSRADIA": STANDARD_HHSRS_MAPPING},
"uncombusted_fuel_gas": {"HHSRSFUEL": STANDARD_HHSRS_MAPPING},
"volatile_organic_compounds": {"HHSRSORGAN": STANDARD_HHSRS_MAPPING},
"crowding_and_space": {"HHSRSCROWD": STANDARD_HHSRS_MAPPING},
"entry_by_intruders": {"HHSRSENTRY": STANDARD_HHSRS_MAPPING},
"lighting": {"HHSRSLIGHT": STANDARD_HHSRS_MAPPING},
"noise": {"HHSRSNOISE": STANDARD_HHSRS_MAPPING},
"domestic_hygiene_pests_and_refuse": {"HHSRSDOMES": STANDARD_HHSRS_MAPPING},
"food_safety": {"HHSRSFOOD": STANDARD_HHSRS_MAPPING},
"personal_hygiene_sanitation_and_drainage": {"HHSRSPERS": STANDARD_HHSRS_MAPPING},
"water_supply": {"HHSRSWATER": STANDARD_HHSRS_MAPPING},
"falls_associated_with_baths": {"HHSRSFBATH": STANDARD_HHSRS_MAPPING},
"falls_on_level_surfaces": {"HHSRSFLEVE": STANDARD_HHSRS_MAPPING},
"falls_on_stairs_and_steps": {"HHSRSFSTAI": STANDARD_HHSRS_MAPPING},
"falls_between_levels": {"HHSRSFBETW": STANDARD_HHSRS_MAPPING},
"electrical_hazards": {"HHSRSELEC": STANDARD_HHSRS_MAPPING},
"fire": {"HHSRSFIRE": STANDARD_HHSRS_MAPPING},
"flames_hot_surfaces_and_materials": {"HHSRSFLAME": STANDARD_HHSRS_MAPPING},
"collision_and_entrapment": {"HHSRSENTRP": STANDARD_HHSRS_MAPPING, "HHSRSCLOW": STANDARD_HHSRS_MAPPING},
"explosions": {"HHSRSEXPLO": STANDARD_HHSRS_MAPPING},
"ergonomics": {"HHSRSPOSI": STANDARD_HHSRS_MAPPING},
"structural_collapse_and_falling_elements": {"HHSRSSTRUC": STANDARD_HHSRS_MAPPING}
}
# print(houses_waltham_forest_data[
# houses_waltham_forest_data["ELEMENT CODE"] == "INTBTHADEQ"
# ][["ATTRIBUTE CODE", "ATTRIBUTE CODE DESCRIPTION"]].drop_duplicates())
# print(flats_waltham_forest_data[
# flats_waltham_forest_data["ELEMENT CODE"] == "INTBTHADEQ"
# ][["ATTRIBUTE CODE", "ATTRIBUTE CODE DESCRIPTION"]].drop_duplicates())
# Criterion B
B_COMPONENT_LABELS = {
# Key components
"wall_structure": [
"Wall Structure in External Area",
],
"lintels": [
"Lintels in External Area",
],
"brickwork_spalling": [
"Wall Spalling in External Area",
],
"wall_finish": [
"Wall Finish 1 in External Area",
"Wall Finish 2 in External Area",
"External Decorations in External Area",
"Brickwork Pointing in External Area",
],
"roof_structure": [
"Roof Structure 1 in External Area",
"Roof Structure 2 in External Area",
"Roof Structure 3 in External Area",
"Garage Roof in External Area",
"Garage and Store Roofs in External Area",
"Store Roof in External Area",
"Fascia / Soffit / Bargeboard in External Area",
"Gutters in External Area",
"Downpipes in External Area",
"Internal Downpipes in External Area"
],
"roof_finish": [
"Roof Covering 1 in External Area",
"Roof Covering 2 in External Area",
"Roof Covering 3 in External Area",
],
"chimneys": [
"Chimneys in External Area",
],
"windows": [
"Windows in Property",
"Windows 1 in External Area",
"Windows 2 in External Area",
"Garage and Store Windows in External Area",
"Garage Windows in External Area",
"Store Windows in External Area",
],
"external_doors": [
"Type and Location of Front Door in Property",
"Front Door Fire Rating in Property",
"Patio and French Doors 1 in External Area",
"Back and Side Doors 1 in External Area",
"Back and Side Doors 2 in External Area",
"Garage and Store Doors in External Area",
"Garage Door in External Area",
"Store Door in External Area",
],
"central_heating_boiler": [
# "Heating Improvement Required in Property",
"Boiler Fuel in Property",
"Type of Water Heating in Property",
],
"heating_other": [
# "Heating Distribution System in Property"
"Boiler Fuel in Property",
"Type of Water Heating in Property",
],
"electrical_systems": [
"Electrics Required in Property",
],
# Other components
"kitchen": [
"Adequacy of Kitchen and Type in Property",
],
"bathroom": [
"Adequacy of Bathroom Location in Property",
],
"central_heating_distribution_system": [
"Heating Distribution System in Property",
],
}
KEY_COMPONENTS = {
"wall_structure", "lintels", "brickwork_spalling", "wall_finish",
"roof_structure", "roof_finish", "chimneys", "windows",
"external_doors", "central_heating_boiler", "heating_other",
"electrical_systems",
}
OTHER_COMPONENTS = {
"kitchen", "bathroom", "central_heating_distribution_system",
}
# Criterion C
COMPONENT_LIFESPANS = {
# Key components
"wall_structure": {
"house": 80, "flat_below_6_storeys": 80, "flat_above_6_storeys": 80
},
"lintels": {
"house": 60, "flat_below_6_storeys": 60, "flat_above_6_storeys": 60
},
"brickwork_spalling": {
"house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
},
"wall_finish": {
"house": 60, "flat_below_6_storeys": 60, "flat_above_6_storeys": 30
},
"roof_structure": {
"house": 50, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
},
"roof_finish": {
"house": 50, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
},
"chimneys": {
"house": 50, "flat_below_6_storeys": 50, "flat_above_6_storeys": None # N/A
},
"windows": {
"house": 40, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
},
"external_doors": {
"house": 40, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
},
"central_heating_boiler": {
"house": 15, "flat_below_6_storeys": 15, "flat_above_6_storeys": 15
},
"heating_other": {
"house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
},
"electrical_systems": {
"house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
},
# Other components
"kitchen": {
"house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
},
"bathroom": {
"house": 40, "flat_below_6_storeys": 40, "flat_above_6_storeys": 40
},
"central_heating_distribution_system": {
"house": 40, "flat_below_6_storeys": 40, "flat_above_6_storeys": 40
},
}
# Database design
# creation_date, uprn, variable, result (pass/fail/nodata), hhsrs_score (optional, numeric), hhsrs_rank (A-J),
# install_date (for components which expire, e.g. kitchen), remaining_life (for components which expire, e.g. kitchen),
# TODO: Add the criterion
decent_homes_meta = []
# Use to capture criterion A, B, C and D. Should be:
# {"uprn": int, "creation_date": datetime, "criterion_a": bool, "criterion_b": bool, "criterion_c": bool,
# "criterion_d": bool, "decent_homes": bool"}
property_decent_homes = []
for fn in filenames:
with open(os.path.join(folder, fn), "rb") as f:
data = json.load(f)
today = pd.Timestamp.today().normalize()
property_info = data["property_info"]
if property_info["PROP TYPE"] in ["HOU"]:
property_type = "house"
elif property_info["PROP TYPE"] == "FLA":
raise NotImplementedError("Implement distrinction between below and above 6 storeys")
# property_type = "flat"
else:
raise NotImplementedError("Unknown property type")
# ---------------- Criterion A ----------------
# Critrion A: pass/fail
# If fail, why?
for hhsrs_variable, mapping in HHSRS_MAPPING.items():
element_code = list(mapping.keys())[0]
# Find the data in the JSON within data["elements"]
check_pass = []
for k, v in data["elements"].items():
if v["ELEMENT CODE"] == element_code:
# We check the attribute code
# Check if pass
if v["ATTRIBUTE CODE"] == mapping[element_code]["pass"]:
result = "pass"
elif v["ATTRIBUTE CODE"] == mapping[element_code]["fail"]:
result = "fail"
elif v["ATTRIBUTE CODE"] == mapping[element_code]["no_data"]:
result = "no_data"
else:
raise ValueError("Unknown attribute code")
check_pass.append(result)
append_result(
decent_homes_meta,
criteria="A",
variable=hhsrs_variable,
sub_variable=ELEMENT_CODE_TO_DESCRIPTION[element_code],
result=result,
install_date=None,
expiry_date=None,
)
# We check if we have a pass, fail or no_data
# if all([x == "pass" for x in check_pass]):
# hhsrs_result = "pass"
# elif any([x == "fail" for x in check_pass]):
# hhsrs_result = "fail"
# elif any([x == "no_data" for x in check_pass]):
# hhsrs_result = "no_data"
# else:
# raise NotImplementedError("Mixed results not implemented")
# ---------------- Criterion B ----------------
# Check each of the components
# ---------------- Criterion B ----------------
property_boiler = get_element(data["elements"], "Boiler Fuel in Property")
for component, labels in B_COMPONENT_LABELS.items():
for label in labels:
label_data = get_element(data["elements"], label)
# Handle no-data or not-applicable
if label_data["ATTRIBUTE CODE"] in ["UNKNOWN", "NONE", "UNKNOWNG", "UNKNOWNS"]:
# append_result(
# decent_homes_meta,
# criteria="B",
# variable=component,
# sub_variable=label,
# result="pass",
# install_date=None,
# expiry_date=None,
# )
continue
# Special skip conditions for heating
no_boiler_condition = (
property_boiler["ATTRIBUTE CODE"] in ["NONENOCH"]
and component == "central_heating_boiler"
)
other_heating_condition = (
label_data["ATTRIBUTE CODE"] in ["NONENOCH"]
and component == "heating_other"
)
if no_boiler_condition or other_heating_condition:
# append_result(
# decent_homes_meta,
# criteria="B",
# variable=component,
# sub_variable=label,
# result="pass",
# install_date=None,
# expiry_date=None,
# )
continue
# Normal case: evaluate install date + lifetime + remaining life
install_date = pd.to_datetime(label_data["INSTALL DATE"])
if pd.isnull(install_date):
raise ValueError(f"Missing install date for {component}/{label}")
component_lifetime = COMPONENT_LIFESPANS[component][property_type]
is_old = years_between(today.to_pydatetime(), install_date.to_pydatetime()) > component_lifetime
if pd.isnull(label_data["REMAINING LIFE"]):
raise ValueError(f"Missing remaining life for {component}/{label}")
has_failed = label_data["REMAINING LIFE"] < 0
expiry_date = install_date + pd.DateOffset(years=component_lifetime)
component_result = "fail" if is_old and has_failed else "pass"
# Push into decent_homes_meta
append_result(
decent_homes_meta,
criteria="B",
variable=component,
sub_variable=label,
result=component_result,
install_date=str(install_date),
expiry_date=str(expiry_date),
)
# ---------------- Criterion C ----------------
# Guard: property type string already set earlier
is_flat = (property_info["PROP TYPE"] == "FLA")
# 1) Kitchen age ≤ 20 years
kitchen = get_element(data["elements"], LABEL_KITCHEN)
if kitchen:
kit_install_raw = kitchen["INSTALL DATE"]
kit_install = pd.to_datetime(kit_install_raw)
kit_age_years = years_between(today.to_pydatetime(), kit_install.to_pydatetime())
kitchen_age_result = "pass" if kit_age_years <= CRITERION_C_AGE_LIMITS["kitchen_years_max"] else "fail"
# For transparency, store next renewal as install + 20 years (criterion C perspective)
kit_next_due = kit_install + pd.DateOffset(years=CRITERION_C_AGE_LIMITS["kitchen_years_max"])
else:
raise NotImplementedError("Kitchen data missing - pls check")
append_result(
decent_homes_meta,
criteria="C",
variable="kitchen_less_than_20_years_old",
sub_variable="kitchen_less_than_20_years_old",
result=kitchen_age_result,
install_date=str(kit_install),
expiry_date=str(kit_next_due)
)
# 2) Kitchen adequate space/layout
# Prefer explicit codes if you have them, fall back to text in ATTRIBUTE CODE DESCRIPTION
if kitchen:
kit_attr_desc = kitchen["ATTRIBUTE CODE"]
if kit_attr_desc == "STDKITADQ":
kitchen_adequacy_result = "pass"
else:
raise NotImplementedError("No other observed codes yet")
else:
raise NotImplementedError("Kitchen data missing - pls check")
append_result(
decent_homes_meta,
criteria="C",
variable="kitchen_adequate_space_and_layout",
sub_variable="kitchen_adequate_space_and_layout",
result=kitchen_adequacy_result,
)
# 3) Bathroom age ≤ 30 years
bath = get_element(data["elements"], LABEL_BATHROOM)
if bath:
bth_install_raw = bath["INSTALL DATE"]
bth_install = pd.to_datetime(bth_install_raw)
bth_age_years = years_between(today.to_pydatetime(), bth_install.to_pydatetime())
bathroom_age_result = "pass" if bth_age_years <= CRITERION_C_AGE_LIMITS["bathroom_years_max"] else "fail"
bth_next_due = bth_install + pd.DateOffset(years=CRITERION_C_AGE_LIMITS["bathroom_years_max"])
else:
raise NotImplementedError("Bathroom data missing - pls check")
append_result(
decent_homes_meta,
criteria="C",
variable="bathroom_less_than_30_years_old",
sub_variable="bathroom_less_than_30_years_old",
result=bathroom_age_result,
install_date=str(bth_install),
expiry_date=bth_next_due
)
# 4) Bathroom/WC appropriately located
if bath:
bth_attr_code = bath["ATTRIBUTE CODE"]
if bth_attr_code in {"STDBTHADQ", "ADPBTHADQ"}:
bathroom_location_result = "pass"
else:
raise NotImplementedError("No other observed codes yet")
else:
raise NotImplementedError("Bathroom data missing - pls check")
append_result(
decent_homes_meta,
criteria="C",
variable="bathroom_wc_appropriately_located",
sub_variable="bathroom_wc_appropriately_located",
result=bathroom_location_result
)
# 5) Adequate external noise insulation
noise = get_element(data["elements"], LABEL_NOISE)
if noise:
noise_code = noise["ATTRIBUTE CODE"]
if noise_code in {"ADEQUATE"}:
noise_result = "pass"
else:
raise NotImplementedError("No other observed codes yet")
else:
raise NotImplementedError("Noise insulation data missing - pls check")
append_result(
decent_homes_meta,
criteria="C",
variable="adequate_external_noise_insulation",
sub_variable="adequate_external_noise_insulation",
result=noise_result
)
# 6) Adequate common entrance areas (flats only)
if is_flat:
raise Exception("Pls check this")
common = get_element(data["elements"], LABEL_COMMON_CIRC)
if common:
circ_desc = common.get("ATTRIBUTE CODE DESCRIPTION", "")
common_areas_result = adequacy_result_by_text(circ_desc)
else:
common_areas_result = "no_data"
append_result(decent_homes_meta, "adequate_common_entrance_areas", common_areas_result)
# ---------------- Criterion D ----------------
# heating system type
heating = get_element(data["elements"], "Heating Improvement Required in Property")
if heating:
heat_type_code = heating["ATTRIBUTE CODE"]
if heat_type_code in {"NOTAPPLIC"}:
heating_type_result = "pass"
elif heat_type_code in {"WETINSFULL"}:
heating_type_result = "fail"
else:
raise NotImplementedError("No other observed codes yet")
else:
raise NotImplementedError("Heating element missing in dataset")
append_result(
decent_homes_meta,
criteria="D",
variable="efficient_heating_system_type",
sub_variable="efficient_heating_system_type",
result=heating_type_result
)
# heating distribution
heating_dist = get_element(data["elements"], "Heating Distribution System in Property")
if heating_dist:
dist_code = heating_dist["ATTRIBUTE CODE"]
if dist_code == "UNKNOWN":
# For the observed case, there was no heating and wet heating needed to be installed in full so the value
# was unknown
heating_dist_result = "no_data"
else:
raise NotImplementedError("No other observed codes yet")
else:
raise NotImplementedError("Heating distribution element missing in dataset")
append_result(
decent_homes_meta,
criteria="D",
variable="efficient_heating_distribution",
sub_variable="efficient_heating_distribution",
result=heating_dist_result
)
# insulation
loft = get_element(data["elements"], "Size in mm of Loft Insulation Thickness in Property")
wall = get_element(data["elements"], "Wall Insulation Improvement in External Area")
# To determine how much loft insulation is required
# Loft insulation check (example threshold: ≥ 270mm = pass)
if loft:
# We have a specific code, where further loft insulation is needed - It appears the heating type check has
# already been completed in this dataset and so we just need to check the code
loft_code = loft["ATTRIBUTE CODE"]
if loft_code == "LOFTINSRQD":
loft_result = "fail"
elif loft_code.isnumeric():
loft_result = "pass"
else:
raise NotImplementedError("Unknown loft insulation code - pls check")
else:
raise NotImplementedError("Loft insulation data missing - pls check")
append_result(
decent_homes_meta,
criteria="D",
variable="loft_insulation_sufficient",
sub_variable="loft_insulation_sufficient",
result=loft_result
)
# Wall insulation check
if wall:
wall_code = wall["ATTRIBUTE CODE"]
if wall_code in {"NONE"}: # Means no insulation improvement required
wall_result = "pass"
else:
raise NotImplementedError("No other observed codes yet")
else:
raise NotImplementedError("Wall insulation data missing - pls check")
append_result(
decent_homes_meta,
criteria="D",
variable="wall_insulation_sufficient",
sub_variable="wall_insulation_sufficient",
result=wall_result
)
# ---------------- Criterion A overall ----------------
a_vars = set(HHSRS_MAPPING.keys())
latest_a_results = {r["variable"]: r["result"] for r in decent_homes_meta if r["variable"] in a_vars}
if any(v == "fail" for v in latest_a_results.values()):
criterion_a_result = "fail"
elif all(v == "pass" for v in latest_a_results.values()):
criterion_a_result = "pass"
else:
criterion_a_result = "no_data"
# ---------------- Criterion B overall ----------------
component_results = {}
for component in B_COMPONENT_LABELS.keys():
comp_rows = [r for r in decent_homes_meta if
r["criteria"] == "B" and r["variable"] == component and r["sub_variable"] is not None]
comp_sub_results = [r["result"] for r in comp_rows]
if not comp_sub_results: # no rows at all
comp_result = "no_data"
elif any(r == "fail" for r in comp_sub_results):
comp_result = "fail"
elif all(r == "pass" for r in comp_sub_results if r != "no_data"):
comp_result = "pass"
elif all(r == "no_data" for r in comp_sub_results):
comp_result = "no_data"
else:
comp_result = "no_data"
component_results[component] = comp_result
key_fails = [c for c, r in component_results.items() if c in KEY_COMPONENTS and r == "fail"]
other_fails = [c for c, r in component_results.items() if c in OTHER_COMPONENTS and r == "fail"]
if key_fails:
criterion_b_result = "fail"
elif len(other_fails) >= 2:
criterion_b_result = "fail"
elif all(r == "no_data" for r in component_results.values()):
criterion_b_result = "no_data"
else:
criterion_b_result = "pass"
# ---------------- Criterion C overall ----------------
criterion_c_vars = [
"kitchen_less_than_20_years_old",
"kitchen_adequate_space_and_layout",
"bathroom_less_than_30_years_old",
"bathroom_wc_appropriately_located",
"adequate_external_noise_insulation",
]
if is_flat:
criterion_c_vars.append("adequate_common_entrance_areas")
latest_c_results = {r["variable"]: r["result"] for r in decent_homes_meta if r["variable"] in criterion_c_vars}
count_fails = sum(1 for v in latest_c_results.values() if v == "fail")
# optionally count no_data too if you want strict interpretation
criterion_c_result = "fail" if count_fails >= 3 else "pass"
# ---------------- Criterion D overall ----------------
# Needs to have both efficient geating and distribution so all should pass
criterion_d_vars = [
"efficient_heating_system_type",
"efficient_heating_distribution",
"loft_insulation_sufficient",
"wall_insulation_sufficient",
]
latest_d_results = {r["variable"]: r["result"] for r in decent_homes_meta if r["variable"] in criterion_d_vars}
if any(v == "fail" for v in latest_d_results.values()):
criterion_d_result = "fail"
elif all(v == "pass" for v in latest_d_results.values()):
criterion_d_result = "pass"
else:
criterion_d_result = "no_data"
# ---------------- Append to property_decent_homes ----------------
property_decent_homes.append({
"uprn": property_info.get("UPRN"), # TODO: Need UPRN
"creation_date": datetime.now().date().isoformat(),
"criterion_a": criterion_a_result,
"criterion_b": criterion_b_result,
"criterion_c": criterion_c_result,
"criterion_d": criterion_d_result,
"decent_homes": (
criterion_a_result == "pass"
and criterion_c_result == "pass"
and criterion_d_result == "pass"
)
})

View file

@ -718,15 +718,26 @@ class RetrieveFindMyEpc:
find_epc_data = searcher.retrieve_newest_find_my_epc_data()
except Exception as e:
logger.error(f"Error retrieving find my epc data: {e}")
if epc["address1"] == epc["address"]:
# There's no benefit of using the same address, so we split on comma
address1 = epc["address"].split(",")[0]
else:
address1 = epc["address1"]
# We attempt with the backup add
searcher = cls(address=address1, postcode=epc["postcode"])
find_epc_data = searcher.retrieve_newest_find_my_epc_data()
logger.info("Successfully retrieved find my epc data using backup address")
# We try two backup approaches. The first is to trim the final section off the end of the address
address1 = ",".join(epc["address"].split(",")[:-1])
try:
searcher = cls(address=address1, postcode=epc["postcode"])
find_epc_data = searcher.retrieve_newest_find_my_epc_data()
logger.info("Successfully retrieved find my epc data using trimmed address")
except Exception as e2:
logger.error(f"Error retrieving find my epc data using trimmed address: {e2}")
# Attempt final approach
if epc["address1"] == epc["address"]:
# There's no benefit of using the same address, so we split on comma
address1 = epc["address"].split(",")[0]
else:
address1 = epc["address1"]
# We attempt with the backup add
searcher = cls(address=address1, postcode=epc["postcode"])
find_epc_data = searcher.retrieve_newest_find_my_epc_data()
logger.info("Successfully retrieved find my epc data using backup address")
non_invasive_recommendations = {
"uprn": epc["uprn"],

View file

@ -1,38 +1,111 @@
# Initial Code
from seleniumbase import SB
from bs4 import BeautifulSoup
import pandas as pd
import time
from stealth_requests import StealthSession
import random
from multiprocessing import Pool
from tqdm import tqdm
uprns = [
100071297618,
100080893397,
100060778033,
200004793081,
100071265143,
100071297618,
100080893397,
100060778033,
200004793081,
100071265143,
]
ENGINES = ["safari", "chrome"]
estimate_list = []
for uprn in uprns:
def scrape_all_estimates(session, url):
# Rotate impersonation per request
resp = session.get(url, impersonate=ENGINES[random.randint(0, 1)])
page_source = BeautifulSoup(resp.text, "html.parser")
estimates = page_source.find_all("div", {"data-testid": "sale-estimate"})
is_blocked = len(estimates) == 0
return estimates, is_blocked
# Probably can change the timings here
time.sleep(5)
with SB(uc=True) as sb:
sb.uc_open_with_reconnect(
f"https://www.zoopla.co.uk/property/uprn/{uprn}/",
3,
def parallel_task(url):
# No impersonate argument here
with StealthSession() as session:
estimates, is_blocked = scrape_all_estimates(session, url)
while is_blocked:
print(f"Blocked by Zoopla for URL: {url}")
time.sleep(random.uniform(0, 1))
estimates, is_blocked = scrape_all_estimates(session, url)
low_estimate = estimates[0].find("span", {"data-testid": "low-estimate-blurred"}).text
middle_estimate = estimates[0].find("p", {"data-testid": "estimate-blurred"}).text
high_estimate = estimates[0].find("span", {"data-testid": "high-estimate-blurred"}).text
return {
"URL": url,
"Low Estimate": low_estimate,
"Middle Estimate": middle_estimate,
"High Estimate": high_estimate,
}
def parse_price(p):
p = p.replace("£", "").strip().lower()
if p.endswith("k"):
return float(p[:-1]) * 1000
elif p.endswith("m"):
return float(p[:-1]) * 1_000_000
else:
return float(p)
# def parallel_task(url):
# with StealthSession(impersonate=ENGINES[random.randint(0, 1)]) as session:
# estimates, is_blocked = scrape_all_estimates(session, url)
#
# while is_blocked:
# # Will need to wait and retry if blocked by Zoopla
# print(f"Blocked by Zoopla for URL: {url}")
# sleep_factor = random.uniform(0, 1) # Random delay to avoid detection
# time.sleep(sleep_factor * 1)
# estimates, is_blocked = scrape_all_estimates(session, url)
#
# low_estimate = (
# estimates[0].find("span", {"data-testid": "low-estimate-blurred"}).text
# ) # Find all span elements with data-testid="low-estimate"
# middle_estimate = (
# estimates[0].find("p", {"data-testid": "estimate-blurred"}).text
# ) # Find all span elements with data-testid="middle-estimate"
# high_estimate = (
# estimates[0].find("span", {"data-testid": "high-estimate-blurred"}).text
# ) # Find all span elements with data-testid="high-estimate-blurred"
#
# return {
# "URL": url,
# "Low Estimate": low_estimate,
# "Middle Estimate": middle_estimate,
# "High Estimate": high_estimate,
# }
if __name__ == "__main__":
# Get a SAL
asset_list = pd.read_excel(
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/NRLA/Property Box/Property Box Finance Portfolio - "
"Standardised.xlsx",
sheet_name="Standardised Asset List"
)
asset_list["epc_os_uprn"] = asset_list["epc_os_uprn"].astype(int).astype(str)
uprns = asset_list["epc_os_uprn"].tolist()
urls = [f"https://www.zoopla.co.uk/property/uprn/{uprn}/" for uprn in uprns]
with Pool(processes=5) as pool:
estimates_list = list(
tqdm(
pool.imap(parallel_task, urls),
total=len(urls),
)
)
soup = sb.get_beautiful_soup()
df = pd.DataFrame(estimates_list)
# Extract UPRN from URL
df["uprn"] = df["URL"].str.extract(r"uprn/(\d+)/")
df["valuation"] = df["Middle Estimate"].apply(parse_price)
df.to_csv("zoopla_estimates.csv", index=False)
estimates = soup.find_all("div", {"data-testid": "sale-estimate"})
# Can change the way we extract the text here
estimate_text = (
estimates[-1].find_all("p")[-1].find_all("span")[-1]["aria-label"]
)
estimate_list.append(estimate_text)
df["uprn"] = df["uprn"].astype(int).astype(str)
asset_list.merge(df[["uprn", "valuation"]], left_on="epc_os_uprn", right_on="uprn", how="left").to_excel(
"Property Box Finance Portfolio - Standardised - with valuations.xlsx", index=False
)

View file

@ -0,0 +1,5 @@
beautifulsoup4>=4.12.0
pandas>=2.0.0
stealth-requests>=1.0.7
tqdm>=4.65.0
openpyxl

View file

@ -66,7 +66,7 @@ resource "aws_security_group" "allow_db" {
resource "aws_db_instance" "default" {
allocated_storage = var.allocated_storage
engine = "postgres"
engine_version = "14.13"
engine_version = "14.17"
instance_class = var.instance_class
db_name = var.database_name
username = jsondecode(data.aws_secretsmanager_secret_version.db_credentials.secret_string)["db_assessment_model_username"]
@ -261,4 +261,17 @@ module "cloudfront_distribution" {
bucket_arn = module.s3.bucket_arn
bucket_domain_name = module.s3.bucket_domain_name
stage = var.stage
}
################################################
# SES - Email sending
################################################
module "ses" {
source = "./modules/ses"
domain_name = "domna.homes"
stage = var.stage
}
output "ses_dns_records" {
value = module.ses.dns_records
}

View file

@ -0,0 +1,50 @@
resource "aws_ses_domain_identity" "this" {
domain = var.domain_name
}
# DKIM signing
resource "aws_ses_domain_dkim" "this" {
domain = aws_ses_domain_identity.this.domain
}
# IAM user for SES SMTP
resource "aws_iam_user" "ses_user" {
name = "${var.stage}-ses-user"
}
resource "aws_iam_user_policy" "ses_send_policy" {
name = "AllowSESSendEmail"
user = aws_iam_user.ses_user.name
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Effect = "Allow"
Action = [
"ses:SendEmail",
"ses:SendRawEmail"
]
Resource = "*"
}
]
})
}
resource "aws_iam_access_key" "ses_user" {
user = aws_iam_user.ses_user.name
}
# Store SMTP credentials in AWS Secrets Manager
resource "aws_secretsmanager_secret" "ses_smtp" {
name = "${var.stage}/ses/smtp_credentials"
description = "SMTP credentials for SES (${var.stage})"
}
resource "aws_secretsmanager_secret_version" "ses_smtp" {
secret_id = aws_secretsmanager_secret.ses_smtp.id
secret_string = jsonencode({
username = aws_iam_access_key.ses_user.id
password = aws_iam_access_key.ses_user.ses_smtp_password_v4
})
}

View file

@ -0,0 +1,66 @@
# These are our DNS records that will need to be added to our Krystal account
# TXT record
output "verification_record" {
description = "TXT record required to verify the domain with SES"
value = {
name = "_amazonses.${aws_ses_domain_identity.this.domain}"
type = "TXT"
value = aws_ses_domain_identity.this.verification_token
}
}
# DKIM CNAME records
output "dkim_records" {
description = "CNAME records required to enable DKIM for SES"
value = [
for dkim in aws_ses_domain_dkim.this.dkim_tokens : {
name = "${dkim}._domainkey.${aws_ses_domain_identity.this.domain}"
type = "CNAME"
value = "${dkim}.dkim.amazonses.com"
}
]
}
# SMTP credentials - send them to secrets manager
output "ses_smtp_secret_arn" {
description = "ARN of the SES SMTP credentials stored in Secrets Manager"
value = aws_secretsmanager_secret.ses_smtp.arn
}
output "smtp_password" {
value = aws_iam_access_key.ses_user.ses_smtp_password_v4
sensitive = true
description = "SMTP password for SES"
}
output "dns_records" {
description = "All DNS records required for SES verification and recommended deliverability"
value = concat(
[
{
name = "_amazonses.${aws_ses_domain_identity.this.domain}"
type = "TXT"
value = aws_ses_domain_identity.this.verification_token
},
{
name = var.domain_name
type = "TXT"
value = "v=spf1 include:amazonses.com -all"
},
{
name = "_dmarc.${var.domain_name}"
type = "TXT"
value = "v=DMARC1; p=quarantine; rua=mailto:postmaster@${var.domain_name}"
}
],
[
for dkim in aws_ses_domain_dkim.this.dkim_tokens : {
name = "${dkim}._domainkey.${aws_ses_domain_identity.this.domain}"
type = "CNAME"
value = "${dkim}.dkim.amazonses.com"
}
]
)
}

View file

@ -0,0 +1,9 @@
variable "domain_name" {
description = "The domain to verify with SES (e.g. domna.homes)"
type = string
}
variable "stage" {
description = "Deployment stage (e.g. dev, prod)"
type = string
}

View file

@ -103,6 +103,7 @@ class HeatingRecommender:
self.property.main_heating["has_electric"] or self.property.main_heating["has_electricaire"]
)
self.has_ashp = self.property.main_heating["has_air_source_heat_pump"]
self.has_gshp = self.property.main_heating["has_ground_source_heat_pump"]
self.has_room_heaters = (
self.property.main_heating["has_room_heaters"] or
self.property.main_heating["has_portable_electric_heaters"]
@ -151,8 +152,10 @@ class HeatingRecommender:
"underfloor heating" not in self.property.main_heating["clean_description"]
)
# If the property has a ground source heat pump, or air source heat pump, we don't recommend HHRSH
return (
hhr_suitable and (not ashp_only_heating_recommendation) and not self.has_ashp and
hhr_suitable and (not ashp_only_heating_recommendation) and not self.has_ashp and not self.has_gshp and
("high_heat_retention_storage_heater" in measures)
)
@ -345,7 +348,7 @@ class HeatingRecommender:
if (
self.property.is_ashp_valid(measures=measures) and
non_invasive_ashp_recommendation["suitable"] and
not self.has_ashp
not self.has_ashp and not self.has_gshp
):
self.recommend_air_source_heat_pump(
phase=phase,

View file

@ -109,7 +109,8 @@ class CostOptimiser:
self.m.optimize()
if self.m.status == OptimizationStatus.INFEASIBLE:
logger.info("We have an infeasible model, setting up slack model")
# Turn off logging - too noisy
# logger.info("We have an infeasible model, setting up slack model")
self.setup_slack()
self.m.optimize()

View file

@ -133,7 +133,8 @@ class GainOptimiser:
(self.m.status == OptimizationStatus.OPTIMAL) and not len(solution)
):
if self.allow_slack:
logger.info("We have an infeasible model, setting up slack model")
# Turn off logging - too noisy
# logger.info("We have an infeasible model, setting up slack model")
self.setup_slack()
self.m.optimize()
solution = [

View file

@ -231,8 +231,8 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
# We now produce a fabric only path for ECO4
# We add in generic insulation funding paths (where there is no fixed measure)
# Heating controls are only eligible if installed as part of a heating upgrade and so we do not include them
# here
if housing_type == "Social":
# here. We don't have an option if the property is a C or above
if housing_type == "Social" and p.data["current-energy-rating"] not in ["C", "B", "A"]:
funding_paths = (
[
{
@ -301,7 +301,6 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
# We log an error and skip this - we should not see any errors but we can probably get a reasonable
# outcome for the end user without a complete termination of the process
logger.error("Skipping fixed selection due to minimum insulation violation: %s", fixed)
blah
continue
scheme = _path_scheme(path_spec)
@ -339,7 +338,7 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
if fixed_gain > target_gain:
picked, sub_cost, sub_gain = ([], 0.0, 0.0)
elif fixed_gain < target_gain and not sub_measures:
elif fixed_gain <= target_gain and not sub_measures:
picked, sub_cost, sub_gain = ([], 0.0, 0.0)
else:
picked, sub_cost, sub_gain = run_optimizer(
@ -829,6 +828,11 @@ def make_funding_paths(p, input_measures, housing_type, funding: Funding):
:param funding: The funding object that provides methods to check eligibility and calculate funding.
:return:
"""
# If the property is currently EPC C, there is no funding availability
if p.data["current-energy-rating"] in ["C", "B", "A"]:
return [], input_measures
# We handle the case of minimum insulation requirements. Whenever we have a heating system recommendation,
# we *must* include an additional insulation measure, unless the property already has sufficient insulation.
@ -892,7 +896,7 @@ def make_funding_paths(p, input_measures, housing_type, funding: Funding):
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# 1) The package must include EWI or IWI if the property is private rental sector
# We check if we have any EWI or IWI measures available - only for EPC E or below
if p.data["current-energy-rating"] not in ["E", "F", "G"]:
if p.data["current-energy-rating"] in ["E", "F", "G"]:
ewi_or_iwi = [{"OR": []}]
reference_measures = []
# If we have EWI we add it in

View file

@ -12,7 +12,10 @@ class TestPrepareInputMeasures:
recs = [
[ # loft insulation measure
{"recommendation_id": "loft1", "type": "loft_insulation", "total": 100, "kwh_savings": 200,
"energy_cost_savings": 10, "has_battery": False, "measure_type": "loft_insulation"},
"energy_cost_savings": 10, "has_battery": False, "measure_type": "loft_insulation",
"partial_project_funding": 0, "partial_project_score": 0,
"uplift_project_score": 0,
},
],
]
measures = optimiser_functions.prepare_input_measures(recs, goal="Energy Savings", needs_ventilation=False)
@ -27,9 +30,12 @@ class TestPrepareInputMeasures:
["internal_wall_insulation"])
recs = [
[{"recommendation_id": "wall1", "type": "internal_wall_insulation", "total": 500, "kwh_savings": 300,
"energy_cost_savings": 5, "has_battery": False, "measure_type": "internal_wall_insulation"}],
"energy_cost_savings": 5, "has_battery": False, "measure_type": "internal_wall_insulation",
"partial_project_funding": 0, "partial_project_score": 0, "uplift_project_score": 0,
}],
[{"recommendation_id": "vent1", "type": "mechanical_ventilation", "total": 50, "kwh_savings": 30,
"energy_cost_savings": 5, "has_battery": False, "measure_type": "mechanical_ventilation"}],
"energy_cost_savings": 5, "has_battery": False, "measure_type": "mechanical_ventilation",
"partial_project_funding": 0, "partial_project_score": 0, "uplift_project_score": 0, }],
]
measures = optimiser_functions.prepare_input_measures(recs, goal="Energy Savings", needs_ventilation=True)
wall_option = measures[0][0]
@ -40,7 +46,8 @@ class TestPrepareInputMeasures:
def test_filters_out_negative_cost_savings(self):
recs = [
[{"recommendation_id": "bad1", "type": "loft_insulation", "total": 200, "kwh_savings": 100,
"energy_cost_savings": -5, "has_battery": False}],
"energy_cost_savings": -5, "has_battery": False,
"partial_project_funding": 0, "partial_project_score": 0, "uplift_project_score": 0, }],
]
measures = optimiser_functions.prepare_input_measures(recs, goal="Energy Savings", needs_ventilation=False)
assert measures == [] # should skip negative cost saving recs
@ -149,14 +156,14 @@ class TestIncreasingEpcE2e:
@pytest.fixture
def setup_case(self):
# Dummy property object
# Dummy property object
p = SimpleNamespace(
id="P1",
has_ventilation=False,
data={"current-energy-efficiency": "52"},
)
# Dummy request body
# Dummy request body
body = SimpleNamespace(
goal="Increasing EPC",
goal_value="C",
@ -165,9 +172,6 @@ class TestIncreasingEpcE2e:
simulate_sap_10=False,
required_measures=[]
)
# ✅ Use your massive measures_to_optimise list
recommendations = {"P1": measures_to_optimise}
return p, body, recommendations
@ -190,6 +194,18 @@ class TestIncreasingEpcE2e:
assert needs_ventilation
# Input the various things we need - set all to 0
for group in measures_to_optimise:
for r in group:
(
r["partial_project_score"],
r["partial_project_funding"],
r["innovation_uplift"],
r["uplift_project_score"],
) = (
0, 0, 0, 0
)
input_measures = optimiser_functions.prepare_input_measures(measures_to_optimise, body.goal, needs_ventilation)
assert input_measures, "Expected measures to optimise"

View file

@ -144,6 +144,15 @@ class DummyProp:
self.has_ventilation = False
self.floor_area = 70.0
self.main_heating_controls = {"clean_description": "time and temperature zone control"}
self.walls = {'original_description': 'Solid brick, as built, no insulation (assumed)',
'thermal_transmittance': None,
'thermal_transmittance_unit': None, 'is_cavity_wall': False, 'is_filled_cavity': False,
'is_solid_brick': True,
'is_system_built': False, 'is_timber_frame': False, 'is_granite_or_whinstone': False,
'is_as_built': True,
'is_cob': False, 'is_assumed': True, 'is_sandstone_or_limestone': False,
'insulation_thickness': 'none',
'external_insulation': False, 'internal_insulation': False}
self.main_heating = {
'original_description': 'Boiler and radiators, mains gas',
@ -230,6 +239,7 @@ def property_recommendations():
'quantity_unit': 'm2', 'total': 19090.810139104888,
'labour_hours': 0.0, 'labour_days': 0.0}],
'type': 'external_wall_insulation', 'measure_type': 'external_wall_insulation',
"innovation_rate": 0,
'description': 'Install 150mm EWI Pro EPS external wall insulation system with Brick '
'Slip finish on external walls',
'starting_u_value': 1.7, 'new_u_value': 0.32, 'already_installed': False,
@ -258,6 +268,7 @@ def property_recommendations():
'quantity_unit': 'm2', 'total': 5694.929118083911, 'labour_hours': 134.37473199973275,
'labour_days': 4.199210374991648}], 'type': 'internal_wall_insulation',
'measure_type': 'internal_wall_insulation',
"innovation_rate": 0,
'description': 'Install 95mm '
'SWIP EcoBatt & '
'Plastered '
@ -314,6 +325,7 @@ def property_recommendations():
'quantity_unit': 'm2', 'total': 645.0, 'labour_hours': 8,
'labour_days': 1}], 'type': 'loft_insulation',
'measure_type': 'loft_insulation',
"innovation_rate": 0,
'description': 'Install 300mm of Knauf Loft Roll 44 glass fibre roll in your loft',
'starting_u_value': 2.3, 'new_u_value': 2.3, 'sap_points': np.float64(2.4),
'already_installed': False,
@ -338,6 +350,7 @@ def property_recommendations():
'plant_cost': 0.0, 'total_cost': 350.0, 'notes': None, 'is_installer_quote': True, 'total': 700.0,
'quantity': 2,
'quantity_unit': 'part'}], 'type': 'mechanical_ventilation', 'measure_type': 'mechanical_ventilation',
"innovation_rate": 0,
'description': 'Install 2 '
'Mechanical '
'Extract '
@ -387,6 +400,7 @@ def property_recommendations():
'labour_hours': 70.08999999999999,
'labour_days': 2.920416666666666}],
'type': 'suspended_floor_insulation', 'measure_type': 'suspended_floor_insulation',
"innovation_rate": 0,
'description': 'Install 75mm Q-bot underfloor insulation insulation in suspended '
'floor',
'starting_u_value': 0.83, 'new_u_value': 0.22, 'sap_points': 2, 'survey': True,
@ -401,6 +415,7 @@ def property_recommendations():
'energy_cost_savings': np.float64(76.04936470588231)}], [
{'phase': 4, 'parts': [], 'type': 'low_energy_lighting',
'measure_type': 'low_energy_lighting',
"innovation_rate": 0,
'description': 'Install low energy lighting in -886 outlets', 'starting_u_value': None,
'new_u_value': None, 'already_installed': False, 'sap_points': 2,
'kwh_savings': -48508.5, 'energy_cost_savings': -12481.237049999998,
@ -413,6 +428,7 @@ def property_recommendations():
'recommendation_id': '5_phase=4', 'efficiency': -1705.5500000000002,
'heat_demand': np.float64(5.099999999999994)}], [
{'type': 'heating', 'phase': 5, 'measure_type': 'time_temperature_zone_control',
"innovation_rate": 0,
'parts': [],
'description': 'Upgrade heating controls to Smart Thermostats, room sensors and '
'smart radiator valves (time & temperature zone control)',
@ -431,6 +447,7 @@ def property_recommendations():
'energy_cost_savings': np.float64(65.29581176470589)}], [
{'phase': 6, 'parts': [], 'type': 'secondary_heating',
'measure_type': 'secondary_heating',
"innovation_rate": 0,
'description': 'Remove the secondary heating system', 'starting_u_value': None,
'new_u_value': None, 'sap_points': np.float64(3.6), 'already_installed': False,
'total': 30.0, 'subtotal': 25.0, 'vat': 5.0, 'labour_hours': 3.0,
@ -443,6 +460,7 @@ def property_recommendations():
'kwh_savings': np.float64(196.29999999999927),
'energy_cost_savings': np.float64(14.61857647058821)}], [
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
"innovation_rate": 0,
'description': 'Install a 4.0 kilowatt-peak (kWp) solar panel system.',
'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(13.0),
'already_installed': False, 'total': 6013.139999999999, 'subtotal': 5010.95, 'vat': 0,
@ -455,6 +473,7 @@ def property_recommendations():
'kwh_savings': np.float64(2040.8566307499998),
'energy_cost_savings': np.float64(525.1124110919749)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
"innovation_rate": 0,
'description': 'Install a 4.0 kilowatt-peak (kWp) solar panel system, with a battery.',
'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(13.0),
'already_installed': False, 'total': 10537.008, 'subtotal': 8780.84, 'vat': 0,
@ -467,6 +486,7 @@ def property_recommendations():
'kwh_savings': np.float64(2857.1992830499994),
'energy_cost_savings': np.float64(735.1573755287648)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
"innovation_rate": 0,
'description': 'Install a 3.6 kilowatt-peak (kWp) solar panel system.',
'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(12.0),
'already_installed': False, 'total': 5826.491999999999, 'subtotal': 4855.41, 'vat': 0,
@ -478,6 +498,7 @@ def property_recommendations():
'heat_demand': np.float64(83.69999999999999), 'kwh_savings': np.float64(1846.33397),
'energy_cost_savings': np.float64(475.0617304809999)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
"innovation_rate": 0,
'description': 'Install a 3.6 kilowatt-peak (kWp) solar panel system, with a battery.',
'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(12.0),
'already_installed': False, 'total': 10350.359999999999, 'subtotal': 8625.3, 'vat': 0,
@ -489,6 +510,7 @@ def property_recommendations():
'heat_demand': np.float64(83.69999999999999), 'kwh_savings': np.float64(2584.867558),
'energy_cost_savings': np.float64(665.0864226734)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
"innovation_rate": 0,
'description': 'Install a 3.2 kilowatt-peak (kWp) solar panel system.',
'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(11.0),
'already_installed': False, 'total': 5642.604, 'subtotal': 4702.17, 'vat': 0,
@ -500,6 +522,7 @@ def property_recommendations():
'kwh_savings': np.float64(1650.2708274),
'energy_cost_savings': np.float64(424.61468389001993)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
"innovation_rate": 0,
'description': 'Install a 3.2 kilowatt-peak (kWp) solar panel system, with a battery.',
'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(11.0),
'already_installed': False, 'total': 10166.472, 'subtotal': 8472.06, 'vat': 0,
@ -511,6 +534,7 @@ def property_recommendations():
'heat_demand': np.float64(78.3), 'kwh_savings': np.float64(2310.3791583599996),
'energy_cost_savings': np.float64(594.4605574460278)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
"innovation_rate": 0,
'description': 'Install a 2.8 kilowatt-peak (kWp) solar panel system.',
'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(9.0),
'already_installed': False, 'total': 5458.727999999999, 'subtotal': 4548.94, 'vat': 0,
@ -522,6 +546,7 @@ def property_recommendations():
'kwh_savings': np.float64(1453.5933906),
'energy_cost_savings': np.float64(374.00957940138)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
"innovation_rate": 0,
'description': 'Install a 2.8 kilowatt-peak (kWp) solar panel system, with a battery.',
'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(9.0),
'already_installed': False, 'total': 9982.596, 'subtotal': 8318.83, 'vat': 0,
@ -533,6 +558,7 @@ def property_recommendations():
'heat_demand': np.float64(64.0), 'kwh_savings': np.float64(2035.03074684),
'energy_cost_savings': np.float64(523.6134111619319)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
"innovation_rate": 0,
'description': 'Install a 2.4 kilowatt-peak (kWp) solar panel system.',
'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(8.0),
'already_installed': False, 'total': 5274.852, 'subtotal': 4395.71, 'vat': 0,
@ -544,6 +570,7 @@ def property_recommendations():
'kwh_savings': np.float64(1255.12594),
'energy_cost_savings': np.float64(322.94390436199996)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
"innovation_rate": 0,
'description': 'Install a 2.4 kilowatt-peak (kWp) solar panel system, with a battery.',
'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(8.0),
'already_installed': False, 'total': 9798.72, 'subtotal': 8165.6, 'vat': 0,
@ -555,6 +582,7 @@ def property_recommendations():
'heat_demand': np.float64(54.3), 'kwh_savings': np.float64(1757.1763159999998),
'energy_cost_savings': np.float64(452.1214661067999)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
"innovation_rate": 0,
'description': 'Install a 2.0 kilowatt-peak (kWp) solar panel system.',
'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(7.0),
'already_installed': False, 'total': 5090.976, 'subtotal': 4242.48, 'vat': 0,
@ -566,6 +594,7 @@ def property_recommendations():
'kwh_savings': np.float64(1048.341318),
'energy_cost_savings': np.float64(269.7382211214)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
"innovation_rate": 0,
'description': 'Install a 2.0 kilowatt-peak (kWp) solar panel system, with a battery.',
'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(7.0),
'already_installed': False, 'total': 9614.844, 'subtotal': 8012.369999999999, 'vat': 0,
@ -586,10 +615,20 @@ def _attach_costs_and_uplifts(recs, funding, p):
for group in out:
for r in group:
if r["type"] in ["mechanical_ventilation", "low_energy_lighting", "secondary_heating"]:
r["innovation_uplift"] = 0
(
r["partial_project_score"],
r["partial_project_funding"],
r["innovation_uplift"],
r["uplift_project_score"],
) = (
0, 0, 0, 0
)
continue
r["uplift"] = 0.0 # fixed for determinism in test
r["innovation_uplift"] = funding.get_innovation_uplift(
(
r["partial_project_score"], r["partial_project_funding"], r["innovation_uplift"],
r["uplift_project_score"]
) = funding.get_innovation_uplift(
measure=r,
starting_sap=55,
floor_area=70.0,
@ -663,3 +702,100 @@ def test_social_fabric_only_returns_only_fabric_types(p, funding, property_recom
unfunded_rows = solutions[
solutions["path"].apply(lambda x: isinstance(x, dict) and x.get("reference") == "unfunded:all")]
assert not unfunded_rows.empty
def test_private_solid_wall_no_innovation_epc_d(p, funding, mock_project_scores_matrix, mock_partial_scores_matrix):
"""
We have a specific test for this case which was implemented incorrectly originally.
This is an EPC D property and so shouldn't be eligible for ECO4. Instead, only GBIS should be considered.
"""
# Overwrite the data - copied from real example
p2 = deepcopy(p)
p2.data = {
"current-energy-rating": "D",
"current-energy-efficiency": 68,
"mainheat-energy-eff": "Good",
}
p2.walls = {'original_description': 'Sandstone or limestone, as built, no insulation (assumed)',
'clean_description': 'Sandstone or limestone, as built, no insulation', 'thermal_transmittance': None,
'thermal_transmittance_unit': None, 'is_cavity_wall': False, 'is_filled_cavity': False,
'is_solid_brick': False, 'is_system_built': False, 'is_timber_frame': False,
'is_granite_or_whinstone': False, 'is_as_built': True, 'is_cob': False, 'is_assumed': True,
'is_sandstone_or_limestone': True, 'is_park_home': False, 'insulation_thickness': 'none',
'external_insulation': False, 'internal_insulation': False}
funding2 = Funding(
tenure="Private",
project_scores_matrix=mock_project_scores_matrix,
partial_project_scores_matrix=mock_partial_scores_matrix,
whlg_eligible_postcodes=pd.DataFrame([{"Postcode": "ab12cd"}]),
eco4_social_cavity_abs_rate=12.5,
eco4_social_solid_abs_rate=17,
eco4_private_cavity_abs_rate=12.5,
eco4_private_solid_abs_rate=17,
gbis_social_cavity_abs_rate=21,
gbis_social_solid_abs_rate=25,
gbis_private_cavity_abs_rate=21,
gbis_private_solid_abs_rate=28,
)
input_measures = [
[{'id': '0_phase=0', 'cost': np.float64(4441.202499013676), 'gain': np.float64(3.4000000000000057),
'type': 'internal_wall_insulation+mechanical_ventilation', 'innovation_uplift': np.float64(0.0),
'cost_minus_uplift': np.float64(4441.202499013676), 'raw_cost': 3881.2024990136756,
'partial_project_funding': np.float64(2300.1000000000004), 'partial_project_score': np.float64(135.3),
'uplift_project_score': np.float64(0.0)}], [
{'id': '2_phase=2', 'cost': np.float64(2280.0), 'gain': np.float64(0.4), 'type': 'secondary_glazing',
'innovation_uplift': np.float64(0.0), 'cost_minus_uplift': np.float64(2280.0),
'raw_cost': np.float64(2280.0), 'partial_project_funding': np.float64(1421.1999999999998),
'partial_project_score': np.float64(83.6), 'uplift_project_score': np.float64(0.0)}], [
{'id': '3_phase=3', 'cost': np.float64(604.5840000000001), 'gain': np.float64(1.2),
'type': 'time_temperature_zone_control', 'innovation_uplift': np.float64(0.0),
'cost_minus_uplift': np.float64(604.5840000000001), 'raw_cost': 604.5840000000001,
'partial_project_funding': np.float64(702.0999999999999), 'partial_project_score': np.float64(41.3),
'uplift_project_score': np.float64(0.0)}], [
{'id': '4_phase=4', 'cost': 60.0, 'gain': np.float64(0.0), 'type': 'secondary_heating',
'innovation_uplift': 0, 'cost_minus_uplift': 60.0, 'raw_cost': 60.0, 'partial_project_funding': 0,
'partial_project_score': 0, 'uplift_project_score': 0}]
]
solutions = optimise_with_funding_paths(
p=p2,
input_measures=input_measures,
housing_type="Private",
budget=None,
target_gain=1.5,
funding=funding2
)
# 3) basic shape assertions
assert isinstance(solutions, pd.DataFrame)
assert not solutions.empty
# We should have 2 rows
assert solutions.shape[0] == 2
# We should only have None or GBIS
assert set(solutions["scheme"].unique()) == {"none", "gbis"}
meets_upgrade_gbis = solutions[solutions["meets_upgrade_target"] & solutions["is_eligible"]]
assert meets_upgrade_gbis.shape[0] == 1
# Check exact result
assert meets_upgrade_gbis.squeeze().to_dict() == {
'fixed_ids': ['0_phase=0'], 'items': [
{'id': '0_phase=0', 'cost': 3881.2024990136756, 'gain': np.float64(3.4000000000000057),
'type': 'internal_wall_insulation+mechanical_ventilation', 'innovation_uplift': np.float64(0.0),
'cost_minus_uplift': np.float64(4441.202499013676), 'raw_cost': 3881.2024990136756,
'partial_project_funding': np.float64(2300.1000000000004), 'partial_project_score': np.float64(135.3),
'uplift_project_score': np.float64(0.0)}], 'total_cost': 3881.2024990136756,
'total_gain': 3.4000000000000057, 'path': [{'AND': ['internal_wall_insulation+mechanical_ventilation'],
'reference':
'internal_wall_insulation+mechanical_ventilation:gbis'}],
'scheme': 'gbis', 'is_eligible': True, 'unfunded_items': [], 'meets_upgrade_target': True, 'starting_sap': 68,
'floor_area': 70.0, 'ending_sap': 71.4, 'starting_band': 'High_D', 'ending_band': 'Low_C',
'floor_area_band': '0-72', 'project_score': 540.0, 'full_project_funding': 0.0,
'partial_project_funding': 2300.1000000000004, 'partial_project_score': 135.3, 'total_uplift': 0.0,
'total_uplift_score': 0.0
}

View file

@ -66,7 +66,7 @@ functions:
- sqs:
arn: arn:aws:sqs:${self:provider.region}:${aws:accountId}:model-engine-queue
batchSize: 1
maximumConcurrency: 2 # Heavily restricts concurrency to avoid overwhelming the ldmbda limits
maximumConcurrency: 5 # Heavily restricts concurrency to avoid overwhelming the ldmbda limits
resources: