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 branch 'main' of https://github.com/Hestia-Homes/Model into feature/ara-rebaselining

Browse source
This commit is contained in:
Khalim Conn-Kowlessar 2026-02-13 12:58:34 +00:00
commit 2400ade256
6 changed files with 303 additions and 62 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
backend/engine/engine.py
View file

@ -871,7 +871,7 @@ async def model_engine(body: PlanTriggerRequest):
check_duplicate_property_ids(input_properties)
logger.info("Inserting property data")
# We now bulk upload all of the EPC data
# We now bulk upload all the EPC data
with db_session() as session:
db_funcs.epc_functions.EpcStoreService.bulk_upsert_epc_data(session, epc_upserts)

18
recommendations/Recommendations.py
View file

@ -768,6 +768,24 @@ class Recommendations:
# Update the current phase values
current_phase_values["sap"] = previous_phase_values["sap"] + property_phase_impact["sap"]
# This is very much an edge case but we also the end result taking the property
# below a SAP rating of 1, which is the minimum SAP rating
if previous_phase_values["sap"] + property_phase_impact["sap"] < 1:
sap_adjustment = 1 - (previous_phase_values["sap"] + property_phase_impact["sap"])
adjustments.append(
{
"recommendation_id": rec["recommendation_id"],
"phase": rec["phase"],
"sap_adjustment": sap_adjustment,
}
)
# The new impact should be the current impact plus the adjustment
property_phase_impact["sap"] = property_phase_impact["sap"] + sap_adjustment
# Update the current phase values
current_phase_values["sap"] = previous_phase_values["sap"] + property_phase_impact["sap"]
elif rec["type"] == "loft_insulation":
# When we have a loft insulation recommendation, where there is an extension and the existing
# amount of loft insulation is already good, we limit the SAP points

41
recommendations/optimiser/funding_optimiser.py
View file

@ -10,6 +10,7 @@ In the future, we will adapt this into a class-based structure to allow for more
from copy import deepcopy
import pandas as pd
import numpy as np
from typing import Mapping, Union
from itertools import product
from backend.app.plan.schemas import (
@ -823,21 +824,23 @@ def optimise_with_scenarios(
# No special path; just exclude ASHP from options and allow us to optimise.
measures_no_heat_pump = exclude_measure_types(optimisation_measures, ["air_source_heat_pump"])
picked, total_cost, total_gain = run_optimizer(
measures_no_heat_pump,
budget=budget,
sub_target_gain=target_gain,
)
if target_gain > 0:
# If we don't have any gain, we don't actually need to do this
picked, total_cost, total_gain = run_optimizer(
measures_no_heat_pump,
budget=budget,
sub_target_gain=target_gain,
)
if picked is not None:
solutions.append({
"scenario": "no_heat_pump",
"items": picked,
"fixed_items": [],
"total_cost": total_cost,
"total_gain": total_gain,
"already_installed_gain": sum([x["gain"] for x in picked if x["already_installed"]])
})
if picked is not None:
solutions.append({
"scenario": "no_heat_pump",
"items": picked,
"fixed_items": [],
"total_cost": total_cost,
"total_gain": total_gain,
"already_installed_gain": sum([x["gain"] for x in picked if x["already_installed"]])
})
solutions_df = append_solution_metrics(solutions, target_gain, p, already_installed_sap)
@ -1101,7 +1104,12 @@ def contributes_min_insulation(opt_types):
})
def run_optimizer(input_measures, budget=None, sub_target_gain=None, allow_slack=False):
def run_optimizer(
input_measures: list[list[Mapping[str, int | float | str]]],
budget: Union[float, None] = None,
sub_target_gain: Union[float, None] = None,
allow_slack: bool = False
):
"""
Thin wrapper over your optimisers.
Returns: list[dict] selected_options
@ -1112,7 +1120,7 @@ def run_optimizer(input_measures, budget=None, sub_target_gain=None, allow_slack
if budget is not None:
opt = GainOptimiser(
input_measures, max_cost=budget, max_gain=(sub_target_gain or float("inf")),
input_measures, max_cost=budget, max_gain=0 if sub_target_gain == 0 else (sub_target_gain or float("inf")),
allow_slack=allow_slack
)
else:
@ -1123,6 +1131,7 @@ def run_optimizer(input_measures, budget=None, sub_target_gain=None, allow_slack
opt.setup()
opt.solve()
cost = sum([x["cost"] for x in opt.solution])
return opt.solution, cost, opt.solution_gain

141
recommendations/tests/test_optimisers.py
View file

@ -1,6 +1,7 @@
import pytest
from recommendations.optimiser.funding_optimiser import build_heat_pump_paths
from recommendations.optimiser.funding_optimiser import run_optimizer
class DummyProp:
@ -68,3 +69,143 @@ def test_build_heat_pump_paths():
assert eg2 == [{'AND': ['internal_wall_insulation', 'loft_insulation', 'air_source_heat_pump']},
{'AND': ['external_wall_insulation', 'loft_insulation', 'air_source_heat_pump']}]
def test_run_optimizer_empty_input():
solution, cost, gain = run_optimizer([])
assert solution is None
assert cost == 0.0
assert gain == 0.0
def test_uses_gain_optimiser_when_budget_provided(monkeypatch):
captured_args = {}
class FakeGainOptimiser:
def __init__(self, measures, max_cost, max_gain, allow_slack):
captured_args["measures"] = measures
captured_args["max_cost"] = max_cost
captured_args["max_gain"] = max_gain
captured_args["allow_slack"] = allow_slack
self.solution = [{"cost": 100}]
self.solution_gain = 5
def setup(self):
pass
def solve(self):
pass
monkeypatch.setattr(
"recommendations.optimiser.funding_optimiser.GainOptimiser",
FakeGainOptimiser
)
measures = [[{"cost": 100, "gain": 5}]]
solution, cost, gain = run_optimizer(
measures,
budget=500,
sub_target_gain=10,
allow_slack=True
)
assert captured_args["max_cost"] == 500
assert captured_args["max_gain"] == 10
assert captured_args["allow_slack"] is True
assert cost == 100
assert gain == 5
def test_sub_target_gain_zero_sets_max_gain_zero(monkeypatch):
captured_args = {}
class FakeGainOptimiser:
def __init__(self, measures, max_cost, max_gain, allow_slack):
captured_args["max_gain"] = max_gain
self.solution = []
self.solution_gain = 0
def setup(self):
pass
def solve(self):
pass
monkeypatch.setattr(
"recommendations.optimiser.funding_optimiser.GainOptimiser",
FakeGainOptimiser
)
measures = [[{"cost": 100, "gain": 5}]]
run_optimizer(
measures,
budget=500,
sub_target_gain=0
)
assert captured_args["max_gain"] == 0
def test_sub_target_gain_none_sets_max_gain_infinity(monkeypatch):
captured_args = {}
class FakeGainOptimiser:
def __init__(self, measures, max_cost, max_gain, allow_slack):
captured_args["max_gain"] = max_gain
self.solution = []
self.solution_gain = 0
def setup(self):
pass
def solve(self):
pass
monkeypatch.setattr(
"recommendations.optimiser.funding_optimiser.GainOptimiser",
FakeGainOptimiser
)
measures = [[{"cost": 100, "gain": 5}]]
run_optimizer(
measures,
budget=500,
sub_target_gain=None
)
assert captured_args["max_gain"] == float("inf")
def test_uses_cost_optimiser_when_no_budget(monkeypatch):
captured_args = {}
class FakeCostOptimiser:
def __init__(self, measures, min_gain):
captured_args["min_gain"] = min_gain
self.solution = [{"cost": 50}]
self.solution_gain = 10
def setup(self):
pass
def solve(self):
pass
monkeypatch.setattr(
"recommendations.optimiser.funding_optimiser.CostOptimiser",
FakeCostOptimiser
)
measures = [[{"cost": 50, "gain": 10}]]
solution, cost, gain = run_optimizer(
measures,
sub_target_gain=10
)
assert captured_args["min_gain"] == 10
assert cost == 50
assert gain == 10

126
recommendations/tests/test_recommendations.py
View file

@ -347,21 +347,21 @@ def property_instance():
"input_data, expected",
[
(
[
{"recommendation_id": "a", "phase": 0, "sap_adjustment": 1.7},
{"recommendation_id": "b", "phase": 0, "sap_adjustment": 1.7},
],
[{"recommendation_id": "a", "phase": 0, "sap_adjustment": 1.7}],
[
{"recommendation_id": "a", "phase": 0, "sap_adjustment": 1.7},
{"recommendation_id": "b", "phase": 0, "sap_adjustment": 1.7},
],
[{"recommendation_id": "a", "phase": 0, "sap_adjustment": 1.7}],
),
(
[
{"recommendation_id": "a", "phase": 1, "sap_adjustment": 2},
{"recommendation_id": "b", "phase": 2, "sap_adjustment": 3},
],
[
{"recommendation_id": "a", "phase": 1, "sap_adjustment": 2},
{"recommendation_id": "b", "phase": 2, "sap_adjustment": 3},
],
[
{"recommendation_id": "a", "phase": 1, "sap_adjustment": 2},
{"recommendation_id": "b", "phase": 2, "sap_adjustment": 3},
],
[
{"recommendation_id": "a", "phase": 1, "sap_adjustment": 2},
{"recommendation_id": "b", "phase": 2, "sap_adjustment": 3},
],
),
],
)
@ -1478,3 +1478,103 @@ def test_lighting_and_loft_adjustment_combined(property_instance, heat_demand_pr
{'recommendation_id': '0_phase=0', 'phase': 0, 'sap_adjustment': np.float64(1.7)},
{'recommendation_id': '4_phase=2', 'phase': 2, 'sap_adjustment': np.float64(4.0)}
]
def test_mechanical_ventilation_sap_floor(property_instance):
rec = {
"type": "mechanical_ventilation",
"recommendation_id": "mv_test",
"phase": 1,
}
previous_phase_values = {"sap": 2.0}
current_phase_values = {"sap": 0.5} # model prediction already below 1
property_phase_impact = {"sap": -1.5, "carbon": 0, "heat_demand": 0}
adjustments = []
updated_impact, updated_current, updated_adjustments = (
Recommendations._apply_measure_specific_rules(
rec=rec,
property_phase_impact=property_phase_impact,
previous_phase_values=previous_phase_values,
current_phase_values=current_phase_values,
adjustments=adjustments,
property_instance=property_instance
)
)
# SAP should be clamped to minimum 1
assert updated_current["sap"] == 1.0
# Original final SAP would have been 0.5 → so adjustment = 1 - 0.5 = 0.5
assert updated_adjustments == [
{
"recommendation_id": "mv_test",
"phase": 1,
"sap_adjustment": 0.5,
}
]
# Impact should now reflect new clamped SAP
assert updated_impact["sap"] == -1.0 # 2.0 → 1.0
def test_mechanical_ventilation_no_floor_adjustment(property_instance):
rec = {
"type": "mechanical_ventilation",
"recommendation_id": "mv_test",
"phase": 1,
}
previous_phase_values = {"sap": 5.0}
current_phase_values = {"sap": 3.0}
property_phase_impact = {"sap": -2.0, "carbon": 0, "heat_demand": 0}
adjustments = []
updated_impact, updated_current, updated_adjustments = (
Recommendations._apply_measure_specific_rules(
rec=rec,
property_phase_impact=property_phase_impact,
previous_phase_values=previous_phase_values,
current_phase_values=current_phase_values,
adjustments=adjustments,
property_instance=property_instance
)
)
# No adjustment expected
assert updated_adjustments == []
# SAP unchanged
assert updated_current["sap"] == 3.0
assert updated_impact["sap"] == -2.0
def test_mechanical_ventilation_exactly_one_no_adjustment(property_instance):
# Test when SAP = 1
rec = {
"type": "mechanical_ventilation",
"recommendation_id": "mv_test",
"phase": 1,
}
previous_phase_values = {"sap": 2.0}
current_phase_values = {"sap": 1.0}
property_phase_impact = {"sap": -1.0, "carbon": 0, "heat_demand": 0}
adjustments = []
updated_impact, updated_current, updated_adjustments = (
Recommendations._apply_measure_specific_rules(
rec=rec,
property_phase_impact=property_phase_impact,
previous_phase_values=previous_phase_values,
current_phase_values=current_phase_values,
adjustments=adjustments,
property_instance=property_instance
)
)
# Exactly 1 → no adjustment
assert updated_adjustments == []
assert updated_current["sap"] == 1.0
assert updated_impact["sap"] == -1.0

37
sfr/principal_pitch/2_export_data.py
View file

@ -28,12 +28,12 @@ from sqlalchemy import func
# PORTFOLIO_ID = 206
# SCENARIOS = [389]
PORTFOLIO_ID = 524
PORTFOLIO_ID = 568
SCENARIOS = [
1009,
1059,
]
scenario_names = {
1009: "EPC C; Most Economic",
1059: "EPC C - 10k budget",
}
@ -230,7 +230,7 @@ for scenario_id in SCENARIOS:
# Get recs for this scenario
recommended_measures_df = recommendations_df[
recommendations_df["scenario_id"] == scenario_id
][["property_id", "measure_type", "estimated_cost", "default"]]
][["property_id", "measure_type", "estimated_cost", "default"]]
recommended_measures_df = recommended_measures_df[
recommended_measures_df["default"]
]
@ -238,7 +238,7 @@ for scenario_id in SCENARIOS:
post_install_sap = recommendations_df[
recommendations_df["scenario_id"] == scenario_id
][["property_id", "default", "sap_points"]]
][["property_id", "default", "sap_points"]]
post_install_sap = post_install_sap[post_install_sap["default"]]
# Sum up the sap points by property id
post_install_sap = (
@ -301,33 +301,6 @@ for scenario_id in SCENARIOS:
)
df["uprn"] = df["uprn"].astype(str)
relevant_plans = plans_df[plans_df["scenario_id"] == scenario_id]
df2 = df.merge(
relevant_plans[["property_id", "post_sap_points", "post_epc_rating"]],
how="left",
on="property_id",
suffixes=("", "_plan"),
)
print(df2["predicted_post_works_epc"].value_counts())
print(df2["post_epc_rating"].value_counts())
z = df2[
(df2["predicted_post_works_epc"] != "D")
& (df2["post_epc_rating"].astype(str) == "Epc.D")
]
df2["predicted_post_works_epc"].value_counts()
df2["post_epc_rating"].astype(str).value_counts()
df2[df2["total_retrofit_cost"] > 0].shape
getting_works = df[df["total_retrofit_cost"] > 0]
getting_works["predicted_post_works_epc"].value_counts()
32565 / getting_works.shape[0]
df[df["predicted_post_works_sap"] == ""]
# Create excel to store to
filename = f"{scenario_names[scenario_id]} - 20250113 final.xlsx"
with pd.ExcelWriter(filename) as writer: