diff --git a/recommendations/tests/test_optimisers.py b/recommendations/tests/test_optimisers.py index 9329b383..2e8186dd 100644 --- a/recommendations/tests/test_optimisers.py +++ b/recommendations/tests/test_optimisers.py @@ -475,7 +475,7 @@ def _find_measure(input_measures, measure_type): return False -def _make_solar_heating_funding_paths(p, input_measures, funding_paths, remaining_insulation_type): +def _make_solar_heating_funding_paths(p, input_measures, funding_paths, remaining_insulation_type, housing_type): # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Solar PV with existing eligible heating system # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ @@ -531,12 +531,20 @@ def _make_solar_heating_funding_paths(p, input_measures, funding_paths, remainin # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Must have an existing eligible heating system + # For private, HHRSH alone, or a boiler upgrade is NOT eligible for ECO4 funding. Boiler upgrade also doesn't + # count as an eligible heating system + if housing_type == "Private": + single_heating_measures = ["air_source_heat_pump"] + else: + single_heating_measures = [ + "boiler_upgrade", "high_heat_retention_storage_heater", "air_source_heat_pump" + ] 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"]: + for heating_upgrade in single_heating_measures: if _find_measure(input_measures, heating_upgrade): if remaining_insulation_type: for insulation_measure in remaining_insulation_type: @@ -575,7 +583,7 @@ def _make_generic_gbis_funding_paths(input_gbis_measures, funding_paths): return funding_paths + gbis_funding_paths -def make_funding_paths(p, input_measures, tenure): +def make_funding_paths(p, input_measures, housing_type): """ 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 @@ -584,7 +592,7 @@ def make_funding_paths(p, input_measures, tenure): 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: + :param housing_type: :return: """ # We handle the case of minimum insulation requirements. Whenever we have a heating system recommendation, @@ -612,7 +620,7 @@ def make_funding_paths(p, input_measures, tenure): funding_paths = [] - if tenure == "Social" and p.data["current-energy-rating"] == "D": + if housing_type == "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 = [] @@ -628,14 +636,14 @@ def make_funding_paths(p, input_measures, tenure): input_gbis_measures_innovation.append([measure]) funding_paths = _make_solar_heating_funding_paths( - p, input_measures_innovation, funding_paths, remaining_insulation_type + p, input_measures_innovation, funding_paths, remaining_insulation_type, housing_type ) # Can only be innovation GBIS measures funding_paths = _make_generic_gbis_funding_paths(input_gbis_measures_innovation, funding_paths) - return funding_paths + return funding_paths, input_measures_innovation - if tenure == "Private": + if housing_type == "Private": # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # EWI or IWI # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ @@ -657,7 +665,7 @@ def make_funding_paths(p, input_measures, tenure): funding_paths.append(ewi_or_iwi) funding_paths = _make_solar_heating_funding_paths( - p, input_measures, funding_paths, remaining_insulation_type + p, input_measures, funding_paths, remaining_insulation_type, housing_type ) # If we have any remaining insulation measures, we add them to the funding paths @@ -669,7 +677,7 @@ def make_funding_paths(p, input_measures, tenure): funding_paths = _make_generic_gbis_funding_paths(input_gbis_measures, funding_paths) - return funding_paths + return funding_paths, input_measures # ---- main wrapper around your optimiser ---------------------------------- @@ -791,26 +799,97 @@ def _is_eligible_funding_package(scheme, starting_sap, total_gain): return True +def _prs_solution_ok(items, p): + # items: list of picked option dicts (after optimisation) + # treat "type" possibly like "x+y" -> split and look at base tokens + types = set() + for opt in items: + for t in opt["type"].split("+"): + types.add(t) + + has_solid_wall = ("external_wall_insulation" in types) or ("internal_wall_insulation" in types) + + # renewable set: + has_ashp = ("air_source_heat_pump" in types) # ASHP alone is renewable + has_solar = ("solar_pv" in types) + has_hhrsh = ("high_heat_retention_storage_heater" in types) # only counts *with* solar + + # solar PV qualifies if paired with eligible existing heating + solar_ok_existing = has_solar and funding.check_solar_eligible_heating_system( + p.main_heating["clean_description"], p.main_heating_controls["clean_description"] + ) + + # or paired with ASHP/HHRSH in the same package + solar_ok_with_installed = has_solar and (has_ashp or has_hhrsh) + + renewable_ok = has_ashp or solar_ok_existing or solar_ok_with_installed + + return has_solid_wall or renewable_ok + + +def _ensure_unfunded_costs(groups): + """Make sure each option’s cost is base+uplift (i.e., no funding). + Safe if fields already match; works on a deepcopy. + """ + for grp in groups: + for opt in grp: + base = opt.get("cost_minus_uplift") + upl = opt.get("innovation_uplift", 0.0) + if base is not None: + opt["cost"] = float(base) + float(upl) + # else: assume opt["cost"] already includes uplift + return groups + + def optimise_with_funding_paths(p, input_measures, housing_type, budget=None, target_gain=None): """ run_optimizer(sub_measures, budget, target_gain) -> (picked_options, sub_cost, sub_gain) """ - funding_paths = make_funding_paths(p, input_measures, housing_type) + solutions = [] + + # unfunded - we utilise all measures + unfunded_measures = input_measures.copy() + unfunded_measures = _ensure_unfunded_costs(unfunded_measures) + picked, total_cost, total_gain = run_optimizer( + unfunded_measures, + budget=budget, + sub_target_gain=target_gain + ) + if picked is not None: + solutions.append({ + "fixed_ids": [], + "items": picked, + "total_cost": total_cost, + "total_gain": total_gain, + "path": {"reference": "unfunded:all"}, + "scheme": "none", + "is_eligible": False, # no funding scheme applied + }) + + # This function will filter down on innovation measures if we are social EPC D + funding_paths, optimisation_input_measures = make_funding_paths(p, input_measures, housing_type) # We now produce a fabric only path for ECO4 # We add in generic insulation funding paths (where there is no fixed measure) # Heating controls are only eligible if installed as part of a heating upgrade and so we do not include them # here - allowed_types = WALL_INSULATION_MEASURES + ROOF_INSULATION_MEASURES + ECO4_ELIGIBILE_FABRIC_MEASURES - funding_paths = [{'reference': 'fabric-only:eco4'}] + funding_paths + if housing_type == "Social": + funding_paths = ( + [ + { + 'reference': 'fabric-only:eco4', + "allowed_types": WALL_INSULATION_MEASURES + ROOF_INSULATION_MEASURES + + ECO4_ELIGIBILE_FABRIC_MEASURES + } + ] + funding_paths + ) - solutions = [] for path_spec in funding_paths: # 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(input_measures, allowed_types) + sub_measures = _filter_measures_by_types(optimisation_input_measures, path_spec["allowed_types"]) if not sub_measures: continue @@ -840,7 +919,7 @@ def optimise_with_funding_paths(p, input_measures, housing_type, budget=None, ta continue # 1) expand fixed selections for this path - fixed_selections = expand_funding_path(input_measures, path_spec) if path_spec else [[]] + fixed_selections = expand_funding_path(optimisation_input_measures, path_spec) if path_spec else [[]] if not fixed_selections: continue @@ -860,7 +939,9 @@ def optimise_with_funding_paths(p, input_measures, housing_type, budget=None, ta fixed_cost, fixed_gain = sum_cost_gain(fixed_items) fixed_groups = {gi for (gi, _, _) in fixed} - sub_measures = deepcopy([grp for gi, grp in enumerate(input_measures) if gi not in fixed_groups]) + sub_measures = deepcopy( + [grp for gi, grp in enumerate(optimisation_input_measures) if gi not in fixed_groups] + ) if scheme == "gbis": # Then for the sub-measures, we need to strip the innovation uplift from the GBIS fixed measures. We @@ -889,6 +970,14 @@ def optimise_with_funding_paths(p, input_measures, housing_type, budget=None, ta total_gain = fixed_gain + sub_gain total_picks = fixed_items + picked + if housing_type == "Private": + if not _prs_solution_ok(total_picks, p): + logger.error( + "Found a solution that does not meet the PRS requirements: %s - this shouldn't be happening", + total_picks + ) + continue + scheme = _path_scheme(path_spec) solutions.append({ @@ -898,7 +987,7 @@ def optimise_with_funding_paths(p, input_measures, housing_type, budget=None, ta "total_gain": total_gain, "path": path_spec, "scheme": scheme, - "is_eligible": _is_eligible_funding_package(scheme, p.data["current-energy-efficiency"], sub_gain) + "is_eligible": _is_eligible_funding_package(scheme, p.data["current-energy-efficiency"], total_gain) }) solutions = pd.DataFrame(solutions)