diff --git a/backend/Funding.py b/backend/Funding.py index d95117c5..4a198bf9 100644 --- a/backend/Funding.py +++ b/backend/Funding.py @@ -715,9 +715,6 @@ class Funding: - all other measures are insulation (can be non-innovation) """ - raise ValueError( - "THis isnt quite right. Band D homes must be pre-insulated OR it should include one of the" - ) # The condition is: # one of the following insulation measures must be installed as part of the # same ECO4 project: diff --git a/recommendations/optimiser/funding_optimiser.py b/recommendations/optimiser/funding_optimiser.py new file mode 100644 index 00000000..65335e02 --- /dev/null +++ b/recommendations/optimiser/funding_optimiser.py @@ -0,0 +1,767 @@ +""" +This script contains a number of functions which are designed to enable optimisation and selection of funding options +for individual properties to improve their energy efficiency + +The main entry point to this is optimise_with_funding_paths + +In the future, we will adapt this into a class-based structure to allow for more flexibility and reusability +""" + +from copy import deepcopy +import pandas as pd + +from backend.app.plan.schemas import ( + WALL_INSULATION_MEASURES, ROOF_INSULATION_MEASURES, ECO4_ELIGIBILE_FABRIC_MEASURES +) +from recommendations.optimiser.CostOptimiser import CostOptimiser +from recommendations.optimiser.GainOptimiser import GainOptimiser +from utils.logger import setup_logger +from backend.Funding import Funding + +logger = setup_logger() + +# measures we DO NOT treat as fundable in the ECO4 'funded' pass +_ECO4_EXCLUDE_TYPES = {"secondary_heating"} + + +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): + """Return True if fixed selection includes a heating/PV measure but no required insulation.""" + picked_types = {opt["type"] for (_, _, opt) in fixed} + + def has_any(substrs): + return any(any(s in t for s in substrs) for t in picked_types) + + # heating (incl. PV) flags + is_heating = has_any([ + "air_source_heat_pump", + "high_heat_retention_storage_heater", + "boiler_upgrade", + "electric_boiler", + "time_temperature_zone_control", + "secondary_heating", + "solar_pv", # PV treated as heating for MIR + ]) + + # MIR insulation (the ones you’re using in path construction) + has_insul = has_any([ + "external_wall_insulation", + "internal_wall_insulation", + "cavity_wall_insulation", + "extension_cavity_wall_insulation", + "loft_insulation", + "flat_roof_insulation", + "room_roof_insulation", + ]) + + return is_heating and not has_insul + + +# Treat "type" like "external_wall_insulation+mechanical_ventilation" → "external_wall_insulation" +def _base_type(s: str) -> str: + return s.split("+", 1)[0] + + +def _filter_measures_by_types(input_measures, allowed_types): + """ + Keep only groups that have ≥1 allowed option; inside each group keep only allowed options. + """ + allowed_set = set(allowed_types) + filtered = [] + for group in input_measures: + kept_opts = [opt for opt in group if _base_type(opt["type"]) in allowed_set] + if kept_opts: + filtered.append(kept_opts) + return filtered + + +def _is_eligible_funding_package(scheme, starting_sap, total_gain): + if scheme == "eco4": + # We check if we meet the upgrade requirements + # If the property is an E or above, we need to upgrade to a C or above + if starting_sap >= 39: # ie. EPC C or above + return starting_sap + total_gain >= 69 + + if scheme == "gbis": + # GBIS is a fixed measure only, so we don't check the gain + return True + + +def _prs_solution_ok(items, p, funding): + # 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, funding: Funding, budget=None, target_gain=None): + """ + run_optimizer(sub_measures, budget, target_gain) -> (picked_options, sub_cost, sub_gain) + """ + + 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, funding) + + # 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 + if housing_type == "Social": + funding_paths = ( + [ + { + 'reference': 'fabric-only:eco4', + "allowed_types": WALL_INSULATION_MEASURES + ROOF_INSULATION_MEASURES + + ECO4_ELIGIBILE_FABRIC_MEASURES + } + ] + funding_paths + ) + + 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(optimisation_input_measures, path_spec["allowed_types"]) + if not sub_measures: + continue + + picked, sub_cost, sub_gain = run_optimizer( + sub_measures, + budget=budget, # no fixed items; budget unchanged + sub_target_gain=target_gain + ) + + if picked is None: + continue + + scheme = _path_scheme([path_spec]) + + solutions.append( + { + "fixed_ids": [], + "items": picked, + "total_cost": sub_cost, + "total_gain": sub_gain, + "path": path_spec, + "scheme": scheme, + "is_eligible": _is_eligible_funding_package(scheme, p.data["current-energy-efficiency"], sub_gain) + } + ) + + continue + + # 1) expand fixed selections for this path + fixed_selections = expand_funding_path(optimisation_input_measures, path_spec) if path_spec else [[]] + if not fixed_selections: + continue + + for fixed in fixed_selections: + + if violates_min_insulation(fixed): + # We log an error and skip this - we should not see any errors but we can probably get a reasonable + # outcome for the end user without a complete termination of the process + 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 = 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 + # 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 + # may already be over budget) and the fixed gain (which may not be achievable) + picked, sub_cost, sub_gain = run_optimizer( + sub_measures, + 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_cost = fixed_cost + sub_cost + total_gain = fixed_gain + sub_gain + total_picks = fixed_items + picked + + if housing_type == "Private": + if not _prs_solution_ok(total_picks, p, funding): + 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({ + "fixed_ids": fixed_ids, + "items": total_picks, + "total_cost": total_cost, + "total_gain": total_gain, + "path": path_spec, + "scheme": scheme, + "is_eligible": _is_eligible_funding_package(scheme, p.data["current-energy-efficiency"], total_gain) + }) + + solutions = pd.DataFrame(solutions) + + # Given the scheme, we now check if the packages are eligible. If they *are* eligible, but they don't meet the + # final upgrade target, we then look to perform a final optimisation pass to meet the target gain. + solutions["meets_upgrade_target"] = solutions["total_gain"] >= target_gain + + # If we have packages that are fundable, but do not meet the upgrade target, we can run a final optimisation pass + if not solutions[solutions["is_eligible"] & ~solutions["meets_upgrade_target"]].empty: + raise NotImplementedError("Implement me") + + return solutions + + +# ---- helpers ------------------------------------------------------------- + + +def sum_cost_gain(items): + c = sum(float(x['cost']) for x in items) + g = sum(float(x['gain']) for x in items) + return c, g + + +# ---- candidate expansion ------------------------------------------------- +def type_matches(option_type: str, required: str) -> bool: + # substring match so "external_wall_insulation+mechanical_ventilation" satisfies "external_wall_insulation" + return required in option_type + + +def candidates_for_type(input_measures, required_type): + """ + Return a list of (gi, oi, opt) where opt['type'] contains required_type. + gi = group index, oi = option index inside that group. + """ + cands = [] + for gi, group in enumerate(input_measures): + for oi, opt in enumerate(group): + if type_matches(opt["type"], required_type): + cands.append((gi, oi, opt)) + return cands + + +def iter_or_candidates(input_measures, types_list): + """ + For OR: pick exactly ONE candidate whose type matches ANY in types_list. + Return a list of dicts: {"fixed": [(gi, oi, opt)]} + """ + union = [] + seen_ids = set() + for t in types_list: + for tup in candidates_for_type(input_measures, t): + # de-dupe by the option id so the same physical option (with multi-type name) isn’t repeated + if tup[2]["id"] not in seen_ids: + seen_ids.add(tup[2]["id"]) + union.append(tup) + return [{"fixed": [t]} for t in union] + + +def iter_and_candidates(input_measures, types_list): + """ + For AND: we must cover ALL required types. + We allow a single option to satisfy multiple types. + We build a simple product but collapse duplicates by (gi, oi). + """ + # Build candidate pools per required type + pools = [candidates_for_type(input_measures, t) for t in types_list] + if any(len(pool) == 0 for pool in pools): + return [] # impossible to satisfy AND + + # Start with one empty selection; accumulate per pool + selections = [[]] # each selection is a list of (gi, oi, opt) + for pool in pools: + new_selections = [] + for sel in selections: + for cand in pool: + # Try adding cand; collapse duplicates by (gi,oi) + gi, oi, opt = cand + replaced = False + conflict = False + merged = [] + for (sgi, soi, sopt) in sel: + if (sgi, soi) == (gi, oi): + # same exact option already in selection (satisfies another required type) – keep one + replaced = True + # keep the existing one (identical) + merged.append((sgi, soi, sopt)) + else: + merged.append((sgi, soi, sopt)) + if not replaced: + merged.append(cand) + if not conflict: + new_selections.append(merged) + selections = new_selections + if not selections: + return [] + + # After accumulation, we may still have duplicate groups with different options (conflict). Drop those. + cleaned = [] + for sel in selections: + seen_by_group = {} + ok = True + for gi, oi, opt in sel: + if gi in seen_by_group and seen_by_group[gi] != oi: + # same group, different option -> conflict for AND; invalid selection + ok = False + break + seen_by_group[gi] = oi + if ok: + # ensure stable order and unique by (gi,oi) + uniq = {} + for gi, oi, opt in sel: + uniq[(gi, oi)] = opt + cleaned.append([(gi, oi, opt) for (gi, oi), opt in uniq.items()]) + return [{"fixed": c} for c in cleaned] + + +def expand_funding_path(input_measures, path_spec): + """ + path_spec is a list of elements; each element is either: + {"OR": [type1, type2, ...], "reference": "..."} or + {"AND": [type1, type2, ...], "reference": "..."} + We cross-product across elements (all required), and produce selections as lists of (gi, oi, opt). + """ + selections = [[]] # list[list[(gi,oi,opt)]] + for elem in path_spec: + if "OR" in elem: + cands = iter_or_candidates(input_measures, elem["OR"]) + elif "AND" in elem: + cands = iter_and_candidates(input_measures, elem["AND"]) + else: + raise ValueError("unknown path element; expected 'OR' or 'AND'") + + if not cands: + return [] + + new_selections = [] + for base in selections: + for cand in cands: + # merge base + cand["fixed"], collapsing duplicate same-option picks + combined = list(base) + # reject if combined picks two different options from the same group + groups_to_oi = {(gi,): oi for gi, oi, _ in combined} # temporary; we’ll refactor below + conflict = False + # simpler: build a dict by group -> (oi, opt), conflict if group exists with different oi + gmap = {gi: (oi, opt) for gi, oi, opt in combined} + for gi, oi, opt in cand["fixed"]: + if gi in gmap: + prev_oi, _ = gmap[gi] + if prev_oi != oi: + conflict = True + break + gmap[gi] = (oi, opt) + if conflict: + continue + # back to list + merged = [(gi, oi, opt) for gi, (oi, opt) in gmap.items()] + new_selections.append(merged) + selections = new_selections + if not selections: + return [] + + # Final tidy: ensure no duplicate groups with different options (already protected), keep stable ordering + deduped = [] + for sel in selections: + gmap = {} + for gi, oi, opt in sel: + # keep the first occurrence + if gi not in gmap: + gmap[gi] = (oi, opt) + else: + # same group, different oi would have been filtered; if same oi, ignore duplicate + pass + deduped.append([(gi, oi, opt) for gi, (oi, opt) in gmap.items()]) + return deduped + + +# ---- tiny utilities ---------------------------------------------------------- + +def parse_types(t): + # e.g. "external_wall_insulation+mechanical_ventilation" -> {"external_wall_insulation","mechanical_ventilation"} + return set(map(str.strip, t.split("+"))) if isinstance(t, str) else set() + + +def includes_heating(opt_types): + return any(x in opt_types for x in { + "air_source_heat_pump", + "high_heat_retention_storage_heater", + "time_temperature_zone_control", # controls count as a heating measure in your pipeline + "solar_pv" # you treat PV as heating for funding logic + }) + + +def contributes_min_insulation(opt_types): + # MIR satisfiers you mentioned (extend as needed) + return any(x in opt_types for x in { + "external_wall_insulation", + "internal_wall_insulation", + "loft_insulation", + "cavity_wall_insulation", + }) + + +def run_optimizer(input_measures, budget=None, sub_target_gain=None, allow_slack=False): + """ + Thin wrapper over your optimisers. + Returns: list[dict] selected_options + """ + if budget is not None: + opt = GainOptimiser( + input_measures, max_cost=budget, max_gain=(sub_target_gain or float("inf")), + allow_slack=allow_slack + ) + else: + if sub_target_gain is None: + raise ValueError("Either budget or target_gain must be provided.") + opt = CostOptimiser(input_measures, min_gain=sub_target_gain) + + opt.setup() + opt.solve() + cost = sum([x["cost"] for x in opt.solution]) + return opt.solution, cost, opt.solution_gain + + +# ---- Define optimisation paths ---------------------------------------------------------- + +def _find_measure(input_measures, measure_type): + for measures in input_measures: + for m in measures: + if measure_type in m["type"]: + return True + return False + + +def _make_solar_heating_funding_paths( + p, input_measures, funding_paths, remaining_insulation_type, housing_type, funding: Funding +): + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + # 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 and _find_measure(input_measures, "solar_pv"): + single_solar_template = [{"AND": ["solar_pv"], "reference": None}] + # 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) + # Make a specific reference for this path + new_solar_path[0]["reference"] = "solar_pv+" + insulation_measure + ":eco4" + 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([{"AND": ["solar_pv"], "reference": "solar_pv:eco4"}]) + + # For each of these, because there is a heating measure begin implemented, we check for minimum insulation + # requirements. + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + # Solar PV + Heating Upgrade combos + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + # We don't include electric boilers as they are not eligible for ECO4 funding + solar_heating_combos = [ + ("high_heat_retention_storage_heater", "solar_pv+hhrsh:eco4"), + ("air_source_heat_pump", "solar_pv+ashp:eco4"), + ] + if _find_measure(input_measures, "solar_pv"): + for heat_type, ref in solar_heating_combos: + if _find_measure(input_measures, heat_type): + if remaining_insulation_type: + for insulation_measure in remaining_insulation_type: + funding_paths.append( + [{"AND": ["solar_pv", heat_type, insulation_measure], + "reference": f"{ref[:-5]}+{insulation_measure}:eco4"}] # keeps naming consistent + ) + else: + funding_paths.append([{"AND": ["solar_pv", heat_type], "reference": ref}]) + + # 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 + + # 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 single_heating_measures: + if _find_measure(input_measures, heating_upgrade): + if remaining_insulation_type: + for insulation_measure in remaining_insulation_type: + path = [ + { + "AND": [heating_upgrade, insulation_measure], + "reference": f"{measure_references[heating_upgrade]}+{insulation_measure}:eco4" + } + ] + funding_paths.append(path) + else: + funding_paths.append( + [{"AND": [heating_upgrade], "reference": f"{measure_references[heating_upgrade]}:eco4"}] + ) + + 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, housing_type, funding: Funding): + """ + 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 housing_type: + :return: + """ + # 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. + + # We determine which insulation measures need to be included + wall_insulation_measures = [ + "internal_wall_insulation", "external_wall_insulation", "cavity_wall_insulation", + "extension_cavity_wall_insulation" + ] + 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 = [] + for insulation_measure in wall_insulation_measures + roof_insulation_measures: + if _find_measure(input_measures, insulation_measure): + remaining_insulation_type.append(insulation_measure) + + remaining_insulation_type = list(set(remaining_insulation_type)) + + funding_paths = [] + + 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 = [] + 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 measure["innovation_uplift"] and measure["type"] in ( + remaining_insulation_type + other_gbis_insulation_measures + ): + input_gbis_measures_innovation.append([measure]) + + funding_paths = _make_solar_heating_funding_paths( + p, input_measures_innovation, funding_paths, remaining_insulation_type, housing_type, funding + ) + + # Can only be innovation GBIS measures + funding_paths = _make_generic_gbis_funding_paths(input_gbis_measures_innovation, funding_paths) + return funding_paths, input_measures_innovation + + if housing_type == "Private": + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + # EWI or IWI + # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + # 1) The package must include EWI or IWI if the property is private rental sector + # We check if we have any EWI or IWI measures available + ewi_or_iwi = [{"OR": []}] + reference_measures = [] + # If we have EWI we add it in + if _find_measure(input_measures, "external_wall_insulation"): + ewi_or_iwi[0]["OR"].append("external_wall_insulation") + reference_measures.append("ewi") + + if _find_measure(input_measures, "internal_wall_insulation"): + ewi_or_iwi[0]["OR"].append("internal_wall_insulation") + reference_measures.append("iwi") + + if ewi_or_iwi[0]["OR"]: + ewi_or_iwi[0]["reference"] = "+".join(reference_measures) + ":eco4" + funding_paths.append(ewi_or_iwi) + + funding_paths = _make_solar_heating_funding_paths( + p, input_measures, funding_paths, remaining_insulation_type, housing_type, funding + ) + + # 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]) + + funding_paths = _make_generic_gbis_funding_paths(input_gbis_measures, funding_paths) + + return funding_paths, input_measures diff --git a/recommendations/tests/test_optimiser_functions.py b/recommendations/tests/test_optimiser_functions.py index b2097422..430acaa8 100644 --- a/recommendations/tests/test_optimiser_functions.py +++ b/recommendations/tests/test_optimiser_functions.py @@ -12,7 +12,7 @@ class TestPrepareInputMeasures: recs = [ [ # loft insulation measure {"recommendation_id": "loft1", "type": "loft_insulation", "total": 100, "kwh_savings": 200, - "energy_cost_savings": 10, "has_battery": False}, + "energy_cost_savings": 10, "has_battery": False, "measure_type": "loft_insulation"}, ], ] measures = optimiser_functions.prepare_input_measures(recs, goal="Energy Savings", needs_ventilation=False) @@ -27,9 +27,9 @@ class TestPrepareInputMeasures: ["internal_wall_insulation"]) recs = [ [{"recommendation_id": "wall1", "type": "internal_wall_insulation", "total": 500, "kwh_savings": 300, - "energy_cost_savings": 5, "has_battery": False}], + "energy_cost_savings": 5, "has_battery": False, "measure_type": "internal_wall_insulation"}], [{"recommendation_id": "vent1", "type": "mechanical_ventilation", "total": 50, "kwh_savings": 30, - "energy_cost_savings": 5, "has_battery": False}] + "energy_cost_savings": 5, "has_battery": False, "measure_type": "mechanical_ventilation"}], ] measures = optimiser_functions.prepare_input_measures(recs, goal="Energy Savings", needs_ventilation=True) wall_option = measures[0][0] diff --git a/recommendations/tests/test_optimisers.py b/recommendations/tests/test_optimisers.py index 2e8186dd..df5cc2e1 100644 --- a/recommendations/tests/test_optimisers.py +++ b/recommendations/tests/test_optimisers.py @@ -1,1212 +1,665 @@ import numpy as np -import pandas as pd +# 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 app.plan.schemas import ( - WALL_INSULATION_MEASURES, ROOF_INSULATION_MEASURES, ECO4_ELIGIBILE_FABRIC_MEASURES, ECO4_ELIGIBLE_HEATING_MEASURES -) -from recommendations.optimiser.CostOptimiser import CostOptimiser -from recommendations.optimiser.GainOptimiser import GainOptimiser +from recommendations.optimiser import optimiser_functions +from recommendations.optimiser.funding_optimiser import optimise_with_funding_paths # wherever you defined it from backend.Funding import Funding +from backend.app.plan.schemas import WALL_INSULATION_MEASURES, ROOF_INSULATION_MEASURES, ECO4_ELIGIBILE_FABRIC_MEASURES -# 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', - '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 -property_recommendations = [ - [{'phase': 0, 'parts': [{'id': 2466, 'type': 'external_wall_insulation', - 'description': 'EWI Pro EPS external wall insulation system with ' - 'Brick Slip finish', - 'depth': 150.0, 'depth_unit': 'mm', 'cost': None, - 'cost_unit': 'gbp_per_m2', 'r_value_per_mm': 0.02631579, - 'r_value_unit': 'square_meter_kelvin_per_watt', - 'thermal_conductivity': 0.038, - 'thermal_conductivity_unit': 'watt_per_meter_kelvin', - 'link': 'SCIS', - 'created_at': Timestamp('2025-03-16 15:26:22.379496'), - 'is_active': True, 'prime_material_cost': None, - 'material_cost': 0.0, 'labour_cost': 0.0, - 'labour_hours_per_unit': 0.0, 'plant_cost': 0.0, - 'total_cost': 298.35, - 'notes': 'This is the quoted value from SCIS', - 'is_installer_quote': True, 'quantity': 63.98796761892035, - 'quantity_unit': 'm2', 'total': 19090.810139104888, - 'labour_hours': 0.0, 'labour_days': 0.0}], - 'type': 'external_wall_insulation', 'measure_type': 'external_wall_insulation', - 'description': 'Install 150mm EWI Pro EPS external wall insulation system with Brick ' - 'Slip finish on external walls', - 'starting_u_value': 1.7, 'new_u_value': 0.32, 'already_installed': False, - 'sap_points': np.float64(9.6), - 'simulation_config': {'is_as_built_ending': False, 'walls_is_assumed_ending': False, - 'walls_insulation_thickness_ending': 'average', - 'external_insulation_ending': True, - 'walls_energy_eff_ending': 'Good', - 'walls_thermal_transmittance_ending': 0.23}, - 'description_simulation': {'walls-description': 'Solid brick, with external insulation', - 'walls-energy-eff': 'Good'}, 'total': 19090.810139104888, - 'labour_hours': 0.0, 'labour_days': 0.0, 'survey': False, - 'recommendation_id': '0_phase=0', 'efficiency': 11229.568317120522, - 'co2_equivalent_savings': np.float64(0.5), 'heat_demand': np.float64(37.099999999999994), - 'kwh_savings': np.float64(1827.8999999999996), - 'energy_cost_savings': np.float64(136.1247882352941)}, {'phase': 0, 'parts': [ - {'id': 2373, 'type': 'internal_wall_insulation', 'description': 'SWIP EcoBatt & Plastered finish', - 'depth': 95.0, - 'depth_unit': 'mm', 'cost': None, 'cost_unit': 'gbp_per_m2', 'r_value_per_mm': 0.03125, - 'r_value_unit': 'square_meter_kelvin_per_watt', 'thermal_conductivity': 0.032, - 'thermal_conductivity_unit': None, - 'link': 'SCIS', 'created_at': Timestamp('2025-03-16 15:26:22.379496'), 'is_active': True, - 'prime_material_cost': None, 'material_cost': 0.0, 'labour_cost': 0.0, 'labour_hours_per_unit': 2.1, - 'plant_cost': 0.0, 'total_cost': 89.0, 'notes': None, 'is_installer_quote': True, - 'quantity': 63.98796761892035, - 'quantity_unit': 'm2', 'total': 5694.929118083911, 'labour_hours': 134.37473199973275, - 'labour_days': 4.199210374991648}], 'type': 'internal_wall_insulation', - 'measure_type': 'internal_wall_insulation', - 'description': 'Install 95mm ' - 'SWIP EcoBatt & ' - 'Plastered ' - 'finish on ' - 'internal walls', - 'starting_u_value': 1.7, - 'new_u_value': 0.32, - 'already_installed': False, - 'sap_points': 6, - 'simulation_config': { - 'is_as_built_ending': False, - 'walls_is_assumed_ending': - False, - 'walls_insulation_thickness_ending': 'average', - 'internal_insulation_ending': True, - 'walls_energy_eff_ending': - 'Good', - 'walls_thermal_transmittance_ending': 0.29}, - 'description_simulation': { - 'walls-description': 'Solid ' - 'brick, with internal ' - 'insulation', - 'walls-energy-eff': 'Good'}, - 'total': 5694.929118083911, - 'labour_hours': 134.37473199973275, - 'labour_days': 4.199210374991648, - 'survey': True, - 'recommendation_id': '1_phase=0', - 'efficiency': 3349.6383047552417, - 'co2_equivalent_savings': np.float64( - 0.5), - 'heat_demand': np.float64( - 35.30000000000001), - 'kwh_savings': np.float64( - 1432.3999999999996), - 'energy_cost_savings': np.float64( - 106.67167058823532)}], [ - {'phase': 1, 'parts': [{'id': 2351, 'type': 'loft_insulation', - 'description': 'Knauf Loft Roll 44 glass fibre roll', - 'depth': 300.0, 'depth_unit': 'mm', 'cost': None, - 'cost_unit': 'gbp_per_m2', 'r_value_per_mm': 0.022727273, - 'r_value_unit': 'square_meter_kelvin_per_watt', - 'thermal_conductivity': 0.044, - 'thermal_conductivity_unit': 'watt_per_meter_kelvin', - 'link': 'SCIS', - 'created_at': Timestamp('2025-03-16 15:26:22.379496'), - 'is_active': True, 'prime_material_cost': None, - 'material_cost': 0.0, 'labour_cost': 0.0, - 'labour_hours_per_unit': 0.11, 'plant_cost': 0.0, - 'total_cost': 15.0, - 'notes': 'This is the cost if there is less than 100mm ' - 'existing insulation', - 'is_installer_quote': True, 'quantity': 63.98796761892035, - 'quantity_unit': 'm2', 'total': 645.0, 'labour_hours': 8, - 'labour_days': 1}], 'type': 'loft_insulation', - 'measure_type': 'loft_insulation', - 'description': 'Install 300mm of Knauf Loft Roll 44 glass fibre roll in your loft', - 'starting_u_value': 2.3, 'new_u_value': 2.3, 'sap_points': np.float64(2.4), - 'already_installed': False, - 'simulation_config': {'is_loft_ending': True, 'roof_is_assumed_ending': False, - 'roof_insulation_thickness_ending': '300', - 'roof_thermal_transmittance_ending': 2.3, - 'roof_energy_eff_ending': 'Very Good'}, - 'description_simulation': {'roof-description': 'Pitched, 300mm loft insulation', - 'roof-energy-eff': 'Very Good'}, 'total': 645.0, - 'labour_hours': 8, 'labour_days': 1, 'survey': False, 'recommendation_id': '2_phase=1', - 'efficiency': 278.1347826086957, - 'co2_equivalent_savings': np.float64(0.10000000000000009), - 'heat_demand': np.float64(1.5), 'kwh_savings': np.float64(566.1499999999996), - 'energy_cost_savings': np.float64(42.16152352941185)}], [{'phase': 2, 'parts': [ - {'id': 2329, 'type': 'mechanical_ventilation', 'description': 'Mechanical Extract Ventilation', 'depth': 0.0, - 'depth_unit': None, 'cost': None, 'cost_unit': 'gbp_per_unit', 'r_value_per_mm': nan, - 'r_value_unit': 'square_meter_kelvin_per_watt', 'thermal_conductivity': None, - 'thermal_conductivity_unit': None, - 'link': 'SCIS', 'created_at': datetime.datetime(2025, 3, 16, 15, 26, 22, 379496), 'is_active': True, - 'prime_material_cost': None, 'material_cost': 0.0, 'labour_cost': 0.0, 'labour_hours_per_unit': 0.0, - 'plant_cost': 0.0, 'total_cost': 350.0, 'notes': None, 'is_installer_quote': True, 'total': 700.0, - 'quantity': 2, - 'quantity_unit': 'part'}], 'type': 'mechanical_ventilation', 'measure_type': 'mechanical_ventilation', - 'description': 'Install 2 ' - 'Mechanical ' - 'Extract ' - 'Ventilation units', - 'starting_u_value': None, - 'new_u_value': None, - 'already_installed': False, - 'sap_points': np.float64( - -0.10000000000000142), - 'heat_demand': np.float64( - -3.3999999999999773), - 'kwh_savings': np.float64( - -53.80000000000018), - 'co2_equivalent_savings': np.float64( - 0.0), - 'energy_cost_savings': np.float64( - -4.0065176470588995), - 'total': 700.0, - 'labour_hours': 8, - 'labour_days': 1.0, - 'simulation_config': { - 'mechanical_ventilation_ending': 'mechanical, ' - 'extract only'}, - 'description_simulation': { - 'mechanical-ventilation': 'mechanical, ' - 'extract only'}, - 'recommendation_id': '3_phase=2', - 'efficiency': 0}], [ - {'phase': 3, 'parts': [{'id': 2409, 'type': 'suspended_floor_insulation', - 'description': 'Q-bot underfloor insulation', 'depth': 75.0, - 'depth_unit': 'mm', 'cost': None, 'cost_unit': 'gbp_per_m2', - 'r_value_per_mm': 0.045454547, - 'r_value_unit': 'square_meter_kelvin_per_watt', - 'thermal_conductivity': 0.022, - 'thermal_conductivity_unit': 'watt_per_meter_kelvin', - 'link': 'SCIS', - 'created_at': Timestamp('2025-03-16 15:26:22.379496'), - 'is_active': True, 'prime_material_cost': None, - 'material_cost': 0.0, 'labour_cost': 0.0, - 'labour_hours_per_unit': 1.63, 'plant_cost': 0.0, - 'total_cost': 93.75, - 'notes': 'Linearly interpolated based on Qbot costs', - 'is_installer_quote': True, 'quantity': 43.0, - 'quantity_unit': 'm2', 'total': 4031.25, - 'labour_hours': 70.08999999999999, - 'labour_days': 2.920416666666666}], - 'type': 'suspended_floor_insulation', 'measure_type': 'suspended_floor_insulation', - 'description': 'Install 75mm Q-bot underfloor insulation insulation in suspended ' - 'floor', - 'starting_u_value': 0.83, 'new_u_value': 0.22, 'sap_points': 2, 'survey': True, - 'already_installed': False, 'simulation_config': {'floor_is_assumed_ending': False, - 'floor_insulation_thickness_ending': 'average', - 'floor_thermal_transmittance_ending': 0.685593}, - 'description_simulation': {'floor-description': 'Suspended, insulated'}, - 'total': 4031.25, 'labour_hours': 70.08999999999999, 'labour_days': 2.920416666666666, - 'recommendation_id': '4_phase=3', 'efficiency': 4856.707710843373, - 'co2_equivalent_savings': np.float64(0.20000000000000018), - 'heat_demand': np.float64(33.5), 'kwh_savings': np.float64(1021.1999999999998), - 'energy_cost_savings': np.float64(76.04936470588231)}], [ - {'phase': 4, 'parts': [], 'type': 'low_energy_lighting', - 'measure_type': 'low_energy_lighting', - 'description': 'Install low energy lighting in -886 outlets', 'starting_u_value': None, - 'new_u_value': None, 'already_installed': False, 'sap_points': 2, - 'kwh_savings': -48508.5, 'energy_cost_savings': -12481.237049999998, - 'co2_equivalent_savings': -7.858377, - 'description_simulation': {'lighting-energy-eff': 'Very Good', - 'lighting-description': 'Low energy lighting in all fixed' - ' outlets', - 'low-energy-lighting': 100}, 'total': -3411.1000000000004, - 'labour_hours': 1, 'labour_days': 0.125, 'survey': True, - 'recommendation_id': '5_phase=4', 'efficiency': -1705.5500000000002, - 'heat_demand': np.float64(5.099999999999994)}], [ - {'type': 'heating', 'phase': 5, 'measure_type': 'time_temperature_zone_control', - 'parts': [], - 'description': 'Upgrade heating controls to Smart Thermostats, room sensors and ' - 'smart radiator valves (time & temperature zone control)', - 'total': 739.576, 'subtotal': 700.48, 'vat': 39.096000000000004, - 'labour_hours': 3.6199999999999997, 'labour_days': np.float64(1.0), - 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(2.9), - 'already_installed': False, 'simulation_config': { - 'thermostatic_control_ending': 'time and temperature zone control', - 'switch_system_ending': None, 'trvs_ending': None, - 'mainheatc_energy_eff_ending': 'Very Good'}, 'description_simulation': { - 'mainheatcont-description': 'Time and temperature zone control', - 'mainheatc-energy-eff': 'Very Good'}, 'recommendation_id': '6_phase=5', - 'efficiency': 739.576, 'co2_equivalent_savings': np.float64(0.30000000000000027), - 'heat_demand': np.float64(6.599999999999994), - 'kwh_savings': np.float64(876.8000000000002), - 'energy_cost_savings': np.float64(65.29581176470589)}], [ - {'phase': 6, 'parts': [], 'type': 'secondary_heating', - 'measure_type': 'secondary_heating', - 'description': 'Remove the secondary heating system', 'starting_u_value': None, - 'new_u_value': None, 'sap_points': np.float64(3.6), 'already_installed': False, - 'total': 30.0, 'subtotal': 25.0, 'vat': 5.0, 'labour_hours': 3.0, - 'labour_days': np.float64(1.0), - 'simulation_config': {'secondheat_description_ending': 'None'}, - 'description_simulation': {'secondheat-description': 'None'}, - 'recommendation_id': '7_phase=6', 'efficiency': 30.0, - 'co2_equivalent_savings': np.float64(0.10000000000000009), - 'heat_demand': np.float64(15.400000000000006), - 'kwh_savings': np.float64(196.29999999999927), - 'energy_cost_savings': np.float64(14.61857647058821)}], [ - {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', - 'description': 'Install a 4.0 kilowatt-peak (kWp) solar panel system.', - 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(13.0), - 'already_installed': False, 'total': 6013.139999999999, 'subtotal': 5010.95, 'vat': 0, - 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(65.0), - 'has_battery': False, 'initial_ac_kwh_per_year': np.float64(4081.7132614999996), - 'description_simulation': {'photo-supply': np.float64(65.0)}, - 'recommendation_id': '8_phase=7', 'efficiency': np.float64(462.54923076923075), - 'co2_equivalent_savings': np.float64(0.47347873833399995), - 'heat_demand': np.float64(88.69999999999999), - 'kwh_savings': np.float64(2040.8566307499998), - 'energy_cost_savings': np.float64(525.1124110919749)}, - {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', - 'description': 'Install a 4.0 kilowatt-peak (kWp) solar panel system, with a battery.', - 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(13.0), - 'already_installed': False, 'total': 10537.008, 'subtotal': 8780.84, 'vat': 0, - 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(65.0), - 'has_battery': True, 'initial_ac_kwh_per_year': np.float64(4081.7132614999996), - 'description_simulation': {'photo-supply': np.float64(65.0)}, - 'recommendation_id': '9_phase=7', 'efficiency': np.float64(810.5390769230769), - 'co2_equivalent_savings': np.float64(0.6628702336675999), - 'heat_demand': np.float64(88.69999999999999), - 'kwh_savings': np.float64(2857.1992830499994), - 'energy_cost_savings': np.float64(735.1573755287648)}, - {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', - 'description': 'Install a 3.6 kilowatt-peak (kWp) solar panel system.', - 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(12.0), - 'already_installed': False, 'total': 5826.491999999999, 'subtotal': 4855.41, 'vat': 0, - 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(60.0), - 'has_battery': False, 'initial_ac_kwh_per_year': np.float64(3692.66794), - 'description_simulation': {'photo-supply': np.float64(60.0)}, - 'recommendation_id': '10_phase=7', 'efficiency': np.float64(485.54099999999994), - 'co2_equivalent_savings': np.float64(0.42834948104), - 'heat_demand': np.float64(83.69999999999999), 'kwh_savings': np.float64(1846.33397), - 'energy_cost_savings': np.float64(475.0617304809999)}, - {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', - 'description': 'Install a 3.6 kilowatt-peak (kWp) solar panel system, with a battery.', - 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(12.0), - 'already_installed': False, 'total': 10350.359999999999, 'subtotal': 8625.3, 'vat': 0, - 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(60.0), - 'has_battery': True, 'initial_ac_kwh_per_year': np.float64(3692.66794), - 'description_simulation': {'photo-supply': np.float64(60.0)}, - 'recommendation_id': '11_phase=7', 'efficiency': np.float64(862.5299999999999), - 'co2_equivalent_savings': np.float64(0.599689273456), - 'heat_demand': np.float64(83.69999999999999), 'kwh_savings': np.float64(2584.867558), - 'energy_cost_savings': np.float64(665.0864226734)}, - {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', - 'description': 'Install a 3.2 kilowatt-peak (kWp) solar panel system.', - 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(11.0), - 'already_installed': False, 'total': 5642.604, 'subtotal': 4702.17, 'vat': 0, - 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(55.0), - 'has_battery': False, 'initial_ac_kwh_per_year': np.float64(3300.5416548), - 'description_simulation': {'photo-supply': np.float64(55.0)}, - 'recommendation_id': '12_phase=7', 'efficiency': np.float64(512.964), - 'co2_equivalent_savings': np.float64(0.3828628319568), 'heat_demand': np.float64(78.3), - 'kwh_savings': np.float64(1650.2708274), - 'energy_cost_savings': np.float64(424.61468389001993)}, - {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', - 'description': 'Install a 3.2 kilowatt-peak (kWp) solar panel system, with a battery.', - 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(11.0), - 'already_installed': False, 'total': 10166.472, 'subtotal': 8472.06, 'vat': 0, - 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(55.0), - 'has_battery': True, 'initial_ac_kwh_per_year': np.float64(3300.5416548), - 'description_simulation': {'photo-supply': np.float64(55.0)}, - 'recommendation_id': '13_phase=7', 'efficiency': np.float64(924.2247272727273), - 'co2_equivalent_savings': np.float64(0.53600796473952), - 'heat_demand': np.float64(78.3), 'kwh_savings': np.float64(2310.3791583599996), - 'energy_cost_savings': np.float64(594.4605574460278)}, - {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', - 'description': 'Install a 2.8 kilowatt-peak (kWp) solar panel system.', - 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(9.0), - 'already_installed': False, 'total': 5458.727999999999, 'subtotal': 4548.94, 'vat': 0, - 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(45.0), - 'has_battery': False, 'initial_ac_kwh_per_year': np.float64(2907.1867812), - 'description_simulation': {'photo-supply': np.float64(45.0)}, - 'recommendation_id': '14_phase=7', 'efficiency': np.float64(606.5253333333333), - 'co2_equivalent_savings': np.float64(0.3372336666192), 'heat_demand': np.float64(64.0), - 'kwh_savings': np.float64(1453.5933906), - 'energy_cost_savings': np.float64(374.00957940138)}, - {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', - 'description': 'Install a 2.8 kilowatt-peak (kWp) solar panel system, with a battery.', - 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(9.0), - 'already_installed': False, 'total': 9982.596, 'subtotal': 8318.83, 'vat': 0, - 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(45.0), - 'has_battery': True, 'initial_ac_kwh_per_year': np.float64(2907.1867812), - 'description_simulation': {'photo-supply': np.float64(45.0)}, - 'recommendation_id': '15_phase=7', 'efficiency': np.float64(1109.1773333333333), - 'co2_equivalent_savings': np.float64(0.47212713326688), - 'heat_demand': np.float64(64.0), 'kwh_savings': np.float64(2035.03074684), - 'energy_cost_savings': np.float64(523.6134111619319)}, - {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', - 'description': 'Install a 2.4 kilowatt-peak (kWp) solar panel system.', - 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(8.0), - 'already_installed': False, 'total': 5274.852, 'subtotal': 4395.71, 'vat': 0, - 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(40.0), - 'has_battery': False, 'initial_ac_kwh_per_year': np.float64(2510.25188), - 'description_simulation': {'photo-supply': np.float64(40.0)}, - 'recommendation_id': '16_phase=7', 'efficiency': np.float64(659.3565), - 'co2_equivalent_savings': np.float64(0.29118921808), 'heat_demand': np.float64(54.3), - 'kwh_savings': np.float64(1255.12594), - 'energy_cost_savings': np.float64(322.94390436199996)}, - {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', - 'description': 'Install a 2.4 kilowatt-peak (kWp) solar panel system, with a battery.', - 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(8.0), - 'already_installed': False, 'total': 9798.72, 'subtotal': 8165.6, 'vat': 0, - 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(40.0), - 'has_battery': True, 'initial_ac_kwh_per_year': np.float64(2510.25188), - 'description_simulation': {'photo-supply': np.float64(40.0)}, - 'recommendation_id': '17_phase=7', 'efficiency': np.float64(1224.84), - 'co2_equivalent_savings': np.float64(0.40766490531199995), - 'heat_demand': np.float64(54.3), 'kwh_savings': np.float64(1757.1763159999998), - 'energy_cost_savings': np.float64(452.1214661067999)}, - {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', - 'description': 'Install a 2.0 kilowatt-peak (kWp) solar panel system.', - 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(7.0), - 'already_installed': False, 'total': 5090.976, 'subtotal': 4242.48, 'vat': 0, - 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(35.0), - 'has_battery': False, 'initial_ac_kwh_per_year': np.float64(2096.682636), - 'description_simulation': {'photo-supply': np.float64(35.0)}, - 'recommendation_id': '18_phase=7', 'efficiency': np.float64(727.2822857142856), - 'co2_equivalent_savings': np.float64(0.243215185776), 'heat_demand': np.float64(48.5), - 'kwh_savings': np.float64(1048.341318), - 'energy_cost_savings': np.float64(269.7382211214)}, - {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', - 'description': 'Install a 2.0 kilowatt-peak (kWp) solar panel system, with a battery.', - 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(7.0), - 'already_installed': False, 'total': 9614.844, 'subtotal': 8012.369999999999, 'vat': 0, - 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(35.0), - 'has_battery': True, 'initial_ac_kwh_per_year': np.float64(2096.682636), - 'description_simulation': {'photo-supply': np.float64(35.0)}, - 'recommendation_id': '19_phase=7', 'efficiency': np.float64(1373.5491428571427), - 'co2_equivalent_savings': np.float64(0.3405012600864), 'heat_demand': np.float64(48.5), - 'kwh_savings': np.float64(1467.6778451999999), - 'energy_cost_savings': np.float64(377.6335095699599)}] -] - -main_heating = { - 'original_description': 'Boiler and radiators, mains gas', 'clean_description': 'Boiler and radiators, mains gas', - 'has_radiators': True, 'has_fan_coil_units': False, 'has_pipes_in_screed_above_insulation': False, - 'has_pipes_in_insulated_timber_floor': False, 'has_pipes_in_concrete_slab': False, 'has_boiler': True, - 'has_air_source_heat_pump': False, 'has_room_heaters': False, 'has_electric_storage_heaters': False, - 'has_warm_air': False, 'has_electric_underfloor_heating': False, 'has_electric_ceiling_heating': False, - 'has_community_scheme': False, 'has_ground_source_heat_pump': False, 'has_no_system_present': False, - 'has_portable_electric_heaters': False, 'has_water_source_heat_pump': False, 'has_electric_heat_pump': False, - 'has_micro-cogeneration': False, 'has_solar_assisted_heat_pump': False, 'has_exhaust_source_heat_pump': False, - 'has_community_heat_pump': False, 'has_hot-water-only': False, 'has_electric': False, 'has_mains_gas': True, - 'has_wood_logs': False, 'has_coal': False, 'has_oil': False, 'has_wood_pellets': False, 'has_anthracite': False, - 'has_dual_fuel_mineral_and_wood': False, 'has_smokeless_fuel': False, 'has_lpg': False, 'has_b30k': False, - 'has_mineral_and_wood': False, 'has_dual_fuel_appliance': False, 'has_assumed': False, 'has_electricaire': False, - 'has_assumed_for_most_rooms': False, 'has_underfloor_heating': False -} - -main_fuel = { - 'original_description': 'mains gas (not community)', 'clean_description': 'Mains gas not community', - 'fuel_type': 'mains gas', 'tariff_type': None, 'is_community': False, - 'no_individual_heating_or_community_network': False, 'complex_fuel_type': None -} - -# 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 -) +ALLOWED_FABRIC_TYPES = set(WALL_INSULATION_MEASURES + ROOF_INSULATION_MEASURES + ECO4_ELIGIBILE_FABRIC_MEASURES) -def _find_measure(input_measures, measure_type): - for measures in input_measures: - for m in measures: - if measure_type in m["type"]: - return True - return False +@pytest.fixture +def mock_project_scores_matrix(): + data = [] + floor_segments = ["0-72", "73-97", "98-199", "200"] + bands = [ + "Low_G", "High_G", "Low_F", "High_F", "Low_E", "High_E", "Low_D", "High_D", "Low_C", "High_C", "Low_B", + "High_B", "Low_A", "High_A" + ] + + cost = 50.0 + for floor in floor_segments: + for start in bands: + for finish in bands: + if start != finish: # skip identical start/finish (no SAP movement) + data.append({ + "Floor Area Segment": floor, + "Starting Band": start, + "Finishing Band": finish, + "Cost Savings": cost + }) + cost += 5.0 # increment to create variety + + return pd.DataFrame(data) -def _make_solar_heating_funding_paths(p, input_measures, funding_paths, remaining_insulation_type, housing_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"] +@pytest.fixture +def mock_partial_scores_matrix(): + df = pd.read_csv("backend/tests/test_data/ECO4_Partial_Project_Scores_Matrix_v6.csv") + df.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'] + return df + + +class DummyProp: + """Minimal property stub exposing just what your code reads.""" + + def __init__(self): + self.data = { + "current-energy-rating": "E", # or "D" for the special Social+D path + "current-energy-efficiency": 55, # numeric SAP points used in eligibility calc + "mainheat-energy-eff": "Very Good", + } + self.has_ventilation = False + self.floor_area = 70.0 + self.main_heating_controls = {"clean_description": "time and temperature zone control"} + + self.main_heating = { + 'original_description': 'Boiler and radiators, mains gas', + 'clean_description': 'Boiler and radiators, mains gas', + 'has_radiators': True, 'has_fan_coil_units': False, 'has_pipes_in_screed_above_insulation': False, + 'has_pipes_in_insulated_timber_floor': False, 'has_pipes_in_concrete_slab': False, 'has_boiler': True, + 'has_air_source_heat_pump': False, 'has_room_heaters': False, 'has_electric_storage_heaters': False, + 'has_warm_air': False, 'has_electric_underfloor_heating': False, 'has_electric_ceiling_heating': False, + 'has_community_scheme': False, 'has_ground_source_heat_pump': False, 'has_no_system_present': False, + 'has_portable_electric_heaters': False, 'has_water_source_heat_pump': False, 'has_electric_heat_pump': + False, + 'has_micro-cogeneration': False, 'has_solar_assisted_heat_pump': False, 'has_exhaust_source_heat_pump': + False, + 'has_community_heat_pump': False, 'has_hot-water-only': False, 'has_electric': False, 'has_mains_gas': + True, + 'has_wood_logs': False, 'has_coal': False, 'has_oil': False, 'has_wood_pellets': False, + 'has_anthracite': False, + 'has_dual_fuel_mineral_and_wood': False, 'has_smokeless_fuel': False, 'has_lpg': False, 'has_b30k': False, + 'has_mineral_and_wood': False, 'has_dual_fuel_appliance': False, 'has_assumed': False, + 'has_electricaire': False, + 'has_assumed_for_most_rooms': False, 'has_underfloor_heating': False + } + + self.main_fuel = { + 'original_description': 'mains gas (not community)', 'clean_description': 'Mains gas not community', + 'fuel_type': 'mains gas', 'tariff_type': None, 'is_community': False, + 'no_individual_heating_or_community_network': False, 'complex_fuel_type': None + } + + +@pytest.fixture +def p(): + return DummyProp() + + +@pytest.fixture +def funding(monkeypatch, mock_partial_scores_matrix, mock_project_scores_matrix): + """Simple Funding that returns zero uplift so costs stay as provided.""" + # Build the Funding with tiny in-memory frames (avoid test I/O) + + f = Funding( + project_scores_matrix=mock_project_scores_matrix, + partial_project_scores_matrix=mock_partial_scores_matrix, + whlg_eligible_postcodes=pd.DataFrame([{"Postcode": "ab12cd"}]), + 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" ) - if has_eligible_heating_system and _find_measure(input_measures, "solar_pv"): - single_solar_template = [{"AND": ["solar_pv"], "reference": None}] - # 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) - # Make a specific reference for this path - new_solar_path[0]["reference"] = "solar_pv+" + insulation_measure + ":eco4" - solar_paths.append(new_solar_path) + # Keep innovation_uplift simple for the first test + # monkeypatch.setattr(f, "get_innovation_uplift", lambda *args, **kwargs: 0.0) - 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([{"AND": ["solar_pv"], "reference": "solar_pv:eco4"}]) + # If your solar precondition matters, you can force True/False here: + # monkeypatch.setattr( + # __import__("backend").Funding, "check_solar_eligible_heating_system", + # staticmethod(lambda mainheat_description, heating_control_description: False) + # ) - # For each of these, because there is a heating measure begin implemented, we check for minimum insulation - # requirements. - # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - # Solar PV + Heating Upgrade combos - # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - # We don't include electric boilers as they are not eligible for ECO4 funding - solar_heating_combos = [ - ("high_heat_retention_storage_heater", "solar_pv+hhrsh:eco4"), - ("air_source_heat_pump", "solar_pv+ashp:eco4"), + return f + + +@pytest.fixture +def property_recommendations(): + """Short sample; replace with your full block if you want.""" + recs = [ + [{'phase': 0, 'parts': [{'id': 2466, 'type': 'external_wall_insulation', + 'description': 'EWI Pro EPS external wall insulation system with ' + 'Brick Slip finish', + 'depth': 150.0, 'depth_unit': 'mm', 'cost': None, + 'cost_unit': 'gbp_per_m2', 'r_value_per_mm': 0.02631579, + 'r_value_unit': 'square_meter_kelvin_per_watt', + 'thermal_conductivity': 0.038, + 'thermal_conductivity_unit': 'watt_per_meter_kelvin', + 'link': 'SCIS', + 'created_at': Timestamp('2025-03-16 15:26:22.379496'), + 'is_active': True, 'prime_material_cost': None, + 'material_cost': 0.0, 'labour_cost': 0.0, + 'labour_hours_per_unit': 0.0, 'plant_cost': 0.0, + 'total_cost': 298.35, + 'notes': 'This is the quoted value from SCIS', + 'is_installer_quote': True, 'quantity': 63.98796761892035, + 'quantity_unit': 'm2', 'total': 19090.810139104888, + 'labour_hours': 0.0, 'labour_days': 0.0}], + 'type': 'external_wall_insulation', 'measure_type': 'external_wall_insulation', + 'description': 'Install 150mm EWI Pro EPS external wall insulation system with Brick ' + 'Slip finish on external walls', + 'starting_u_value': 1.7, 'new_u_value': 0.32, 'already_installed': False, + 'sap_points': np.float64(9.6), + 'simulation_config': {'is_as_built_ending': False, 'walls_is_assumed_ending': False, + 'walls_insulation_thickness_ending': 'average', + 'external_insulation_ending': True, + 'walls_energy_eff_ending': 'Good', + 'walls_thermal_transmittance_ending': 0.23}, + 'description_simulation': {'walls-description': 'Solid brick, with external insulation', + 'walls-energy-eff': 'Good'}, 'total': 19090.810139104888, + 'labour_hours': 0.0, 'labour_days': 0.0, 'survey': False, + 'recommendation_id': '0_phase=0', 'efficiency': 11229.568317120522, + 'co2_equivalent_savings': np.float64(0.5), 'heat_demand': np.float64(37.099999999999994), + 'kwh_savings': np.float64(1827.8999999999996), + 'energy_cost_savings': np.float64(136.1247882352941)}, {'phase': 0, 'parts': [ + {'id': 2373, 'type': 'internal_wall_insulation', 'description': 'SWIP EcoBatt & Plastered finish', + 'depth': 95.0, + 'depth_unit': 'mm', 'cost': None, 'cost_unit': 'gbp_per_m2', 'r_value_per_mm': 0.03125, + 'r_value_unit': 'square_meter_kelvin_per_watt', 'thermal_conductivity': 0.032, + 'thermal_conductivity_unit': None, + 'link': 'SCIS', 'created_at': Timestamp('2025-03-16 15:26:22.379496'), 'is_active': True, + 'prime_material_cost': None, 'material_cost': 0.0, 'labour_cost': 0.0, 'labour_hours_per_unit': 2.1, + 'plant_cost': 0.0, 'total_cost': 89.0, 'notes': None, 'is_installer_quote': True, + 'quantity': 63.98796761892035, + 'quantity_unit': 'm2', 'total': 5694.929118083911, 'labour_hours': 134.37473199973275, + 'labour_days': 4.199210374991648}], 'type': 'internal_wall_insulation', + 'measure_type': 'internal_wall_insulation', + 'description': 'Install 95mm ' + 'SWIP EcoBatt & ' + 'Plastered ' + 'finish on ' + 'internal walls', + 'starting_u_value': 1.7, + 'new_u_value': 0.32, + 'already_installed': False, + 'sap_points': 6, + 'simulation_config': { + 'is_as_built_ending': False, + 'walls_is_assumed_ending': + False, + 'walls_insulation_thickness_ending': 'average', + 'internal_insulation_ending': True, + 'walls_energy_eff_ending': + 'Good', + 'walls_thermal_transmittance_ending': 0.29}, + 'description_simulation': { + 'walls-description': 'Solid ' + 'brick, with internal ' + 'insulation', + 'walls-energy-eff': 'Good'}, + 'total': 5694.929118083911, + 'labour_hours': 134.37473199973275, + 'labour_days': 4.199210374991648, + 'survey': True, + 'recommendation_id': '1_phase=0', + 'efficiency': 3349.6383047552417, + 'co2_equivalent_savings': np.float64( + 0.5), + 'heat_demand': np.float64( + 35.30000000000001), + 'kwh_savings': np.float64( + 1432.3999999999996), + 'energy_cost_savings': np.float64( + 106.67167058823532)}], [ + {'phase': 1, 'parts': [{'id': 2351, 'type': 'loft_insulation', + 'description': 'Knauf Loft Roll 44 glass fibre roll', + 'depth': 300.0, 'depth_unit': 'mm', 'cost': None, + 'cost_unit': 'gbp_per_m2', 'r_value_per_mm': 0.022727273, + 'r_value_unit': 'square_meter_kelvin_per_watt', + 'thermal_conductivity': 0.044, + 'thermal_conductivity_unit': 'watt_per_meter_kelvin', + 'link': 'SCIS', + 'created_at': Timestamp('2025-03-16 15:26:22.379496'), + 'is_active': True, 'prime_material_cost': None, + 'material_cost': 0.0, 'labour_cost': 0.0, + 'labour_hours_per_unit': 0.11, 'plant_cost': 0.0, + 'total_cost': 15.0, + 'notes': 'This is the cost if there is less than 100mm ' + 'existing insulation', + 'is_installer_quote': True, 'quantity': 63.98796761892035, + 'quantity_unit': 'm2', 'total': 645.0, 'labour_hours': 8, + 'labour_days': 1}], 'type': 'loft_insulation', + 'measure_type': 'loft_insulation', + 'description': 'Install 300mm of Knauf Loft Roll 44 glass fibre roll in your loft', + 'starting_u_value': 2.3, 'new_u_value': 2.3, 'sap_points': np.float64(2.4), + 'already_installed': False, + 'simulation_config': {'is_loft_ending': True, 'roof_is_assumed_ending': False, + 'roof_insulation_thickness_ending': '300', + 'roof_thermal_transmittance_ending': 2.3, + 'roof_energy_eff_ending': 'Very Good'}, + 'description_simulation': {'roof-description': 'Pitched, 300mm loft insulation', + 'roof-energy-eff': 'Very Good'}, 'total': 645.0, + 'labour_hours': 8, 'labour_days': 1, 'survey': False, 'recommendation_id': '2_phase=1', + 'efficiency': 278.1347826086957, + 'co2_equivalent_savings': np.float64(0.10000000000000009), + 'heat_demand': np.float64(1.5), 'kwh_savings': np.float64(566.1499999999996), + 'energy_cost_savings': np.float64(42.16152352941185)}], [{'phase': 2, 'parts': [ + {'id': 2329, 'type': 'mechanical_ventilation', 'description': 'Mechanical Extract Ventilation', + 'depth': 0.0, + 'depth_unit': None, 'cost': None, 'cost_unit': 'gbp_per_unit', 'r_value_per_mm': nan, + 'r_value_unit': 'square_meter_kelvin_per_watt', 'thermal_conductivity': None, + 'thermal_conductivity_unit': None, + 'link': 'SCIS', 'created_at': datetime.datetime(2025, 3, 16, 15, 26, 22, 379496), 'is_active': True, + 'prime_material_cost': None, 'material_cost': 0.0, 'labour_cost': 0.0, 'labour_hours_per_unit': 0.0, + 'plant_cost': 0.0, 'total_cost': 350.0, 'notes': None, 'is_installer_quote': True, 'total': 700.0, + 'quantity': 2, + 'quantity_unit': 'part'}], 'type': 'mechanical_ventilation', 'measure_type': 'mechanical_ventilation', + 'description': 'Install 2 ' + 'Mechanical ' + 'Extract ' + 'Ventilation units', + 'starting_u_value': None, + 'new_u_value': None, + 'already_installed': False, + 'sap_points': np.float64( + -0.10000000000000142), + 'heat_demand': np.float64( + -3.3999999999999773), + 'kwh_savings': np.float64( + -53.80000000000018), + 'co2_equivalent_savings': np.float64( + 0.0), + 'energy_cost_savings': np.float64( + -4.0065176470588995), + 'total': 700.0, + 'labour_hours': 8, + 'labour_days': 1.0, + 'simulation_config': { + 'mechanical_ventilation_ending': + 'mechanical, ' + 'extract ' + 'only'}, + 'description_simulation': { + 'mechanical-ventilation': 'mechanical, ' + 'extract only'}, + 'recommendation_id': '3_phase=2', + 'efficiency': 0}], [ + {'phase': 3, 'parts': [{'id': 2409, 'type': 'suspended_floor_insulation', + 'description': 'Q-bot underfloor insulation', 'depth': 75.0, + 'depth_unit': 'mm', 'cost': None, 'cost_unit': 'gbp_per_m2', + 'r_value_per_mm': 0.045454547, + 'r_value_unit': 'square_meter_kelvin_per_watt', + 'thermal_conductivity': 0.022, + 'thermal_conductivity_unit': 'watt_per_meter_kelvin', + 'link': 'SCIS', + 'created_at': Timestamp('2025-03-16 15:26:22.379496'), + 'is_active': True, 'prime_material_cost': None, + 'material_cost': 0.0, 'labour_cost': 0.0, + 'labour_hours_per_unit': 1.63, 'plant_cost': 0.0, + 'total_cost': 93.75, + 'notes': 'Linearly interpolated based on Qbot costs', + 'is_installer_quote': True, 'quantity': 43.0, + 'quantity_unit': 'm2', 'total': 4031.25, + 'labour_hours': 70.08999999999999, + 'labour_days': 2.920416666666666}], + 'type': 'suspended_floor_insulation', 'measure_type': 'suspended_floor_insulation', + 'description': 'Install 75mm Q-bot underfloor insulation insulation in suspended ' + 'floor', + 'starting_u_value': 0.83, 'new_u_value': 0.22, 'sap_points': 2, 'survey': True, + 'already_installed': False, 'simulation_config': {'floor_is_assumed_ending': False, + 'floor_insulation_thickness_ending': 'average', + 'floor_thermal_transmittance_ending': 0.685593}, + 'description_simulation': {'floor-description': 'Suspended, insulated'}, + 'total': 4031.25, 'labour_hours': 70.08999999999999, 'labour_days': 2.920416666666666, + 'recommendation_id': '4_phase=3', 'efficiency': 4856.707710843373, + 'co2_equivalent_savings': np.float64(0.20000000000000018), + 'heat_demand': np.float64(33.5), 'kwh_savings': np.float64(1021.1999999999998), + 'energy_cost_savings': np.float64(76.04936470588231)}], [ + {'phase': 4, 'parts': [], 'type': 'low_energy_lighting', + 'measure_type': 'low_energy_lighting', + 'description': 'Install low energy lighting in -886 outlets', 'starting_u_value': None, + 'new_u_value': None, 'already_installed': False, 'sap_points': 2, + 'kwh_savings': -48508.5, 'energy_cost_savings': -12481.237049999998, + 'co2_equivalent_savings': -7.858377, + 'description_simulation': {'lighting-energy-eff': 'Very Good', + 'lighting-description': 'Low energy lighting in all fixed' + ' outlets', + 'low-energy-lighting': 100}, 'total': -3411.1000000000004, + 'labour_hours': 1, 'labour_days': 0.125, 'survey': True, + 'recommendation_id': '5_phase=4', 'efficiency': -1705.5500000000002, + 'heat_demand': np.float64(5.099999999999994)}], [ + {'type': 'heating', 'phase': 5, 'measure_type': 'time_temperature_zone_control', + 'parts': [], + 'description': 'Upgrade heating controls to Smart Thermostats, room sensors and ' + 'smart radiator valves (time & temperature zone control)', + 'total': 739.576, 'subtotal': 700.48, 'vat': 39.096000000000004, + 'labour_hours': 3.6199999999999997, 'labour_days': np.float64(1.0), + 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(2.9), + 'already_installed': False, 'simulation_config': { + 'thermostatic_control_ending': 'time and temperature zone control', + 'switch_system_ending': None, 'trvs_ending': None, + 'mainheatc_energy_eff_ending': 'Very Good'}, 'description_simulation': { + 'mainheatcont-description': 'Time and temperature zone control', + 'mainheatc-energy-eff': 'Very Good'}, 'recommendation_id': '6_phase=5', + 'efficiency': 739.576, 'co2_equivalent_savings': np.float64(0.30000000000000027), + 'heat_demand': np.float64(6.599999999999994), + 'kwh_savings': np.float64(876.8000000000002), + 'energy_cost_savings': np.float64(65.29581176470589)}], [ + {'phase': 6, 'parts': [], 'type': 'secondary_heating', + 'measure_type': 'secondary_heating', + 'description': 'Remove the secondary heating system', 'starting_u_value': None, + 'new_u_value': None, 'sap_points': np.float64(3.6), 'already_installed': False, + 'total': 30.0, 'subtotal': 25.0, 'vat': 5.0, 'labour_hours': 3.0, + 'labour_days': np.float64(1.0), + 'simulation_config': {'secondheat_description_ending': 'None'}, + 'description_simulation': {'secondheat-description': 'None'}, + 'recommendation_id': '7_phase=6', 'efficiency': 30.0, + 'co2_equivalent_savings': np.float64(0.10000000000000009), + 'heat_demand': np.float64(15.400000000000006), + 'kwh_savings': np.float64(196.29999999999927), + 'energy_cost_savings': np.float64(14.61857647058821)}], [ + {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', + 'description': 'Install a 4.0 kilowatt-peak (kWp) solar panel system.', + 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(13.0), + 'already_installed': False, 'total': 6013.139999999999, 'subtotal': 5010.95, 'vat': 0, + 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(65.0), + 'has_battery': False, 'initial_ac_kwh_per_year': np.float64(4081.7132614999996), + 'description_simulation': {'photo-supply': np.float64(65.0)}, + 'recommendation_id': '8_phase=7', 'efficiency': np.float64(462.54923076923075), + 'co2_equivalent_savings': np.float64(0.47347873833399995), + 'heat_demand': np.float64(88.69999999999999), + 'kwh_savings': np.float64(2040.8566307499998), + 'energy_cost_savings': np.float64(525.1124110919749)}, + {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', + 'description': 'Install a 4.0 kilowatt-peak (kWp) solar panel system, with a battery.', + 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(13.0), + 'already_installed': False, 'total': 10537.008, 'subtotal': 8780.84, 'vat': 0, + 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(65.0), + 'has_battery': True, 'initial_ac_kwh_per_year': np.float64(4081.7132614999996), + 'description_simulation': {'photo-supply': np.float64(65.0)}, + 'recommendation_id': '9_phase=7', 'efficiency': np.float64(810.5390769230769), + 'co2_equivalent_savings': np.float64(0.6628702336675999), + 'heat_demand': np.float64(88.69999999999999), + 'kwh_savings': np.float64(2857.1992830499994), + 'energy_cost_savings': np.float64(735.1573755287648)}, + {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', + 'description': 'Install a 3.6 kilowatt-peak (kWp) solar panel system.', + 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(12.0), + 'already_installed': False, 'total': 5826.491999999999, 'subtotal': 4855.41, 'vat': 0, + 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(60.0), + 'has_battery': False, 'initial_ac_kwh_per_year': np.float64(3692.66794), + 'description_simulation': {'photo-supply': np.float64(60.0)}, + 'recommendation_id': '10_phase=7', 'efficiency': np.float64(485.54099999999994), + 'co2_equivalent_savings': np.float64(0.42834948104), + 'heat_demand': np.float64(83.69999999999999), 'kwh_savings': np.float64(1846.33397), + 'energy_cost_savings': np.float64(475.0617304809999)}, + {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', + 'description': 'Install a 3.6 kilowatt-peak (kWp) solar panel system, with a battery.', + 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(12.0), + 'already_installed': False, 'total': 10350.359999999999, 'subtotal': 8625.3, 'vat': 0, + 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(60.0), + 'has_battery': True, 'initial_ac_kwh_per_year': np.float64(3692.66794), + 'description_simulation': {'photo-supply': np.float64(60.0)}, + 'recommendation_id': '11_phase=7', 'efficiency': np.float64(862.5299999999999), + 'co2_equivalent_savings': np.float64(0.599689273456), + 'heat_demand': np.float64(83.69999999999999), 'kwh_savings': np.float64(2584.867558), + 'energy_cost_savings': np.float64(665.0864226734)}, + {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', + 'description': 'Install a 3.2 kilowatt-peak (kWp) solar panel system.', + 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(11.0), + 'already_installed': False, 'total': 5642.604, 'subtotal': 4702.17, 'vat': 0, + 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(55.0), + 'has_battery': False, 'initial_ac_kwh_per_year': np.float64(3300.5416548), + 'description_simulation': {'photo-supply': np.float64(55.0)}, + 'recommendation_id': '12_phase=7', 'efficiency': np.float64(512.964), + 'co2_equivalent_savings': np.float64(0.3828628319568), 'heat_demand': np.float64(78.3), + 'kwh_savings': np.float64(1650.2708274), + 'energy_cost_savings': np.float64(424.61468389001993)}, + {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', + 'description': 'Install a 3.2 kilowatt-peak (kWp) solar panel system, with a battery.', + 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(11.0), + 'already_installed': False, 'total': 10166.472, 'subtotal': 8472.06, 'vat': 0, + 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(55.0), + 'has_battery': True, 'initial_ac_kwh_per_year': np.float64(3300.5416548), + 'description_simulation': {'photo-supply': np.float64(55.0)}, + 'recommendation_id': '13_phase=7', 'efficiency': np.float64(924.2247272727273), + 'co2_equivalent_savings': np.float64(0.53600796473952), + 'heat_demand': np.float64(78.3), 'kwh_savings': np.float64(2310.3791583599996), + 'energy_cost_savings': np.float64(594.4605574460278)}, + {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', + 'description': 'Install a 2.8 kilowatt-peak (kWp) solar panel system.', + 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(9.0), + 'already_installed': False, 'total': 5458.727999999999, 'subtotal': 4548.94, 'vat': 0, + 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(45.0), + 'has_battery': False, 'initial_ac_kwh_per_year': np.float64(2907.1867812), + 'description_simulation': {'photo-supply': np.float64(45.0)}, + 'recommendation_id': '14_phase=7', 'efficiency': np.float64(606.5253333333333), + 'co2_equivalent_savings': np.float64(0.3372336666192), 'heat_demand': np.float64(64.0), + 'kwh_savings': np.float64(1453.5933906), + 'energy_cost_savings': np.float64(374.00957940138)}, + {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', + 'description': 'Install a 2.8 kilowatt-peak (kWp) solar panel system, with a battery.', + 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(9.0), + 'already_installed': False, 'total': 9982.596, 'subtotal': 8318.83, 'vat': 0, + 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(45.0), + 'has_battery': True, 'initial_ac_kwh_per_year': np.float64(2907.1867812), + 'description_simulation': {'photo-supply': np.float64(45.0)}, + 'recommendation_id': '15_phase=7', 'efficiency': np.float64(1109.1773333333333), + 'co2_equivalent_savings': np.float64(0.47212713326688), + 'heat_demand': np.float64(64.0), 'kwh_savings': np.float64(2035.03074684), + 'energy_cost_savings': np.float64(523.6134111619319)}, + {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', + 'description': 'Install a 2.4 kilowatt-peak (kWp) solar panel system.', + 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(8.0), + 'already_installed': False, 'total': 5274.852, 'subtotal': 4395.71, 'vat': 0, + 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(40.0), + 'has_battery': False, 'initial_ac_kwh_per_year': np.float64(2510.25188), + 'description_simulation': {'photo-supply': np.float64(40.0)}, + 'recommendation_id': '16_phase=7', 'efficiency': np.float64(659.3565), + 'co2_equivalent_savings': np.float64(0.29118921808), 'heat_demand': np.float64(54.3), + 'kwh_savings': np.float64(1255.12594), + 'energy_cost_savings': np.float64(322.94390436199996)}, + {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', + 'description': 'Install a 2.4 kilowatt-peak (kWp) solar panel system, with a battery.', + 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(8.0), + 'already_installed': False, 'total': 9798.72, 'subtotal': 8165.6, 'vat': 0, + 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(40.0), + 'has_battery': True, 'initial_ac_kwh_per_year': np.float64(2510.25188), + 'description_simulation': {'photo-supply': np.float64(40.0)}, + 'recommendation_id': '17_phase=7', 'efficiency': np.float64(1224.84), + 'co2_equivalent_savings': np.float64(0.40766490531199995), + 'heat_demand': np.float64(54.3), 'kwh_savings': np.float64(1757.1763159999998), + 'energy_cost_savings': np.float64(452.1214661067999)}, + {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', + 'description': 'Install a 2.0 kilowatt-peak (kWp) solar panel system.', + 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(7.0), + 'already_installed': False, 'total': 5090.976, 'subtotal': 4242.48, 'vat': 0, + 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(35.0), + 'has_battery': False, 'initial_ac_kwh_per_year': np.float64(2096.682636), + 'description_simulation': {'photo-supply': np.float64(35.0)}, + 'recommendation_id': '18_phase=7', 'efficiency': np.float64(727.2822857142856), + 'co2_equivalent_savings': np.float64(0.243215185776), 'heat_demand': np.float64(48.5), + 'kwh_savings': np.float64(1048.341318), + 'energy_cost_savings': np.float64(269.7382211214)}, + {'phase': 7, 'parts': [], 'type': 'solar_pv', 'measure_type': 'solar_pv', + 'description': 'Install a 2.0 kilowatt-peak (kWp) solar panel system, with a battery.', + 'starting_u_value': None, 'new_u_value': None, 'sap_points': np.float64(7.0), + 'already_installed': False, 'total': 9614.844, 'subtotal': 8012.369999999999, 'vat': 0, + 'labour_hours': 48, 'labour_days': 2, 'photo_supply': np.float64(35.0), + 'has_battery': True, 'initial_ac_kwh_per_year': np.float64(2096.682636), + 'description_simulation': {'photo-supply': np.float64(35.0)}, + 'recommendation_id': '19_phase=7', 'efficiency': np.float64(1373.5491428571427), + 'co2_equivalent_savings': np.float64(0.3405012600864), 'heat_demand': np.float64(48.5), + 'kwh_savings': np.float64(1467.6778451999999), + 'energy_cost_savings': np.float64(377.6335095699599)}] ] - if _find_measure(input_measures, "solar_pv"): - for heat_type, ref in solar_heating_combos: - if _find_measure(input_measures, heat_type): - if remaining_insulation_type: - for insulation_measure in remaining_insulation_type: - funding_paths.append( - [{"AND": ["solar_pv", heat_type, insulation_measure], - "reference": f"{ref[:-5]}+{insulation_measure}:eco4"}] # keeps naming consistent - ) - else: - funding_paths.append([{"AND": ["solar_pv", heat_type], "reference": ref}]) - - # 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 - - # 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 single_heating_measures: - if _find_measure(input_measures, heating_upgrade): - if remaining_insulation_type: - for insulation_measure in remaining_insulation_type: - path = [ - { - "AND": [heating_upgrade, insulation_measure], - "reference": f"{measure_references[heating_upgrade]}+{insulation_measure}:eco4" - } - ] - funding_paths.append(path) - else: - funding_paths.append( - [{"AND": [heating_upgrade], "reference": f"{measure_references[heating_upgrade]}:eco4"}] - ) - - return funding_paths + return recs -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 _attach_costs_and_uplifts(recs, funding, p): + """Mimic what your script did: add cost fields & innovation uplift.""" + out = deepcopy(recs) + for group in out: + for r in group: + if r["type"] in ["mechanical_ventilation", "low_energy_lighting", "secondary_heating"]: + r["innovation_uplift"] = 0 + continue + r["uplift"] = 0.0 # fixed for determinism in test + r["innovation_uplift"] = funding.get_innovation_uplift( + measure=r, + starting_sap=55, + floor_area=70.0, + 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="Very Good", + ) + # the optimiser_functions.prepare_input_measures will translate these to input format; but + # for safety add explicit cost fields some downstream code expects: + r["total"] = float(r["total"]) + return out -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 - to meet minimum insulation requirements, particularly when a heating system is recommended. +def _to_input_measures(recs, p): + """Use your own helper so we test the full pipeline.""" + property_measure_types = {rec["type"] for grp in recs for rec in grp} + needs_ventilation = any( + x in property_measure_types for x in optimiser_functions.assumptions.measures_needing_ventilation + ) and not getattr(p, "has_ventilation", False) - 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 housing_type: - :return: - """ - # 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. - - # We determine which insulation measures need to be included - wall_insulation_measures = [ - "internal_wall_insulation", "external_wall_insulation", "cavity_wall_insulation", - "extension_cavity_wall_insulation" - ] - 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 = [] - for insulation_measure in wall_insulation_measures + roof_insulation_measures: - if _find_measure(input_measures, insulation_measure): - remaining_insulation_type.append(insulation_measure) - - remaining_insulation_type = list(set(remaining_insulation_type)) - - funding_paths = [] - - 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 = [] - 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 measure["innovation_uplift"] and measure["type"] in ( - remaining_insulation_type + other_gbis_insulation_measures - ): - input_gbis_measures_innovation.append([measure]) - - funding_paths = _make_solar_heating_funding_paths( - 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, input_measures_innovation - - if housing_type == "Private": - # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - # EWI or IWI - # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - # 1) The package must include EWI or IWI if the property is private rental sector - # We check if we have any EWI or IWI measures available - ewi_or_iwi = [{"OR": []}] - reference_measures = [] - # If we have EWI we add it in - if _find_measure(input_measures, "external_wall_insulation"): - ewi_or_iwi[0]["OR"].append("external_wall_insulation") - reference_measures.append("ewi") - - if _find_measure(input_measures, "internal_wall_insulation"): - ewi_or_iwi[0]["OR"].append("internal_wall_insulation") - reference_measures.append("iwi") - - if ewi_or_iwi[0]["OR"]: - ewi_or_iwi[0]["reference"] = "+".join(reference_measures) + ":eco4" - funding_paths.append(ewi_or_iwi) - - funding_paths = _make_solar_heating_funding_paths( - p, input_measures, funding_paths, remaining_insulation_type, housing_type + # goal="Increasing EPC", add_uplift=True for Social path + return optimiser_functions.prepare_input_measures( + recs, goal="Increasing EPC", needs_ventilation=needs_ventilation, funding=True ) - # 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]) - funding_paths = _make_generic_gbis_funding_paths(input_gbis_measures, funding_paths) - - return funding_paths, input_measures +def _types_of(picked_items): + return {item["type"] for item in picked_items} -# ---- main wrapper around your optimiser ---------------------------------- +def test_social_fabric_only_returns_only_fabric_types(p, funding, property_recommendations, monkeypatch): + # 1) prepare data like your script + recs = _attach_costs_and_uplifts(property_recommendations, funding, p) + input_measures = _to_input_measures(recs, p) -# Run inputs: -target_gain = 18.5 + # 2) run optimiser wrapper (budget and target_gain can be modest for the test) + budget = 30000.0 + target_gain = 8.0 - -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): - """Return True if fixed selection includes a heating/PV measure but no required insulation.""" - picked_types = {opt["type"] for (_, _, opt) in fixed} - - def has_any(substrs): - return any(any(s in t for s in substrs) for t in picked_types) - - # heating (incl. PV) flags - is_heating = has_any([ - "air_source_heat_pump", - "high_heat_retention_storage_heater", - "boiler_upgrade", - "electric_boiler", - "time_temperature_zone_control", - "secondary_heating", - "solar_pv", # PV treated as heating for MIR - ]) - - # MIR insulation (the ones you’re using in path construction) - has_insul = has_any([ - "external_wall_insulation", - "internal_wall_insulation", - "cavity_wall_insulation", - "extension_cavity_wall_insulation", - "loft_insulation", - "flat_roof_insulation", - "room_roof_insulation", - ]) - - return is_heating and not has_insul - - -# Treat "type" like "external_wall_insulation+mechanical_ventilation" → "external_wall_insulation" -def _base_type(s: str) -> str: - return s.split("+", 1)[0] - - -def _filter_measures_by_types(input_measures, allowed_types): - """ - Keep only groups that have ≥1 allowed option; inside each group keep only allowed options. - """ - allowed_set = set(allowed_types) - filtered = [] - for group in input_measures: - kept_opts = [opt for opt in group if _base_type(opt["type"]) in allowed_set] - if kept_opts: - filtered.append(kept_opts) - return filtered - - -def _is_eligible_funding_package(scheme, starting_sap, total_gain): - if scheme == "eco4": - # We check if we meet the upgrade requirements - # If the property is an E or above, we need to upgrade to a C or above - if starting_sap >= 39: # ie. EPC C or above - return starting_sap + total_gain >= 69 - - if scheme == "gbis": - # GBIS is a fixed measure only, so we don't check the 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) - """ - - 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, + solutions = optimise_with_funding_paths( + p=p, + input_measures=input_measures, + housing_type="Social", budget=budget, - sub_target_gain=target_gain + target_gain=target_gain, + funding=funding ) - 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) + # 3) basic shape assertions + assert isinstance(solutions, pd.DataFrame) + assert not solutions.empty - # 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 - if housing_type == "Social": - funding_paths = ( - [ - { - 'reference': 'fabric-only:eco4', - "allowed_types": WALL_INSULATION_MEASURES + ROOF_INSULATION_MEASURES + - ECO4_ELIGIBILE_FABRIC_MEASURES - } - ] + funding_paths - ) + # 4) find the fabric-only ECO4 row + fabric_rows = solutions[ + solutions["path"].apply(lambda x: isinstance(x, dict) and x.get("reference") == "fabric-only:eco4")] + assert not fabric_rows.empty, "Expected a fabric-only:eco4 solution for Social tenure" - for path_spec in funding_paths: + # 5) ensure only fabric measure types are present in that solution + picked_types = _types_of(fabric_rows.iloc[0]["items"]) + assert picked_types == {'internal_wall_insulation+mechanical_ventilation', + 'suspended_floor_insulation'}, "incorrect types selected" - # 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(optimisation_input_measures, path_spec["allowed_types"]) - if not sub_measures: - continue + # 6) respect budget + assert fabric_rows.iloc[0]["total_cost"] <= budget + 1e-9 - picked, sub_cost, sub_gain = run_optimizer( - sub_measures, - budget=budget, # no fixed items; budget unchanged - sub_target_gain=target_gain - ) - - if picked is None: - continue - - scheme = _path_scheme([path_spec]) - - solutions.append( - { - "fixed_ids": [], - "items": picked, - "total_cost": sub_cost, - "total_gain": sub_gain, - "path": path_spec, - "scheme": scheme, - "is_eligible": _is_eligible_funding_package(scheme, p.data["current-energy-efficiency"], sub_gain) - } - ) - - continue - - # 1) expand fixed selections for this path - fixed_selections = expand_funding_path(optimisation_input_measures, path_spec) if path_spec else [[]] - if not fixed_selections: - continue - - for fixed in fixed_selections: - - if violates_min_insulation(fixed): - # We log an error and skip this - we should not see any errors but we can probably get a reasonable - # outcome for the end user without a complete termination of the process - 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 = 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 - # 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 - # may already be over budget) and the fixed gain (which may not be achievable) - picked, sub_cost, sub_gain = run_optimizer( - sub_measures, - 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_cost = fixed_cost + sub_cost - 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({ - "fixed_ids": fixed_ids, - "items": total_picks, - "total_cost": total_cost, - "total_gain": total_gain, - "path": path_spec, - "scheme": scheme, - "is_eligible": _is_eligible_funding_package(scheme, p.data["current-energy-efficiency"], total_gain) - }) - - solutions = pd.DataFrame(solutions) - - # Given the scheme, we now check if the packages are eligible. If they *are* eligible, but they don't meet the - # final upgrade target, we then look to perform a final optimisation pass to meet the target gain. - solutions["meets_upgrade_target"] = solutions["total_gain"] >= target_gain - - # If we have packages that are fundable, but do not meet the upgrade target, we can run a final optimisation pass - if not solutions[solutions["is_eligible"] & ~solutions["meets_upgrade_target"]].empty: - raise NotImplementedError("Implement me") - - return solutions - - -# ---- helpers ------------------------------------------------------------- - - -def sum_cost_gain(items): - c = sum(float(x['cost']) for x in items) - g = sum(float(x['gain']) for x in items) - return c, g - - -# ---- candidate expansion ------------------------------------------------- -def type_matches(option_type: str, required: str) -> bool: - # substring match so "external_wall_insulation+mechanical_ventilation" satisfies "external_wall_insulation" - return required in option_type - - -def candidates_for_type(input_measures, required_type): - """ - Return a list of (gi, oi, opt) where opt['type'] contains required_type. - gi = group index, oi = option index inside that group. - """ - cands = [] - for gi, group in enumerate(input_measures): - for oi, opt in enumerate(group): - if type_matches(opt["type"], required_type): - cands.append((gi, oi, opt)) - return cands - - -def iter_or_candidates(input_measures, types_list): - """ - For OR: pick exactly ONE candidate whose type matches ANY in types_list. - Return a list of dicts: {"fixed": [(gi, oi, opt)]} - """ - union = [] - seen_ids = set() - for t in types_list: - for tup in candidates_for_type(input_measures, t): - # de-dupe by the option id so the same physical option (with multi-type name) isn’t repeated - if tup[2]["id"] not in seen_ids: - seen_ids.add(tup[2]["id"]) - union.append(tup) - return [{"fixed": [t]} for t in union] - - -def iter_and_candidates(input_measures, types_list): - """ - For AND: we must cover ALL required types. - We allow a single option to satisfy multiple types. - We build a simple product but collapse duplicates by (gi, oi). - """ - # Build candidate pools per required type - pools = [candidates_for_type(input_measures, t) for t in types_list] - if any(len(pool) == 0 for pool in pools): - return [] # impossible to satisfy AND - - # Start with one empty selection; accumulate per pool - selections = [[]] # each selection is a list of (gi, oi, opt) - for pool in pools: - new_selections = [] - for sel in selections: - for cand in pool: - # Try adding cand; collapse duplicates by (gi,oi) - gi, oi, opt = cand - replaced = False - conflict = False - merged = [] - for (sgi, soi, sopt) in sel: - if (sgi, soi) == (gi, oi): - # same exact option already in selection (satisfies another required type) – keep one - replaced = True - # keep the existing one (identical) - merged.append((sgi, soi, sopt)) - else: - merged.append((sgi, soi, sopt)) - if not replaced: - merged.append(cand) - if not conflict: - new_selections.append(merged) - selections = new_selections - if not selections: - return [] - - # After accumulation, we may still have duplicate groups with different options (conflict). Drop those. - cleaned = [] - for sel in selections: - seen_by_group = {} - ok = True - for gi, oi, opt in sel: - if gi in seen_by_group and seen_by_group[gi] != oi: - # same group, different option -> conflict for AND; invalid selection - ok = False - break - seen_by_group[gi] = oi - if ok: - # ensure stable order and unique by (gi,oi) - uniq = {} - for gi, oi, opt in sel: - uniq[(gi, oi)] = opt - cleaned.append([(gi, oi, opt) for (gi, oi), opt in uniq.items()]) - return [{"fixed": c} for c in cleaned] - - -def expand_funding_path(input_measures, path_spec): - """ - path_spec is a list of elements; each element is either: - {"OR": [type1, type2, ...], "reference": "..."} or - {"AND": [type1, type2, ...], "reference": "..."} - We cross-product across elements (all required), and produce selections as lists of (gi, oi, opt). - """ - selections = [[]] # list[list[(gi,oi,opt)]] - for elem in path_spec: - if "OR" in elem: - cands = iter_or_candidates(input_measures, elem["OR"]) - elif "AND" in elem: - cands = iter_and_candidates(input_measures, elem["AND"]) - else: - raise ValueError("unknown path element; expected 'OR' or 'AND'") - - if not cands: - return [] - - new_selections = [] - for base in selections: - for cand in cands: - # merge base + cand["fixed"], collapsing duplicate same-option picks - combined = list(base) - # reject if combined picks two different options from the same group - groups_to_oi = {(gi,): oi for gi, oi, _ in combined} # temporary; we’ll refactor below - conflict = False - # simpler: build a dict by group -> (oi, opt), conflict if group exists with different oi - gmap = {gi: (oi, opt) for gi, oi, opt in combined} - for gi, oi, opt in cand["fixed"]: - if gi in gmap: - prev_oi, _ = gmap[gi] - if prev_oi != oi: - conflict = True - break - gmap[gi] = (oi, opt) - if conflict: - continue - # back to list - merged = [(gi, oi, opt) for gi, (oi, opt) in gmap.items()] - new_selections.append(merged) - selections = new_selections - if not selections: - return [] - - # Final tidy: ensure no duplicate groups with different options (already protected), keep stable ordering - deduped = [] - for sel in selections: - gmap = {} - for gi, oi, opt in sel: - # keep the first occurrence - if gi not in gmap: - gmap[gi] = (oi, opt) - else: - # same group, different oi would have been filtered; if same oi, ignore duplicate - pass - deduped.append([(gi, oi, opt) for gi, (oi, opt) in gmap.items()]) - return deduped - - -# ---- tiny utilities ---------------------------------------------------------- - -def parse_types(t): - # e.g. "external_wall_insulation+mechanical_ventilation" -> {"external_wall_insulation","mechanical_ventilation"} - return set(map(str.strip, t.split("+"))) if isinstance(t, str) else set() - - -def includes_heating(opt_types): - return any(x in opt_types for x in { - "air_source_heat_pump", - "high_heat_retention_storage_heater", - "time_temperature_zone_control", # controls count as a heating measure in your pipeline - "solar_pv" # you treat PV as heating for funding logic - }) - - -def contributes_min_insulation(opt_types): - # MIR satisfiers you mentioned (extend as needed) - return any(x in opt_types for x in { - "external_wall_insulation", - "internal_wall_insulation", - "loft_insulation", - "cavity_wall_insulation", - }) - - -def run_optimizer(input_measures, budget=None, sub_target_gain=None, allow_slack=False): - """ - Thin wrapper over your optimisers. - Returns: list[dict] selected_options - """ - if budget is not None: - opt = GainOptimiser( - input_measures, max_cost=budget, max_gain=(sub_target_gain or float("inf")), - allow_slack=allow_slack - ) - else: - if sub_target_gain is None: - raise ValueError("Either budget or target_gain must be provided.") - opt = CostOptimiser(sub_measures, min_gain=sub_target_gain) - - opt.setup() - opt.solve() - cost = sum([x["cost"] for x in opt.solution]) - return opt.solution, cost, opt.solution_gain + # (optional) ensure unfunded baseline also appears + unfunded_rows = solutions[ + solutions["path"].apply(lambda x: isinstance(x, dict) and x.get("reference") == "unfunded:all")] + assert not unfunded_rows.empty