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

added funding calulator to backend

Browse source
This commit is contained in:
Khalim Conn-Kowlessar 2025-08-20 17:37:26 +01:00
parent 2595d3a2de
commit b715402999
5 changed files with 212 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
backend/Funding.py
View file

@ -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":

34
backend/Property.py
View file

@ -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):

172
backend/engine/engine.py
View file

@ -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

29
recommendations/optimiser/funding_optimiser.py
View file

@ -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)

4
recommendations/optimiser/optimiser_functions.py
View file

@ -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"],
}
)