diff --git a/recommendations/optimiser/optimiser_functions.py b/recommendations/optimiser/optimiser_functions.py index 95309a19..5175c6b6 100644 --- a/recommendations/optimiser/optimiser_functions.py +++ b/recommendations/optimiser/optimiser_functions.py @@ -90,7 +90,10 @@ def prepare_input_measures(property_recommendations, goal, needs_ventilation, fu if rec["measure_type"] in assumptions.measures_needing_ventilation and needs_ventilation else rec["total"] ) - total = 0 if total < 0 else total + + # If the innovation uplift being removed make this negative, we keep the total so we can re-engineer + # the original cost + non_negative_total = 0 if total < 0 else total gain = ( rec[goal_key] + ventilation_recommendation[goal_key] if rec["measure_type"] in assumptions.measures_needing_ventilation and needs_ventilation @@ -105,8 +108,9 @@ def prepare_input_measures(property_recommendations, goal, needs_ventilation, fu # 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": non_negative_total, "gain": gain, "type": rec_type, "innovation_uplift": rec["innovation_uplift"] if funding else 0, + "cost_minus_uplift": total } ) diff --git a/recommendations/tests/test_optimisers.py b/recommendations/tests/test_optimisers.py index 07b42882..252ad675 100644 --- a/recommendations/tests/test_optimisers.py +++ b/recommendations/tests/test_optimisers.py @@ -9,6 +9,9 @@ from recommendations.optimiser.CostOptimiser import CostOptimiser from recommendations.optimiser.GainOptimiser import GainOptimiser from backend.Funding import Funding +# measures we DO NOT treat as fundable in the ECO4 'funded' pass +_ECO4_EXCLUDE_TYPES = {"secondary_heating"} + 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', @@ -460,30 +463,6 @@ input_measures = optimiser_functions.prepare_input_measures( measures_to_optimise, "Increasing EPC", needs_ventilation, True ) -# ---- rule definitions you can tweak ------------------------------------- - -HEATING_TYPES = {"air_source_heat_pump", "high_heat_retention_storage_heater", "solar_pv"} -MIN_INSULATION_OR = [{"loft_insulation"}, {"cavity_wall_insulation"}] # extend if needed - -# “Funding paths”: each is a list of elements; each element is: -# - {"OR": {"types": {..}}} means choose one option from any group whose type is in that set -# - {"AND": [{"types": {..}}, {"types": {..}}]} means choose one from each of those -FUNDING_PATHS = [ - # Path A: IWI OR EWI - [ - { - "OR": { - "types": {"internal_wall_insulation", "external_wall_insulation"} - } - } - ], - # Path B: Solar PV AND HHRSH - [{"AND": [{"types": {"solar_pv"}}, {"types": {"high_heat_retention_storage_heater"}}]}], - # Path C: ASHP alone (may still trigger min insulation rule below) - [{"OR": {"types": {"air_source_heat_pump"}}}], - # -] - def _find_measure(input_measures, measure_type): for measures in input_measures: @@ -695,8 +674,56 @@ def make_funding_paths(p, input_measures, tenure): # Run inputs: target_gain = 18.5 -from itertools import product -import math + +def _path_scheme(path_spec): + """ + Infer scheme from any 'reference' tag in the path. + Defaults to 'eco4' if not specified. + """ + for elem in path_spec or []: + ref = elem.get("reference") + if isinstance(ref, str): + if ref.endswith(":gbis"): + return "gbis" + if ref.endswith(":eco4"): + return "eco4" + return "eco4" + + +def _filter_fundable_subgroups(groups, scheme): + """ + Keep only options eligible for the funded pass of the given scheme. + - ECO4: drop excluded types (e.g., secondary_heating) + - GBIS: funded pass is the GBIS fixed measure only, so return empty sub-groups + """ + if scheme == "gbis": + return [] # we won't optimise 'the rest' under GBIS here + + # ECO4 case + filtered = [] + for grp in groups: + kept = [opt for opt in grp + if not any(ex in opt["type"] for ex in _ECO4_EXCLUDE_TYPES)] + if kept: + filtered.append(kept) + return filtered + + +def _sum_cost_gain_with_scheme(items, scheme): + """ + Sum cost/gain of fixed items, adjusting for scheme rules. + - GBIS: strip innovation uplift from GBIS-funded fixed measures only. + """ + total_cost = 0.0 + total_gain = 0.0 + for it in items: + cost = float(it["cost"]) + if scheme == "gbis": + # innovation uplifts are not paid under GBIS + cost -= float(it.get("innovation_uplift", 0.0)) + total_cost += cost + total_gain += float(it["gain"]) + return total_cost, total_gain def violates_min_insulation(fixed): @@ -740,11 +767,6 @@ def optimise_with_funding_paths(input_measures, budget=None, target_gain=None, s solutions = [] for path_spec in funding_paths: - # TODO: If the path spec is GBIS, need to handle this differently. There is no funding associated - # with the other measures we're optimising. Instead, we fix the GBIS measure (which is funded) - # and then run the optimiser on the remaining measures which are NOT funded. The key change is all - # measures in input_measures right now have costs adjusted with innovation uplift, which we don't want - # to apply to the GBIS measures. So we need to strip the innovation uplift from the GBIS measures # 1) expand fixed selections for this path fixed_selections = expand_funding_path(input_measures, path_spec) if path_spec else [[]] if not fixed_selections: @@ -758,13 +780,26 @@ def optimise_with_funding_paths(input_measures, budget=None, target_gain=None, s logger.error("Skipping fixed selection due to minimum insulation violation: %s", fixed) continue + scheme = _path_scheme(path_spec) + # 3) compute fixed cost/gain, and strip those groups from subproblem fixed_items = [opt for (_, _, opt) in fixed] fixed_ids = [opt['id'] for opt in fixed_items] fixed_cost, fixed_gain = sum_cost_gain(fixed_items) fixed_groups = {gi for (gi, _, _) in fixed} - sub_measures = [grp for gi, grp in enumerate(input_measures) if gi not in fixed_groups] + sub_measures = deepcopy([grp for gi, grp in enumerate(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 + # do this by adding innovation back onto the cost + for grp in sub_measures: + for opt in grp: + opt["cost"] = opt["cost_minus_uplift"] + opt.get("innovation_uplift", 0.0) + + if scheme == "eco4": + # Need to strip out any measure types that are not eligible for ECO4 funding (e.g. secondary heating) + raise ValueError() # 4) run your existing optimiser for the remaining groups # If we have a budget, we need to ensure the subproblem respects it so we remove the fixed cost (which