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 #480 from Hestia-Homes/funding-engine

Browse source 
Funding engine debugging
This commit is contained in:
KhalimCK 2025-08-22 03:26:34 +01:00 • committed by GitHub
commit ff98f87db0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 176 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

40
backend/Funding.py
View file

@ -109,7 +109,7 @@ class Funding:
return "73-97"
if floor_area <= 199:
return "98-199"
return "200"
return "200+"
@staticmethod
def _split_measures(measures: List[dict]):
@ -322,6 +322,10 @@ class Funding:
return data["Cost Savings"].values[0]
def _calculate_full_project_abs(self, floor_area_band: str, starting_sap_band: str, ending_sap_band: str):
if starting_sap_band == ending_sap_band:
return 0
data = self.project_scores_matrix[
(self.project_scores_matrix["Floor Area Segment"] == floor_area_band) &
(self.project_scores_matrix["Starting Band"] == starting_sap_band) &
@ -610,7 +614,7 @@ class Funding:
raise ValueError("something went wrong, more than one pps for ashp")
return pps.squeeze()["Cost Savings"]
if measure_type == "high_heat_retention_storage_heater":
if measure_type == "high_heat_retention_storage_heaters":
pps_data = filtered_pps_matrix[
filtered_pps_matrix["Post_Main_Heating_Source"] == "High Heat Retention Storage Heaters"
]
@ -638,6 +642,38 @@ class Funding:
# If we don't have a pre heating system, we assume the measure is not applicable
return 0
if measure_type in ["double_glazing", "secondary_glazing"]:
# pps is under the WG_singletodouble Measure_Type
pps = filtered_pps_matrix[
filtered_pps_matrix["Measure_Type"] == "WG_singletodouble"
]
return pps.squeeze()["Cost Savings"]
if measure_type == "roomstat_programmer_trvs":
# We can get funding for TRVs
pps = filtered_pps_matrix[
filtered_pps_matrix["Measure_Type"] == "TRV"
]
if pre_heating_system in pps["Pre_Main_Heating_Source"].values:
pps = pps[pps["Pre_Main_Heating_Source"] == pre_heating_system]
if pps.shape[0] != 1:
raise ValueError("something went wrong, more than one pps for TRV")
return pps.squeeze()["Cost Savings"]
# If we don't have a pre heating system, we assume the measure is not applicable
return 0
if measure_type == "time_temperature_zone_control":
pps = filtered_pps_matrix[
filtered_pps_matrix["Measure_Type"] == "TTZC"
]
if pre_heating_system in pps["Pre_Main_Heating_Source"].values:
pps = pps[pps["Pre_Main_Heating_Source"] == pre_heating_system]
if pps.shape[0] != 1:
raise ValueError("something went wrong, more than one pps for TTZC")
return pps.squeeze()["Cost Savings"]
# If we don't have a pre heating system, we assume the measure is not applicable
return 0
raise ValueError(f"Invalid measure type for partial project ABS calculation: {measure_type}")
# -----------------------

8
backend/app/db/functions/funding_functions.py
View file

@ -34,9 +34,15 @@ def upload_funding(session: Session, p, plan_id, recommendations_to_upload):
material_id = None
if recommendation["parts"]:
material_id = recommendation["parts"][0]["id"]
part_type = part["type"]
if part_type == "extension_cavity_wall_insulation":
part_type = "cavity_wall_insulation"
if part_type == "sealing_open_fireplace":
part_type = "sealing_fireplace"
funding_measures_data.append({
"funding_package_id": funding_package_id,
"measure": part["type"],
"measure": part_type,
"material_id": material_id,
"innovation_uplift": float(part["innovation_uplift"]),
"partial_project_score": float(part["partial_project_score"]),

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

@ -23,7 +23,7 @@ ECO4_ELIGIBLE_HEATING_MEASURES = [
SPECIFIC_MEASURES = (
WALL_INSULATION_MEASURES + ROOF_INSULATION_MEASURES + ECO4_ELIGIBILE_FABRIC_MEASURES +
ECO4_ELIGIBLE_HEATING_MEASURES + [
"secondary_heating" "ventilation", "low_energy_lighting", "fireplace",
"secondary_heating", "ventilation", "low_energy_lighting", "fireplace",
"hot_water_tank_insulation",
"cylinder_thermostat"
]

14
backend/engine/engine.py
View file

@ -438,6 +438,10 @@ def get_funding_data():
'Post_Main_Heating_Source', 'Total Floor Area Band', 'Starting Band',
'Average Treatable Factor', 'Cost Savings', 'SAP Savings'
]
# Replace 200 with 200+ in floor area band
partial_project_scores_matrix["Total Floor Area Band"] = partial_project_scores_matrix[
"Total Floor Area Band"
].replace({"200": "200+"})
partial_project_scores_matrix["Cost Savings"] = partial_project_scores_matrix["Cost Savings"].astype(float)
whlg_eligible_postcodes = read_csv_from_s3(
@ -850,9 +854,9 @@ async def model_engine(body: PlanTriggerRequest):
project_scores_matrix=project_scores_matrix,
partial_project_scores_matrix=partial_project_scores_matrix,
whlg_eligible_postcodes=whlg_eligible_postcodes,
eco4_social_cavity_abs_rate=13,
eco4_social_cavity_abs_rate=12.5,
eco4_social_solid_abs_rate=17,
eco4_private_cavity_abs_rate=13,
eco4_private_cavity_abs_rate=12.5,
eco4_private_solid_abs_rate=17,
gbis_social_cavity_abs_rate=21,
gbis_social_solid_abs_rate=25,
@ -879,7 +883,8 @@ async def model_engine(body: PlanTriggerRequest):
for group in measures_to_optimise_with_uplift:
for r in group:
if r["type"] in ["mechanical_ventilation", "low_energy_lighting", "secondary_heating"]:
if r["type"] in ["mechanical_ventilation", "low_energy_lighting", "secondary_heating",
"extension_cavity_wall_insulation", "draught_proofing", "sealing_open_fireplace"]:
(
r["partial_project_score"],
r["partial_project_funding"],
@ -949,7 +954,7 @@ async def model_engine(body: PlanTriggerRequest):
total_uplift = optimal_solution["total_uplift"]
# This is the funding scheme selected
# This is the full project ABS
full_project_score = optimal_solution["full_project_funding"]
full_project_score = optimal_solution["project_score"]
# This is the partial project ABS
partial_project_score = optimal_solution["partial_project_score"]
# This is the uplift score ABS
@ -1170,6 +1175,7 @@ async def model_engine(body: PlanTriggerRequest):
session.rollback()
print("Failed i = %s" % str(i))
logger.error(f"An error occurred during batch starting at index {i}: {e}")
logger.error(f"property is uprn {p.uprn} id {p.id} address {p.address}")
logger.info("Creating portfolio aggregations")
# We implement this in the simplest way possible which will be just to query the database for all

111
etl/customers/remote_assessments/app.py
View file

@ -4,7 +4,7 @@ from dotenv import load_dotenv
from utils.s3 import save_csv_to_s3
from etl.find_my_epc.AssetListEpcData import AssetListEpcData
PORTFOLIO_ID = 212
PORTFOLIO_ID = 235
USER_ID = 8
load_dotenv(dotenv_path="backend/.env")
@ -17,15 +17,45 @@ def app():
:return:
"""
asset_list = pd.read_excel(
"/Users/khalimconn-kowlessar/Downloads/Energy Information MASTER June 2025 - Standardised.xlsx",
sheet_name="Solar Properties",
)
asset_list = asset_list[~asset_list["estimated"]]
asset_list["domna_address_1"] = asset_list["domna_address_1"].astype(str)
asset_list = asset_list[["domna_address_1", "domna_postcode", "epc_os_uprn"]].rename(
columns={"domna_address_1": "address", "domna_postcode": "postcode", "epc_os_uprn": "uprn"}
)
# asset_list = pd.read_excel(
# "/Users/khalimconn-kowlessar/Downloads/Energy Information MASTER June 2025 - Standardised.xlsx",
# sheet_name="Solar Properties",
# )
# asset_list = asset_list[~asset_list["estimated"]]
# asset_list["domna_address_1"] = asset_list["domna_address_1"].astype(str)
# asset_list = asset_list[["domna_address_1", "domna_postcode", "epc_os_uprn"]].rename(
# columns={"domna_address_1": "address", "domna_postcode": "postcode", "epc_os_uprn": "uprn"}
# )
asset_list = [
{
"address": "9 Reeds Place",
"postcode": "PO12 3HR",
"uprn": 37017508
},
{
"address": "7 Crawley Road",
"postcode": "N22 6AN",
"uprn": 100021169757
},
{
"address": "20 Main Street",
"postcode": "NG32 1SE",
"uprn": 200002698370
},
{
"address": "19 Wolley Avenue",
"postcode": "LS12 5DX",
"uprn": 72234517
},
{
"address": "45 Bolton Lane, Hose",
"postcode": "LE14 4JE",
"uprn": 100030535501
}
]
asset_list = pd.DataFrame(asset_list)
# Store the asset list in s3
filename = f"{USER_ID}/{PORTFOLIO_ID}/asset_list.csv"
@ -64,16 +94,24 @@ def app():
valuation_data = [
{
"valuation": 339_000,
"uprn": 200003423454,
"valuation": 201000,
"uprn": 37017508,
},
{
"valuation": 374_000,
"uprn": 200003423194
"valuation": 810000,
"uprn": 100021169757,
},
{
"valuation": 719_000,
"uprn": 200003423607
"valuation": 228_000,
"uprn": 72234517
},
{
"valuation": 236_000,
"uprn": 100030535501
},
{
"valuation": 509000,
"uprn": 200002698370
},
]
# Store valuation data to s3
@ -84,19 +122,42 @@ def app():
file_name=valuation_filename
)
body = {
body1 = {
"portfolio_id": str(PORTFOLIO_ID),
"housing_type": "Private",
"housing_type": "Social",
"goal": "Increasing EPC",
"goal_value": "A",
"goal_value": "B",
"trigger_file_path": filename,
"already_installed_file_path": "",
"patches_file_path": patches_filename,
"non_invasive_recommendations_file_path": non_invasive_recommendations_filename,
"valuation_file_path": "",
"scenario_name": "Full package remote assessment",
"patches_file_path": "",
"non_invasive_recommendations_file_path": "",
"valuation_file_path": valuation_filename,
"scenario_name": "EPC B",
"multi_plan": True,
"budget": None,
"inclusions": ["cavity_wall_insulation", "ventilation"]
"ashp_cop": 3.5,
"event_type": "remote_assessment",
"default_u_values": True,
}
print(body)
print(body1)
body2 = {
"portfolio_id": str(PORTFOLIO_ID),
"housing_type": "Social",
"goal": "Increasing EPC",
"goal_value": "C",
"trigger_file_path": filename,
"already_installed_file_path": "",
"patches_file_path": "",
"non_invasive_recommendations_file_path": "",
"valuation_file_path": valuation_filename,
"scenario_name": "EPC C",
"multi_plan": True,
"budget": None,
"ashp_cop": 3.5,
"event_type": "remote_assessment",
"default_u_values": True,
}
print(body2)

8
etl/epc_clean/EpcClean.py
View file

@ -75,6 +75,9 @@ class EpcClean:
]
]
# Average
filtered_data.groupby("lighting-description")["low-energy-lighting"].mean().reset_index()
# Convert low-energy-lighting to float
for row in filtered_data:
row["low-energy-lighting"] = float(row["low-energy-lighting"])
@ -88,9 +91,10 @@ class EpcClean:
sums[description] += row["low-energy-lighting"]
counts[description] += 1
# Scale to between 0 and 1
averages = [{
"lighting-description": correct_spelling(description.lower()),
"low-energy-lighting": total / counts[description]
"lighting-description": correct_spelling(description.lower()) / 100,
"low-energy-lighting": total / counts[description] / 100
} for description, total in sums.items()]
return averages

1
recommendations/Costs.py
View file

@ -181,6 +181,7 @@ class Costs:
"solid_floor_insulation": 0.26,
"low_energy_lighting": 0.26,
"high_heat_retention_storage_heaters": 0.1,
"windows_glazing": 0.15,
}
# Preliminaries are a percentage of the total cost of the work and covers the cost of site-specific costs

2
recommendations/LightingRecommendations.py
View file

@ -100,7 +100,7 @@ class LightingRecommendations:
:return:
"""
if self.property.lighting["low_energy_proportion"] == 100:
if self.property.lighting["low_energy_proportion"] >= 1:
return
leds_recommendation_config = next(

2
recommendations/SolarPvRecommendations.py
View file

@ -264,7 +264,7 @@ class SolarPvRecommendations:
scaffolding_options=self.scaffolding_options,
n_floors=self.property.number_of_floors
)
description = f"Install a {solar_pv_product['description']}"
description = solar_pv_product['description']
if self.property.in_conservation_area:
description += " Property is in a consevation area - please check with local planning authority."

28
recommendations/optimiser/funding_optimiser.py
View file

@ -227,6 +227,19 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
# ECO4 fabric only path = special case
if isinstance(path_spec, dict) and path_spec.get("reference") == "fabric-only:eco4":
sub_measures = _filter_measures_by_types(optimisation_input_measures, path_spec["allowed_types"])
# If the property is EPC D and socil, we also include just innovation measures
if housing_type == "Social" and p.data["current-energy-rating"] == "D":
# We add in a second option which is just innovation measures
sub_measures_innovation = []
for measures in sub_measures:
group = []
for measure in measures:
if measure["innovation_uplift"]:
group.append(measure)
if group:
sub_measures_innovation.append(group)
sub_measures = deepcopy(sub_measures_innovation)
if not sub_measures:
continue
@ -380,7 +393,7 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
# If we have packages that are fundable, but do not meet the upgrade target, we can run a final optimisation pass
if not solutions[solutions["is_eligible"] & ~solutions["meets_upgrade_target"]].empty:
raise NotImplementedError("Implement me")
logger.info("We have some packages that are fundable but do not meet the target gain")
# We now can calculate the project ABS, which subtracts from the cost, but this is only relevant for ECO4
solutions["starting_sap"] = p.data["current-energy-efficiency"]
@ -397,6 +410,7 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
),
axis=1
)
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
@ -809,14 +823,22 @@ def make_funding_paths(p, input_measures, housing_type, funding: Funding):
input_measures_innovation = []
input_gbis_measures_innovation = []
for measures in input_measures:
group_of_innovation_measures = []
group_of_gbis_innovation_measures = []
for measure in measures:
if measure["innovation_uplift"] or measure["type"] in remaining_insulation_type:
input_measures_innovation.append([measure])
group_of_innovation_measures.append(measure)
if measure["innovation_uplift"] and measure["type"] in (
remaining_insulation_type + other_gbis_insulation_measures
):
input_gbis_measures_innovation.append([measure])
group_of_gbis_innovation_measures.append([measure])
if group_of_innovation_measures:
input_measures_innovation.append(group_of_innovation_measures)
if group_of_gbis_innovation_measures:
input_gbis_measures_innovation.extend(group_of_gbis_innovation_measures)
funding_paths = _make_solar_heating_funding_paths(
p, input_measures_innovation, funding_paths, remaining_insulation_type, housing_type, funding