junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-08 11:17:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #476 from Hestia-Homes/debugging-api

Browse source 
Debugging api
This commit is contained in:
KhalimCK 2025-08-01 14:12:17 +01:00 • committed by GitHub
commit 6c6a44abfe
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
20 changed files with 1967 additions and 253 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

77
asset_list/AssetList.py
View file

@ -887,6 +887,9 @@ class AssetList:
self.landlord_year_built
].apply(extract_year)
for x in self.standardised_asset_list[self.landlord_year_built].values:
extract_year(x)
# We now create standard lookups
to_remap = {
self.landlord_property_type: {
@ -1099,6 +1102,13 @@ class AssetList:
)
# Estimate the perimeter
# Handle funky edge case
self.standardised_asset_list[self.EPC_API_DATA_NAMES["total-floor-area"]] = np.where(
(self.standardised_asset_list[self.EPC_API_DATA_NAMES["total-floor-area"]] == 0),
self.standardised_asset_list[self.EPC_API_DATA_NAMES["total-floor-area"]].mean(),
self.standardised_asset_list[self.EPC_API_DATA_NAMES["total-floor-area"]]
)
self.standardised_asset_list[self.ATTRIBUTE_ESTIMATED_PERIMETER] = self.standardised_asset_list.apply(
lambda x: estimate_perimeter(
floor_area=x[self.EPC_API_DATA_NAMES["total-floor-area"]] / x[self.ATTRIBUTE_NUMBER_OF_FLOORS],
@ -1753,7 +1763,9 @@ class AssetList:
# It's empty cavity
self.standardised_asset_list["cavity_is_empty"] |
# It's a cavity wall
(self.standardised_asset_list[self.STANDARD_WALL_CONSTRUCTION].str.contains("cavity"))
self.standardised_asset_list[self.STANDARD_WALL_CONSTRUCTION].isin(
["filled cavity", "partial insulated cavity"]
)
)
not_a_flat = (
@ -2097,6 +2109,7 @@ class AssetList:
RANGE_RE = re.compile(r'\b(\d+[A-Za-z]?)\s*[-–]\s*(\d+[A-Za-z]?)\b')
NUM_RE = re.compile(r'\b\d+[A-Za-z]?\b') # captures 12, 12A, etc.
TO_RANGE_RE = re.compile(r'\b(\d+[A-Za-z]?)\s+(?:to|To|TO)\s+(\d+[A-Za-z]?)\b') # captures "13 to 15"
expanded_rows = []
@ -2121,11 +2134,12 @@ class AssetList:
# 1 ─ Range (e.g. 1-7)
m_range = RANGE_RE.search(addr)
if m_range:
to_range = TO_RANGE_RE.search(addr)
start, end = m_range.groups()
if m_range or to_range:
start, end = m_range.groups() if m_range else to_range.groups()
start, end = int(re.match(r'\d+', start)[0]), int(re.match(r'\d+', end)[0])
if start > end or (end - start) > 100:
if start > end or (end - start) > 200:
raise ValueError(f"Suspicious range '{addr}'")
# We define the looping range on whether we have odd, even or all numbers
@ -2137,10 +2151,12 @@ class AssetList:
for n in house_number_range:
new = row.copy()
new_addr = RANGE_RE.sub(str(n), addr, count=1)
range_text = m_range.group(0) if m_range else to_range.group(0)
new_addr = addr.replace(range_text, str(n))
# Build the new full address by also swapping out the range_text
original_full_address = new[self.STANDARD_FULL_ADDRESS]
new_full_address = original_full_address.replace(addr, new_addr)
new[self.STANDARD_ADDRESS_1] = new_addr
new_full_address = original_full_address.replace(range_text, str(n))
new[self.STANDARD_ADDRESS_1] = str(n)
new[self.STANDARD_FULL_ADDRESS] = new_full_address
new[self.STANDARD_PROPERTY_TYPE] = "flat"
# Keep a record of the previous address 1
@ -2155,7 +2171,7 @@ class AssetList:
# 2 ─ Explicit list (e.g. 1, 2, 5 Block) or split by an ampersand (e.g. 1 & 2 Block)
nums = NUM_RE.findall(addr)
if len(nums) > 1 and (',' in addr or '&' in addr):
if len(nums) > 1 and (',' in addr or '&' in addr or ' and ' in addr.lower()):
for n in nums:
new = row.copy()
new_addr = re.sub(NUM_RE, n, addr, count=1) # replace the first number only
@ -2174,6 +2190,10 @@ class AssetList:
expanded_blocks = pd.DataFrame(expanded_rows)
# Check for duplicated domna ids
if expanded_blocks[self.DOMNA_PROPERTY_ID].duplicated().sum():
raise ValueError("expanded blocks has duplicated IDs")
# We drop the blocks from the standardised asset list and append on the expanded blocks
self.standardised_asset_list = self.standardised_asset_list[
self.standardised_asset_list[self.STANDARD_PROPERTY_TYPE] != "block of flats"
@ -2318,18 +2338,37 @@ class AssetList:
(~group["cavity_reason"].str.contains("(unlikely to quality)", case=False, na=False, regex=False))
).sum()
n_empties_high_confidence = (
(group["identified_empty_cavity"] == True) &
(~group["SAP Category"].isin(["SAP Rating 69-75", "SAP Rating 76 or more"])) &
(~pd.isnull(group["cavity_reason"])) &
(~group["cavity_reason"].str.contains("(unlikely to quality)", case=False, na=False, regex=False))
).sum()
# Average age of the EPCs
group["time_since_epc"] = (
pd.to_datetime("now") - pd.to_datetime(
group[self.EPC_API_DATA_NAMES["inspection-date"]])
).dt.days
average_age_of_epc = group["time_since_epc"].mean()
works = group["hubspot_status"]
above_threshold = works.map(LABEL_TO_ENUM.get).dropna()
count_above = (above_threshold >= threshold).sum()
proportion_surveyed = count_above / len(works)
proportion_empty = n_empties / len(works)
proportion_empty_high_confidence = n_empties_high_confidence / len(works)
# We auto-populate any blocks that have greater than 50% proportion empty
block_analysis.append(
{
"Block Reference": block_reference,
"Block Size": len(group),
"average_age_of_epc": average_age_of_epc,
"Proportion of properties suryeyed": proportion_surveyed,
"Percentage of Empties": proportion_empty,
"Percentage of Empties (high confidence)": proportion_empty_high_confidence,
**cavity_breakdown.to_dict(),
}
)
@ -3345,6 +3384,8 @@ class AssetList:
property_type_col = "PROPERTY TYPE As per table emailed"
elif "PROPERTY TYPE" in master_data.columns:
property_type_col = "PROPERTY TYPE"
elif 'Property Type' in master_data.columns:
property_type_col = 'Property Type'
else:
property_type_col = "PROPERTY TYPE (SEE DEEMED SCORES SHEET) Eg. 3W_Flat_1 (As per Matrix)"
@ -3496,8 +3537,20 @@ class AssetList:
]
if df.shape[0] != 1:
# We have multiple matches
raise NotImplementedError("FIX ME")
# We have multiple matches - it's likely because the landlord has a duplicate
# that has been referenced in totally different ways so we just match to both
for _, x in df.iterrows():
matched.append(
{
"row_id": row["row_id"],
"original_house_no": original_house_no,
"original_street": original_street,
"original_postcode": original_postcode,
self.STANDARD_LANDLORD_PROPERTY_ID: x[self.STANDARD_LANDLORD_PROPERTY_ID],
}
)
continue
matched.append(
{
"row_id": row["row_id"],
@ -3594,6 +3647,10 @@ class AssetList:
self.master_surveyed, how="left", on=self.STANDARD_LANDLORD_PROPERTY_ID
)
# Make sure no dupes
if self.standardised_asset_list[self.DOMNA_PROPERTY_ID].duplicated().sum():
raise ValueError("duplicated ids!")
# Finally, we keep a record of the unmatched
if unmatched_submissions:
self.unmatched_submissions = pd.concat(

112
asset_list/app.py
View file

@ -59,6 +59,110 @@ def app():
Property UPRN
"""
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Broadlands"
data_filename = "Broadlands Asset List.xlsx"
sheet_name = "Assets"
postcode_column = 'POSTCODE'
fulladdress_column = None
address1_column = "Address1"
address1_method = None
address_cols_to_concat = ["Address1"]
missing_postcodes_method = None
landlord_year_built = "DATEBUILT"
landlord_os_uprn = None
landlord_property_type = "PropertyType"
landlord_built_form = "PropertyType"
landlord_wall_construction = None
landlord_heating_system = "Heating Fuel"
landlord_existing_pv = None
landlord_property_id = "Row ID"
outcomes_filename = [os.path.join(data_folder, "outcomes.xlsx")]
outcomes_sheetname = ["Sheet1"]
outcomes_postcode = ["Postcode"]
outcomes_houseno = ["No."]
outcomes_address = ["Address"]
outcomes_id = [None]
master_filepaths = [
os.path.join(data_folder, "eco3 submissions.csv"),
os.path.join(data_folder, "eco4 submissions.csv"),
]
master_to_asset_list_filepath = None
asset_list_header = 0
landlord_block_reference = None
master_id_colnames = [None, None]
landlord_roof_construction = None
phase = False
landlord_sap = None
ecosurv_landlords = "broadland"
#
# Community:
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Community Housing/New Programme"
data_filename = "SUB EPC C to DOMNA - 24.07.25.xlsx"
sheet_name = "Sheet1"
postcode_column = 'POSTCODE'
fulladdress_column = "ADDRESS"
address1_column = None
address1_method = "house_number_extraction"
address_cols_to_concat = []
missing_postcodes_method = None
landlord_year_built = "BUILD DATE"
landlord_os_uprn = None
landlord_property_type = "PROPERTY TYPE"
landlord_built_form = "Archetype" # Using the inspections archetype
landlord_wall_construction = "CONSTRUCTION TYPE"
landlord_roof_construction = None
landlord_heating_system = None
landlord_existing_pv = None
landlord_property_id = "UPRN"
landlord_sap = None
outcomes_filename = []
outcomes_sheetname = []
outcomes_postcode = []
outcomes_houseno = []
outcomes_id = []
outcomes_address = []
master_filepaths = []
master_to_asset_list_filepath = None
phase = False
ecosurv_landlords = None
asset_list_header = 1
landlord_block_reference = None
master_id_colnames = []
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Ealing/Programme Analysis"
data_filename = "EalingProjectRebuildJW210725.xlsx"
sheet_name = "Refine & Houses"
postcode_column = 'Postcode'
fulladdress_column = "Address"
address1_column = None
address1_method = "house_number_extraction"
address_cols_to_concat = []
missing_postcodes_method = None
landlord_year_built = None
landlord_os_uprn = None
landlord_property_type = None # Using the inspections property type
landlord_built_form = None
landlord_wall_construction = None
landlord_roof_construction = None
landlord_heating_system = None
landlord_existing_pv = None
landlord_property_id = "Property ref"
landlord_sap = None
outcomes_filename = []
outcomes_sheetname = []
outcomes_postcode = []
outcomes_houseno = []
outcomes_id = []
outcomes_address = []
master_filepaths = []
master_to_asset_list_filepath = None
phase = False
ecosurv_landlords = None
asset_list_header = 0
landlord_block_reference = "Block Reference"
master_id_colnames = []
# TODO: Delete me
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/NRLA/"
data_filename = "20250716 Asset List.xlsx"
@ -148,7 +252,7 @@ def app():
landlord_existing_pv = None
landlord_property_id = "PropertyCode"
outcomes_filename = [os.path.join(data_folder, "Rooftop_Outcomes.xlsx")]
outcomes_sheetname = ["OUTCOMESs"]
outcomes_sheetname = ["OUTCOMES"]
outcomes_postcode = ["POSTCODE"]
outcomes_houseno = ["NO"]
outcomes_address = ["ADDRESS"]
@ -221,15 +325,15 @@ def app():
outcomes_houseno = []
outcomes_address = []
outcomes_id = []
master_filepaths = []
master_filepaths = [os.path.join(data_folder, "submissions.csv")]
master_to_asset_list_filepath = None
asset_list_header = 0
landlord_block_reference = None
master_id_colnames = []
master_id_colnames = [None]
landlord_roof_construction = None
phase = False
landlord_sap = None
ecosurv_landlords = None
ecosurv_landlords = "cds"
# Plus Dane
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Plus Dane/New Programme July 2025/"

55
asset_list/mappings/built_form.py
View file

@ -385,6 +385,59 @@ BUILT_FORM_MAPPINGS = {
'Maisonette Over Shop': 'mid-floor',
'Medium Rise Flat': 'mid-floor',
'Maisonette Medium Rise': 'unknown',
'End-terraced house': 'end-terrace'
'End-terraced house': 'end-terrace',
'Ground floor study bedroom': 'ground floor',
'End terrace bungalow': 'end-terrace',
'End terrace house': 'end-terrace',
'Ground floor bedsit': 'ground floor',
'Detached bungalow': 'detached',
'Lower ground floor flat': 'ground floor',
'Mid terrace bungalow': 'mid-terrace',
'Mid terrace house': 'mid-terrace',
'Basement bedsit': 'basement',
'Ground floor flat': 'ground floor',
'Ground floor flat with study': 'ground floor',
'Basement flat': 'basement',
'Semi bungalow': 'semi-detached',
'2nd floor flat': 'mid-floor',
'General/Communal': 'unknown',
'Semi house': 'semi-detached',
'2nd floor flat with study': 'mid-floor',
'1st floor flat with study room': 'mid-floor',
'Cluster House': 'detached',
'Utility pod': 'unknown',
'3rd floor flat': 'mid-floor',
'4th floor flat': 'mid-floor',
'2nd floor study bedroom': 'mid-floor',
'1st floor study bedroom': 'mid-floor',
'Dormer bungalow': 'detached',
'1st floor flat': 'mid-floor',
'Block property': 'unknown',
'Utility pod - DDA compliant': 'unknown',
'2nd floor bedsit': 'mid-floor',
'1st floor bedsit': 'mid-floor',
'2nd/3rd floor duplex flat': 'mid-floor',
'Bungalow - Detached': 'detached',
'Maisonette - Detached': 'detached',
'Bedsit - Mid Terrace': 'mid-terrace',
'House - End Terrace': 'end-terrace',
'House - Mid Terrace': 'mid-terrace',
'Bungalow - End Terrace': 'end-terrace',
'Maisonette - End Terrace': 'end-terrace',
'Maisonette - Semi Detached': 'semi-detached',
'House - Detached': 'detached',
'Bedsit - End Terrace': 'end-terrace',
'House - Semi detached': 'semi-detached',
'Studio Flat - Mid Terrace': 'mid-terrace',
'Bungalow - Semi detached': 'semi-detached',
'Amenity Block - Detached': 'detached',
'Bungalow - Mid Terrace': 'mid-terrace',
'Amenity Block - Semi detached': 'semi-detached',
'Maisonette - Mid Terrace': 'mid-terrace',
'Chalet - Wheelchair': 'unknown',
'Studio Flat': 'unknown',
'Bungalow - Attached': 'semi-detached'
}

56
asset_list/mappings/heating_systems.py
View file

@ -377,6 +377,60 @@ HEATING_MAPPINGS = {
'Warm air Electricity': 'warm air heating',
'None': 'no heating',
'Boiler None': 'unknown',
'Storage heaters Electricity': 'electric storage heaters'
'Storage heaters Electricity': 'electric storage heaters',
'Unknown when old solid fuel system was removed': 'solid fuel',
'Storage Heater': 'electric storage heaters',
'Combi': 'gas condensing combi',
'Combi condensing': 'gas condensing combi',
'Combi Condensing': 'gas condensing combi',
'Tenant Burner': 'unknown',
'Wall Mounted Condens': 'gas condensing boiler',
'Gas Pipework': 'unknown',
'Open Fire Bck Boiler': 'solid fuel',
'Back Boiler Unit': 'solid fuel',
'Sharedgasboiler': 'communal gas boiler',
'Wall Mntd Condensing': 'gas condensing boiler',
'Flr Standing Combi': 'gas combi boiler',
'Oil - Tenant': 'oil boiler',
'Open Flue Fire': 'solid fuel',
'Wall Mounted Fire': 'room heaters',
'Gas - Unvented Cylinder': 'gas boiler, radiators',
'Commercial Pipework': 'unknown',
'Wall Mntd Condensin': 'gas condensing boiler',
'Offpeakelectric': 'electric storage heaters',
'Closed Burner': 'unknown',
'Domesticgasboiler': 'gas boiler, radiators',
'Elec - Storage': 'electric storage heaters',
'Share Common Boiler': 'communal heating',
'Down Flow Heater': 'electric radiators',
'Inset Flame Effect': 'electric radiators',
'Closedmulti': 'unknown',
'Open Fire': 'solid fuel',
'Lpgas - Domesticgasboiler': 'gas boiler, radiators',
'Solarpvpanels': 'other',
'Renew - Ashp': 'air source heat pump',
'Room Sealed App': 'unknown',
'5 Year Periodic Insp': 'unknown',
'Solarthermal': 'other',
'Wall Mounted Combi': 'gas combi boiler',
'Woodburner': 'solid fuel',
'Sealed System Wl Mtd': 'unknown',
'Room Seal App': 'unknown',
'Shared Gas Boiler': 'communal gas boiler',
'Heating Distribution': 'unknown',
'Flr Standing Boiler': 'boiler - other fuel',
'Multifuel Burner': 'solid fuel',
'Gas - Shared': 'communal gas boiler',
'Wall Mounted Boiler': 'gas boiler, radiators',
'Tenant Boiler': 'gas boiler, radiators',
'Gas - Domesticgasboiler': 'gas boiler, radiators',
'Domestic gas boiler': 'gas boiler, radiators',
'Combination': 'unknown',
'Mains Electric': 'electric fuel',
'Unvented cylinder': 'other',
'MVHR & Heat Recovery': 'other',
'Solar': 'other'
}

55
asset_list/mappings/property_type.py
View file

@ -283,6 +283,59 @@ PROPERTY_MAPPING = {
'Flat Over Shop': 'flat',
'Medium Rise Flat': 'flat',
'End Terraced Town House': 'house',
'Maisonette Medium Rise': 'maisonette'
'Maisonette Medium Rise': 'maisonette',
'Semi bungalow': 'bungalow',
'2nd floor flat': 'flat',
'End terrace bungalow': 'bungalow',
'End terrace house': 'house',
'Ground floor bedsit': 'bedsit',
'Detached bungalow': 'bungalow',
'Semi house': 'house',
'2nd floor flat with study': 'flat',
'1st floor flat with study room': 'flat',
'Lower ground floor flat': 'flat',
'Cluster House': 'house',
'Mid terrace bungalow': 'bungalow',
'Mid terrace house': 'house',
'Basement bedsit': 'bedsit',
'Detached house': 'house',
'3rd floor flat': 'flat',
'4th floor flat': 'flat',
'Dormer bungalow': 'bungalow',
'1st floor flat': 'flat',
'Ground floor flat': 'flat',
'Ground floor flat with study': 'flat',
'Basement flat': 'flat',
'2nd floor bedsit': 'bedsit',
'1st floor bedsit': 'bedsit',
'2nd/3rd floor duplex flat': 'flat',
'Ground floor study bedroom': 'other',
'General/Communal': 'other',
'Utility pod': 'other',
'2nd floor study bedroom': 'other',
'1st floor study bedroom': 'other',
'Block property': 'block of flats',
'Utility pod - DDA compliant': 'other',
'Bungalow - Detached': 'bungalow',
'Maisonette - Detached': 'maisonette',
'Bedsit - Mid Terrace': 'bedsit',
'Studio Flat': 'flat',
'House - End Terrace': 'house',
'House - Mid Terrace': 'house',
'Bungalow - End Terrace': 'bungalow',
'Bungalow - Attached': 'bungalow',
'Maisonette - End Terrace': 'maisonette',
'Maisonette - Semi Detached': 'maisonette',
'House - Detached': 'house',
'Bedsit - End Terrace': 'bedsit',
'House - Semi detached': 'house',
'Studio Flat - Mid Terrace': 'flat',
'Bungalow - Semi detached': 'bungalow',
'Bungalow - Mid Terrace': 'bungalow',
'Maisonette - Mid Terrace': 'maisonette',
'Chalet - Wheelchair': 'other',
'Amenity Block - Detached': 'other',
'Amenity Block - Semi detached': 'other'
}

26
backend/Funding.py
View file

@ -1,3 +1,4 @@
from enum import Enum
import pandas as pd
import numpy as np
from typing import List
@ -413,6 +414,10 @@ class FundingOld:
self.whlg()
class EligibilityCaveats(Enum):
TENANT_ON_BENEFITS_OR_LOW_INCOME = "tenant_on_benefits_or_low_income"
class Funding:
"""
New class to handle funding calculation
@ -440,6 +445,9 @@ class Funding:
self.project_scores_matrix = project_scores_matrix
self.whlg_eligible_postcodes = whlg_eligible_postcodes
self.eco4_eligible = False
self.eligbility_caveat = None
@staticmethod
def get_sap_band(sap_score_number):
bands = [
@ -478,9 +486,8 @@ class Funding:
return "200"
@staticmethod
def eco4_prs_eligibility(
starting_sap: int, measures: List, mainheat_description: str, heating_control_description: str
self, starting_sap: int, measures: List, mainheat_description: str, heating_control_description: str
):
"""
Handles the eligibility criteria for private rental properties under eco
@ -509,11 +516,19 @@ class Funding:
# Is a renewable heating
ashp = "air_source_heat_pump" in measures
# Meets the EPC criteria, has the measure requirement and tenant must be on benefits
if meets_epc & (solar_renweable_heating or ashp or has_solid_wall):
return True
self.eco4_eligible = True
self.eligbility_caveat = EligibilityCaveats.TENANT_ON_BENEFITS_OR_LOW_INCOME
return
return False
def gbis_prs_eligibiltiy(self):
"""
Determines if a project is eligible for GBIS funding for private rental properties
"""
def calculate_full_project_abs(self):
# Filter the project scores matrix
@ -568,7 +583,7 @@ class Funding:
# 2) GBIS
if self.tenure == "Private":
is_eco4_eligible = self.eco4_prs_eligibility(
self.eco4_prs_eligibility(
starting_sap=starting_sap,
measures=measures,
mainheat_description=mainheat_description,
@ -578,7 +593,8 @@ class Funding:
# Need to implement
# 1) Package has to include an insulation measure
# 2) We should use the funding for the measure that has the largest partial project score
is_gbis_eligible = ()
# TODO: check the rules around GBIS eligibility and heating controls
self.gbis_prs_eligibiltiy()
if not is_eco4_eligible:
return

17
backend/app/plan/schemas.py
View file

@ -18,6 +18,12 @@ SPECIFIC_MEASURES = [
"cylinder_thermostat"
]
INSULATION_MEASURES = [
"internal_wall_insulation", "external_wall_insulation", "cavity_wall_insulation",
"loft_insulation", "flat_roof_insulation", "room_roof_insulation",
"suspended_floor_insulation", "solid_floor_insulation",
]
NON_INVASIVE_SPECIFIC_MEASURES = [
"trickle_vents", "draught_proofing", "mixed_glazing", "cavity_extract_and_refill",
"extension_cavity_wall_insulation"
@ -36,7 +42,7 @@ MEASURE_MAP = {
"heating_controls": ["roomstat_programmer_trvs", "time_temperature_zone_control"]
}
VALID_GOALS = ["Increasing EPC"]
VALID_GOALS = ["Increasing EPC", "Energy Savings", "Reducing CO2 emissions"]
VALID_HOUSING_TYPES = ["Social", "Private"]
VALID_EVENT_TYPES = ["remote_assessment"]
@ -74,7 +80,7 @@ class PlanTriggerRequest(BaseModel):
budget: Optional[float] = None
goal: Goal
housing_type: HousingType
goal_value: str
goal_value: Optional[str] = None
portfolio_id: int
trigger_file_path: str
already_installed_file_path: Optional[str] = None
@ -118,3 +124,10 @@ class PlanTriggerRequest(BaseModel):
if (self.index_start is None) != (self.index_end is None):
raise ValueError("Both index_start and index_end must be set or both must be None")
return self
@model_validator(mode="after")
def check_goal_value_requirement(self):
# Make sure that goal_value is set when goal is "Increasing EPC"
if self.goal == "Increasing EPC" and not self.goal_value:
raise ValueError("goal_value is required when goal is 'Increasing EPC'")
return self

186
backend/engine/engine.py
View file

@ -26,7 +26,7 @@ from backend.app.db.functions.energy_assessment_functions import get_latest_asse
from backend.app.db.models.portfolio import rating_lookup
from backend.app.plan.schemas import PlanTriggerRequest
from backend.app.plan.utils import get_cleaned
from backend.app.utils import epc_to_sap_lower_bound, sap_to_epc
from backend.app.utils import sap_to_epc
import backend.app.assumptions as assumptions
from backend.ml_models.api import ModelApi
@ -35,7 +35,7 @@ from backend.apis.GoogleSolarApi import GoogleSolarApi
from recommendations.optimiser.CostOptimiser import CostOptimiser
from recommendations.optimiser.GainOptimiser import GainOptimiser
from recommendations.optimiser.optimiser_functions import prepare_input_measures
import recommendations.optimiser.optimiser_functions as optimiser_functions
from recommendations.Recommendations import Recommendations
from utils.logger import setup_logger
from utils.s3 import read_dataframe_from_s3_parquet, read_csv_from_s3, read_excel_from_s3
@ -798,157 +798,59 @@ async def model_engine(body: PlanTriggerRequest):
# we need to double unlist because we have a list of lists
property_measure_types = {rec["type"] for recs in recommendations[p.id] for rec in recs}
property_required_measures = [m for m in recommendations[p.id] if m[0]["type"] in body.required_measures]
measures_to_optimise = [m for m in recommendations[p.id] if m[0]["type"] not in body.required_measures]
property_required_measures = [
m for m in recommendations[p.id] if m[0]["type"] in body.required_measures
]
measures_to_optimise = [
m for m in recommendations[p.id] if m[0]["type"] not in body.required_measures
]
# If we have a wall insulation measure, we MUST include mechanical ventilation
# Additionally, if we have required measures, they should also be included. Therefore
# we can discount the number of points required to get to the target SAP band (or increase)
# in the case of ventilation
# If a measure requiring ventilation is selected, and the property does not have ventilation, we enfore
# its inclusion
needs_ventilation = any(
x in property_measure_types for x in assumptions.measures_needing_ventilation) and not p.has_ventilation
x in property_measure_types for x in assumptions.measures_needing_ventilation
) and not p.has_ventilation
input_measures = prepare_input_measures(measures_to_optimise, body.goal, needs_ventilation)
input_measures = optimiser_functions.prepare_input_measures(
measures_to_optimise, body.goal, needs_ventilation
)
if not input_measures[0]:
# This means that we have no defaults
selected_recommendations = {}
solution = []
# Nothing to do, we just reshape the recommendations
recommendations[p.id] = optimiser_functions.flatten_recommendations_with_defaults(
p.id, recommendations, set()
)
continue
fixed_gain = optimiser_functions.calculate_fixed_gain(
property_required_measures, recommendations, p, needs_ventilation
)
gain = optimiser_functions.calculate_gain(body=body, p=p, fixed_gain=fixed_gain)
if not body.optimise:
if body.goal != "Increasing EPC":
raise NotImplementedError("Only EPC optimisation is currently supported")
solution = [max(sub_list, key=lambda x: (x['gain'], -x['cost'])) for sub_list in input_measures]
else:
optimiser = (
GainOptimiser(
input_measures, max_cost=body.budget, max_gain=gain, allow_slack=body.goal == "Increasing EPC"
) if body.budget else CostOptimiser(input_measures, min_gain=gain)
)
optimiser.setup()
optimiser.solve()
solution = optimiser.solution
fixed_gain = 0
if property_required_measures:
# We get the SAP points for the required measures
if body.goal != "Increasing EPC":
raise NotImplementedError("Only EPC optimisation is currently supported")
sap_by_type = [
{"type": rec["type"], "sap_points": rec["sap_points"]} for recs in property_required_measures
for rec in recs
]
# We get a MAX sap points per type
max_per_type = (
pd.DataFrame(sap_by_type).groupby("type")["sap_points"].max().to_dict()
)
fixed_gain = sum(max_per_type.values())
property_required_measure_types = {rec["type"] for rec in sap_by_type}
# if the property needs ventilation, but the measure we optimise didn't include
# venilation we add the points for ventilation as a fixed gain
if needs_ventilation and any(
r in property_required_measure_types for r in assumptions.measures_needing_ventilation
):
fixed_gain += next(
(r[0]["sap_points"] for r in recommendations[p.id] if
r[0]["type"] == "mechanical_ventilation"),
0
)
current_sap_points = int(p.data["current-energy-efficiency"])
sap_gain = CostOptimiser.calculate_sap_gain_with_slack(
epc_to_sap_lower_bound(body.goal_value) - current_sap_points
) - fixed_gain
if body.simulate_sap_10:
# We add 3 additional SAP points to the required gain to account for SAP 10
sap_gain += 3
if not body.optimise:
if body.goal != "Increasing EPC":
raise NotImplementedError("Only EPC optimisation is currently supported")
solution = []
for sub_list in input_measures:
# Select the entry with the highest gain, and if tied, choose the one with the lowest cost
best_measure = max(sub_list, key=lambda x: (x['gain'], -x['cost']))
solution.append(best_measure)
else:
if body.budget:
optimiser = GainOptimiser(
input_measures, max_cost=body.budget, max_gain=sap_gain if sap_gain > 0 else 0
)
else:
# The minimum gain is the minimum number of SAP points required to get to the target SAP band
# If the gain is negative, the optimiser will return an empty solution
optimiser = CostOptimiser(
input_measures,
min_gain=sap_gain
)
optimiser.setup()
optimiser.solve()
solution = optimiser.solution
selected_recommendations = {r["id"] for r in solution}
selected = {r["id"] for r in solution}
if property_required_measures:
# We select the cheapest of the required measures, into selected
for recs in property_required_measures:
# We select the cheapest of the required measures
cost_to_id = {
rec["recommendation_id"]: rec["total"] for rec in recs
if rec["recommendation_id"] not in selected_recommendations
}
# Take the recommendation id with the lowers cost
selected_recommendations.add(min(cost_to_id, key=cost_to_id.get))
# Update the solution with the selected recommendaitons
solution = []
for recs in recommendations[p.id]:
for rec in recs:
if rec["recommendation_id"] in selected_recommendations:
solution.append(
{
"id": rec["recommendation_id"],
"cost": rec["total"],
"gain": rec["sap_points"],
"type": rec["type"]
}
)
# If wall insulation is selected, we also include mechanical ventilation as a best practice measure
ventilation_selected = [
r for r in solution if "+mechanical_ventilation" in r["type"]
]
if (any(x in [r["type"] for r in solution] for x in assumptions.measures_needing_ventilation) or
len(ventilation_selected)):
ventilation_rec = next(
(r[0] for r in recommendations[p.id] if r[0]["type"] == "mechanical_ventilation"),
None
solution = optimiser_functions.add_required_measures(
property_id=p.id, property_required_measures=property_required_measures,
recommendations=recommendations, selected=selected,
)
# If a matching recommendation was found, add its ID to the selected recommendations
if ventilation_rec:
selected_recommendations.add(ventilation_rec["recommendation_id"])
# If we have a trickle vents recommendation, we also switch it on. We don't just check the solution
trickle_vents_rec = next(
(r[0] for r in recommendations[p.id] if r[0]["type"] == "trickle_vents"),
None
# Add best practice measures (ventilation/trickle vents)
selected = optimiser_functions.add_best_practice_measures(p.id, solution, recommendations, selected)
# Final flattening
recommendations[p.id] = optimiser_functions.flatten_recommendations_with_defaults(
p.id, recommendations, selected
)
# If a matching recommendation was found, add its ID to the selected recommendations
if trickle_vents_rec:
selected_recommendations.add(trickle_vents_rec["recommendation_id"])
# We'll use the set of selected recommendations to filter the recommendations to upload
final_recommendations = [
[
{**rec, "default": True if rec["recommendation_id"] in selected_recommendations else False}
for rec in recommendations_by_type
]
for recommendations_by_type in recommendations[p.id]
]
# We'll also unlist the recommendations so they're a bit easier to handle from here onwards
recommendations[p.id] = [
rec for recommendations_by_type in final_recommendations for rec in recommendations_by_type
]
# when we have buildings, we tweak our solar PV recommendations as if one unit needs it, we apply it to all
# of them
@ -1111,6 +1013,8 @@ async def model_engine(body: PlanTriggerRequest):
[sum(r["labour_days"] for r in rec_group if r["default"]) for p_id, rec_group in recommendations.items()]
))
# TODO - This code only pulls in the properties that have been updated in this run, but we need to
# aggregate all properties in the portfolio. We likely need to trigger a re-aggregation
aggregated_data = extract_portfolio_aggregation_data(
input_properties=input_properties,
total_valuation_increase=total_valuation_increase,

49
backend/tests/test_funding.py
View file

@ -25,28 +25,27 @@ def get_funding_data():
return project_scores_matrix, whlg_eligible_postcodes
class TestFunding:
def test_prs(self):
eco_project_scores_matrix, whlg_eligible_postcodes = get_funding_data()
funding = Funding(
project_scores_matrix=eco_project_scores_matrix,
whlg_eligible_postcodes=whlg_eligible_postcodes,
social_cavity_abs_rate=13.5,
social_solid_abs_rate=17,
private_cavity_abs_rate=13.5,
private_solid_abs_rate=17,
tenure="Private",
)
measures_1 = ["internal_wall_insulation", "solar_pv"]
funding.check_funding(
measures=measures_1,
starting_sap=54,
ending_sap=69,
floor_area=73,
mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True
)
# class TestFunding:
#
# def test_prs(self):
# eco_project_scores_matrix, whlg_eligible_postcodes = get_funding_data()
# funding = Funding(
# project_scores_matrix=eco_project_scores_matrix,
# whlg_eligible_postcodes=whlg_eligible_postcodes,
# social_cavity_abs_rate=13.5,
# social_solid_abs_rate=17,
# private_cavity_abs_rate=13.5,
# private_solid_abs_rate=17,
# tenure="Private",
# )
#
# measures_1 = ["internal_wall_insulation", "solar_pv"]
# funding.check_funding(
# measures=measures_1,
# starting_sap=54,
# ending_sap=69,
# floor_area=73,
# mainheat_description="Boiler and radiators, mains gas",
# heating_control_description="Programmer, room thermostat and TRVs",
# is_cavity=True
# )

567
epr_data_exports/app.py Normal file
View file

@ -0,0 +1,567 @@
"""
This is a placeholder script to extract epr data from files, where we can
"""
"""
July 2025 LiveWest Heating Upgrades
"""
import os
import re
import PyPDF2
import pandas as pd
from tqdm import tqdm
from collections import Counter
def extract_window_age_description(windows_text):
"""
Extracts the most common window age description and its proportion.
Parameters:
windows_text (str): The text section containing window data.
Returns:
dict: A dictionary with the most common window age description and its proportion.
"""
# Clean up windows_text by removing line breaks for better pattern matching
windows_text = windows_text.replace("\n", "")
# Define possible window age descriptions
window_descriptions = [
"Double post or during 2002",
"Double pre 2002",
"Double with unknown install date",
"Secondary glazing",
"Triple glazing",
"Single glazing",
"Double between 2002 \nand 2021",
"Double between 2002 and 2021"
]
# Count occurrences of each description
description_counts = Counter()
for description in window_descriptions:
matches = re.findall(re.escape(description), windows_text)
description_counts[description] = len(matches)
if not description_counts or not sum(description_counts.values()):
raise ValueError("Failed to extract window data.")
# Determine the most common description and calculate its proportion
most_common_description, window_count = description_counts.most_common(1)[0]
window_proportion = window_count / sum(description_counts.values()) * 100
# Get the second most common and the proportion
if window_proportion == 100:
second_most_common_description = None
second_most_common_proportion = 0
else:
second_most_common_description, second_window_count = description_counts.most_common(2)[1]
second_most_common_proportion = second_window_count / sum(description_counts.values()) * 100
return {
"Window Age Description": most_common_description,
"Window Age Description Proportion (%)": window_proportion,
"Secondary Window Age Description": second_most_common_description,
"Secondary Window Age Description Proportion (%)": second_most_common_proportion,
"Number of Windows": sum(description_counts.values())
}
def extract_building_parts_summary(text):
"""
Extracts building parts and associated dimensions from the summary report PDF.
This includes Main Property, multiple extensions if they exist, and Room in Roof areas.
"""
data = []
# Locate the Dimensions section
dimensions_section = re.search(
r"Dimensions:\s*Dimension type: Internal\n(.*?)\n5\.0 Conservatory:", text, re.DOTALL
)
if not dimensions_section:
dimensions_section = re.search(
r"Dimensions:\s*Dimension type: External\n(.*?)\n5\.0 Conservatory:", text, re.DOTALL
)
if not dimensions_section:
raise ValueError("Failed to locate dimensions section in the text.")
dimensions_text = dimensions_section.group(1)
# Pattern to extract each building part, starting from Main Property and including extensions
building_part_pattern = re.compile(
r"(Main Property|\d+(?:st|nd|rd|th) Extension)\s*"
r"(.*?)(?=\d+(?:st|nd|rd|th) Extension|5\.0 Conservatory|$)",
re.DOTALL
)
# Loop through each building part match, including Main Property and extensions
for match in building_part_pattern.finditer(dimensions_text):
part_name = match.group(1)
floor_data = match.group(2)
# Pattern to extract floor details: Floor Level, Floor Area, Room Height, Perimeter, Party Wall Length
floor_pattern = re.compile(
r"(1st Floor|Lowest Floor|Second floor):\s*([\d.]+)\s+([\d.]+)\s+([\d.]+)\s+([\d.]+)"
)
# Extract data for each floor within the building part
for floor_match in floor_pattern.finditer(floor_data):
floor_level = floor_match.group(1)
floor_area = float(floor_match.group(2))
room_height = float(floor_match.group(3))
perimeter = float(floor_match.group(4))
party_wall_length = float(floor_match.group(5))
# Append to data list
data.append({
"Building Part": part_name,
"Floor Level": floor_level,
"Floor Area (m2)": floor_area,
"Room Height (m)": room_height,
"Perimeter (m)": perimeter,
"Party Wall Length (m)": party_wall_length
})
# Check specifically for "Room(s) in Roof" entries, which only have Floor Area
room_in_roof_pattern = re.compile(r"Room\(s\) in Roof:\s*([\d.]+)")
room_in_roof_match = room_in_roof_pattern.search(floor_data)
if room_in_roof_match:
floor_area = float(room_in_roof_match.group(1))
data.append({
"Building Part": part_name,
"Floor Level": "Room in Roof",
"Floor Area (m2)": floor_area,
"Room Height (m)": None, # Placeholder for missing data
"Perimeter (m)": None, # Placeholder for missing data
"Party Wall Length (m)": None # Placeholder for missing data
})
# Calculate aggregated dimensions
main_property = [part for part in data if "Main Property" in part["Building Part"]]
first_extensions = [part for part in data if "1st Extension" in part["Building Part"]]
dimensions = {
"Total Floor Area (m2)": sum([part["Floor Area (m2)"] for part in data]),
"Total Ground Floor Area (m2)": sum(
[part["Floor Area (m2)"] for part in data if "Lowest Floor" in part["Floor Level"]]
),
"RIR Floor Area": sum(
[part["Floor Area (m2)"] for part in data if "Room in Roof" in part["Floor Level"]]
),
"Main Building Wall Area (m2)": sum([x["Perimeter (m)"] * x["Room Height (m)"] for x in main_property if
x["Perimeter (m)"] and x["Room Height (m)"]]),
"First Extension Wall Area (m2)": sum(
[x["Perimeter (m)"] * x["Room Height (m)"] for x in first_extensions if
x["Perimeter (m)"] and x["Room Height (m)"]]
),
}
return dimensions
def extract_roof_details_summary(text):
"""
Extracts roof type, insulation, and insulation thickness for each building part
in the 8.0 Roofs section of the summary report.
"""
# Define data structure to hold results
roof_data = []
# Locate the entire 8.0 Roofs section
roof_section_match = re.search(r"8\.0 Roofs:\n(.*?)(?=\n9\.0 Floors:|$)", text, re.DOTALL)
if not roof_section_match:
return roof_data # Return empty if no roof section is found
# Extract the roof section and append "9.0 Floors:" as the boundary
roof_section = roof_section_match.group(1).strip() + "\n9.0 Floors:"
# Define pattern to match each building part's roof entry
building_part_pattern = re.compile(
r"(Main Property|1st Extension|2nd Extension|[\w\s]+)\n" # Matches each building part label
r"Type\s+(.*?)(?=\n(?:Insulation|9\.0 Floors:|[A-Z]))" # Matches Roof Type until the next field, label, or end
r"(?:\nInsulation\s+(.*?)(?=\n(?:Insulation Thickness|9\.0 Floors:|[A-Z])))?" # Optional Insulation
r"(?:\nInsulation Thickness\s+(.*?)(?=\n(?:9\.0 Floors:|[A-Z])))?", # Optional Insulation Thickness
re.DOTALL
)
# Extract each building part's data
for match in building_part_pattern.finditer(roof_section):
part_name = match.group(1).strip() # Building part label
roof_type = match.group(2).strip() # Roof Type
roof_insulation = match.group(3).strip() if match.group(3) else None # Optional Insulation
roof_insulation_thickness = match.group(4).strip() if match.group(4) else None # Optional Thickness
# Cleaning to handle annoying cases when it comes out like this:
# 'A Another dwelling above\n1st Extension'
if roof_type.startswith("A Another dwelling above"):
roof_type = "A Another dwelling above"
# Store results for this building part
roof_data.append({
"Building Part": part_name,
"Roof Type": roof_type,
"Roof Insulation": roof_insulation,
"Roof Insulation Thickness": roof_insulation_thickness,
})
return roof_data
def extract_wall_details_summary(text):
"""
Extracts wall type, insulation, dry-lining, and thickness for each building part,
including any alternative wall details within the 7.0 Walls section of the summary PDF text.
"""
# Define data structure to hold all building part wall entries
wall_data = []
# Locate the entire 7.0 Walls section
wall_section = re.search(r"7\.0 Walls:\n(.*?)\n8\.0 Roofs:", text, re.DOTALL).group(1)
# Define pattern to match each building part's wall entry within the section
building_part_pattern = re.compile(
r"(Main Property|1st Extension|2nd Extension|[\w\s]+)\n" # Matches each building part label
r"Type\s+(.*?)\n" # Matches main wall Type
r"Insulation\s+(.*?)\n", # Matches main wall Insulation
# r"(Dry-lining\s+(.*?)\n)?" # Optional main wall Dry-lining
# r"Wall Thickness Unknown\s+(.*?)\n" # Matches main wall Thickness Unknown
# r"Wall Thickness \[mm\]\s+(\d+)", # Matches main wall Thickness
re.DOTALL
)
# Define pattern to capture alternative wall details, if present
alternative_wall_pattern = re.compile(
r"Alternative Wall Area.*?\n" # Matches start of alternative wall section
r"Alternative Type\s+(.*?)\n" # Matches alternative wall Type
r"Alternative Insulation\s+(.*?)\n" # Matches alternative wall Insulation
r"(Alternative Dry-lining\s+(.*?)\n)?" # Optional Alternative Dry-lining
r"Alternative Wall Thickness Unknown\s+(.*?)\n" # Matches alternative wall Thickness Unknown
r"Alternative Wall Thickness\s+(\d+)", # Matches alternative wall Thickness
re.DOTALL
)
# Find all building part entries within the 7.0 Walls section
for match in building_part_pattern.finditer(wall_section):
wall_label = match.group(1).strip()
main_wall_type = match.group(2).strip()
main_wall_insulation = match.group(3).strip()
# main_wall_dry_lining = match.group(5).strip() if match.group(5) else "N/A"
# main_wall_thickness_unknown = match.group(6).strip()
# main_wall_thickness = int(match.group(7))
# Initialize dictionary for this wall entry
wall_entry = {
"Building Part": wall_label,
"Wall Type": main_wall_type,
"Wall Insulation": main_wall_insulation,
# "Wall Dry-lining": main_wall_dry_lining,
# "Wall Thickness Unknown": main_wall_thickness_unknown,
# "Wall Thickness (mm)": main_wall_thickness,
"Alternative Wall Type": None,
"Alternative Wall Insulation": None,
"Alternative Wall Dry-lining": "N/A",
"Alternative Wall Thickness Unknown": None,
"Alternative Wall Thickness (mm)": None,
}
# Check if there's an alternative wall section following this wall entry
alt_match = alternative_wall_pattern.search(wall_section, match.end())
if alt_match:
wall_entry["Alternative Wall Type"] = alt_match.group(1).strip()
wall_entry["Alternative Wall Insulation"] = alt_match.group(2).strip()
wall_entry["Alternative Wall Dry-lining"] = alt_match.group(4).strip() if alt_match.group(4) else "N/A"
wall_entry["Alternative Wall Thickness Unknown"] = alt_match.group(5).strip()
wall_entry["Alternative Wall Thickness (mm)"] = int(alt_match.group(6))
# Append each building part as a dictionary in the wall_data list
wall_data.append(wall_entry)
return wall_data
def extract_summary_report(pdf_path):
"""
Extracts specific data from the provided PDF file.
Data includes:
- Current SAP rating
- Fuel Bill
- Address
"""
data = {
"Address": None,
"Postcode": None,
"Current SAP Rating": None,
"Current EPC Band": None,
"Fuel Bill": None,
"Main Building Age Band": None,
"Number of Storeys": None,
"Window Age Description": None,
"Window Age Description Proportion (%)": None,
"Secondary Window Age Description": None,
"Secondary Window Age Description Proportion (%)": None,
"Number of Windows": None,
"Total Number of Doors": None,
"Number of Insulated Doors": None,
"Existing Primary Heating System": None,
"Existing Primary Heating PCDF Reference": None,
"Existing Primary Heating Controls": None,
"Existing Primary Heating % of Heat": None,
"Existing Secondary Heating System": None,
"Existing Secondary Heating PCDF Reference": None,
"Existing Secondary Heating Controls": None,
"Existing Secondary Heating % of Heat": None,
"Secondary Heating Code": None,
"Water Heating Code": None,
'Total Floor Area (m2)': None,
'Total Ground Floor Area (m2)': None,
'RIR Floor Area': None,
'Main Building Wall Area (m2)': None,
'First Extension Wall Area (m2)': None,
"Number of Light Fittings": None,
"Number of LEL Fittings": None,
"Number of fittings needing LEL": None,
"Main Roof Type": None,
"Main Roof Insulation": None,
"Main Roof Insulation Thickness": None,
"Main Wall Type": None,
"Main Wall Insulation": None,
"Main Wall Dry-lining": None,
"Main Wall Thickness": None,
"Main Building Alternative Wall Type": None,
"Main Building Alternative Wall Insulation": None,
"Main Building Alternative Wall Dry-lining": None,
"Main Building Alternative Wall Thickness": None,
}
with (open(pdf_path, "rb") as file):
reader = PyPDF2.PdfReader(file)
text = ""
for page in reader.pages:
text += page.extract_text()
# Extract Current SAP rating
sap_match = re.search(r"Current SAP rating:\s*([A-Z] \d+)", text)
data["Current SAP Rating"] = sap_match.group(1).split(" ")[1]
data["Property Type"] = (
re.search(r"Property type:\s*(.*?)\n2\.0", text, re.DOTALL)
.group(1).replace('\n', ' ').strip().replace(" ", " ")
)
# Extract age
age_band_match = re.search(
r"3\.0 Date Built:\s*Main Property\s*[A-Z]?\s*(\d{4}-\d{4}|before \d{4}|\d{4} onwards)",
text
)
data["Main Building Age Band"] = age_band_match.group(1)
# Number of storeys
storeys_match = re.search(r"Number of Storeys:\s*(\d+)", text)
data["Number of Storeys"] = int(storeys_match.group(1))
# Grab number of heated rooms, number of habitable rooms
data["Number of Heated Rooms"] = int(re.search(r"Heated Habitable Rooms:\s*(\d+)", text).group(1))
data["Number of Habitable rooms"] = int(re.search(r"Habitable Rooms:\s*(\d+)", text).group(1))
# Extract Carbon Emissions
# carbon_match = re.search(r"Emissions \(t/year\):\s*([\d.]+)\s*tonnes", text)
# data["Carbon Emissions (t/year)"] = float(carbon_match.group(1))
# Extract Fuel Bill
fuel_bill_match = re.search(r"Fuel Bill:\s*£(\d+)", text)
data["Fuel Bill"] = f"£{fuel_bill_match.group(1)}"
# Extract individual address components
postcode = re.search(r"Postcode:\s*(.*?)\nRegion:", text)
# region = re.search(r"Region:\s*(.*?)\nHouse Name:", text)
house_name = re.search(r"House Name:\s*(.*?)\nHouse No:", text)
house_no = re.search(r"House No:\s*(.*?)\nStreet:", text)
street = re.search(r"Street:\s*(.*?)\nLocality:", text)
locality = re.search(r"Locality:\s*(.*?)\nTown:", text)
town = re.search(r"Town:\s*(.*?)\nCounty:", text)
county = re.search(r"County:\s*(.*?)\nProperty Tenure:", text)
# Clean extracted values and remove any prefixes
address_parts = [
house_no.group(1).strip() if house_no else "",
house_name.group(1).strip() if house_name else "",
street.group(1).strip() if street else "",
locality.group(1).strip() if locality else "",
town.group(1).strip() if town else "",
county.group(1).strip() if county else "",
postcode.group(1).strip() if postcode else ""
]
# Join non-empty parts with a comma
data["Address"] = ", ".join([part for part in address_parts if part])
data["Postcode"] = postcode.group(1).strip()
# windows_section = re.search(r"Windows\s*(.*?)\s*Draught Proofing", text, re.DOTALL)
# windows_text = windows_section.group(1)
# window_data = extract_window_age_description(windows_text)
# data.update(window_data)
# Extract Total Number of Doors
total_doors_match = re.search(r"Total Number of Doors\s*(\d+)", text)
data["Total Number of Doors"] = int(total_doors_match.group(1))
# Extract Number of Insulated Doors
insulated_doors_match = re.search(r"Number of Insulated Doors\s*(\d+)", text)
data["Number of Insulated Doors"] = int(insulated_doors_match.group(1))
# Extract heating system
# Extract Primary Heating Data
# Extract Primary Heating Section
primary_heating_section1 = re.search(r"Main\s*Heating1\s*(.*?)\s*Main\s*Heating2", text, re.DOTALL)
primary_heating_section2 = re.search(r"Main\s*Heating1\s*(.*?)\s*Water\s*Heating", text, re.DOTALL)
primary_heating_section = primary_heating_section1 if primary_heating_section1 else primary_heating_section2
primary_text = primary_heating_section.group(1)
# Handle extracting main heating code:
mainheat_search = re.search(r"Main Heating Code\s*(.*?)\n", primary_text)
if mainheat_search is None:
mainheat_search = re.search(r"Main Heating EES Code\s*(.*?)\n", primary_text)
if mainheat_search is None:
mainheat_search = re.search(r"PCDF boiler Reference\s*(.*?)\n", primary_text)
data["Existing Primary Heating System"] = mainheat_search.group(1).strip()
data["Existing Primary Heating PCDF Reference"] = re.search(
r"PCDF boiler Reference\s*(\d+)", primary_text
).group(1)
controls_search = re.search(
r"Main Heating Controls Sap\s*(.*?)\n", primary_text
)
if controls_search is None:
controls_search = re.search(
r"Main Heating Controls\s*(.*?)\n", primary_text
)
data["Existing Primary Heating Controls"] = controls_search.group(1).strip()
data["Existing Primary Heating % of Heat"] = int(
re.search(r"Percentage of Heat\s*(\d+)\s*%", primary_text).group(1)
)
# Extract Secondary Heating Section
secondary_heating_section = re.search(r"Main\s*Heating2\s*(.*?)\s*Water\s*Heating", text, re.DOTALL)
if secondary_heating_section is None:
data["Existing Secondary Heating System"] = ""
data["Existing Secondary Heating PCDF Reference"] = ""
data["Existing Secondary Heating Controls"] = ""
data["Existing Secondary Heating % of Heat"] = 0
else:
secondary_text = secondary_heating_section.group(1)
main_heating_code_match_secondary = re.search(
r"Main Heating Code\s*(.*?)(?=\n|Percentage of Heat)", secondary_text
)
if main_heating_code_match_secondary is None:
main_heating_code_match_secondary = re.search(
r"Main Heating EES Code\s*(.*?)(?=\n|Percentage of Heat)", secondary_text
)
data["Existing Secondary Heating System"] = main_heating_code_match_secondary.group(1).strip()
data["Existing Secondary Heating PCDF Reference"] = re.search(r"PCDF boiler Reference\s*(\d+)",
secondary_text).group(1)
second_heating_controls_match = re.search(r"Main Heating Controls\s*(.*?)\n", secondary_text)
data["Existing Secondary Heating Controls"] = (
second_heating_controls_match.group(1).strip() if second_heating_controls_match else ""
)
data["Existing Secondary Heating % of Heat"] = int(
re.search(r"Percentage of Heat\s*(\d+)\s*%", secondary_text).group(1)
)
# Extract Secondary Heating and Water Heating Codes
secondary_heating_code_match = re.search(r"Secondary Heating Code\s*(.*?)\n", text)
water_heating_code_match = re.search(r"Water Heating Code\s*(.*?)\n", text)
if data["Existing Secondary Heating System"] == "":
data["Secondary Heating Code"] = ""
else:
data["Secondary Heating Code"] = secondary_heating_code_match.group(
1).strip() if secondary_heating_code_match else ""
data["Water Heating Code"] = water_heating_code_match.group(1).strip()
dimensions = extract_building_parts_summary(text)
data.update(dimensions)
# Need to get the hot water
section_match = re.search(r"15\.0.*?\n(.*?)15\.1", text, re.DOTALL)
section_text = section_match.group(1)
# Extract Water Heating Code
code_match = re.search(r"Water Heating Code\s+(\S+)", section_text)
fuel_match = re.search(r"Water Heating Fuel Type\s+(.+)", section_text)
if fuel_match is None:
fuel_type = None
else:
fuel_type = fuel_match.group(1).strip()
code = code_match.group(1)
data["Hot Water System"] = code
data["Hot Water Fuel"] = fuel_type
# data["Number of Light Fittings"] = int(re.search(r"Total number of light fittings\s*(\d+)", text).group(1))
# data["Number of LEL Fittings"] = int(re.search(r"Total number of L.E.L. fittings\s*(\d+)", text).group(1))
# data["Number of fittings needing LEL"] = data["Number of Light Fittings"] - data["Number of LEL Fittings"]
extracted_roof_data = extract_roof_details_summary(text)
main_roof_data = [roof for roof in extracted_roof_data if "Main" in roof["Building Part"]][0]
data["Main Roof Type"] = main_roof_data["Roof Type"]
data["Main Roof Insulation"] = main_roof_data["Roof Insulation"]
data["Main Roof Insulation Thickness"] = main_roof_data["Roof Insulation Thickness"]
walls_data = extract_wall_details_summary(text)
# Get the main building wall data
main_building_walls = [wall for wall in walls_data if "Main" in wall["Building Part"]][0]
data["Main Wall Type"] = main_building_walls["Wall Type"]
data["Main Wall Insulation"] = main_building_walls["Wall Insulation"]
# data["Main Wall Dry-lining"] = main_building_walls["Wall Dry-lining"]
# data["Main Wall Thickness"] = main_building_walls["Wall Thickness (mm)"]
# data["Main Building Alternative Wall Type"] = main_building_walls["Alternative Wall Type"]
# data["Main Building Alternative Wall Insulation"] = main_building_walls["Alternative Wall Insulation"]
# data["Main Building Alternative Wall Dry-lining"] = main_building_walls["Alternative Wall Dry-lining"]
# data["Main Building Alternative Wall Thickness"] = main_building_walls["Alternative Wall Thickness (mm)"]
return data
folder_location = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Livewest/July 2025 Heating Upgrades"
df = pd.read_csv("/Users/khalimconn-kowlessar/Documents/hestia/July 2025 Surveys/export_summary_table.csv")
property_data = []
for _, x in tqdm(df.iterrows(), total=len(df)):
if not pd.isnull(x["error"]):
continue
filepath = x["filepath"]
if filepath in ["No summary file found"]:
continue
summary_data = extract_summary_report(pdf_path=filepath)
property_data.append(
{
**x.to_dict(),
**summary_data
}
)
property_data = pd.DataFrame(property_data)
# Store as excel
property_data.to_excel(
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/Livewest/July 2025 Heating "
"Upgrades/property_table_24th_july.xlsx"
)
sandwell_data = property_data[property_data["company"] == "sandwell.gov.uk"]
sandwell_data.to_csv(
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/Livewest/July 2025 Heating "
"Upgrades/Sandwell EPR data (WIP).xlsx"
)

1
recommendations/VentilationRecommendations.py
View file

@ -17,7 +17,6 @@ class VentilationRecommendations(Definitions):
):
self.property = property_instance
self.has_ventilaion = None
self.recommendation = None
self.materials = [part for part in materials if part["type"] == "mechanical_ventilation"]

26
recommendations/optimiser/GainOptimiser.py
View file

@ -9,7 +9,7 @@ class GainOptimiser:
This class is used to maximise gain, given a constrained cost
"""
def __init__(self, components, max_cost, max_gain):
def __init__(self, components, max_cost, max_gain, allow_slack=True):
"""
This function will try and maximise the gain, given a constrained cost. If we specific a max_gain, then the
optimisation routine is constained to try not to exceed a maximum increase
@ -21,6 +21,8 @@ class GainOptimiser:
:param components: List of components, where each component is a dictionary with keys "id", "cost" and "gain"
:param max_cost: Maximum cost constraint
:param max_gain: Maximum gain constraint
:param allow_slack: If True, allows the model to use slack variables to relax the cost constraint if the model
is infeasible. Defaults to True.
"""
self.components = components
self.max_cost = max_cost
@ -32,6 +34,7 @@ class GainOptimiser:
self.solution = []
self.solution_gain = None
self.solution_cost = None
self.allow_slack = allow_slack
def setup(self):
# Initialize Model
@ -124,15 +127,18 @@ class GainOptimiser:
if (self.m.status == OptimizationStatus.INFEASIBLE) or (
(self.m.status == OptimizationStatus.OPTIMAL) and not len(solution)
):
logger.info("We have an infeasible model, setting up slack model")
self.setup_slack()
self.m.optimize()
solution = [
item for group, group_vars in zip(self.components, self.variables) for item, var in
zip(group, group_vars)
if
var.x >= 0.99
]
if self.allow_slack:
logger.info("We have an infeasible model, setting up slack model")
self.setup_slack()
self.m.optimize()
solution = [
item for group, group_vars in zip(self.components, self.variables) for item, var in
zip(group, group_vars)
if
var.x >= 0.99
]
else:
logger.info("Infeasible but slack disabled - returning empty solution")
self.solution = solution

285
recommendations/optimiser/optimiser_functions.py
View file

@ -1,26 +1,56 @@
import pandas as pd
import backend.app.assumptions as assumptions
from backend.Property import Property
from backend.app.plan.schemas import PlanTriggerRequest
from backend.app.utils import epc_to_sap_lower_bound
from recommendations.optimiser.CostOptimiser import CostOptimiser
def prepare_input_measures(property_recommendations, goal, needs_ventilation):
"""
Basic function to convert recommendations_to_upload to a format that is
suitable for the optimiser - large
:param property_recommendations: object containing the recommendations, created in the plan trigger api
:param goal: goal to be optimised for, should be one of the keys in gain_map. E.g. if the gain is SAP points,
the goal should reflect that desired gain
:param needs_ventilation: boolean to indicate if the property needs ventilation
:return: Nested list of input measures
Prepares a nested list of measure options for optimisation.
Each sublist represents all available variants of a single measure type (e.g. all solar PV options).
Within each sublist, each measure is represented as a dictionary containing:
- id: unique recommendation identifier
- cost: total cost of the measure (including ventilation if bundled)
- gain: the relevant gain metric based on the selected goal
- type: the measure type, optionally combined with ventilation (e.g. "wall_insulation+mechanical_ventilation")
Ventilation bundling:
- If a property needs ventilation, and a measure type requires it (as defined in
assumptions.measures_needing_ventilation),
the ventilation cost and gain are added to that measure’s values.
Filtering:
- Measures with negative `energy_cost_savings` are excluded.
- Solar PV options with batteries are excluded (currently handled by a placeholder bitwise NOT).
Parameters
----------
property_recommendations : list[list[dict]]
Nested list of recommendations for a property. Each inner list represents variations of the same measure type.
goal : str
Optimisation goal, one of: "Increasing EPC", "Energy Savings", "Reducing CO2 emissions".
needs_ventilation : bool
Whether the property requires mechanical ventilation to accompany certain measures.
Returns
-------
list[list[dict]]
Nested list of prepared measure options, ready for input into the optimiser.
"""
goal_map = {
"Increasing EPC": "sap_points"
"Increasing EPC": "sap_points",
"Energy Savings": "kwh_savings",
"Reducing CO2 emissions": "co2_equivalent_savings",
}
goal_key = goal_map[goal]
if not goal_key:
raise NotImplementedError("Not implemented this gain type - investigate me")
# We ony ever have one ventilation measure with now
ventilation_recommendation = next(
(measure[0] for measure in property_recommendations if measure[0]["type"] == "mechanical_ventilation"),
{}
@ -29,22 +59,22 @@ def prepare_input_measures(property_recommendations, goal, needs_ventilation):
input_measures = []
for recs in property_recommendations:
# Skip ventilation as a standalone optimisation option (it will be bundled)
if needs_ventilation and recs[0]["type"] == "mechanical_ventilation":
# If we house needs ventilation, ventilation will be packaged with the fabric measure so
# we don't need to optimise it independently
continue
# Filter out solar PV with batteries
if recs[0]["type"] == "solar_pv":
# if the recommendation is a solar recommendation with a battery, we exclude it from the optimisation.
recs = [r for r in recs if ~r["has_battery"]]
# Only include measures with non-negative cost savings
recs_to_append = [rec for rec in recs if rec["energy_cost_savings"] >= 0]
if not recs_to_append:
continue
# Build enriched measure data
to_append = []
for rec in recs:
# We bundle the impact of ventilation with the measure
total = (
rec["total"] + ventilation_recommendation["total"]
if rec["type"] in assumptions.measures_needing_ventilation and needs_ventilation
@ -55,23 +85,232 @@ def prepare_input_measures(property_recommendations, goal, needs_ventilation):
if rec["type"] in assumptions.measures_needing_ventilation and needs_ventilation
else rec[goal_key]
)
rec_type = (
"+".join(
[rec["type"], ventilation_recommendation["type"]]
) if rec["type"] in assumptions.measures_needing_ventilation and needs_ventilation
f"{rec['type']}+{ventilation_recommendation['type']}"
if rec["type"] in assumptions.measures_needing_ventilation and needs_ventilation
else rec["type"]
)
to_append.append(
{
"id": rec["recommendation_id"],
"cost": total,
"gain": gain,
"type": rec_type
}
{"id": rec["recommendation_id"], "cost": total, "gain": gain, "type": rec_type}
)
input_measures.append(to_append)
return input_measures
def calculate_fixed_gain(property_required_measures, recommendations, p, needs_ventilation):
"""
Calculates the total "fixed gain" from required measures for a property.
Required measures are applied regardless of optimisation. This function:
- Finds the maximum SAP points for each required measure type.
- Sums those max SAP values into a fixed gain total.
- Adds the SAP points for mechanical ventilation if:
* The property needs ventilation, and
* At least one required measure needs ventilation.
Parameters
----------
property_required_measures : list[list[dict]]
Nested list of required measures for the property.
recommendations : dict
All recommendations for all properties, keyed by property id.
p : object
Property object (must have .id).
needs_ventilation : bool
Whether ventilation should be bundled with certain measures.
Returns
-------
float
Total fixed SAP gain from required measures (and ventilation, if applicable).
"""
if not property_required_measures:
return 0
sap_by_type = [
{"type": rec["type"], "sap_points": rec["sap_points"]}
for recs in property_required_measures for rec in recs
]
max_per_type = pd.DataFrame(sap_by_type).groupby("type")["sap_points"].max().to_dict()
fixed_gain = sum(max_per_type.values())
required_types = {rec["type"] for rec in sap_by_type}
if needs_ventilation and any(r in required_types for r in assumptions.measures_needing_ventilation):
fixed_gain += next(
(r[0]["sap_points"] for r in recommendations[p.id] if r[0]["type"] == "mechanical_ventilation"),
0
)
return fixed_gain
def calculate_gain(body: PlanTriggerRequest, p: Property, fixed_gain: float) -> float | None:
"""
Calculates the target gain value for optimisation based on the goal.
- For "Increasing EPC": Computes the SAP gain needed to reach the target EPC,
applies a slack adjustment (via CostOptimiser), and subtracts fixed gains from required measures.
- For "Energy Savings" or "Reducing CO2 emissions": Returns None,
which signals the optimiser to simply maximise gain under a budget.
Parameters
----------
body : object
Request body object containing optimisation settings (goal, goal_value, simulate_sap_10, etc.)
p : object
Property object with EPC data (must have p.data["current-energy-efficiency"]).
fixed_gain : float
Total fixed gain from required measures (returned by calculate_fixed_gain).
Returns
-------
float or None
Required SAP gain for EPC, or None for non-EPC goals.
"""
if body.goal == "Increasing EPC":
current_sap = int(p.data["current-energy-efficiency"])
gain = CostOptimiser.calculate_sap_gain_with_slack(
epc_to_sap_lower_bound(body.goal_value) - current_sap
) - fixed_gain
if body.simulate_sap_10:
gain += 3
return max(gain, 0)
elif body.goal in ["Energy Savings", "Reducing CO2 emissions"]:
return None
else:
raise NotImplementedError(f"Goal {body.goal} is not supported")
def add_required_measures(property_id, property_required_measures, recommendations, selected):
"""
Ensures the cheapest variant of each required measure is added to the selected recommendations.
For each required measure type, this function:
- Finds the lowest-cost variant not already selected.
- Adds it to the selected recommendation IDs.
- Returns a flattened list of all selected measure details for final output.
Parameters
----------
property_id : int
Unique identifier for the property.
property_required_measures : list[list[dict]]
Nested list of required measures for the property.
recommendations : dict
All recommendations for all properties, keyed by property id.
selected : set
Set of already selected recommendation IDs from the optimiser.
Returns
-------
list[dict]
Flat list of selected measure details, each containing:
{"id", "cost", "gain", "type"}
"""
for recs in property_required_measures:
cheapest = min(
(rec for rec in recs if rec["recommendation_id"] not in selected),
key=lambda rec: rec["total"],
)
selected.add(cheapest["recommendation_id"])
return [
{"id": rec["recommendation_id"], "cost": rec["total"], "gain": rec["sap_points"], "type": rec["type"]}
for recs in recommendations[property_id] for rec in recs
if rec["recommendation_id"] in selected
]
def add_best_practice_measures(property_id, solution, recommendations, selected):
"""
Ensures best-practice measures like ventilation and trickle vents are included
in the selected recommendations when appropriate.
Rules:
- If a measure requiring ventilation is selected AND ventilation is not already present,
add the corresponding mechanical ventilation recommendation.
- Always add trickle vents if they exist in the recommendations.
Parameters
----------
property_id : int
The unique identifier for the property.
solution : list[dict]
The current list of selected measures (each containing id, type, gain, cost).
recommendations : dict
All recommendations for all properties, keyed by property id.
selected : set
Set of already selected recommendation IDs.
Returns
-------
set
Updated set of selected recommendation IDs, including ventilation and trickle vents if applicable.
"""
# Check if any selected measure requires ventilation
ventilation_selected = [r for r in solution if "+mechanical_ventilation" in r["type"]]
# If ventilation has been selected, or one of the measures needs ventilation, we need to ensure ventilation is
# included
needs_ventilation = any(
x in [r["type"] for r in solution] for x in assumptions.measures_needing_ventilation
) or len(ventilation_selected) > 0
if needs_ventilation:
ventilation_rec = next(
(r[0] for r in recommendations[property_id] if r[0]["type"] == "mechanical_ventilation"),
None
)
if ventilation_rec:
selected.add(ventilation_rec["recommendation_id"])
# Always add trickle vents if available
trickle_vents_rec = next(
(r[0] for r in recommendations[property_id] if r[0]["type"] == "trickle_vents"),
None
)
if trickle_vents_rec:
selected.add(trickle_vents_rec["recommendation_id"])
return selected
def flatten_recommendations_with_defaults(property_id, recommendations, selected):
"""
Flattens nested recommendation lists for a property and marks which
recommendations were selected.
Each recommendation dict is copied and an extra key `default` is added:
- True if the recommendation ID is in `selected`
- False otherwise
Parameters
----------
property_id : int
The unique identifier for the property.
recommendations : dict
All recommendations for all properties, keyed by property id.
Each value is a list of lists (grouped by measure type).
selected : set
Set of selected recommendation IDs.
Returns
-------
list[dict]
A flattened list of recommendation dicts for the given property,
each with an added `default` field.
"""
final_recommendations = [
[
{**rec, "default": rec["recommendation_id"] in selected}
for rec in recommendations_by_type
]
for recommendations_by_type in recommendations[property_id]
]
# Flatten the nested list of lists into a single list
return [rec for recommendations_by_type in final_recommendations for rec in recommendations_by_type]

19
recommendations/tests/test_data/heating_recommendations_data.py
View file

@ -1193,11 +1193,10 @@ testing_examples = [
'uprn': 100070685908, 'uprn-source': 'Address Matched', 'sheating-energy-eff': None,
'sheating-env-eff': None
},
"heating_measure_types": [
'high_heat_retention_storage_heater'
],
"notes": "This property is a flag, without mains gas connection. Currently has underfloor electric heating"
"so we recommend HHR"
"heating_measure_types": [],
"notes": "This property is a flat, without mains gas connection. Currently has underfloor electric heating"
"don't recommend anything. HHRSH isn't recommended as with underfloor heating, it's quite"
"disruptive"
},
{
"epc": {
@ -1239,12 +1238,9 @@ testing_examples = [
},
"heating_measure_types": [
'air_source_heat_pump',
'boiler_upgrade',
'boiler_upgrade',
'high_heat_retention_storage_heater'
],
"notes": "The property has warm air electricaire heating, so we recommend ASHP and HHR. It also has a mains"
"connection so we recommend a gas condensing boiler"
"notes": "The property has warm air electricaire heating, so we recommend ASHP and HHR"
},
{
"epc": {
@ -1287,9 +1283,8 @@ testing_examples = [
'sheating-env-eff': None
},
"heating_measure_types": [
'boiler_upgrade',
'boiler_upgrade',
],
"notes": "This property has warm air mains gas heating, so we recommend a gas condensing boiler"
"notes": "This property has warm air mains gas heating; we recommend no heating upgrades as the efficiency is"
"good"
}
]

350
recommendations/tests/test_data/measures_to_optimise.py Normal file
View file

@ -0,0 +1,350 @@
import datetime
import numpy as np
from numpy import nan
from pandas import Timestamp
measures_to_optimise = [
[{'phase': 0, 'parts': [{'id': 2466, 'type': 'external_wall_insulation',
'description': 'EWI Pro EPS external wall insulation system with '
'Brick Slip finish',
'depth': 150.0, 'depth_unit': 'mm', 'cost': None,
'cost_unit': 'gbp_per_m2', 'r_value_per_mm': 0.02631579,
'r_value_unit': 'square_meter_kelvin_per_watt',
'thermal_conductivity': 0.038,
'thermal_conductivity_unit': 'watt_per_meter_kelvin', 'link': 'SCIS',
'created_at': Timestamp('2025-03-16 15:26:22.379496'),
'is_active': True, 'prime_material_cost': None, 'material_cost': 0.0,
'labour_cost': 0.0, 'labour_hours_per_unit': 0.0, 'plant_cost': 0.0,
'total_cost': 298.35, 'notes': 'This is the quoted value from SCIS',
'is_installer_quote': True, 'quantity': 63.98796761892035,
'quantity_unit': 'm2', 'total': 19090.810139104888,
'labour_hours': 0.0, 'labour_days': 0.0}],
'type': 'external_wall_insulation', 'measure_type': 'external_wall_insulation',
'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,
'sap_points': np.float64(9.6),
'simulation_config': {'is_as_built_ending': False, 'walls_is_assumed_ending': False,
'walls_insulation_thickness_ending': 'average',
'external_insulation_ending': True, 'walls_energy_eff_ending': 'Good',
'walls_thermal_transmittance_ending': 0.23},
'description_simulation': {'walls-description': 'Solid brick, with external insulation',
'walls-energy-eff': 'Good'}, 'total': 19090.810139104888,
'labour_hours': 0.0, 'labour_days': 0.0, 'survey': False, 'recommendation_id': '0_phase=0',
'efficiency': 11229.568317120522, 'co2_equivalent_savings': np.float64(0.5),
'heat_demand': np.float64(37.099999999999994), 'kwh_savings': np.float64(1813.199999999999),
'energy_cost_savings': np.float64(135.03007058823516)}, {'phase': 0, 'parts': [
{'id': 2373, 'type': 'internal_wall_insulation', 'description': 'SWIP EcoBatt & Plastered finish',
'depth': 95.0,
'depth_unit': 'mm', 'cost': None, 'cost_unit': 'gbp_per_m2', 'r_value_per_mm': 0.03125,
'r_value_unit': 'square_meter_kelvin_per_watt', 'thermal_conductivity': 0.032,
'thermal_conductivity_unit': None,
'link': 'SCIS', 'created_at': Timestamp('2025-03-16 15:26:22.379496'), 'is_active': True,
'prime_material_cost': None, 'material_cost': 0.0, 'labour_cost': 0.0, 'labour_hours_per_unit': 2.1,
'plant_cost': 0.0, 'total_cost': 89.0, 'notes': None, 'is_installer_quote': True,
'quantity': 63.98796761892035,
'quantity_unit': 'm2', 'total': 5694.929118083911, 'labour_hours': 134.37473199973275,
'labour_days': 4.199210374991648}], 'type': 'internal_wall_insulation',
'measure_type': 'internal_wall_insulation',
'description': 'Install 95mm SWIP '
'EcoBatt & '
'Plastered finish '
'on internal walls',
'starting_u_value': 1.7,
'new_u_value': 0.32,
'already_installed': False,
'sap_points': 6,
'simulation_config': {
'is_as_built_ending': False,
'walls_is_assumed_ending': False,
'walls_insulation_thickness_ending': 'average',
'internal_insulation_ending':
True,
'walls_energy_eff_ending':
'Good',
'walls_thermal_transmittance_ending': 0.29},
'description_simulation': {
'walls-description': 'Solid '
'brick, '
'with '
'internal '
'insulation',
'walls-energy-eff': 'Good'},
'total': 5694.929118083911,
'labour_hours': 134.37473199973275,
'labour_days': 4.199210374991648,
'survey': True,
'recommendation_id': '1_phase=0',
'efficiency': 3349.6383047552417,
'co2_equivalent_savings': np.float64(
0.5), 'heat_demand': np.float64(
35.30000000000001), 'kwh_savings': np.float64(1424.699999999999), 'energy_cost_savings': np.float64(
106.09824705882352)}], [{'phase': 1, 'parts': [
{'id': 2351, 'type': 'loft_insulation', 'description': 'Knauf Loft Roll 44 glass fibre roll', 'depth': 300.0,
'depth_unit': 'mm', 'cost': None, 'cost_unit': 'gbp_per_m2', 'r_value_per_mm': 0.022727273,
'r_value_unit': 'square_meter_kelvin_per_watt', 'thermal_conductivity': 0.044,
'thermal_conductivity_unit': 'watt_per_meter_kelvin', 'link': 'SCIS',
'created_at': Timestamp('2025-03-16 15:26:22.379496'), 'is_active': True, 'prime_material_cost': None,
'material_cost': 0.0, 'labour_cost': 0.0, 'labour_hours_per_unit': 0.11, 'plant_cost': 0.0, 'total_cost': 15.0,
'notes': 'This is the cost if there is less than 100mm existing insulation', 'is_installer_quote': True,
'quantity': 63.98796761892035, 'quantity_unit': 'm2', 'total': 645.0, 'labour_hours': 8, 'labour_days': 1}],
'type': 'loft_insulation', 'measure_type': 'loft_insulation',
'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,
'simulation_config': {'is_loft_ending': True, 'roof_is_assumed_ending': False,
'roof_insulation_thickness_ending': '300',
'roof_thermal_transmittance_ending': 2.3,
'roof_energy_eff_ending': 'Very Good'},
'description_simulation': {'roof-description': 'Pitched, 300mm loft insulation',
'roof-energy-eff': 'Very Good'}, 'total': 645.0,
'labour_hours': 8, 'labour_days': 1, 'survey': False,
'recommendation_id': '2_phase=1',
'efficiency': 278.1347826086957,
'co2_equivalent_savings': np.float64(0.10000000000000009),
'heat_demand': np.float64(1.5), 'kwh_savings': np.float64(572.5500000000002),
'energy_cost_savings': np.float64(42.638135294117774)}], [{'phase': 2, 'parts': [
{'id': 2329, 'type': 'mechanical_ventilation', 'description': 'Mechanical Extract Ventilation', 'depth': 0.0,
'depth_unit': None, 'cost': None, 'cost_unit': 'gbp_per_unit', 'r_value_per_mm': nan,
'r_value_unit': 'square_meter_kelvin_per_watt', 'thermal_conductivity': None,
'thermal_conductivity_unit': None,
'link': 'SCIS', 'created_at': datetime.datetime(2025, 3, 16, 15, 26, 22, 379496), 'is_active': True,
'prime_material_cost': None, 'material_cost': 0.0, 'labour_cost': 0.0, 'labour_hours_per_unit': 0.0,
'plant_cost': 0.0, 'total_cost': 350.0, 'notes': None, 'is_installer_quote': True, 'total': 700.0,
'quantity': 2,
'quantity_unit': 'part'}
], 'type': 'mechanical_ventilation', 'measure_type': 'mechanical_ventilation',
'description':
'Install 2 '
'Mechanical '
'Extract '
'Ventilation units',
'starting_u_value': None,
'new_u_value': None,
'already_installed': False,
'sap_points': np.float64(
-0.10000000000000142),
'heat_demand': np.float64(
-3.3999999999999773),
'kwh_savings': np.float64(
-45.899999999999636),
'co2_equivalent_savings': np.float64(
0.0),
'energy_cost_savings': np.float64(
-3.4181999999999846),
'total': 700.0,
'labour_hours': 8,
'labour_days': 1.0,
'simulation_config': {
'mechanical_ventilation_ending': 'mechanical, extract only'},
'description_simulation': {
'mechanical-ventilation': 'mechanical, extract only'},
'recommendation_id': '3_phase=2',
'efficiency': 0}
], [
{'phase': 3, 'parts': [{'id': 2409, 'type': 'suspended_floor_insulation',
'description': 'Q-bot underfloor insulation', 'depth': 75.0,
'depth_unit': 'mm', 'cost': None, 'cost_unit': 'gbp_per_m2',
'r_value_per_mm': 0.045454547,
'r_value_unit': 'square_meter_kelvin_per_watt',
'thermal_conductivity': 0.022,
'thermal_conductivity_unit': 'watt_per_meter_kelvin',
'link': 'SCIS',
'created_at': Timestamp('2025-03-16 15:26:22.379496'),
'is_active': True, 'prime_material_cost': None,
'material_cost': 0.0, 'labour_cost': 0.0,
'labour_hours_per_unit': 1.63, 'plant_cost': 0.0,
'total_cost': 93.75,
'notes': 'Linearly interpolated based on Qbot costs',
'is_installer_quote': True, 'quantity': 43.0, 'quantity_unit': 'm2',
'total': 4031.25, 'labour_hours': 70.08999999999999,
'labour_days': 2.920416666666666}],
'type': 'suspended_floor_insulation', 'measure_type': 'suspended_floor_insulation',
'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,
'already_installed': False, 'simulation_config': {'floor_is_assumed_ending': False,
'floor_insulation_thickness_ending':
'average',
'floor_thermal_transmittance_ending':
0.685593},
'description_simulation': {'floor-description': 'Suspended, insulated'}, 'total': 4031.25,
'labour_hours': 70.08999999999999, 'labour_days': 2.920416666666666,
'recommendation_id': '4_phase=3', 'efficiency': 4856.707710843373,
'co2_equivalent_savings': np.float64(0.20000000000000018), 'heat_demand': np.float64(33.5),
'kwh_savings': np.float64(1018.0999999999995),
'energy_cost_savings': np.float64(75.8185058823529)}], [
{'phase': 4, 'parts': [], 'type': 'low_energy_lighting',
'measure_type': 'low_energy_lighting',
'description': 'Install low energy lighting in 14 outlets', 'starting_u_value': None,
'new_u_value': None, 'already_installed': False, 'sap_points': 2, 'kwh_savings': 766.5,
'energy_cost_savings': 197.22044999999997,
'co2_equivalent_savings': np.float64(0.09999999999999964),
'description_simulation': {'lighting-energy-eff': 'Very Good',
'lighting-description': 'Low energy lighting in all fixed '
'outlets',
'low-energy-lighting': 100}, 'total': 58.8, 'labour_hours': 1,
'labour_days': 0.125, 'survey': True, 'recommendation_id': '5_phase=4', 'efficiency': 29.4,
'heat_demand': np.float64(5.099999999999994)}], [
{'type': 'heating', 'phase': 5, 'measure_type': 'time_temperature_zone_control',
'parts': [],
'description': 'Upgrade heating controls to Smart Thermostats, room sensors and smart '
'radiator valves (time & temperature zone control)',
'total': 739.576, 'subtotal': 700.48, 'vat': 39.096000000000004,
'labour_hours': 3.6199999999999997, 'labour_days': np.float64(1.0),
'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(2.9),
'already_installed': False,
'simulation_config': {'thermostatic_control_ending': 'time and temperature zone control',
'switch_system_ending': None, 'trvs_ending': None,
'mainheatc_energy_eff_ending': 'Very Good'},
'description_simulation': {'mainheatcont-description': 'Time and temperature zone control',
'mainheatc-energy-eff': 'Very Good'},
'recommendation_id': '6_phase=5', 'efficiency': 739.576,
'co2_equivalent_savings': np.float64(0.30000000000000027),
'heat_demand': np.float64(6.599999999999994), 'kwh_savings': np.float64(853.6999999999998),
'energy_cost_savings': np.float64(63.57554117647055)}], [
{'phase': 6, 'parts': [], 'type': 'secondary_heating', 'measure_type': 'secondary_heating',
'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,
'labour_days': np.float64(1.0),
'simulation_config': {'secondheat_description_ending': 'None'},
'description_simulation': {'secondheat-description': 'None'},
'recommendation_id': '7_phase=6', 'efficiency': 30.0,
'co2_equivalent_savings': np.float64(0.10000000000000009),
'heat_demand': np.float64(15.400000000000006),
'kwh_savings': np.float64(202.30000000000018),
'energy_cost_savings': np.float64(15.065400000000011)}], [
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
'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,
'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(65.0),
'has_battery': False, 'initial_ac_kwh_per_year': np.float64(4081.7132614999996),
'description_simulation': {'photo-supply': np.float64(65.0)},
'recommendation_id': '8_phase=7', 'efficiency': np.float64(462.54923076923075),
'co2_equivalent_savings': np.float64(0.47347873833399995),
'heat_demand': np.float64(88.69999999999999),
'kwh_savings': np.float64(2040.8566307499998),
'energy_cost_savings': np.float64(525.1124110919749)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
'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,
'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(65.0),
'has_battery': True, 'initial_ac_kwh_per_year': np.float64(4081.7132614999996),
'description_simulation': {'photo-supply': np.float64(65.0)},
'recommendation_id': '9_phase=7', 'efficiency': np.float64(810.5390769230769),
'co2_equivalent_savings': np.float64(0.6628702336675999),
'heat_demand': np.float64(88.69999999999999),
'kwh_savings': np.float64(2857.1992830499994),
'energy_cost_savings': np.float64(735.1573755287648)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
'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,
'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(60.0),
'has_battery': False, 'initial_ac_kwh_per_year': np.float64(3692.66794),
'description_simulation': {'photo-supply': np.float64(60.0)},
'recommendation_id': '10_phase=7', 'efficiency': np.float64(485.54099999999994),
'co2_equivalent_savings': np.float64(0.42834948104),
'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',
'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,
'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(60.0),
'has_battery': True, 'initial_ac_kwh_per_year': np.float64(3692.66794),
'description_simulation': {'photo-supply': np.float64(60.0)},
'recommendation_id': '11_phase=7', 'efficiency': np.float64(862.5299999999999),
'co2_equivalent_savings': np.float64(0.599689273456),
'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',
'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,
'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(55.0),
'has_battery': False, 'initial_ac_kwh_per_year': np.float64(3300.5416548),
'description_simulation': {'photo-supply': np.float64(55.0)},
'recommendation_id': '12_phase=7', 'efficiency': np.float64(512.964),
'co2_equivalent_savings': np.float64(0.3828628319568), 'heat_demand': np.float64(78.3),
'kwh_savings': np.float64(1650.2708274),
'energy_cost_savings': np.float64(424.61468389001993)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
'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,
'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(55.0),
'has_battery': True, 'initial_ac_kwh_per_year': np.float64(3300.5416548),
'description_simulation': {'photo-supply': np.float64(55.0)},
'recommendation_id': '13_phase=7', 'efficiency': np.float64(924.2247272727273),
'co2_equivalent_savings': np.float64(0.53600796473952), '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',
'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,
'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(45.0),
'has_battery': False, 'initial_ac_kwh_per_year': np.float64(2907.1867812),
'description_simulation': {'photo-supply': np.float64(45.0)},
'recommendation_id': '14_phase=7', 'efficiency': np.float64(606.5253333333333),
'co2_equivalent_savings': np.float64(0.3372336666192), 'heat_demand': np.float64(64.0),
'kwh_savings': np.float64(1453.5933906),
'energy_cost_savings': np.float64(374.00957940138)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
'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,
'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(45.0),
'has_battery': True, 'initial_ac_kwh_per_year': np.float64(2907.1867812),
'description_simulation': {'photo-supply': np.float64(45.0)},
'recommendation_id': '15_phase=7', 'efficiency': np.float64(1109.1773333333333),
'co2_equivalent_savings': np.float64(0.47212713326688), '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',
'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,
'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(40.0),
'has_battery': False, 'initial_ac_kwh_per_year': np.float64(2510.25188),
'description_simulation': {'photo-supply': np.float64(40.0)},
'recommendation_id': '16_phase=7', 'efficiency': np.float64(659.3565),
'co2_equivalent_savings': np.float64(0.29118921808), 'heat_demand': np.float64(54.3),
'kwh_savings': np.float64(1255.12594),
'energy_cost_savings': np.float64(322.94390436199996)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
'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,
'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(40.0),
'has_battery': True, 'initial_ac_kwh_per_year': np.float64(2510.25188),
'description_simulation': {'photo-supply': np.float64(40.0)},
'recommendation_id': '17_phase=7', 'efficiency': np.float64(1224.84),
'co2_equivalent_savings': np.float64(0.40766490531199995), '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',
'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,
'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(35.0),
'has_battery': False, 'initial_ac_kwh_per_year': np.float64(2096.682636),
'description_simulation': {'photo-supply': np.float64(35.0)},
'recommendation_id': '18_phase=7', 'efficiency': np.float64(727.2822857142856),
'co2_equivalent_savings': np.float64(0.243215185776), 'heat_demand': np.float64(48.5),
'kwh_savings': np.float64(1048.341318), 'energy_cost_savings': np.float64(269.7382211214)},
{'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv',
'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,
'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(35.0),
'has_battery': True, 'initial_ac_kwh_per_year': np.float64(2096.682636),
'description_simulation': {'photo-supply': np.float64(35.0)},
'recommendation_id': '19_phase=7', 'efficiency': np.float64(1373.5491428571427),
'co2_equivalent_savings': np.float64(0.3405012600864), 'heat_demand': np.float64(48.5),
'kwh_savings': np.float64(1467.6778451999999),
'energy_cost_savings': np.float64(377.6335095699599)}]
]

17
recommendations/tests/test_lighting_recommendations.py
View file

@ -40,15 +40,14 @@ class TestLightingRecommendations:
lr.recommend()
assert len(lr.recommendation) == 1
# Note - this test may be dependent on the ofgem price caps
assert lr.recommendation == [
{'phase': 0, 'parts': [], 'type': 'low_energy_lighting', 'measure_type': 'low_energy_lighting',
'description': 'Install low energy lighting in 4 outlets', 'starting_u_value': None, 'new_u_value': None,
'already_installed': False, 'sap_points': 0.4, 'kwh_savings': 219.0, 'energy_cost_savings': 54.4434,
'co2_equivalent_savings': 0.035478, 'description_simulation': {'lighting-energy-eff': 'Very Good',
'lighting-description': 'Low energy '
'lighting in all '
'fixed outlets',
'low-energy-lighting': 100},
'total': 188.76000000000002, 'subtotal': 157.3, 'vat': 31.460000000000004, 'contingency': 14.3,
'material': 80.0, 'labour_hours': 3.2, 'labour_days': 0.4, 'labour_cost': 63.0, 'survey': False}
]
'already_installed': False, 'sap_points': 0.4, 'kwh_savings': 219.0,
'energy_cost_savings': 56.348699999999994, 'co2_equivalent_savings': 0.035478,
'description_simulation': {'lighting-energy-eff': 'Very Good',
'lighting-description': 'Low energy lighting in all fixed outlets',
'low-energy-lighting': 100}, 'total': 188.76000000000002, 'subtotal': 157.3,
'vat': 31.460000000000004, 'contingency': 14.3, 'material': 80.0, 'labour_hours': 3.2, 'labour_days': 0.4,
'labour_cost': 63.0, 'survey': False}]

255
recommendations/tests/test_optimiser_functions.py Normal file
View file

@ -0,0 +1,255 @@
import pytest
import numpy as np
from types import SimpleNamespace
from recommendations.tests.test_data.measures_to_optimise import measures_to_optimise
from recommendations.optimiser import optimiser_functions
from recommendations.optimiser.GainOptimiser import GainOptimiser
from recommendations.optimiser.CostOptimiser import CostOptimiser
class TestPrepareInputMeasures:
def test_returns_expected_structure_without_ventilation(self):
recs = [
[ # loft insulation measure
{"recommendation_id": "loft1", "type": "loft_insulation", "total": 100, "kwh_savings": 200,
"energy_cost_savings": 10, "has_battery": False},
],
]
measures = optimiser_functions.prepare_input_measures(recs, goal="Energy Savings", needs_ventilation=False)
assert isinstance(measures, list)
assert measures[0][0]["id"] == "loft1"
assert measures[0][0]["cost"] == 100
assert measures[0][0]["gain"] == 200
def test_bundles_ventilation_when_needed(self, monkeypatch):
# patch measures_needing_ventilation so that "wall_insulation" needs ventilation
monkeypatch.setattr(optimiser_functions.assumptions, "measures_needing_ventilation",
["internal_wall_insulation"])
recs = [
[{"recommendation_id": "wall1", "type": "internal_wall_insulation", "total": 500, "kwh_savings": 300,
"energy_cost_savings": 5, "has_battery": False}],
[{"recommendation_id": "vent1", "type": "mechanical_ventilation", "total": 50, "kwh_savings": 30,
"energy_cost_savings": 5, "has_battery": False}]
]
measures = optimiser_functions.prepare_input_measures(recs, goal="Energy Savings", needs_ventilation=True)
wall_option = measures[0][0]
assert wall_option["cost"] == 550
assert wall_option["gain"] == 330
assert "+mechanical_ventilation" in wall_option["type"]
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}],
]
measures = optimiser_functions.prepare_input_measures(recs, goal="Energy Savings", needs_ventilation=False)
assert measures == [] # should skip negative cost saving recs
class TestCalculateFixedGain:
def test_no_required_measures_returns_zero(self):
fixed_gain = optimiser_functions.calculate_fixed_gain(
[], {}, SimpleNamespace(id="P1"), needs_ventilation=False
)
assert fixed_gain == 0
def test_sums_max_sap_points_per_type(self, monkeypatch):
monkeypatch.setattr(optimiser_functions.assumptions, "measures_needing_ventilation",
["internal_wall_insulation"])
required_measures = [
[{"type": "internal_wall_insulation", "sap_points": 5},
{"type": "internal_wall_insulation", "sap_points": 10}],
[{"type": "loft_insulation", "sap_points": 3}]
]
recommendations = {"P1": [[{"type": "mechanical_ventilation", "sap_points": 2}]]}
prop = SimpleNamespace(id="P1")
gain = optimiser_functions.calculate_fixed_gain(
required_measures, recommendations, prop, needs_ventilation=True
)
# Should take max of wall (10) + loft (3) + ventilation (2)
assert gain == 15
class TestCalculateGain:
def test_returns_none_for_energy_savings_goal(self):
body = SimpleNamespace(goal="Energy Savings")
prop = SimpleNamespace(data={"current-energy-efficiency": "50"})
gain = optimiser_functions.calculate_gain(body, prop, fixed_gain=0)
assert gain is None
def test_calculates_gain_for_epc(self, monkeypatch):
# patch cost optimiser calculation
monkeypatch.setattr(optimiser_functions, "epc_to_sap_lower_bound", lambda goal_value: 69)
body = SimpleNamespace(goal="Increasing EPC", goal_value="C", simulate_sap_10=False)
prop = SimpleNamespace(data={"current-energy-efficiency": "50"})
gain = optimiser_functions.calculate_gain(body, prop, fixed_gain=2)
assert gain == 18.5
class TestAddRequiredMeasures:
def test_adds_cheapest_required_measure(self):
property_id = "P1"
required_measures = [
[{"recommendation_id": "a", "total": 100, "sap_points": 5, "type": "loft_insulation"},
{"recommendation_id": "b", "total": 80, "sap_points": 6, "type": "loft_insulation"}]
]
recommendations = {
"P1": [[{"recommendation_id": "a", "total": 100, "sap_points": 5, "type": "loft_insulation"},
{"recommendation_id": "b", "total": 80, "sap_points": 6, "type": "loft_insulation"}]]
}
selected = set()
result = optimiser_functions.add_required_measures(property_id, required_measures, recommendations, selected)
# cheapest should be b
assert "b" in selected
assert any(rec["id"] == "b" for rec in result)
class TestAddBestPracticeMeasures:
def test_adds_ventilation_and_trickle_vents(self, monkeypatch):
monkeypatch.setattr(optimiser_functions.assumptions, "measures_needing_ventilation",
["internal_wall_insulation"])
property_id = "P1"
solution = [{"type": "internal_wall_insulation", "id": "w1", "gain": 10, "cost": 100}]
recommendations = {
"P1": [
[{"type": "mechanical_ventilation", "recommendation_id": "vent1"}],
[{"type": "trickle_vents", "recommendation_id": "trickle1"}]
]
}
selected = set()
updated = optimiser_functions.add_best_practice_measures(property_id, solution, recommendations, selected)
assert "vent1" in updated
assert "trickle1" in updated
class TestFlattenRecommendationsWithDefaults:
def test_marks_selected_and_flattens(self):
property_id = "P1"
recommendations = {
"P1": [
[{"recommendation_id": "a", "foo": 1}, {"recommendation_id": "b", "foo": 2}],
[{"recommendation_id": "c", "foo": 3}]
]
}
selected = {"b", "c"}
result = optimiser_functions.flatten_recommendations_with_defaults(property_id, recommendations, selected)
# All recs should now have a default key
assert all("default" in rec for rec in result)
assert next(r for r in result if r["recommendation_id"] == "b")["default"] is True
assert next(r for r in result if r["recommendation_id"] == "a")["default"] is False
class TestIncreasingEpcE2e:
"""
Test out the classic increasing EPC optimisation flow end-to-end.
We have a goal (Increasing EPC), no budget, and we expect the optimiser to choose
the best set of measures and include best-practice ventilation.
"""
@pytest.fixture
def setup_case(self):
# ✅ Dummy property object
p = SimpleNamespace(
id="P1",
has_ventilation=False,
data={"current-energy-efficiency": "52"},
)
# ✅ Dummy request body
body = SimpleNamespace(
goal="Increasing EPC",
goal_value="C",
optimise=True,
budget=None,
simulate_sap_10=False,
required_measures=[]
)
# ✅ Use your massive measures_to_optimise list
recommendations = {"P1": measures_to_optimise}
return p, body, recommendations
def test_end_to_end_increasing_epc(self, setup_case):
p, body, recommendations = setup_case
# ---------------------
# RUN THE OPTIMISATION LOOP
# ---------------------
property_measure_types = {rec["type"] for recs in recommendations[p.id] for rec in recs}
property_required_measures = [m for m in recommendations[p.id] if m[0]["type"] in body.required_measures]
measures_to_optimise = [m for m in recommendations[p.id] if m[0]["type"] not in body.required_measures]
# ventilation flag
needs_ventilation = any(
x in property_measure_types for x in optimiser_functions.assumptions.measures_needing_ventilation
) and not p.has_ventilation
assert needs_ventilation
input_measures = optimiser_functions.prepare_input_measures(measures_to_optimise, body.goal, needs_ventilation)
assert input_measures, "Expected measures to optimise"
assert len(input_measures) == 7
fixed_gain = optimiser_functions.calculate_fixed_gain(
property_required_measures, recommendations, p, needs_ventilation
)
assert fixed_gain == 0, "No required measures should mean fixed gain is 0"
gain = optimiser_functions.calculate_gain(body=body, p=p, fixed_gain=fixed_gain)
assert gain == 18.5, "Expected gain to be calculated correctly based on fixed gain and SAP target"
optimiser = (
GainOptimiser(
input_measures, max_cost=body.budget, max_gain=gain,
allow_slack=body.goal == "Increasing EPC"
) if body.budget else CostOptimiser(input_measures, min_gain=gain)
)
optimiser.setup()
optimiser.solve()
solution = optimiser.solution
assert solution, "Optimiser should return a non-empty solution"
assert all("id" in m for m in solution)
assert any("solar_pv" in m["type"] for m in solution), "Expected solar PV to be included"
# Collect selected measure IDs
selected = {r["id"] for r in solution}
assert selected == {'8_phase=7', '5_phase=4', '7_phase=6'}
# Add required measures (none here)
solution = optimiser_functions.add_required_measures(
property_id=p.id, property_required_measures=property_required_measures,
recommendations=recommendations, selected=selected,
)
assert solution == [
{'id': '5_phase=4', 'cost': 58.8, 'gain': 2, 'type': 'low_energy_lighting'},
{'id': '7_phase=6', 'cost': 30.0, 'gain': np.float64(3.6), 'type': 'secondary_heating'},
{'id': '8_phase=7', 'cost': 6013.139999999999, 'gain': np.float64(13.0), 'type': 'solar_pv'}
]
total_optimised_gain = sum(m["gain"] for m in solution)
assert total_optimised_gain == 18.6, "Total gain of optimised measures should meet or exceed target gain"
selected = optimiser_functions.add_best_practice_measures(p.id, solution, recommendations, selected)
# Flatten recommendations for output
flattened = optimiser_functions.flatten_recommendations_with_defaults(p.id, recommendations, selected)
# ---------------------
# FINAL ASSERTIONS
# ---------------------
assert isinstance(flattened, list)
assert all("default" in rec for rec in flattened)
assert any(rec["default"] for rec in flattened), "Some measures should be marked as default"
# We don't add ventilation as major insulation work isn't done
ventilation_added = any(rec["recommendation_id"] == "3_phase=2" and rec["default"] for rec in flattened)
assert not ventilation_added, "Ventilation should not be added without major insulation work"

6
recommendations/tests/test_ventilation_recommendations.py
View file

@ -76,6 +76,8 @@ class TestVentilationRecommendations:
epc_record = EPCRecord()
epc_record.prepared_epc = {"mechanical-ventilation": "mechanical, extract only"}
input_property4 = Property(id=1, postcode="F4k3 6", address="623 fake street", epc_record=epc_record)
input_property4.identify_ventilation()
assert input_property4.has_ventilation
recommender4 = VentilationRecommendations(
property_instance=input_property4,
@ -87,12 +89,13 @@ class TestVentilationRecommendations:
recommender4.recommend(phase=None)
assert not recommender4.recommendation
assert recommender4.has_ventilaion
def test_existing_ventilation_2(self):
epc_record = EPCRecord()
epc_record.prepared_epc = {"mechanical-ventilation": "mechanical, supply and extract"}
input_property5 = Property(id=1, postcode="F4k3 6", address="623 fake street", epc_record=epc_record)
input_property5.identify_ventilation()
assert input_property5.has_ventilation
recommender5 = VentilationRecommendations(
property_instance=input_property5,
@ -104,4 +107,3 @@ class TestVentilationRecommendations:
recommender5.recommend(phase=None)
assert not recommender5.recommendation
assert recommender5.has_ventilaion

2
serverless.yml
View file

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

59
sfr/principal_pitch/2_export_data.py
View file

@ -7,10 +7,12 @@ from backend.app.utils import sap_to_epc
from sqlalchemy.orm import sessionmaker
from backend.app.db.connection import db_engine
from backend.app.db.models.recommendations import Recommendation, Plan, PlanRecommendations
from backend.app.db.models.portfolio import PropertyModel, PropertyDetailsEpcModel
from backend.app.db.models.portfolio import PropertyModel, PropertyDetailsEpcModel, PropertyDetailsSpatial
PORTFOLIO_ID = 206
SCENARIOS = [389]
# PORTFOLIO_ID = 206
# SCENARIOS = [389]
PORTFOLIO_ID = 221
SCENARIOS = [427]
def get_data(portfolio_id, scenario_ids):
@ -125,17 +127,64 @@ df["predicted_post_works_sap"] = df["predicted_post_works_sap"].round()
df["predicted_post_works_epc"] = df["predicted_post_works_sap"].apply(lambda x: sap_to_epc(x))
# We merge this back to the main dataframe, which will contain the bathrooms
from utils.s3 import read_csv_from_s3
from utils.s3 import read_csv_from_s3, read_excel_from_s3
asset_list = read_csv_from_s3(bucket_name="retrofit-plan-inputs-dev", filepath='8/206/asset_list.csv')
# asset_list = read_csv_from_s3(bucket_name="retrofit-plan-inputs-dev", filepath='8/206/asset_list.csv')
asset_list = read_excel_from_s3(
bucket_name="retrofit-plan-inputs-dev", file_key='8/221/20250722T202328736Z/asset_list.xlsx',
header_row=0, sheet_name="320 - edited"
)
asset_list = pd.DataFrame(asset_list)
asset_list = asset_list[["domna_full_address", "domna_postcode", "epc_os_uprn", ]].copy()
asset_list = asset_list.rename(columns={"epc_os_uprn": "uprn"})
df["uprn"] = df["uprn"].astype(str)
asset_list["uprn"] = asset_list["uprn"].astype("Int64").astype(str)
asset_list = asset_list.merge(
df.drop(columns=["address", "postcode", "property_type", "total_floor_area"]),
how="left",
on="uprn"
)
# Get conservation area data from property details spatial. based on the UPRNs
def get_conservation_area_data(uprns):
session = sessionmaker(bind=db_engine)()
session.begin()
# Query to get conservation area data
spatial_query = session.query(
PropertyDetailsSpatial
).filter(
PropertyDetailsSpatial.uprn.in_(uprns) # Filter by UPRNs
).all()
# Transform spatial data to include all fields dynamically
spatial_data = [
{col.name: getattr(spatial, col.name) for col in PropertyDetailsSpatial.__table__.columns}
for spatial in spatial_query
]
session.close()
return pd.DataFrame(spatial_data)
uprns = asset_list[
~pd.isna(asset_list["uprn"]) & (asset_list["uprn"] != "<NA>")
]["uprn"].astype(int).unique().tolist()
conservation_area_data = get_conservation_area_data(uprns)
conservation_area_data["uprn"] = conservation_area_data["uprn"].astype(str)
asset_list = asset_list.merge(
conservation_area_data[["uprn", "conservation_status", "is_listed_building", "is_heritage_building"]],
how="left",
on="uprn"
)
# For exporting NCHA
asset_list.to_excel(
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/NCHA/320 Portfolio/asset_list_epc_b.xlsx",
index=False
)
condition_costs = pd.read_excel(
"/Users/khalimconn-kowlessar/Documents/hestia/sfr/Spring JV/Condition costs.xlsx",
sheet_name="Prices - Khalim",