From b715402999041dba52fd9d33269d8755b31fd6f4 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Wed, 20 Aug 2025 17:37:26 +0100 Subject: [PATCH] added funding calulator to backend --- backend/Funding.py | 48 ++++- backend/Property.py | 34 +++- backend/engine/engine.py | 172 +++++++++++++----- .../optimiser/funding_optimiser.py | 29 ++- .../optimiser/optimiser_functions.py | 4 +- 5 files changed, 212 insertions(+), 75 deletions(-) diff --git a/backend/Funding.py b/backend/Funding.py index 9ab301cd..05c2921d 100644 --- a/backend/Funding.py +++ b/backend/Funding.py @@ -70,6 +70,7 @@ class Funding: self.gbis_funding = None self.eco4_funding = None self.eco4_uplift = 0 + self.gbis_uplift = 0 self.partial_project_abs = None @@ -118,7 +119,7 @@ class Funding: """ measure_types = [m["type"] for m in measures] innovation_flags = [m.get("is_innovation", False) for m in measures] - uplifts = [m["uplift"] for m in measures] + uplifts = [m["innovation_uplift"] for m in measures] innovation_measures = [m["type"] for m in measures if m.get("is_innovation", False)] return measure_types, uplifts, innovation_flags, innovation_measures @@ -952,8 +953,7 @@ class Funding: existing_li_thickness=existing_li_thickness, ) - self.full_project_abs += self.eco4_uplift - self.eco4_funding = self.full_project_abs * ( + self.eco4_funding = (self.full_project_abs + self.eco4_uplift) * ( self.eco4_social_cavity_abs_rate if is_cavity else self.eco4_social_solid_abs_rate ) @@ -966,7 +966,22 @@ class Funding: filtered_pps_matrix=filtered_pps_matrix, pre_heating_system=pre_heating_system ) - self.gbis_funding = self.partial_project_abs * ( + + self.gbis_uplift = self.calc_innovation_uplift( + measure_types=measure_types, + innovation_flags=innovation_flags, + uplifts=uplifts, + filtered_pps_matrix=filtered_pps_matrix, + pre_heating_system=pre_heating_system, + mainheating=mainheating, + main_fuel=main_fuel, + mainheat_energy_eff=mainheat_energy_eff, + current_wall_uvalue=current_wall_uvalue, + is_partial=is_partial, + existing_li_thickness=existing_li_thickness, + ) + + self.gbis_funding = (self.partial_project_abs + self.gbis_uplift) * ( self.gbis_private_cavity_abs_rate if is_cavity else self.gbis_private_solid_abs_rate ) @@ -997,8 +1012,7 @@ class Funding: existing_li_thickness=existing_li_thickness, ) - self.full_project_abs += self.eco4_uplift - self.eco4_funding = self.full_project_abs * ( + self.eco4_funding = (self.full_project_abs + self.eco4_uplift) * ( self.eco4_social_cavity_abs_rate if is_cavity else self.eco4_social_solid_abs_rate ) @@ -1012,7 +1026,21 @@ class Funding: filtered_pps_matrix=filtered_pps_matrix, pre_heating_system=pre_heating_system ) - self.gbis_funding = self.partial_project_abs * ( + self.gbis_uplift = self.calc_innovation_uplift( + measure_types=measure_types, + innovation_flags=innovation_flags, + uplifts=uplifts, + filtered_pps_matrix=filtered_pps_matrix, + pre_heating_system=pre_heating_system, + mainheating=mainheating, + main_fuel=main_fuel, + mainheat_energy_eff=mainheat_energy_eff, + current_wall_uvalue=current_wall_uvalue, + is_partial=is_partial, + existing_li_thickness=existing_li_thickness, + ) + + self.gbis_funding = (self.partial_project_abs + self.gbis_uplift) * ( self.gbis_social_cavity_abs_rate if is_cavity else self.gbis_social_solid_abs_rate ) @@ -1061,7 +1089,7 @@ class Funding: else self.eco4_private_solid_abs_rate ) - return pps, pps * rate, innovation_uplift * rate + return pps, pps * rate, innovation_uplift * rate, innovation_uplift if self.tenure == "Social": # We return ECO4 rates @@ -1069,11 +1097,11 @@ class Funding: self.eco4_social_cavity_abs_rate if is_cavity else self.eco4_social_solid_abs_rate ) - return pps, pps * rate, innovation_uplift * rate + return pps, pps * rate, innovation_uplift * rate, innovation_uplift raise ValueError("Invalid tenure type for innovation uplift calculation: {}".format(self.tenure)) - def get_abs_rate(self, is_cavity: bool) -> float: + def get_eco4_abs_rate(self, is_cavity: bool) -> float: if self.tenure == "Social": return self.eco4_social_cavity_abs_rate if is_cavity else self.eco4_social_solid_abs_rate if self.tenure == "Private": diff --git a/backend/Property.py b/backend/Property.py index d6f43b8a..5b96d413 100644 --- a/backend/Property.py +++ b/backend/Property.py @@ -213,9 +213,16 @@ class Property: self.parse_kwargs(kwargs) # Funding - self.gbis_eligibiltiy = None - self.eco4_eligibility = None - self.whlg_eligibility = None + # self.gbis_eligibiltiy = None + # self.eco4_eligibility = None + # self.whlg_eligibility = None + self.scheme = None + self.funded_measures = None + self.full_project_funding = None + self.total_uplift = None + self.full_project_score = None + self.partial_project_score = None + self.uplift_project_score = None # Ventilation self.has_ventilation = self.identify_ventilation() @@ -1336,13 +1343,26 @@ class Property: return electric_consumption - def insert_funding(self, funding_calulator: Funding): + def insert_funding( + self, + scheme, + funded_measures, + full_project_funding, + total_uplift, + full_project_score, + partial_project_score, + uplift_project_score + ): """ This method inserts the funding into the property object """ - self.gbis_eligibiltiy = funding_calulator.gbis_eligibiltiy - self.eco4_eligibility = funding_calulator.eco4_eligibility - self.whlg_eligibility = funding_calulator.whlg_eligibility + self.scheme = scheme + self.funded_measures = funded_measures, + self.full_project_funding = full_project_funding, + self.total_uplift = total_uplift, + self.full_project_score = full_project_score, + self.partial_project_score = partial_project_score, + self.uplift_project_score = uplift_project_score def identify_ventilation(self): diff --git a/backend/engine/engine.py b/backend/engine/engine.py index b5330c47..1a94919c 100644 --- a/backend/engine/engine.py +++ b/backend/engine/engine.py @@ -25,7 +25,7 @@ from backend.app.db.functions.recommendations_functions import ( ) from backend.app.db.functions.energy_assessment_functions import get_latest_assessment_by_uprn from backend.app.db.models.portfolio import rating_lookup -from backend.app.plan.schemas import PlanTriggerRequest +from backend.app.plan.schemas import PlanTriggerRequest, WALL_INSULATION_MEASURES, ROOF_INSULATION_MEASURES from backend.app.plan.utils import get_cleaned from backend.app.utils import sap_to_epc import backend.app.assumptions as assumptions @@ -46,6 +46,10 @@ from etl.bill_savings.KwhData import KwhData from etl.spatial.OpenUprnClient import OpenUprnClient from etl.find_my_epc.RetrieveFindMyEpc import RetrieveFindMyEpc +from backend.Funding import Funding +from recommendations.optimiser.funding_optimiser import optimise_with_funding_paths +from recommendations.recommendation_utils import convert_thickness_to_numeric, get_wall_u_value + logger = setup_logger() BATCH_SIZE = 5 @@ -833,10 +837,6 @@ async def model_engine(body: PlanTriggerRequest): ) gain = optimiser_functions.calculate_gain(body=body, p=p, fixed_gain=fixed_gain) - from backend.Funding import Funding - from recommendations.optimiser.funding_optimiser import optimise_with_funding_paths - from recommendations.recommendation_utils import convert_thickness_to_numeric, get_wall_u_value - funding = Funding( tenure=body.housing_type, project_scores_matrix=project_scores_matrix, @@ -852,51 +852,57 @@ async def model_engine(body: PlanTriggerRequest): gbis_private_solid_abs_rate=28, ) - # When the goal is Increasing EPC, we can run the funding optimiser - if body.goal == "Increasing EPC": - - # We insert the innovation uplift - measures_to_optimise_with_uplift = deepcopy(measures_to_optimise) - - li_thickness = convert_thickness_to_numeric( - p.roof["insulation_thickness"], p.roof["is_pitched"], p.roof["is_flat"] + li_thickness = convert_thickness_to_numeric( + p.roof["insulation_thickness"], p.roof["is_pitched"], p.roof["is_flat"] + ) + current_wall_u_value = p.walls["thermal_transmittance"] + if current_wall_u_value is None: + current_wall_u_value = get_wall_u_value( + clean_description=p.walls["clean_description"], + age_band=p.age_band, + is_granite_or_whinstone=p.walls["is_granite_or_whinstone"], + is_sandstone_or_limestone=p.walls["is_sandstone_or_limestone"], ) - current_wall_u_value = p.walls["thermal_transmittance"] - if current_wall_u_value is None: - current_wall_u_value = get_wall_u_value( - clean_description=p.walls["clean_description"], - age_band=p.age_band, - is_granite_or_whinstone=p.walls["is_granite_or_whinstone"], - is_sandstone_or_limestone=p.walls["is_sandstone_or_limestone"], + + # We insert the innovation uplift + measures_to_optimise_with_uplift = deepcopy(measures_to_optimise) + + # TODO: Turn this into a function and store the innovaiton uplift + for group in measures_to_optimise_with_uplift: + for r in group: + + if r["type"] in ["mechanical_ventilation", "low_energy_lighting", "secondary_heating"]: + ( + r["partial_project_score"], + r["partial_project_funding"], + r["innovation_uplift"], + r["uplift_project_score"], + ) = ( + 0, 0, 0, 0 + ) + continue + ( + r["partial_project_score"], r["partial_project_funding"], r["innovation_uplift"], + r["uplift_project_score"] + ) = funding.get_innovation_uplift( + measure=r, + starting_sap=p.data["current-energy-efficiency"], + floor_area=p.floor_area, + is_cavity=p.walls["is_cavity_wall"], + current_wall_uvalue=current_wall_u_value, + is_partial="partial" in p.walls["clean_description"].lower(), + existing_li_thickness=li_thickness, + mainheating=p.main_heating, + main_fuel=p.main_fuel, + mainheat_energy_eff=p.data["mainheat-energy-eff"], ) - # TODO: Turn this into a function and store the innovaiton uplift - for group in measures_to_optimise_with_uplift: - for r in group: - if r["type"] in ["mechanical_ventilation", "low_energy_lighting", "secondary_heating"]: - r["partial_project_score"], r["partial_project_funding"], r["innovation_uplift"] = ( - 0, 0, 0 - ) - continue + input_measures = optimiser_functions.prepare_input_measures( + measures_to_optimise_with_uplift, body.goal, needs_ventilation, funding=True + ) - ( - r["partial_project_score"], r["partial_project_funding"], r["innovation_uplift"] - ) = funding.get_innovation_uplift( - measure=r, - starting_sap=p.data["current-energy-efficiency"], - floor_area=p.floor_area, - is_cavity=p.walls["is_cavity_wall"], - current_wall_uvalue=current_wall_u_value, - is_partial="partial" in p.walls["clean_description"].lower(), - existing_li_thickness=li_thickness, - mainheating=p.main_heating, - main_fuel=p.main_fuel, - mainheat_energy_eff=p.data["mainheat-energy-eff"], - ) - - input_measures = optimiser_functions.prepare_input_measures( - measures_to_optimise_with_uplift, body.goal, needs_ventilation, funding=True - ) + # When the goal is Increasing EPC, we can run the funding optimiser + if body.goal == "Increasing EPC": solutions = optimise_with_funding_paths( p=p, @@ -925,10 +931,21 @@ async def model_engine(body: PlanTriggerRequest): else: optimal_solution = solutions.iloc[0] - solution = optimal_solution["items"] + optimal_solution["unfunded_items"] - full_project_funding = optimal_solution["eco4_full_project_funding"] + # This is the list of measures that we will recommend + funded_measures = optimal_solution["items"] + solution = funded_measures + optimal_solution["unfunded_items"] + # This is the total amount of funding that the project will product (£) + full_project_funding = optimal_solution["full_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 scheme = optimal_solution["scheme"] - + # This is the full project ABS + full_project_score = optimal_solution["full_project_funding"] + # This is the partial project ABS + partial_project_score = optimal_solution["partial_project_score"] + # This is the uplift score ABS + uplift_project_score = optimal_solution["total_uplift_score"] else: # We optimise and then we determine eligibility for funding, based on the measures selected optimiser = ( @@ -940,6 +957,53 @@ async def model_engine(body: PlanTriggerRequest): optimiser.solve() solution = optimiser.solution + recommendation_types = [] + for measures in input_measures: + for measure in measures: + recommendation_types.append(measure["type"]) + recommendation_types = set(recommendation_types) + + has_wall_insulation_recommendation = any( + (m in recommendation_types or "+".join([m, "mechanical_ventilation"])) for m in + WALL_INSULATION_MEASURES + ) + has_roof_insulation_recommendation = any( + (m in recommendation_types or "+".join([m, "mechanical_ventilation"])) for m in + ROOF_INSULATION_MEASURES + ) + + funding.check_funding( + measures=solution, + starting_sap=p.data["current-energy-efficiency"], + ending_sap=p.data["current-energy-efficiency"] + sum([x["gain"] for x in solution]), + floor_area=p.floor_area, + mainheat_description=p.main_heating["clean_description"], + heating_control_description=p.main_heating_controls["clean_description"], + is_cavity=p.walls["is_cavity_wall"], + current_wall_uvalue=current_wall_u_value, + is_partial="partial" in p.walls["clean_description"].lower(), + existing_li_thickness=li_thickness, + mainheating=p.main_heating, + main_fuel=p.main_fuel, + mainheat_energy_eff=p.data["mainheat-energy-eff"], + has_wall_insulation_recommendation=has_wall_insulation_recommendation, + has_roof_insulation_recommendation=has_roof_insulation_recommendation, + ) + + # Determine the scheme + scheme = "none" + if funding.eco4_eligible: + scheme = "eco4" + if scheme == "none" and funding.gbis_eligible: + scheme = "gbis" + + funded_measures = solution if scheme in ["gbis", "eco4"] else [] + full_project_funding = 0 if funding.full_project_abs is not None else funding.full_project_abs + total_uplift = funding.eco4_uplift + full_project_score = 0 if funding.full_project_abs is not None else funding.full_project_abs + partial_project_score = funding.partial_project_abs + uplift_project_score = funding.eco4_uplift if scheme == "eco4" else funding.gbis_uplift + selected = {r["id"] for r in solution} if property_required_measures: @@ -955,6 +1019,16 @@ async def model_engine(body: PlanTriggerRequest): p.id, recommendations, selected ) + p.insert_funding( + scheme=scheme, + funded_measures=funded_measures, + full_project_funding=full_project_funding, + total_uplift=total_uplift, + full_project_score=full_project_score, + partial_project_score=partial_project_score, + uplift_project_score=uplift_project_score + ) + # when we have buildings, we tweak our solar PV recommendations as if one unit needs it, we apply it to all # of them # TODO: We can probably do better and optimise at the building level - this is temp diff --git a/recommendations/optimiser/funding_optimiser.py b/recommendations/optimiser/funding_optimiser.py index 999355ee..060826cd 100644 --- a/recommendations/optimiser/funding_optimiser.py +++ b/recommendations/optimiser/funding_optimiser.py @@ -201,6 +201,7 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin "path": {"reference": "unfunded:all"}, "scheme": "none", "is_eligible": False, # no funding scheme applied + "unfunded_items": [] }) # This function will filter down on innovation measures if we are social EPC D @@ -291,7 +292,7 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin # do this by adding innovation back onto the cost for grp in sub_measures: for opt in grp: - opt["cost"] = x["raw_cost"] + opt["cost"] = opt["raw_cost"] if scheme == "eco4": # Need to strip out any measure types that are not eligible for ECO4 funding (e.g. secondary heating) @@ -396,17 +397,16 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin ), axis=1 ) - rate = funding.get_abs_rate(is_cavity=p.walls["is_cavity_wall"]) + 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 solutions.loc[solutions["scheme"] != "eco4", "full_project_funding"] = 0.0 - solutions["partial_project_funding"] = solutions.apply( - lambda x: get_gbis_pps(x), - axis=1 - ) + solutions["partial_project_funding"] = solutions.apply(lambda x: get_gbis_pp_funding(x), axis=1) + solutions["partial_project_score"] = solutions.apply(lambda x: get_gbis_pps(x), axis=1) # We pull out uplifts solutions["total_uplift"] = solutions.apply(lambda x: get_total_uplift(x), axis=1) + solutions["total_uplift_score"] = solutions.apply(lambda x: get_total_innovation_score(x), axis=1) return solutions @@ -414,19 +414,32 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin # ---- helpers ------------------------------------------------------------- +def get_gbis_pp_funding(x): + if x["scheme"] != "gbis": + return 0 + fixed_ids = x["fixed_ids"] + if len(fixed_ids) != 1: + raise ValueError("More than one fixed ID for GBIS") + return [x for x in x["items"] if x["id"] in fixed_ids][0]["partial_project_funding"] + + def get_gbis_pps(x): if x["scheme"] != "gbis": return 0 - fixed_ids = row["fixed_ids"] + fixed_ids = x["fixed_ids"] if len(fixed_ids) != 1: raise ValueError("More than one fixed ID for GBIS") - return [x for x in row["items"] if x["id"] in fixed_ids][0]["partial_project_funding"] + return [x for x in x["items"] if x["id"] in fixed_ids][0]["partial_project_score"] def get_total_uplift(x): return sum([y["innovation_uplift"] for y in x["items"]]) +def get_total_innovation_score(x): + return sum([y["uplift_project_score"] for y in x["items"]]) + + def sum_cost_gain(items): c = sum(float(x['cost']) for x in items) g = sum(float(x['gain']) for x in items) diff --git a/recommendations/optimiser/optimiser_functions.py b/recommendations/optimiser/optimiser_functions.py index 4ffe9ef3..98725138 100644 --- a/recommendations/optimiser/optimiser_functions.py +++ b/recommendations/optimiser/optimiser_functions.py @@ -117,7 +117,9 @@ def prepare_input_measures(property_recommendations, goal, needs_ventilation, fu "innovation_uplift": rec["innovation_uplift"] if funding else 0, "cost_minus_uplift": total, "raw_cost": raw_cost, - "partial_project_funding": rec["partial_project_funding"] + "partial_project_funding": rec["partial_project_funding"], + "partial_project_score": rec["partial_project_score"], + "uplift_project_score": rec["uplift_project_score"], } )