From 00f4b4190766cfc228ab609f64b90945c7895c91 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Tue, 12 Aug 2025 21:00:19 +0100 Subject: [PATCH] completed mvp for make funding paths --- .../optimiser/optimiser_functions.py | 6 +- recommendations/tests/test_optimisers.py | 205 ++++++++++++------ 2 files changed, 146 insertions(+), 65 deletions(-) diff --git a/recommendations/optimiser/optimiser_functions.py b/recommendations/optimiser/optimiser_functions.py index a4f32c41..95309a19 100644 --- a/recommendations/optimiser/optimiser_functions.py +++ b/recommendations/optimiser/optimiser_functions.py @@ -102,8 +102,12 @@ def prepare_input_measures(property_recommendations, goal, needs_ventilation, fu else rec["measure_type"] ) + # We also include the innovation uplift to_append.append( - {"id": rec["recommendation_id"], "cost": total, "gain": gain, "type": rec_type} + { + "id": rec["recommendation_id"], "cost": total, "gain": gain, "type": rec_type, + "innovation_uplift": rec["innovation_uplift"] if funding else 0, + } ) input_measures.append(to_append) diff --git a/recommendations/tests/test_optimisers.py b/recommendations/tests/test_optimisers.py index c265bb99..4256bb31 100644 --- a/recommendations/tests/test_optimisers.py +++ b/recommendations/tests/test_optimisers.py @@ -457,7 +457,7 @@ needs_ventilation = any( ) and not p.has_ventilation input_measures = optimiser_functions.prepare_input_measures( - measures_to_optimise, "Increasing EPC", needs_ventilation + measures_to_optimise, "Increasing EPC", needs_ventilation, True ) # ---- rule definitions you can tweak ------------------------------------- @@ -493,19 +493,118 @@ def _find_measure(input_measures, measure_type): return False -def make_funding_paths(input_measures, tenure): +def _make_generic_eco4_funding_paths(p, input_measures, funding_paths, remaining_insulation_type): + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + # Solar PV with existing eligible heating system + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + has_eligible_heating_system = funding.check_solar_eligible_heating_system( + mainheat_description=p.main_heating["clean_description"], + heating_control_description=p.main_heating_controls["clean_description"] + ) + + if has_eligible_heating_system: + single_solar_template = [{"AND": ["solar_pv"], "reference": "solar_pv"}] + # We now look to pair this with any lingering insulation measures + solar_paths = [] + for insulation_measure in remaining_insulation_type: + new_solar_path = deepcopy(single_solar_template) + new_solar_path[0]["AND"].append(insulation_measure) + solar_paths.append(new_solar_path) + + if solar_paths: + funding_paths.extend(solar_paths) + else: + # If we have no insulation measures, we just add the solar PV path + funding_paths.append(single_solar_template) + + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + # Solar PV + HHRSH + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + if (_find_measure(input_measures, "solar_pv") and + _find_measure(input_measures, "high_heat_retention_storage_heater")): + funding_paths.append( + [{"AND": ["solar_pv", "high_heat_retention_storage_heater"], "reference": "solar_pv+hhrsh"}] + ) + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + # Solar PV + ASHP + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + if (_find_measure(input_measures, "solar_pv") and + _find_measure(input_measures, "air_source_heat_pump")): + funding_paths.append([{"AND": ["solar_pv", "air_source_heat_pump"]}]) + + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + # Solar PV + Electric Boiler + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + if (_find_measure(input_measures, "solar_pv") and + _find_measure(input_measures, "electric_boiler")): + funding_paths.append([{"AND": ["solar_pv", "electric_boiler"]}]) + + # We've actually covered all possible options where solar PV can be included in a funded package, so where + # solar PV is not in a reference, we can exclude it + + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + # Heating Upgrades + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + # Must have an existing eligible heating system + + measure_references = { + "boiler_upgrade": "boiler_upgrade", + "high_heat_retention_storage_heater": "hhrsh", + "air_source_heat_pump": "ashp" + } + for heating_upgrade in ["boiler_upgrade", "high_heat_retention_storage_heater", "air_source_heat_pump"]: + if _find_measure(input_measures, ""): + # We check if we have any remaining insulation measures to be applied to the property + if remaining_insulation_type: + hhrsh_template = [ + {"AND": [heating_upgrade], "reference": measure_references[heating_upgrade]} + ] + hhrsh_paths = [] + for insulation_measure in remaining_insulation_type: + new_hhrsh_path = deepcopy(hhrsh_template) + new_hhrsh_path[0]["AND"].append(insulation_measure) + hhrsh_paths.append(new_hhrsh_path) + + funding_paths.extend(hhrsh_paths) + else: + # If we have no insulation measures, we just add the HHRSH path + funding_paths.append([{"AND": [measure_references[heating_upgrade]]}]) + + return funding_paths + + +def _make_generic_gbis_funding_paths(input_gbis_measures, funding_paths): + """ + For GBIS, the packages are single insulation measure. + + We also have potential GBIS packages that allow heating controls as a secondary measure, however this + is not currently implemented in the optimiser due to not being certain about the heating controls pre conditions + :param input_gbis_measures: + :param funding_paths: + :return: + """ + + gbis_funding_paths = [] + for input_measure in input_gbis_measures: + for measure in input_measure: + # We create a path for each measure + gbis_funding_paths.append([{"AND": [measure["type"]], "reference": measure["type"] + ":gbis"}]) + + return funding_paths + gbis_funding_paths + + +def make_funding_paths(p, input_measures, tenure): """ This function generates funding paths based on the input measures and the tenure of the property. It checks for the presence of specific measures and creates paths that include necessary insulation measures to meet minimum insulation requirements, particularly when a heating system is recommended. Remaining measures that are not fixed as part of the package are then optimised + :param p: The property object containing details about the property, including main heating and controls. :param input_measures: :param tenure: :return: """ - funding_paths = [] - # We handle the case of minimum insulation requirements. Whenever we have a heating system recommendation, # we *must* include an additional insulation measure, unless the property already has sufficient insulation. @@ -517,6 +616,9 @@ def make_funding_paths(input_measures, tenure): roof_insulation_measures = [ "loft_insulation", "flat_roof_insulation", "room_roof_insulation" ] + other_gbis_insulation_measures = [ + "suspended_floor_insulation", "solid_floor_insulation", + ] # These are the insulation measures that the property still needs and so will be considered for # filling the minimum insulation requirements remaining_insulation_type = [] @@ -524,37 +626,36 @@ def make_funding_paths(input_measures, tenure): if _find_measure(input_measures, insulation_measure): remaining_insulation_type.append(insulation_measure) - # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - # Air source heat pump - # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + funding_paths = [] - # For all tenures, if we have an air source heat pump it's a funded measure and so we must consider the minimum - # insulation requirements - if _find_measure(input_measures, "air_source_heat_pump"): - ashp_template = [{"AND": ["air_source_heat_pump"]}] - ashp_paths = [] - for insulation_measure in remaining_insulation_type: - new_ashp_path = deepcopy(ashp_template) - new_ashp_path[0]["AND"].append(insulation_measure) - ashp_paths.append(new_ashp_path) + if tenure == "Social" and p.data["current-energy-rating"] == "D": + # If the property is currently EPC D, we can only include innovation measures or measures to meet the + # minimum insulation requirements + input_measures_innovation = [] + input_gbis_measures_innovation = [] + for measures in input_measures: + for measure in measures: + if measure["innovation_uplift"] or measure["type"] in remaining_insulation_type: + input_measures_innovation.append([measure]) - if ashp_paths: - # If we have any insulation measures, we add them to the funding paths - funding_paths.extend(ashp_paths) - else: - # If we have no insulation measures, we just add the ASHP path - funding_paths.append(ashp_template) + if measure["innovation_uplift"] and measure["type"] in ( + remaining_insulation_type + other_gbis_insulation_measures + ): + input_gbis_measures_innovation.append([measure]) - # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - # Solar PV with existing eligible heating system - # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + funding_paths = _make_generic_eco4_funding_paths( + p, input_measures_innovation, funding_paths, remaining_insulation_type + ) - if tenure == "Social": - raise NotImplementedError("Implement me!") + # Can only be innovation GBIS measures + funding_paths = _make_generic_gbis_funding_paths(input_gbis_measures_innovation, funding_paths) + return funding_paths if tenure == "Private": - # We cover off the main funding paths - # 1) The package must include EWI or IWI + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + # EWI or IWI + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + # 1) The package must include EWI or IWI if the property is privately owned # We check if we have any EWI or IWI measures available ewi_or_iwi = [{"OR": []}] # If we have EWI we add it in @@ -567,44 +668,20 @@ def make_funding_paths(input_measures, tenure): if ewi_or_iwi[0]["OR"]: funding_paths.append(ewi_or_iwi) - # 3) The package must have an existing eligible heating system. We test this with the funding checker - # If we have any remaining insulation measure to be applied to the property, we also need to include that in - # the package - single_solar_template = [{"AND": []}] - has_eligible_heating_system = funding.check_solar_eligible_heating_system( - mainheat_description=p.main_heating["clean_description"], - heating_control_description=p.main_heating_controls["clean_description"] - ) + funding_paths = _make_generic_eco4_funding_paths( + p, input_measures, funding_paths, remaining_insulation_type + ) - if has_eligible_heating_system: - single_solar_template[0]["AND"].append("solar_pv") - # We now look to pair this with any lingering insulation measures - solar_paths_with_insulation = [] - for insulation_measure in wall_insulation_measures + roof_insulation_measures: - if _find_measure(input_measures, insulation_measure): - new_solar_path = deepcopy(single_solar_template) - new_solar_path[0]["OR"].append(insulation_measure) - solar_paths_with_insulation.append(new_solar_path) + # If we have any remaining insulation measures, we add them to the funding paths + input_gbis_measures = [] + for measures in input_measures: + for measure in measures: + if measure["type"] in remaining_insulation_type + other_gbis_insulation_measures: + input_gbis_measures.append([measure]) - if not solar_paths_with_insulation: - # If we have no insulation measures, we're good with just single solar - solar_paths_with_insulation.append(single_solar_template) + funding_paths = _make_generic_gbis_funding_paths(input_gbis_measures, funding_paths) - funding_paths.extend(solar_paths_with_insulation) - else: - # If we don't have an eligible heating system, we check if we have an eligible heating system - # (HHRSH/ASHP/Electric boiler) + solar PV - if (_find_measure(input_measures, "solar_pv") and - _find_measure(input_measures, "high_heat_retention_storage_heater")): - funding_paths.append([{"AND": ["solar_pv", "high_heat_retention_storage_heater"]}]) - - if (_find_measure(input_measures, "solar_pv") and - _find_measure(input_measures, "air_source_heat_pump")): - funding_paths.append([{"AND": ["solar_pv", "air_source_heat_pump"]}]) - - if (_find_measure(input_measures, "solar_pv") and - _find_measure(input_measures, "electric_boiler")): - funding_paths.append([{"AND": ["solar_pv", "electric_boiler"]}]) + return funding_paths # ---- main wrapper around your optimiser ----------------------------------