Merge pull request #492 from Hestia-Homes/funding-engine

Funding engine - fixing GBIS bugs
This commit is contained in:
KhalimCK 2025-08-26 18:14:26 +01:00 committed by GitHub
commit 3e4f33dfa3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 49 additions and 31 deletions

View file

@ -40,6 +40,9 @@ def upload_funding(session: Session, p, plan_id, recommendations_to_upload):
part_type = "cavity_wall_insulation"
if part_type == "sealing_open_fireplace":
part_type = "sealing_fireplace"
if part == "low_energy_lighting":
part_type = "low_energy_lighting_installation"
funding_measures_data.append({
"funding_package_id": funding_package_id,
"measure": part_type,

View file

@ -943,6 +943,7 @@ async def model_engine(body: PlanTriggerRequest):
# If we have a solution that meets the upgrade target, we select that one
optimal_solution = solutions[solutions["meets_upgrade_target"]].iloc[0]
else:
# Pick the cheapest
optimal_solution = solutions.iloc[0]
# This is the list of measures that we will recommend
@ -950,7 +951,8 @@ async def model_engine(body: PlanTriggerRequest):
funded_measures = optimal_solution["items"] if scheme != "none" else []
solution = optimal_solution["items"] + optimal_solution["unfunded_items"]
# This is the total amount of funding that the project will produce (including uplifts) (£)
project_funding = optimal_solution["full_project_funding"]
project_funding = optimal_solution["full_project_funding"] if scheme == "eco4" else \
optimal_solution["partial_project_funding"]
# This is the total amount of funding associated to the uplift (£)
total_uplift = optimal_solution["total_uplift"]
# This is the funding scheme selected

View file

@ -1,5 +1,8 @@
import numpy as np
from recommendations.county_to_region import county_to_region_map
from utils.logger import setup_logger
logger = setup_logger()
# This data comes from SPONs 2023
regional_labour_variations = [
@ -224,7 +227,9 @@ class Costs:
}.get(self.property.data["local-authority-label"].lower(), None)
if self.region is None:
raise ValueError("Region not found in county map")
logger.warning("No region found for county %s, defaulting to South East England",
self.property.data["county"])
self.region = "South East England"
self.labour_adjustment_factor = [
x["Adjustment_Factor"] for x in self.regional_labour_variations if

View file

@ -110,7 +110,9 @@ county_to_region_map = {
'West Oxfordshire': 'South East England', 'West Sussex': 'South East England', 'Winchester': 'South East England',
'Windsor and Maidenhead': 'South East England', 'Woking': 'South East England', 'Wokingham': 'South East England',
'Worthing': 'South East England', 'Wycombe': 'South East England',
'Bath and North East Somerset': 'South West England', 'Bournemouth': 'South West England',
'Bath and North East Somerset': 'South West England',
'Bournemouth': 'South West England',
'Bournemouth, Christchurch and Poole': 'South West England',
'Bristol': 'South West England',
'Cheltenham': 'South West England', 'Christchurch': 'South West England',
'City of Bristol': 'South West England',

View file

@ -21,7 +21,9 @@ from backend.Funding import Funding
logger = setup_logger()
# measures we DO NOT treat as fundable in the ECO4 'funded' pass
_ECO4_EXCLUDE_TYPES = {"secondary_heating", "extension_cavity_wall_insulation", "sealing_open_fireplace"}
_ECO4_EXCLUDE_TYPES = {
"secondary_heating", "extension_cavity_wall_insulation", "sealing_open_fireplace", "low_energy_lighting"
}
def _path_scheme(path_spec):
@ -329,25 +331,31 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
if picked is None:
continue
scheme = _path_scheme(path_spec)
total_cost = fixed_cost + sub_cost
total_gain = fixed_gain + sub_gain
total_picks = fixed_items + picked
unfunded_picked = []
if scheme == "gbis":
# The fixed items are fundded, everything else is unfunded
total_picks = fixed_items
unfunded_picked = picked
else:
total_picks = fixed_items + picked
if housing_type == "Private":
if not _prs_solution_ok(total_picks, p, funding):
if not _prs_solution_ok(total_picks, p, funding) and scheme == "eco4":
logger.error(
"Found a solution that does not meet the PRS requirements: %s - this shouldn't be happening",
total_picks
)
continue
scheme = _path_scheme(path_spec)
unfunded_picked = []
if total_gain - target_gain < -0.1:
# In this case, we have a funded package that does not meet the target gain, so we look at the remaining
# measures and see if we can include them
picked_types = {opt["type"] for opt in total_picks}
picked_types = {opt["type"] for opt in total_picks + unfunded_picked}
# We find the indexes of the picked types
picked_group_index = {}
@ -371,11 +379,13 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
if remaining:
# If we have remaining measures we can optimise, we run them down an unfunded route
unfunded_picked, unfunded_cost, unfunded_gain = run_optimizer(
unfunded_picked_remaining, unfunded_cost, unfunded_gain = run_optimizer(
remaining,
budget - total_cost if budget is not None else None,
sub_target_gain=target_gain - total_gain if target_gain is not None else None
)
if unfunded_picked_remaining is not None:
unfunded_picked += unfunded_picked_remaining
total_cost += unfunded_cost
total_gain += unfunded_gain
@ -417,13 +427,6 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
axis=1
)
for _, x in solutions.iterrows():
funding._calculate_full_project_abs(
floor_area_band=x["floor_area_band"],
starting_sap_band=x["starting_band"],
ending_sap_band=x["ending_band"],
)
rate = funding.get_eco4_abs_rate(is_cavity=p.walls["is_cavity_wall"])
solutions["full_project_funding"] = solutions["project_score"] * rate
# if the scheme is not ECO4, we set the funding to 0 with iloc
@ -803,6 +806,7 @@ def make_funding_paths(p, input_measures, housing_type, funding: Funding):
:param p: The property object containing details about the property, including main heating and controls.
:param input_measures:
:param housing_type:
:param funding: The funding object that provides methods to check eligibility and calculate funding.
:return:
"""
# We handle the case of minimum insulation requirements. Whenever we have a heating system recommendation,
@ -862,25 +866,27 @@ def make_funding_paths(p, input_measures, housing_type, funding: Funding):
return funding_paths, input_measures_innovation
if housing_type == "Private":
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# EWI or IWI
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# 1) The package must include EWI or IWI if the property is private rental sector
# We check if we have any EWI or IWI measures available
ewi_or_iwi = [{"OR": []}]
reference_measures = []
# If we have EWI we add it in
if _find_measure(input_measures, "external_wall_insulation"):
ewi_or_iwi[0]["OR"].append("external_wall_insulation")
reference_measures.append("ewi")
# We check if we have any EWI or IWI measures available - only for EPC E or below
if p.data["current-energy-rating"] not in ["E", "F", "G"]:
ewi_or_iwi = [{"OR": []}]
reference_measures = []
# If we have EWI we add it in
if _find_measure(input_measures, "external_wall_insulation"):
ewi_or_iwi[0]["OR"].append("external_wall_insulation")
reference_measures.append("ewi")
if _find_measure(input_measures, "internal_wall_insulation"):
ewi_or_iwi[0]["OR"].append("internal_wall_insulation")
reference_measures.append("iwi")
if _find_measure(input_measures, "internal_wall_insulation"):
ewi_or_iwi[0]["OR"].append("internal_wall_insulation")
reference_measures.append("iwi")
if ewi_or_iwi[0]["OR"]:
ewi_or_iwi[0]["reference"] = "+".join(reference_measures) + ":eco4"
funding_paths.append(ewi_or_iwi)
if ewi_or_iwi[0]["OR"]:
ewi_or_iwi[0]["reference"] = "+".join(reference_measures) + ":eco4"
funding_paths.append(ewi_or_iwi)
funding_paths = _make_solar_heating_funding_paths(
p, input_measures, funding_paths, remaining_insulation_type, housing_type, funding