junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-30 13:10:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactored the handling of dual heating recommendations and fixing coverage of heating types in property class

Browse source
This commit is contained in:
Khalim Conn-Kowlessar 2025-11-14 22:20:03 +00:00
parent 3624b34dd0
commit 005c6b844a
3 changed files with 140 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
backend/Property.py
View file

@ -1221,7 +1221,8 @@ class Property:
None: "Natural Gas (Community Scheme)", None: "Natural Gas (Community Scheme)",
"mains gas": "Natural Gas (Community Scheme)", "mains gas": "Natural Gas (Community Scheme)",
"biomass": "Smokeless Fuel", "biomass": "Smokeless Fuel",
"electricity": "Electricity" "electricity": "Electricity",
"biogas": "Smokeless Fuel",
} }
if self.main_fuel["fuel_type"] in fuel_map: # We assume when None as it's unknown if self.main_fuel["fuel_type"] in fuel_map: # We assume when None as it's unknown
self.heating_energy_source = fuel_map[self.main_fuel["fuel_type"]] self.heating_energy_source = fuel_map[self.main_fuel["fuel_type"]]

66
backend/tests/test_integration.py
View file

@ -1,36 +1,36 @@
import ast # import ast
import json # import json
from copy import deepcopy from copy import deepcopy
from dataclasses import replace # from dataclasses import replace
from datetime import datetime # from datetime import datetime
import random import random
from tqdm import tqdm from tqdm import tqdm
import pandas as pd # import pandas as pd
import numpy as np import numpy as np
from etl.epc.Record import EPCRecord from etl.epc.Record import EPCRecord
from backend.SearchEpc import SearchEpc # from backend.SearchEpc import SearchEpc
from sqlalchemy.exc import IntegrityError, OperationalError # from sqlalchemy.exc import IntegrityError, OperationalError
from sqlalchemy.orm import sessionmaker # from sqlalchemy.orm import sessionmaker
from starlette.responses import Response # from starlette.responses import Response
from backend.app.config import get_settings, get_prediction_buckets # from backend.app.config import get_settings, get_prediction_buckets
from backend.app.db.connection import db_engine # from backend.app.db.connection import db_engine
from backend.app.db.functions.materials_functions import get_materials # from backend.app.db.functions.materials_functions import get_materials
from backend.app.db.functions.portfolio_functions import aggregate_portfolio_recommendations # from backend.app.db.functions.portfolio_functions import aggregate_portfolio_recommendations
from backend.app.db.functions.property_functions import ( # from backend.app.db.functions.property_functions import (
create_property, create_property_details_epc, create_property_targets, update_property_data, # create_property, create_property_details_epc, create_property_targets, update_property_data,
update_or_create_property_spatial_details # update_or_create_property_spatial_details
) # )
from backend.app.db.functions.recommendations_functions import ( # from backend.app.db.functions.recommendations_functions import (
create_plan, upload_recommendations, create_scenario # create_plan, upload_recommendations, create_scenario
) # )
from backend.app.db.functions.funding_functions import upload_funding # from backend.app.db.functions.funding_functions import upload_funding
from backend.app.db.functions.energy_assessment_functions import get_latest_assessment_by_uprn # from backend.app.db.functions.energy_assessment_functions import get_latest_assessment_by_uprn
from backend.app.db.models.portfolio import rating_lookup # from backend.app.db.models.portfolio import rating_lookup
from backend.app.plan.schemas import PlanTriggerRequest, WALL_INSULATION_MEASURES, ROOF_INSULATION_MEASURES from backend.app.plan.schemas import PlanTriggerRequest, WALL_INSULATION_MEASURES, ROOF_INSULATION_MEASURES
from backend.app.plan.utils import get_cleaned # from backend.app.plan.utils import get_cleaned
from backend.app.utils import sap_to_epc # from backend.app.utils import sap_to_epc
import backend.app.assumptions as assumptions import backend.app.assumptions as assumptions
from backend.ml_models.api import ModelApi from backend.ml_models.api import ModelApi
@ -41,13 +41,13 @@ from recommendations.optimiser.CostOptimiser import CostOptimiser
from recommendations.optimiser.GainOptimiser import GainOptimiser from recommendations.optimiser.GainOptimiser import GainOptimiser
import recommendations.optimiser.optimiser_functions as optimiser_functions import recommendations.optimiser.optimiser_functions as optimiser_functions
from recommendations.Recommendations import Recommendations from recommendations.Recommendations import Recommendations
from utils.logger import setup_logger # from utils.logger import setup_logger
from utils.s3 import read_dataframe_from_s3_parquet, read_csv_from_s3, read_excel_from_s3 # from utils.s3 import read_dataframe_from_s3_parquet, read_csv_from_s3, read_excel_from_s3
from backend.ml_models.Valuation import PropertyValuation # from backend.ml_models.Valuation import PropertyValuation
#
from etl.bill_savings.KwhData import KwhData # from etl.bill_savings.KwhData import KwhData
from etl.spatial.OpenUprnClient import OpenUprnClient # from etl.spatial.OpenUprnClient import OpenUprnClient
from etl.find_my_epc.RetrieveFindMyEpc import RetrieveFindMyEpc # from etl.find_my_epc.RetrieveFindMyEpc import RetrieveFindMyEpc
from backend.Funding import Funding from backend.Funding import Funding
from recommendations.optimiser.funding_optimiser import optimise_with_funding_paths from recommendations.optimiser.funding_optimiser import optimise_with_funding_paths
@ -91,7 +91,7 @@ costs_by_floor_area = costs_by_floor_area.groupby("current-energy-efficiency")[
].mean().reset_index() ].mean().reset_index()
sample_epc_data = epc_data[pd.to_datetime(epc_data["LODGEMENT_DATE"]) >= "2015-01-01"].drop_duplicates("UPRN").sample( sample_epc_data = epc_data[pd.to_datetime(epc_data["LODGEMENT_DATE"]) >= "2015-01-01"].drop_duplicates("UPRN").sample(
20000).reset_index(drop=True) 3000).reset_index(drop=True)
# TODO: In Property find_energy_sources, sort out biomass community heating - what fuel type # TODO: In Property find_energy_sources, sort out biomass community heating - what fuel type
# TODO: We might be able to remove find_energy_sources entirely and remove estimate_electrical_consumption. It's used # TODO: We might be able to remove find_energy_sources entirely and remove estimate_electrical_consumption. It's used

131
recommendations/HeatingRecommender.py
View file

@ -10,6 +10,9 @@ from etl.epc_clean.epc_attributes.MainheatAttributes import MainHeatAttributes
from etl.epc_clean.epc_attributes.HotWaterAttributes import HotWaterAttributes from etl.epc_clean.epc_attributes.HotWaterAttributes import HotWaterAttributes
from etl.epc_clean.epc_attributes.MainFuelAttributes import MainFuelAttributes from etl.epc_clean.epc_attributes.MainFuelAttributes import MainFuelAttributes
from recommendations.HeatingControlRecommender import HeatingControlRecommender from recommendations.HeatingControlRecommender import HeatingControlRecommender
from utils.logger import setup_logger
logger = setup_logger()
class HeatingRecommender: class HeatingRecommender:
@ -44,6 +47,22 @@ class HeatingRecommender:
] ]
} }
}, },
"Boiler and radiators, mains gas, electric underfloor heating": {
"boiler": {
"mainheating_description": "Boiler and radiators, mains gas, electric underfloor heating",
"recommendation_description": "Upgrade the existing boiler to a new, more efficient condensing "
"boiler. ",
"controls_suffix": "Manual charge controls"
},
# These are the heating types we need to produce a dual heating recommendation
"dual": {
"recommendation_description": "Upgrade the existing boiler to a new condensing boiler",
"types": [
# type 1
"boiler_upgrade",
]
}
},
"Portable electric heaters assumed for most rooms, room heaters, electric": { "Portable electric heaters assumed for most rooms, room heaters, electric": {
"hhr": { "hhr": {
"mainheating_description": "Electric storage heaters, radiators", "mainheating_description": "Electric storage heaters, radiators",
@ -127,7 +146,7 @@ class HeatingRecommender:
n_trues += 1 n_trues += 1
if n_trues > 2 or n_trues == 0: if n_trues > 2 or n_trues == 0:
raise Exception("Implement me") raise NotImplementedError("Implement me, zero or more than two heating systemss")
if n_trues == 1: if n_trues == 1:
return False return False
@ -917,9 +936,11 @@ class HeatingRecommender:
if self.property.main_heating_controls["clean_description"] != self.high_heat_retention_contols_desc: if self.property.main_heating_controls["clean_description"] != self.high_heat_retention_contols_desc:
if self.dual_heating: if self.dual_heating:
controls_prefix = self.DUAL_HEATING_DESCRIPTIONS[ controls_prefix = self._map_dual_heating_description(
self.property.main_heating["clean_description"] backup_map_to_description="current_controls",
]["hhr"]["controls_prefix"] output_type="controls_prefix",
recommendation_type="hhr"
)
if controls_prefix == "current_controls": if controls_prefix == "current_controls":
description_prefix = self.property.main_heating_controls["clean_description"] description_prefix = self.property.main_heating_controls["clean_description"]
@ -951,9 +972,11 @@ class HeatingRecommender:
# We check if the property has dual heating in place with a boiler and storage heaters # We check if the property has dual heating in place with a boiler and storage heaters
if self.dual_heating: if self.dual_heating:
new_heating_description = self.DUAL_HEATING_DESCRIPTIONS[ new_heating_description = self._map_dual_heating_description(
self.property.main_heating["clean_description"] backup_map_to_description="Electric storage heaters",
]["hhr"]["mainheating_description"] output_type="mainheating_description",
recommendation_type="hhr"
)
new_hot_water_description = self.property.hotwater["clean_description"] # We keep the hot water system new_hot_water_description = self.property.hotwater["clean_description"] # We keep the hot water system
else: else:
new_heating_description = "Electric storage heaters" new_heating_description = "Electric storage heaters"
@ -1010,10 +1033,12 @@ class HeatingRecommender:
product=hhrsh_product product=hhrsh_product
) )
if self.dual_heating: if self.dual_heating:
description = self.DUAL_HEATING_DESCRIPTIONS[ description = self._map_dual_heating_description(
self.property.main_heating["clean_description"] backup_map_to_description="Install high heat retention electric storage heaters with an appropriate "
]["hhr"]["recommendation_description"] "off-peak tariff.",
output_type="recommendation_description",
recommendation_type="hhr"
)
else: else:
description = "Install high heat retention electric storage heaters with an appropriate off-peak tariff." description = "Install high heat retention electric storage heaters with an appropriate off-peak tariff."
@ -1102,6 +1127,60 @@ class HeatingRecommender:
return max(num_heated_rooms * 1.5, 6) return max(num_heated_rooms * 1.5, 6)
def _map_dual_heating_description(
self, backup_map_to_description, output_type, recommendation_type
):
"""
Utility function to handle dual heating systems
:param backup_map_to_description:
:return:
"""
if backup_map_to_description not in [
# Recommendation descriptions - these are the textual descriptions shown in the front end
"Upgrade to a new condensing boiler.",
"Install high heat retention electric storage heaters with an appropriate off-peak tariff.",
# Simulation descriptions - this is the new EPC description we simulate with in the case
# of single heating
"Boiler and radiators, mains gas",
"Electric storage heaters",
# Suffixes allowed
"",
# Controls prefixes
"current_controls"
]:
raise ValueError(f"Invalid backup_map_to_description, given {backup_map_to_description}")
if output_type not in [
"recommendation_description",
"mainheating_description",
"controls_suffix",
"controls_prefix",
]:
raise ValueError(f"Invalid output_type, given {output_type}")
if recommendation_type not in [
"boiler",
]:
raise ValueError(f"Given invalid recommendation type {recommendation_type}")
# "Upgrade to a new condensing boiler."
if self.dual_heating:
# We check if we have a mapped description
if self.property.main_heating["clean_description"] not in self.DUAL_HEATING_DESCRIPTIONS:
logger.warning(
f"We have a dual heating system that hasn't been mapped, defaulting to single "
f"{self.property.main_heating['clean_description']}"
)
return backup_map_to_description
return self.DUAL_HEATING_DESCRIPTIONS[
self.property.main_heating["clean_description"]
][recommendation_type][output_type]
return backup_map_to_description
def recommend_boiler_upgrades(self, phase, system_change, exising_room_heaters): def recommend_boiler_upgrades(self, phase, system_change, exising_room_heaters):
""" """
This boiler recommendation will only recommend a like-for-like upgrade, since changing the system This boiler recommendation will only recommend a like-for-like upgrade, since changing the system
@ -1137,12 +1216,11 @@ class HeatingRecommender:
if has_inefficient_space_heating or has_inefficient_water: if has_inefficient_space_heating or has_inefficient_water:
if self.dual_heating: description = self._map_dual_heating_description(
description = self.DUAL_HEATING_DESCRIPTIONS[ backup_map_to_description="Upgrade to a new condensing boiler.",
self.property.main_heating["clean_description"] output_type="recommendation_description",
]["boiler"]["recommendation_description"] recommendation_type="boiler"
else: )
description = "Upgrade to a new condensing boiler."
new_heating_eff = ( new_heating_eff = (
"Good" if self.property.data["mainheat-energy-eff"] in ["Very Poor", "Poor", "Average"] "Good" if self.property.data["mainheat-energy-eff"] in ["Very Poor", "Poor", "Average"]
@ -1167,13 +1245,12 @@ class HeatingRecommender:
if system_change: if system_change:
# Installation of a boiler improves the hot water system so we need to reflect this in # Installation of a boiler improves the hot water system so we need to reflect this in
# the outcome of the recommendation # the outcome of the recommendation
if self.dual_heating:
new_heating_description = self.DUAL_HEATING_DESCRIPTIONS[
self.property.main_heating["clean_description"]
]["boiler"]["mainheating_description"]
else:
new_heating_description = "Boiler and radiators, mains gas"
new_heating_description = self._map_dual_heating_description(
backup_map_to_description="Boiler and radiators, mains gas",
output_type="mainheating_description",
recommendation_type="boiler"
)
new_hotwater_description = "From main system" new_hotwater_description = "From main system"
new_fuel_description = "mains gas (not community)" new_fuel_description = "mains gas (not community)"
@ -1239,9 +1316,11 @@ class HeatingRecommender:
# If the property did not previously have a boiler, we combine # If the property did not previously have a boiler, we combine
controls_recommender = HeatingControlRecommender(self.property) controls_recommender = HeatingControlRecommender(self.property)
if self.dual_heating: if self.dual_heating:
description_suffix = self.DUAL_HEATING_DESCRIPTIONS[ description_suffix = self._map_dual_heating_description(
self.property.main_heating["clean_description"] backup_map_to_description="",
]["boiler"]["controls_suffix"] output_type="controls_suffix",
recommendation_type="boiler"
)
else: else:
description_suffix = "" description_suffix = ""
controls_recommender.recommend( controls_recommender.recommend(