mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
handling large floor area
This commit is contained in:
Khalim Conn-Kowlessar
parent
ff5bc2f834
commit
90c5f12671
4 changed files with 289 additions and 98 deletions
|
|
@ -555,7 +555,9 @@ class HeatingRecommender:
|
|||
for kw in models_kw:
|
||||
if kw >= target:
|
||||
return kw
|
||||
return None
|
||||
|
||||
# Return the largest
|
||||
return max(models_kw)
|
||||
|
||||
def recommend_air_source_heat_pump(self, phase, has_cavity_or_loft_recommendations, _return=False):
|
||||
"""
|
||||
|
|
@ -586,7 +588,15 @@ class HeatingRecommender:
|
|||
)
|
||||
ashp_size = self.pick_model(estimated_load)
|
||||
|
||||
ashp_costs = self.costs.air_source_heat_pump(ashp_size)
|
||||
number_heated_rooms = self._estimate_n_heated_rooms()
|
||||
# We now adjust this depending on the floor area to get number of communcal rooms (e.g. hallways)
|
||||
communal_heated_rooms = self._estimate_n_communal_heated_rooms()
|
||||
|
||||
ashp_costs = self.costs.air_source_heat_pump(
|
||||
ashp_size,
|
||||
number_heated_rooms=number_heated_rooms + communal_heated_rooms,
|
||||
total_floor_area=self.property.floor_area
|
||||
)
|
||||
if non_intrusive_recommendation:
|
||||
# Update with non-intrusive recommendation
|
||||
if non_intrusive_recommendation.get("cost"):
|
||||
|
|
@ -907,6 +917,56 @@ class HeatingRecommender:
|
|||
|
||||
return already_has_hhr and already_has_hhr_contols
|
||||
|
||||
def _estimate_n_heated_rooms(self):
|
||||
# If the property is off-gas and has no heating system in place, the number of heated rooms will actually
|
||||
# be 0, so we use the number of rooms as the figure
|
||||
number_heated_rooms = (
|
||||
self.property.data["number-heated-rooms"] if self.property.data["number-heated-rooms"] > 0
|
||||
else (
|
||||
self.property.number_of_rooms - 1 if self.property.number_of_rooms > 1 else
|
||||
self.property.number_of_rooms
|
||||
)
|
||||
)
|
||||
# To be conservative, we adjust if we still have 1 room
|
||||
if (number_heated_rooms == 1) and (self.property.number_of_rooms > 2):
|
||||
number_heated_rooms = self.property.number_of_rooms - 1
|
||||
|
||||
return number_heated_rooms
|
||||
|
||||
def _estimate_n_communal_heated_rooms(self) -> int:
|
||||
"""
|
||||
Estimate number of communal circulation rooms (hallways / landings) that may reasonably contain a heater
|
||||
"""
|
||||
|
||||
# Base assumptions
|
||||
base_by_type = {
|
||||
"Flat": 1,
|
||||
"Maisonette": 1,
|
||||
"Bungalow": 1,
|
||||
"House": 2,
|
||||
}
|
||||
|
||||
# Fallback if property type unknown
|
||||
base = base_by_type.get(self.property.data["property-type"], 1)
|
||||
|
||||
# Area-based adjustments
|
||||
if self.property.data["property-type"] in ("Flat", "Maisonette"):
|
||||
if self.property.floor_area > 90:
|
||||
return base + 1 # duplex or very large flat
|
||||
return base
|
||||
|
||||
if self.property.data["property-type"] == "Bungalow":
|
||||
if self.property.floor_area > 100:
|
||||
return base + 1 # secondary corridor
|
||||
return base
|
||||
|
||||
if self.property.data["property-type"] == "House":
|
||||
if self.property.floor_area > 140:
|
||||
return base + 1 # extra landing / circulation
|
||||
return base
|
||||
|
||||
return base
|
||||
|
||||
def recommend_hhr_storage_heaters(self, phase, system_change, heating_controls_only, _return=False):
|
||||
"""
|
||||
We will recommend upgrading to a high heat retention storage system, if the current system is not already
|
||||
|
|
@ -1010,18 +1070,7 @@ class HeatingRecommender:
|
|||
else:
|
||||
heating_simulation_config["hot_water_energy_eff_ending"] = self.property.data["hot-water-energy-eff"]
|
||||
|
||||
# If the property is off-gas and has no heating system in place, the number of heated rooms will actually
|
||||
# be 0, so we use the number of rooms as the figure
|
||||
number_heated_rooms = (
|
||||
self.property.data["number-heated-rooms"] if self.property.data["number-heated-rooms"] > 0
|
||||
else (
|
||||
self.property.number_of_rooms - 1 if self.property.number_of_rooms > 1 else
|
||||
self.property.number_of_rooms
|
||||
)
|
||||
)
|
||||
# To be conservative, we adjust if we still have 1 room
|
||||
if (number_heated_rooms == 1) and (self.property.number_of_rooms > 2):
|
||||
number_heated_rooms = self.property.number_of_rooms - 1
|
||||
number_heated_rooms = self._estimate_n_heated_rooms()
|
||||
|
||||
# We focus on the 700 watt product
|
||||
hhrsh_product = next((x for x in self.hhrsh_products if x["size"] == 700), {})
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ In the future, we will adapt this into a class-based structure to allow for more
|
|||
from copy import deepcopy
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
from itertools import product
|
||||
|
||||
from backend.app.plan.schemas import (
|
||||
WALL_INSULATION_MEASURES, ROOF_INSULATION_MEASURES, ECO4_ELIGIBILE_FABRIC_MEASURES
|
||||
|
|
@ -587,6 +588,218 @@ def optimise_with_funding_paths(
|
|||
return solutions
|
||||
|
||||
|
||||
def build_heat_pump_paths(
|
||||
remaining_wall_measures,
|
||||
remaining_roof_measures,
|
||||
):
|
||||
"""
|
||||
Build AND-paths using cartesian products.
|
||||
|
||||
Rules:
|
||||
- Always include air_source_heat_pump
|
||||
- Choose 1 wall measure if any exist
|
||||
- Choose 1 roof measure if any exist
|
||||
"""
|
||||
|
||||
# If a category is empty, use [None] so product still works
|
||||
wall_choices = remaining_wall_measures or [None]
|
||||
roof_choices = remaining_roof_measures or [None]
|
||||
|
||||
paths = []
|
||||
|
||||
for wall, roof in product(wall_choices, roof_choices):
|
||||
parts = []
|
||||
|
||||
if wall is not None:
|
||||
parts.append(wall)
|
||||
if roof is not None:
|
||||
parts.append(roof)
|
||||
|
||||
parts.append("air_source_heat_pump")
|
||||
|
||||
paths.append({"AND": parts})
|
||||
|
||||
return paths
|
||||
|
||||
|
||||
def exclude_measure_types(input_measures, excluded_types):
|
||||
excluded = set(excluded_types)
|
||||
filtered = []
|
||||
|
||||
for group in input_measures:
|
||||
kept = [
|
||||
opt for opt in group
|
||||
if opt["type"] not in excluded
|
||||
]
|
||||
if kept:
|
||||
filtered.append(kept)
|
||||
|
||||
return filtered
|
||||
|
||||
|
||||
def optimise_with_scenarios(
|
||||
input_measures,
|
||||
budget=None,
|
||||
target_gain=None,
|
||||
enforce_heat_pump_insulation=True,
|
||||
enforce_fabric_first=False
|
||||
):
|
||||
"""
|
||||
Scenario-based optimiser (funding-agnostic).
|
||||
|
||||
Currently implemented scenarios:
|
||||
1) With air source heat pump AND required insulation
|
||||
"""
|
||||
|
||||
solutions = []
|
||||
paths = []
|
||||
# Produce the unique list of measure types
|
||||
all_measure_types = []
|
||||
for inputs in input_measures:
|
||||
all_measure_types.extend([x["type"] for x in inputs])
|
||||
all_measure_types = list(set(all_measure_types))
|
||||
|
||||
if enforce_fabric_first:
|
||||
# If this is true, it means we only want to consider a fabric first approach. This means that
|
||||
# - We treat the fabric of the house first
|
||||
# - Only once the fabric has been upgraded, do we consider heating upgrades
|
||||
|
||||
# This should be wall insulation, roof insulation, floor insulation and windows
|
||||
fabric_measures = WALL_INSULATION_MEASURES + ROOF_INSULATION_MEASURES + ECO4_ELIGIBILE_FABRIC_MEASURES
|
||||
|
||||
fabric_only_measures = [[opt for opt in group if opt["type"] in fabric_measures] for group in input_measures]
|
||||
fabric_only_measures = [g for g in fabric_only_measures if g]
|
||||
|
||||
if not fabric_only_measures:
|
||||
# If we have no fabric measures, it means the work has already been done and we can proceed
|
||||
# straight to heating optimisation
|
||||
picked_fabric, fabric_cost, fabric_gain = [], 0, 0
|
||||
else:
|
||||
picked_fabric, fabric_cost, fabric_gain = run_optimizer(
|
||||
input_measures=fabric_only_measures,
|
||||
budget=budget,
|
||||
sub_target_gain=target_gain,
|
||||
# If we can achieve the target gain with just insulation measures, we're done
|
||||
)
|
||||
|
||||
picked_fabric_types = {m["type"] for m in picked_fabric}
|
||||
|
||||
remaining_measures = []
|
||||
for group in input_measures:
|
||||
kept = [m for m in group if m["type"] not in picked_fabric_types]
|
||||
if kept:
|
||||
remaining_measures.append(kept)
|
||||
|
||||
picked_extra, extra_cost, extra_gain = run_optimizer(
|
||||
remaining_measures,
|
||||
budget=budget - fabric_cost if budget is not None else None,
|
||||
sub_target_gain=(
|
||||
target_gain - fabric_gain
|
||||
if target_gain is not None
|
||||
else None
|
||||
)
|
||||
)
|
||||
|
||||
if picked_extra is None:
|
||||
picked_extra, extra_cost, extra_gain = [], 0, 0
|
||||
|
||||
solutions.append({
|
||||
"scenario": "fabric_first",
|
||||
"items": picked_fabric + picked_extra,
|
||||
"fixed_items": picked_fabric,
|
||||
"total_cost": fabric_cost + extra_cost,
|
||||
"total_gain": fabric_gain + extra_gain,
|
||||
})
|
||||
return solutions
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Scenario 1: Air source heat pump with required insulation
|
||||
# ------------------------------------------------------------------
|
||||
if enforce_heat_pump_insulation:
|
||||
# Wall measures could be IWI or EWI
|
||||
remaining_wall_measures = [x for x in all_measure_types if x in WALL_INSULATION_MEASURES]
|
||||
remaining_roof_measures = [x for x in all_measure_types if x in ROOF_INSULATION_MEASURES]
|
||||
|
||||
# Mandatory structure:
|
||||
# - must include ASHP
|
||||
# - must include >=1 wall insulation (if still needed)
|
||||
# - must include >=1 roof insulation (if still needed)
|
||||
# We need all of the combinations of remaining wall and remaining roof measures
|
||||
heat_pump_paths = build_heat_pump_paths(remaining_wall_measures, remaining_roof_measures)
|
||||
paths.extend(heat_pump_paths)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Scenario 2: Optimise without air source heat pump
|
||||
# ------------------------------------------------------------------
|
||||
# No special path; just exclude ASHP from options and allow us to optimise.
|
||||
measures_no_heat_pump = exclude_measure_types(input_measures, ["air_source_heat_pump"])
|
||||
|
||||
picked, total_cost, total_gain = run_optimizer(
|
||||
measures_no_heat_pump,
|
||||
budget=budget,
|
||||
sub_target_gain=target_gain,
|
||||
)
|
||||
|
||||
if picked is not None:
|
||||
solutions.append({
|
||||
"scenario": "no_heat_pump",
|
||||
"items": picked,
|
||||
"fixed_items": [],
|
||||
"total_cost": total_cost,
|
||||
"total_gain": total_gain,
|
||||
})
|
||||
|
||||
fixed_selections = expand_funding_path(input_measures, paths)
|
||||
|
||||
for fixed in fixed_selections:
|
||||
|
||||
# fixed = [(gi, oi, opt), ...]
|
||||
fixed_items = [opt for (_, _, opt) in fixed]
|
||||
fixed_groups = {gi for (gi, _, _) in fixed}
|
||||
|
||||
fixed_cost, fixed_gain = sum_cost_gain(fixed_items)
|
||||
|
||||
# Remaining measures (all other groups)
|
||||
remaining_measures = [
|
||||
grp for gi, grp in enumerate(input_measures)
|
||||
if gi not in fixed_groups
|
||||
]
|
||||
|
||||
# Optimise remaining measures
|
||||
if (
|
||||
target_gain is not None
|
||||
and fixed_gain >= target_gain
|
||||
):
|
||||
picked, sub_cost, sub_gain = [], 0, 0
|
||||
else:
|
||||
picked, sub_cost, sub_gain = run_optimizer(
|
||||
remaining_measures,
|
||||
budget=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_items = fixed_items + picked
|
||||
total_cost = fixed_cost + sub_cost
|
||||
total_gain = fixed_gain + sub_gain
|
||||
|
||||
solutions.append({
|
||||
"scenario": "heat_pump_with_insulation",
|
||||
"items": total_items,
|
||||
"fixed_items": fixed_items,
|
||||
"total_cost": total_cost,
|
||||
"total_gain": total_gain,
|
||||
})
|
||||
|
||||
return solutions
|
||||
|
||||
|
||||
# ---- helpers -------------------------------------------------------------
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ from recommendations.optimiser.CostOptimiser import CostOptimiser
|
|||
|
||||
|
||||
class TestPrepareInputMeasures:
|
||||
|
||||
def test_returns_expected_structure_without_ventilation(self):
|
||||
recs = [
|
||||
[ # loft insulation measure
|
||||
|
|
|
|||
|
|
@ -1,97 +1,14 @@
|
|||
import numpy as np
|
||||
# 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 recommendations.optimiser import optimiser_functions
|
||||
from recommendations.optimiser.funding_optimiser import optimise_with_funding_paths # wherever you defined it
|
||||
from recommendations.optimiser.funding_optimiser import optimise_with_funding_paths, build_heat_pump_paths
|
||||
from backend.Funding import Funding
|
||||
from backend.app.plan.schemas import WALL_INSULATION_MEASURES, ROOF_INSULATION_MEASURES, ECO4_ELIGIBILE_FABRIC_MEASURES
|
||||
|
||||
|
|
@ -799,3 +716,14 @@ def test_private_solid_wall_no_innovation_epc_d(p, funding, mock_project_scores_
|
|||
'partial_project_funding': 2300.1000000000004, 'partial_project_score': 135.3, 'total_uplift': 0.0,
|
||||
'total_uplift_score': 0.0
|
||||
}
|
||||
|
||||
|
||||
def test_build_heat_pump_paths():
|
||||
eg1 = build_heat_pump_paths([], ["loft_insulation"])
|
||||
|
||||
assert eg1 == [{'AND': ['loft_insulation', 'air_source_heat_pump']}]
|
||||
|
||||
eg2 = build_heat_pump_paths(["internal_wall_insulation", "external_wall_insulation"], ["loft_insulation"])
|
||||
|
||||
assert eg2 == [{'AND': ['internal_wall_insulation', 'loft_insulation', 'air_source_heat_pump']},
|
||||
{'AND': ['external_wall_insulation', 'loft_insulation', 'air_source_heat_pump']}]
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue