diff --git a/recommendations/HeatingRecommender.py b/recommendations/HeatingRecommender.py index 409f9ec6..b1f6205c 100644 --- a/recommendations/HeatingRecommender.py +++ b/recommendations/HeatingRecommender.py @@ -555,7 +555,9 @@ class HeatingRecommender: for kw in models_kw: if kw >= target: return kw - return None + + # Return the largest + return max(models_kw) def recommend_air_source_heat_pump(self, phase, has_cavity_or_loft_recommendations, _return=False): """ @@ -586,7 +588,15 @@ class HeatingRecommender: ) ashp_size = self.pick_model(estimated_load) - ashp_costs = self.costs.air_source_heat_pump(ashp_size) + number_heated_rooms = self._estimate_n_heated_rooms() + # We now adjust this depending on the floor area to get number of communcal rooms (e.g. hallways) + communal_heated_rooms = self._estimate_n_communal_heated_rooms() + + ashp_costs = self.costs.air_source_heat_pump( + ashp_size, + number_heated_rooms=number_heated_rooms + communal_heated_rooms, + total_floor_area=self.property.floor_area + ) if non_intrusive_recommendation: # Update with non-intrusive recommendation if non_intrusive_recommendation.get("cost"): @@ -907,6 +917,56 @@ class HeatingRecommender: return already_has_hhr and already_has_hhr_contols + def _estimate_n_heated_rooms(self): + # If the property is off-gas and has no heating system in place, the number of heated rooms will actually + # be 0, so we use the number of rooms as the figure + number_heated_rooms = ( + self.property.data["number-heated-rooms"] if self.property.data["number-heated-rooms"] > 0 + else ( + self.property.number_of_rooms - 1 if self.property.number_of_rooms > 1 else + self.property.number_of_rooms + ) + ) + # To be conservative, we adjust if we still have 1 room + if (number_heated_rooms == 1) and (self.property.number_of_rooms > 2): + number_heated_rooms = self.property.number_of_rooms - 1 + + return number_heated_rooms + + def _estimate_n_communal_heated_rooms(self) -> int: + """ + Estimate number of communal circulation rooms (hallways / landings) that may reasonably contain a heater + """ + + # Base assumptions + base_by_type = { + "Flat": 1, + "Maisonette": 1, + "Bungalow": 1, + "House": 2, + } + + # Fallback if property type unknown + base = base_by_type.get(self.property.data["property-type"], 1) + + # Area-based adjustments + if self.property.data["property-type"] in ("Flat", "Maisonette"): + if self.property.floor_area > 90: + return base + 1 # duplex or very large flat + return base + + if self.property.data["property-type"] == "Bungalow": + if self.property.floor_area > 100: + return base + 1 # secondary corridor + return base + + if self.property.data["property-type"] == "House": + if self.property.floor_area > 140: + return base + 1 # extra landing / circulation + return base + + return base + def recommend_hhr_storage_heaters(self, phase, system_change, heating_controls_only, _return=False): """ We will recommend upgrading to a high heat retention storage system, if the current system is not already @@ -1010,18 +1070,7 @@ class HeatingRecommender: else: heating_simulation_config["hot_water_energy_eff_ending"] = self.property.data["hot-water-energy-eff"] - # If the property is off-gas and has no heating system in place, the number of heated rooms will actually - # be 0, so we use the number of rooms as the figure - number_heated_rooms = ( - self.property.data["number-heated-rooms"] if self.property.data["number-heated-rooms"] > 0 - else ( - self.property.number_of_rooms - 1 if self.property.number_of_rooms > 1 else - self.property.number_of_rooms - ) - ) - # To be conservative, we adjust if we still have 1 room - if (number_heated_rooms == 1) and (self.property.number_of_rooms > 2): - number_heated_rooms = self.property.number_of_rooms - 1 + number_heated_rooms = self._estimate_n_heated_rooms() # We focus on the 700 watt product hhrsh_product = next((x for x in self.hhrsh_products if x["size"] == 700), {}) diff --git a/recommendations/optimiser/funding_optimiser.py b/recommendations/optimiser/funding_optimiser.py index a8b998ae..dd7184d0 100644 --- a/recommendations/optimiser/funding_optimiser.py +++ b/recommendations/optimiser/funding_optimiser.py @@ -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 itertools import product from backend.app.plan.schemas import ( WALL_INSULATION_MEASURES, ROOF_INSULATION_MEASURES, ECO4_ELIGIBILE_FABRIC_MEASURES @@ -587,6 +588,218 @@ def optimise_with_funding_paths( return solutions +def build_heat_pump_paths( + remaining_wall_measures, + remaining_roof_measures, +): + """ + Build AND-paths using cartesian products. + + Rules: + - Always include air_source_heat_pump + - Choose 1 wall measure if any exist + - Choose 1 roof measure if any exist + """ + + # If a category is empty, use [None] so product still works + wall_choices = remaining_wall_measures or [None] + roof_choices = remaining_roof_measures or [None] + + paths = [] + + for wall, roof in product(wall_choices, roof_choices): + parts = [] + + if wall is not None: + parts.append(wall) + if roof is not None: + parts.append(roof) + + parts.append("air_source_heat_pump") + + paths.append({"AND": parts}) + + return paths + + +def exclude_measure_types(input_measures, excluded_types): + excluded = set(excluded_types) + filtered = [] + + for group in input_measures: + kept = [ + opt for opt in group + if opt["type"] not in excluded + ] + if kept: + filtered.append(kept) + + return filtered + + +def optimise_with_scenarios( + input_measures, + budget=None, + target_gain=None, + enforce_heat_pump_insulation=True, + enforce_fabric_first=False +): + """ + Scenario-based optimiser (funding-agnostic). + + Currently implemented scenarios: + 1) With air source heat pump AND required insulation + """ + + solutions = [] + paths = [] + # Produce the unique list of measure types + all_measure_types = [] + for inputs in input_measures: + all_measure_types.extend([x["type"] for x in inputs]) + all_measure_types = list(set(all_measure_types)) + + if enforce_fabric_first: + # If this is true, it means we only want to consider a fabric first approach. This means that + # - We treat the fabric of the house first + # - Only once the fabric has been upgraded, do we consider heating upgrades + + # This should be wall insulation, roof insulation, floor insulation and windows + fabric_measures = WALL_INSULATION_MEASURES + ROOF_INSULATION_MEASURES + ECO4_ELIGIBILE_FABRIC_MEASURES + + fabric_only_measures = [[opt for opt in group if opt["type"] in fabric_measures] for group in input_measures] + fabric_only_measures = [g for g in fabric_only_measures if g] + + if not fabric_only_measures: + # If we have no fabric measures, it means the work has already been done and we can proceed + # straight to heating optimisation + picked_fabric, fabric_cost, fabric_gain = [], 0, 0 + else: + picked_fabric, fabric_cost, fabric_gain = run_optimizer( + input_measures=fabric_only_measures, + budget=budget, + sub_target_gain=target_gain, + # If we can achieve the target gain with just insulation measures, we're done + ) + + picked_fabric_types = {m["type"] for m in picked_fabric} + + remaining_measures = [] + for group in input_measures: + kept = [m for m in group if m["type"] not in picked_fabric_types] + if kept: + remaining_measures.append(kept) + + picked_extra, extra_cost, extra_gain = run_optimizer( + remaining_measures, + budget=budget - fabric_cost if budget is not None else None, + sub_target_gain=( + target_gain - fabric_gain + if target_gain is not None + else None + ) + ) + + if picked_extra is None: + picked_extra, extra_cost, extra_gain = [], 0, 0 + + solutions.append({ + "scenario": "fabric_first", + "items": picked_fabric + picked_extra, + "fixed_items": picked_fabric, + "total_cost": fabric_cost + extra_cost, + "total_gain": fabric_gain + extra_gain, + }) + return solutions + + # ------------------------------------------------------------------ + # Scenario 1: Air source heat pump with required insulation + # ------------------------------------------------------------------ + if enforce_heat_pump_insulation: + # Wall measures could be IWI or EWI + remaining_wall_measures = [x for x in all_measure_types if x in WALL_INSULATION_MEASURES] + remaining_roof_measures = [x for x in all_measure_types if x in ROOF_INSULATION_MEASURES] + + # Mandatory structure: + # - must include ASHP + # - must include >=1 wall insulation (if still needed) + # - must include >=1 roof insulation (if still needed) + # We need all of the combinations of remaining wall and remaining roof measures + heat_pump_paths = build_heat_pump_paths(remaining_wall_measures, remaining_roof_measures) + paths.extend(heat_pump_paths) + + # ------------------------------------------------------------------ + # Scenario 2: Optimise without air source heat pump + # ------------------------------------------------------------------ + # No special path; just exclude ASHP from options and allow us to optimise. + measures_no_heat_pump = exclude_measure_types(input_measures, ["air_source_heat_pump"]) + + 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, + }) + + fixed_selections = expand_funding_path(input_measures, paths) + + for fixed in fixed_selections: + + # fixed = [(gi, oi, opt), ...] + fixed_items = [opt for (_, _, opt) in fixed] + fixed_groups = {gi for (gi, _, _) in fixed} + + fixed_cost, fixed_gain = sum_cost_gain(fixed_items) + + # Remaining measures (all other groups) + remaining_measures = [ + grp for gi, grp in enumerate(input_measures) + if gi not in fixed_groups + ] + + # Optimise remaining measures + if ( + target_gain is not None + and fixed_gain >= target_gain + ): + picked, sub_cost, sub_gain = [], 0, 0 + else: + picked, sub_cost, sub_gain = run_optimizer( + remaining_measures, + budget=budget - fixed_cost if budget is not None else None, + sub_target_gain=( + target_gain - fixed_gain + if target_gain is not None + else None + ) + ) + + if picked is None: + continue + + total_items = fixed_items + picked + total_cost = fixed_cost + sub_cost + total_gain = fixed_gain + sub_gain + + solutions.append({ + "scenario": "heat_pump_with_insulation", + "items": total_items, + "fixed_items": fixed_items, + "total_cost": total_cost, + "total_gain": total_gain, + }) + + return solutions + + # ---- helpers ------------------------------------------------------------- diff --git a/recommendations/tests/test_optimiser_functions.py b/recommendations/tests/test_optimiser_functions.py index 031bb9ac..865e3398 100644 --- a/recommendations/tests/test_optimiser_functions.py +++ b/recommendations/tests/test_optimiser_functions.py @@ -8,6 +8,7 @@ from recommendations.optimiser.CostOptimiser import CostOptimiser class TestPrepareInputMeasures: + def test_returns_expected_structure_without_ventilation(self): recs = [ [ # loft insulation measure diff --git a/recommendations/tests/test_optimisers.py b/recommendations/tests/test_optimisers.py index e81aac69..ecc6ea56 100644 --- a/recommendations/tests/test_optimisers.py +++ b/recommendations/tests/test_optimisers.py @@ -1,97 +1,14 @@ -import numpy as np -# import pandas as pd from pandas import Timestamp from numpy import nan import datetime -# import backend.app.assumptions as assumptions -# import recommendations.optimiser.optimiser_functions as optimiser_functions -# -# from backend.Funding import Funding -# -# project_scores_matrix = pd.read_csv("/Users/khalimconn-kowlessar/Downloads/ECO4 Full Project Scores Matrix.csv") -# partial_project_scores_matrix = pd.read_csv("backend/tests/test_data/ECO4_Partial_Project_Scores_Matrix_v6.csv") -# partial_project_scores_matrix.columns = ['Measure category', 'Measure_Type', 'Pre_Main_Heating_Source', -# 'Post_Main_Heating_Source', 'Total Floor Area Band', 'Starting Band', -# 'Average Treatable Factor', 'Cost Savings', 'SAP Savings'] -# whlg_eligible_postcodes = pd.DataFrame([{"Postcode": "ab12cd"}]) -# -# funding = Funding( -# 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.5, -# eco4_social_solid_abs_rate=17, -# eco4_private_cavity_abs_rate=13.5, -# eco4_private_solid_abs_rate=17, -# gbis_social_cavity_abs_rate=21, -# gbis_social_solid_abs_rate=25, -# gbis_private_cavity_abs_rate=22, -# gbis_private_solid_abs_rate=28, -# tenure="Social" -# ) -# -# # Assume these costs have been adjusted - - -# -# # Insert the funding uplifts -# for recs in property_recommendations: -# for r in recs: -# # Insert randomly -# # Select one of 0, 0.25 or 0.45 -# r["uplift"] = np.random.choice([0, 0.25, 0.45]) -# -# # We calculate the innovation uplift against each measure -# for recs in property_recommendations: -# for r in recs: -# if r["type"] in ["mechanical_ventilation", "low_energy_lighting", "secondary_heating"]: -# r["innovation_uplift"] = 0 -# continue -# r["innovation_uplift"] = funding.get_innovation_uplift( -# measure=r, -# starting_sap=p.data["current-energy-efficiency"], -# floor_area=p.floor_area, -# is_cavity=False, -# current_wall_uvalue=1.7, -# is_partial=False, -# existing_li_thickness=150, -# mainheating=p.main_heating, -# main_fuel=p.main_fuel, -# mainheat_energy_eff=p.data["mainheat-energy-eff"], -# ) -# print(r["innovation_uplift"]) -# -# property_measure_types = {rec["type"] for recs in property_recommendations for rec in recs} -# property_required_measures = [m for m in property_recommendations if m[0]["type"] in []] -# measures_to_optimise = [m for m in property_recommendations if m[0]["type"] not in []] -# -# # If a measure requiring ventilation is selected, and the property does not have ventilation, we enfore -# # its inclusion -# needs_ventilation = any( -# x in property_measure_types for x in assumptions.measures_needing_ventilation -# ) and not p.has_ventilation -# -# input_measures = optimiser_functions.prepare_input_measures( -# measures_to_optimise, "Increasing EPC", needs_ventilation, True -# ) -# -# # ---- main wrapper around your optimiser ---------------------------------- -# -# # Run inputs: -# target_gain = 18.5 -# -# # Run the optimiser with these inouts - - -# tests/test_social_fabric_only.py import numpy as np import pandas as pd import pytest from copy import deepcopy from recommendations.optimiser import optimiser_functions -from recommendations.optimiser.funding_optimiser import optimise_with_funding_paths # wherever you defined it +from recommendations.optimiser.funding_optimiser import optimise_with_funding_paths, build_heat_pump_paths from backend.Funding import Funding from backend.app.plan.schemas import WALL_INSULATION_MEASURES, ROOF_INSULATION_MEASURES, ECO4_ELIGIBILE_FABRIC_MEASURES @@ -799,3 +716,14 @@ def test_private_solid_wall_no_innovation_epc_d(p, funding, mock_project_scores_ 'partial_project_funding': 2300.1000000000004, 'partial_project_score': 135.3, 'total_uplift': 0.0, 'total_uplift_score': 0.0 } + + +def test_build_heat_pump_paths(): + eg1 = build_heat_pump_paths([], ["loft_insulation"]) + + assert eg1 == [{'AND': ['loft_insulation', 'air_source_heat_pump']}] + + eg2 = build_heat_pump_paths(["internal_wall_insulation", "external_wall_insulation"], ["loft_insulation"]) + + assert eg2 == [{'AND': ['internal_wall_insulation', 'loft_insulation', 'air_source_heat_pump']}, + {'AND': ['external_wall_insulation', 'loft_insulation', 'air_source_heat_pump']}]