junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-08 11:17:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

implemented unfunded route

Browse source
This commit is contained in:
Khalim Conn-Kowlessar 2025-08-13 14:10:37 +01:00
parent 213c0f5600
commit 50de86e379
1 changed files with 107 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

125
recommendations/tests/test_optimisers.py
View file

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