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

Merge pull request #290 from Hestia-Homes/immo-pilot

Browse source 
Immo pilot
This commit is contained in:
KhalimCK 2024-04-15 13:39:51 +01:00 • committed by GitHub
commit 4cad8e243b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
35 changed files with 1146 additions and 109 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.idea/Model.iml generated
View file

@ -7,7 +7,7 @@
<sourceFolder url="file://$MODULE_DIR$/open_uprn" isTestSource="false" />
<sourceFolder url="file://$MODULE_DIR$/recommendations" isTestSource="false" />
</content>
<orderEntry type="jdk" jdkName="Python 3.10 (model_data)" jdkType="Python SDK" />
<orderEntry type="jdk" jdkName="Python 3.10 (backend)" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
<component name="PyNamespacePackagesService">

2
.idea/misc.xml generated
View file

@ -3,7 +3,7 @@
<component name="Black">
<option name="sdkName" value="Python 3.10 (backend)" />
</component>
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.10 (model_data)" project-jdk-type="Python SDK" />
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.10 (backend)" project-jdk-type="Python SDK" />
<component name="PythonCompatibilityInspectionAdvertiser">
<option name="version" value="3" />
</component>

49
backend/Property.py
View file

@ -1,4 +1,5 @@
import os
import ast
from itertools import groupby
import pandas as pd
@ -55,7 +56,13 @@ class Property:
DATA_ANOMALY_MATCHES = DATA_ANOMALY_MATCHES
def __init__(self, id, postcode, address, epc_record):
# Surplus information, that can be provided as optional inputs, by a customer
n_bathrooms = None
n_bedrooms = None
def __init__(
self, id, postcode, address, epc_record, already_installed=None, **kwargs
):
self.epc_record = epc_record
@ -68,6 +75,11 @@ class Property:
}
self.old_data = epc_record.get("old_data")
self.property_dimensions = None
# This is a list of measures that have already been installed in the property, typically found as a result
# of the non-invasive surveys. We reflect that this has been installed in the recommendations, but remove the
# cost and instead, provide a message that the measure has already been installed
self.already_installed = ast.literal_eval(already_installed['already_installed']) if already_installed else []
self.uprn = epc_record.get("uprn")
self.full_sap_epc = epc_record.get("full_sap_epc")
@ -133,6 +145,35 @@ class Property:
self.recommendations_scoring_data = []
self.parse_kwargs(kwargs)
@classmethod
def extract_kwargs(cls, kwargs):
"""
This method is to be used in the router, to extract the kwargs from the request and prevent any errors such as
non-integer values, or inputs that clash with the __init__ method of this class
:param kwargs:
:return:
"""
n_bathrooms = kwargs.get("n_bathrooms", None)
if n_bathrooms is not None:
# We add on a small value to ensure that the number of bathrooms is rounded up, in case the value is 0.5
n_bathrooms = int(round(float(n_bathrooms) + 1e-5))
n_bedrooms = kwargs.get("n_bedrooms", None)
if n_bedrooms is not None:
n_bedrooms = int(round(float(n_bedrooms) + 1e-5))
return {
"n_bathrooms": n_bathrooms,
"n_bedrooms": n_bedrooms,
}
def parse_kwargs(self, kwargs):
# We extract the elements from kwargs that we recognise. Anything additional is ignored
self.n_bathrooms = kwargs.get("n_bathrooms", None)
self.n_bedrooms = kwargs.get("n_bedrooms", None)
def create_base_difference_epc_record(self, cleaned_lookup: dict):
"""
Creates a EPCDifferenceRecord object, which is used to store the difference between the current and
@ -421,7 +462,9 @@ class Property:
"double glazing installed during or after 2002"
)
if recommendation["type"] in ["heating", "hot_water_tank_insulation", "heating_control"]:
if recommendation["type"] in [
"heating", "hot_water_tank_insulation", "heating_control", "secondary_heating"
]:
# We update the data, as defined in the recommendaton
simulation_config = recommendation["simulation_config"]
@ -442,7 +485,7 @@ class Property:
"loft_insulation", "room_roof_insulation", "flat_roof_insulation",
"solid_floor_insulation", "suspended_floor_insulation", "exposed_floor_insulation",
"windows_glazing", "solar_pv", "heating", "hot_water_tank_insulation",
"heating_control",
"heating_control", "secondary_heating"
]:
raise NotImplementedError(
"Implement me, given type %s" % recommendation["type"]

50
backend/app/db/functions/non_intrusive_surveys.py Normal file
View file

@ -0,0 +1,50 @@
from sqlalchemy.orm import Session
from backend.app.db.models.non_intrusive_surveys import NonIntrusiveSurvey, NonIntrusiveSurveyNotes
def upload_non_intrusive_survey_notes(session: Session, non_invasive_notes, batch_size=500):
"""
Uploads a list of non-intrusive survey notes into the database in batches. Each dictionary in the list represents
one survey and its associated notes.
:param session: SQLAlchemy Session object through which all database transactions are handled.
:param non_invasive_notes: List of dictionaries where each dictionary contains survey details including 'uprn',
'survey_date', 'surveyor', and other notes as key-value pairs.
:param batch_size: The size of each batch to be processed (default is 500).
:return: None
"""
# Helper function to process each batch
def process_batch(batch):
surveys = []
notes = []
for note in batch:
survey = NonIntrusiveSurvey(
uprn=note['uprn'],
survey_date=note['survey_date'],
surveyor=note['surveyor']
)
surveys.append(survey)
session.add_all(surveys)
session.flush() # Get IDs for surveys
for note, survey in zip(batch, surveys):
for key, value in note.items():
if key not in ['uprn', 'survey_date', 'surveyor']:
notes.append(NonIntrusiveSurveyNotes(
survey_id=survey.id,
title=key,
note=value
))
session.bulk_save_objects(notes)
session.commit()
# Split the data into batches and process each batch
total = len(non_invasive_notes)
for start in range(0, total, batch_size):
end = min(start + batch_size, total)
batch = non_invasive_notes[start:end]
process_batch(batch)

3
backend/app/db/functions/recommendations_functions.py
View file

@ -85,7 +85,8 @@ def upload_recommendations(session: Session, recommendations_to_upload, property
"co2_equivalent_savings": rec["co2_equivalent_savings"],
"total_work_hours": rec["labour_hours"],
"energy_cost_savings": rec["energy_cost_savings"],
"labour_days": rec["labour_days"]
"labour_days": rec["labour_days"],
"already_installed": rec["already_installed"],
}
for rec in recommendations_to_upload
]

22
backend/app/db/models/non_intrusive_surveys.py Normal file
View file

@ -0,0 +1,22 @@
from sqlalchemy import Column, BigInteger, String, TIMESTAMP, ForeignKey, Integer
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class NonIntrusiveSurvey(Base):
__tablename__ = 'non_intrusive_survey'
id = Column(BigInteger, primary_key=True, autoincrement=True)
uprn = Column(Integer, nullable=False)
survey_date = Column(TIMESTAMP, nullable=False)
surveyor = Column(String, nullable=False)
class NonIntrusiveSurveyNotes(Base):
__tablename__ = 'non_intrusive_survey_notes'
id = Column(BigInteger, primary_key=True, autoincrement=True)
survey_id = Column(BigInteger, ForeignKey('non_intrusive_survey.id'), nullable=False)
title = Column(String, nullable=False)
note = Column(String, nullable=False)

1
backend/app/db/models/recommendations.py
View file

@ -30,6 +30,7 @@ class Recommendation(Base):
rental_yield_increase = Column(Float)
total_work_hours = Column(Float)
labour_days = Column(Float)
already_installed = Column(Boolean, nullable=False, default=False)
class RecommendationMaterials(Base):

44
backend/app/plan/router.py
View file

@ -44,20 +44,15 @@ BATCH_SIZE = 5
SCORING_BATCH_SIZE = 400
def patch_epc(config, epc_records):
def patch_epc(patch, epc_records):
"""
This utility function is useful to patch the epc data if we have data from the customer
:return:
"""
number_habitable_rooms = config.get("number-habitable-rooms", None)
number_heated_rooms = config.get("number-heated-rooms", None)
if number_habitable_rooms is not None:
epc_records["original_epc"]["number-habitable-rooms"] = int(number_habitable_rooms)
if number_heated_rooms is not None:
epc_records["original_epc"]["number-heated-rooms"] = int(number_heated_rooms)
for patch_variable, patch_value in patch.items():
if patch_variable in epc_records["original_epc"]:
epc_records["original_epc"][patch_variable] = patch_value
return epc_records
@ -79,12 +74,23 @@ async def trigger_plan(body: PlanTriggerRequest):
# TODO: We should store the trigger file path in the database with the plan so we can track the file that
# triggered the plan
# TODO: Create the ability to congigure/switch off certain measures
# TODO: if the measure is already installed, it should actually be the very first phase
try:
session.begin()
logger.info("Getting the inputs")
plan_input = read_csv_from_s3(bucket_name=get_settings().PLAN_TRIGGER_BUCKET, filepath=body.trigger_file_path)
# If we have patches or overrides, we should read them in here
patches = []
if body.patches_file_path:
patches = read_csv_from_s3(bucket_name=get_settings().PLAN_TRIGGER_BUCKET, filepath=body.patches_file_path)
already_installed = []
if body.already_installed_file_path:
already_installed = read_csv_from_s3(
bucket_name=get_settings().PLAN_TRIGGER_BUCKET, filepath=body.already_installed_file_path
)
cleaning_data = read_dataframe_from_s3_parquet(
bucket_name=get_settings().DATA_BUCKET, file_key="sap_change_model/cleaning_dataset.parquet",
)
@ -108,7 +114,6 @@ async def trigger_plan(body: PlanTriggerRequest):
property_id, is_new = create_property(
session, body.portfolio_id, epc_searcher.address_clean, epc_searcher.postcode_clean, epc_searcher.uprn
)
# if a new record was not created, we don't produduce recommendations
if not is_new:
continue
@ -125,7 +130,11 @@ async def trigger_plan(body: PlanTriggerRequest):
'full_sap_epc': epc_searcher.full_sap_epc.copy(),
'old_data': epc_searcher.older_epcs.copy(),
}
epc_records = patch_epc(config, epc_records)
patch = next((
x for x in patches if (x["address"] == config["address"]) and (x["postcode"] == config["postcode"])
), {})
epc_records = patch_epc(patch, epc_records)
prepared_epc = EPCRecord(
epc_records=epc_records,
@ -133,12 +142,18 @@ async def trigger_plan(body: PlanTriggerRequest):
cleaning_data=cleaning_data
)
property_already_installed = next((
x for x in already_installed if
(x["address"] == config["address"]) and (x["postcode"] == config["postcode"])
), {})
input_properties.append(
Property(
id=property_id,
address=epc_searcher.address_clean,
postcode=epc_searcher.postcode_clean,
epc_record=prepared_epc,
already_installed=property_already_installed,
**Property.extract_kwargs(config)
)
)
@ -242,7 +257,7 @@ async def trigger_plan(body: PlanTriggerRequest):
expected_adjusted_energy=expected_adjusted_energy
)
input_measures = prepare_input_measures(recommendations_with_impact, body.goal, body.housing_type)
input_measures = prepare_input_measures(recommendations_with_impact, body.goal)
current_sap_points = int(property_instance.data["current-energy-efficiency"])
target_sap_points = epc_to_sap_lower_bound(body.goal_value)
@ -279,9 +294,6 @@ async def trigger_plan(body: PlanTriggerRequest):
if ventilation_rec:
selected_recommendations.add(ventilation_rec["recommendation_id"])
# We check if the selected recommendation is wall ventilation and if so, we make sure
# mechanical ventilation is selected
# We'll use the set of selected recommendations to filter the recommendations to upload
final_recommendations = [
[

2
backend/app/plan/schemas.py
View file

@ -9,6 +9,8 @@ class PlanTriggerRequest(BaseModel):
goal_value: str
portfolio_id: int
trigger_file_path: str
already_installed_file_path: Optional[str] = None
patches_file_path: Optional[str] = None
exclusions: Optional[conlist(str, min_items=1)] = None
# Pre-defined list of possibilities for exclusions

11
backend/ml_models/Valuation.py
View file

@ -52,6 +52,17 @@ class PropertyValuation:
10070056829: 76_000,
10070056920: 76_000,
10023345463: 76_000,
# IMMO Dudley Pilot - search by going to https://www.zoopla.co.uk/property/uprn/{uprn}/
90070461: 172_000, # Based on Zoopla
90022227: 181_000, # Based on Zoopla
90106884: 180_000, # Based on Zoopla
90051858: 201_000, # Based on Zoopla
90060989: 172_000, # Based on Zoopla
90048026: 196_000, # Based on Zoopla
90077535: 192_000, # Based on Zoopla
90093693: 279_000, # Based on Zoopla
90055152: 149_000, # Based on Zoopla
90028499: 238_000, # Based on Zoopla
}
# We base our valuation uplifts on a number of sources

78
etl/air_source_heat_pump/AirSourceHeatPumpEfficiency.py Normal file
View file

@ -0,0 +1,78 @@
import pandas as pd
from tqdm import tqdm
from utils.s3 import save_dataframe_to_s3_parquet, read_dataframe_from_s3_parquet
from utils.logger import setup_logger
from etl.epc.settings import EARLIEST_EPC_DATE
logger = setup_logger()
class AirSourceHeatPumpEfficiency:
def __init__(self, file_directories, cleaned_lookup):
"""
:param file_directories: A list of directories where files are stored.
:param cleaned_lookup: A dictionary containing cleaned lookup data.
"""
self.file_directories = file_directories
self.cleaned_lookup = cleaned_lookup
self.results = []
def create_dataset(self):
logger.info("Creating solar photo supply dataset")
for dir in tqdm(self.file_directories):
filepath = dir / "certificates.csv"
df = pd.read_csv(filepath, low_memory=False)
df = df[~pd.isnull(df["UPRN"])]
df["UPRN"] = df["UPRN"].astype(int).astype(str)
# Take entries after SAP12
df["LODGEMENT_DATE"] = pd.to_datetime(df["LODGEMENT_DATE"])
df = df[df["LODGEMENT_DATE"] > EARLIEST_EPC_DATE]
df = df[
~df["TENURE"].isin(
[
"unknown",
"Not defined - use in the case of a new dwelling for which the intended tenure in not known. "
"It is not to be used for an existing dwelling"
]
)
]
# Take entries that contain an air source heat pump
df = df[
df["MAINHEAT_DESCRIPTION"].str.contains("air source heat pump", case=False, na=False)
]
# Get the columns we're interested in
df = df[
[
"MAINHEAT_DESCRIPTION",
"MAINHEAT_ENERGY_EFF",
"MAINHEATCONT_DESCRIPTION",
"MAINHEATC_ENERGY_EFF",
"MAIN_FUEL",
"HOTWATER_DESCRIPTION",
"HOT_WATER_ENERGY_EFF",
"MAINS_GAS_FLAG"
]
]
counts = df.groupby(
[
"MAINHEAT_DESCRIPTION",
"MAINHEAT_ENERGY_EFF",
"MAINHEATCONT_DESCRIPTION",
"MAINHEATC_ENERGY_EFF",
"MAIN_FUEL",
"HOTWATER_DESCRIPTION",
"HOT_WATER_ENERGY_EFF",
"MAINS_GAS_FLAG"
]
).size().reset_index(name="count")
# Drop rows that have a missing PROPERTY_TYPE, BUILT_FORM, CONSTRUCTION_AGE_BAND, TOTAL_FLOOR_AREA
for col in ["PROPERTY_TYPE", "BUILT_FORM", "CONSTRUCTION_AGE_BAND", "TOTAL_FLOOR_AREA"]:
df = df[~pd.isnull(df[col])]
# Take newest LODGEMENT_DATE per UPRN
df = df.sort_values(by="LODGEMENT_DATE", ascending=False).drop_duplicates(subset=["UPRN"])

24
etl/air_source_heat_pump/app.py Normal file
View file

@ -0,0 +1,24 @@
from pathlib import Path
from backend.app.plan.utils import get_cleaned
from etl.air_source_heat_pump.AirSourceHeatPumpEfficiency import AirSourceHeatPumpEfficiency
DATA_DIRECTORY = Path(__file__).parent / "local_data" / "all-domestic-certificates"
def app():
"""
This code reads in the EPC dataset and looks at the efficiency values for heating systems that inclue air source
heat pumps. This dataset is then used to inform the recommendations for the air source heat pump, so we know
how to set the simulation
:return:
"""
directories = [entry for entry in DATA_DIRECTORY.iterdir() if entry.is_dir()]
cleaned_lookup = get_cleaned()
ashp_data_client = AirSourceHeatPumpEfficiency(
file_directories=directories,
cleaned_lookup=cleaned_lookup
)
ashp_data_client.create_dataset()

129
etl/customers/immo/pilot/asset_list.py Normal file
View file

@ -0,0 +1,129 @@
import pandas as pd
from utils.s3 import read_excel_from_s3
from utils.s3 import save_csv_to_s3
USER_ID = 8
PORTFOLIO_ID = 70
council_tax_bands = [
{'address': '8 Corporation Road', 'postcode': 'DY2 7PX', 'band': 'A'},
{'address': '21 Wells Road', 'postcode': 'DY5 3TB', 'band': 'A'},
{'address': '27 Milton Road', 'postcode': 'WV14 8HZ', 'band': 'A'},
{'address': '195 Ashenhurst Road', 'postcode': 'DY1 2JB', 'band': 'A'},
{'address': '53 Bromley', 'postcode': 'DY5 4PJ', 'band': 'A'},
{'address': '91 Osprey Drive', 'postcode': 'DY1 2JS', 'band': 'B'},
{'address': '47 Fairfield Road', 'postcode': 'DY8 5UJ', 'band': 'B'},
{'address': '150 Huntingtree Road', 'postcode': 'B63 4HP', 'band': 'C'},
{'address': '6 Beech Road', 'postcode': 'DY1 4BP', 'band': 'A'},
{'address': '5 Oaklands', 'postcode': 'B62 0JA', 'band': 'A'},
]
council_tax_bands = pd.DataFrame(council_tax_bands)
# This is information we need to override on the EPC itself, for instance if a new survey has been conducted and
# that has not reached the API
patches = [
{
'address': '6 Beech Road', 'postcode': 'DY1 4BP',
'walls-description': 'Cavity wall, filled cavity',
'walls-energy-eff': 'Good',
'roof-description': 'Pitched, 12 mm loft insulation',
'roof-energy-eff': 'Very Poor',
'windows-description': 'Fully double glazed',
'windows-energy-eff': 'Good',
'mainheat-description': 'Room heaters, electric',
'mainheat-energy-eff': 'Very Poor',
'mainheatcont-description': 'Appliance thermostats',
'mainheatc-energy-eff': 'Good',
'lighting-description': 'Low energy lighting in 25% of fixed outlets',
'lighting-energy-eff': 'Good',
'floor-description': 'Solid, no insulation (assumed)',
'secondheat-description': 'None',
'current-energy-efficiency': '32',
'energy-consumption-current': '491',
'co2-emissions-current': '5.0',
'potential-energy-efficiency': '87'
}
]
# This is information that is found as a result of the non-invasives, that mean that certain measures
# have been installed already. To reflect this in the front end, it is included in the recommendation, however
# the cost is removed and instead, a message is presented saying that the measure is already installed.
already_installed = [
{
'address': '5 Oaklands',
'postcode': 'B62 0JA',
"already_installed": ["windows_glazing"]
}
]
def app():
raw_asset_list = read_excel_from_s3(
bucket_name="retrofit-datalake-dev",
file_key="customers/Immo/IMMO Sample Assets_Dudley.xlsx",
header_row=0
)
raw_asset_list = raw_asset_list.drop(columns=["Unnamed: 0"])
# Extract address and postcode
raw_asset_list["address"] = raw_asset_list["Full Address"].str.split(",").str[0]
raw_asset_list["postcode"] = raw_asset_list["Full Address"].str.split(",").str[-1].str.strip()
asset_list = raw_asset_list.merge(council_tax_bands, how="left", on=["address", "postcode"])
# We're provided with number of bathrooms and number of bedrooms.
asset_list = asset_list.rename(
columns={
"No. of Beds": "n_bedrooms",
"No. of WC's": "n_bathrooms"
}
)
# Store the asset list in s3
filename = f"{USER_ID}/{PORTFOLIO_ID}/pilot.csv"
save_csv_to_s3(
dataframe=asset_list,
bucket_name="retrofit-plan-inputs-dev",
file_name=filename
)
# Store overrides in s3
already_installed_filename = f"{USER_ID}/{PORTFOLIO_ID}/already_installed.json"
save_csv_to_s3(
dataframe=pd.DataFrame(already_installed),
bucket_name="retrofit-plan-inputs-dev",
file_name=already_installed_filename
)
# Store patches in s3
patches_filename = f"{USER_ID}/{PORTFOLIO_ID}/patches.json"
save_csv_to_s3(
dataframe=pd.DataFrame(patches),
bucket_name="retrofit-plan-inputs-dev",
file_name=patches_filename
)
# EPC C portoflio
body = {
"portfolio_id": str(PORTFOLIO_ID),
"housing_type": "Private",
"goal": "Increase EPC",
"goal_value": "C",
"trigger_file_path": filename,
"already_installed_file_path": already_installed_filename,
"patches_file_path": patches_filename,
"budget": None,
}
print(body)
# EPC B portoflio
body = {
"portfolio_id": str(PORTFOLIO_ID + 1),
"housing_type": "Private",
"goal": "Increase EPC",
"goal_value": "B",
"trigger_file_path": filename,
"already_installed_file_path": already_installed_filename,
"patches_file_path": patches_filename,
"budget": None,
}
print(body)

210
etl/customers/immo/pilot/non_invasive.py Normal file
View file

@ -0,0 +1,210 @@
# import extract_msg
from datetime import datetime
from sqlalchemy.orm import sessionmaker
from backend.app.db.connection import db_engine
from backend.app.db.functions.non_intrusive_surveys import upload_non_intrusive_survey_notes
def parse_msg_body(text):
# Split the text into lines
lines = text.split('\r\n')
# Dictionary to hold the parsed data
data = {}
# Process each line
for line in lines:
# Remove all asterisks and extra whitespace
clean_line = line.replace('*', '').strip()
if clean_line: # Ensure the line is not empty after cleaning
# Attempt to split clean '=' if present
if '=' in clean_line:
clean_line = clean_line.replace(' = ', ': ')
# Use line content as a key with a default value indicating presence
# Generate a unique key for lines without '='
data[f"Info{len(data) + 1}"] = clean_line
return data
def app():
"""
This code retrieves the results of the non-invasive surveys, to be stored in S3
:return:
"""
# filepath = ("/Users/khalimconn-kowlessar/Downloads/IMMO - Dudley Pilot - non-invasive raw data/5 Oaklands B62 "
# "0JA/Immo - 5 Oaklands Halesowen B62 0JA.msg")
# filepath = ("/Users/khalimconn-kowlessar/Downloads/IMMO - Dudley Pilot - non-invasive raw data/6 Beech Rd DY1 "
# "4BP/IMMO - 6 Beech Road Dudley DY1 4BP.msg")
# filepath = (
# "/Users/khalimconn-kowlessar/Downloads/IMMO - Dudley Pilot - non-invasive raw data/8 Corporation Rd DY2 "
# "7PX/IMMO - 8 Corporation Road Dudley DY2 7PX.msg"
# )
# filepath = (
# "/Users/khalimconn-kowlessar/Downloads/IMMO - Dudley Pilot - non-invasive raw data/21 Wells Rd DY5 3TB/"
# "IMMO - 21 Wells Road Brierley Hill DY5 3TB.msg"
# )
# filepath = (
# "/Users/khalimconn-kowlessar/Downloads/IMMO - Dudley Pilot - non-invasive raw data/47 Fairfield Rd DY8 "
# "5UJ/IMMO - 47 Fairfield Road Wordsley Stourbridge DY8 5UJ.msg"
# )
# filepath = (
# "/Users/khalimconn-kowlessar/Downloads/IMMO - Dudley Pilot - non-invasive raw data/91 Osprey Drive DY1 "
# "2JS/IMMO - 91 Osprey Drive Dudley DY1 2JS.msg"
# )
# filepath = (
# "/Users/khalimconn-kowlessar/Downloads/IMMO - Dudley Pilot - non-invasive raw data/195 Ashenhurst Rd DY1 "
# "2JB/IMMO - 195 Ashenhurst Road Dudley DY1 2JB.msg"
# )
# filepath = (
# "/Users/khalimconn-kowlessar/Downloads/IMMO - Dudley Pilot - non-invasive raw data/27 Milton Rd DY1 2JB/IMMO "
# "- 27 Milton Road Coseley Bilston WV14 8HZ.msg"
# )
#
# with extract_msg.Message(filepath) as msg:
# body = msg.body
#
# from pprint import pprint
# pprint(parse_msg_body(body))
# We manually create the non-invasive notes for the pilot
non_invasive_notes = [
{
'uprn': 90028499,
# 'address': '5 Oaklands',
# 'postcode': 'B62 0JA',
'surveyor': 'Carl Fitzgerald - The Warmfront Team',
'survey_date': datetime.strptime('2024-04-11', '%Y-%m-%d'),
'Wall Insulation': 'Cavity wall, retro drilled, containing loose fibre insulation. Consider getting a '
'CIGA check and extracting the cavity, replacing with bead insulation. '
'There is a shared alleyway with the neighbour, that is a solid brick wall.',
'Wall Render': 'Partial render between top of ground floor window and bottom of 1st floor window',
'Existing solar PV': 'No existing solar',
'Orientation': 'Front house direction: North East, Back house direction: South West',
'Access to mains?': 'Property has access to the mains',
},
{
'uprn': 90055152,
# 'address': '6 Beech Road',
# 'postcode': 'DY1 4BP',
'surveyor': 'Carl Fitzgerald - The Warmfront Team',
'survey_date': datetime.strptime('2024-04-11', '%Y-%m-%d'),
'Wall Insulation': '1st floor is solid brick with external wall insulation. 2nd floor is cavity, '
'retro drilled, containing loose fibre insulation. Consider getting a '
'CIGA check and extracting the cavity, replacing with bead insulation.',
'Wall Render': None,
'Existing solar PV': 'No existing solar',
'Orientation': 'Side house direction: North East',
'Access to mains?': 'Property has access to the mains',
},
{
'uprn': 90070461,
# 'address': '8 Corporation Road',
# 'postcode': 'DY2 7PX',
'surveyor': 'Carl Fitzgerald - The Warmfront Team',
'survey_date': datetime.strptime('2024-04-11', '%Y-%m-%d'),
'Wall Insulation': "External wall insulation",
'Wall Render': "Render finish throughout",
'Existing solar PV': 'No existing solar',
'Orientation': 'Front house direction: North East, Back house direction: South West',
'Access to mains?': None,
},
{
'uprn': 90022227,
# 'address': '21 Wells Road',
# 'postcode': 'DY5 3TB',
'surveyor': 'Carl Fitzgerald - The Warmfront Team',
'survey_date': datetime.strptime('2024-04-11', '%Y-%m-%d'),
'Wall Insulation': 'Cavity wall, retro drilled, containing loose fibre insulation. Consider getting a '
'CIGA check and extracting the cavity, replacing with bead insulation.',
'Wall Render': None,
'Existing solar PV': 'No existing solar',
'Orientation': 'Front house direction: East, Back house direction: West',
'Access to mains?': 'Property has access to the mains',
},
{
'uprn': 90077535,
# 'address': '47 Fairfield Road',
# 'postcode': 'DY8 5UJ',
'surveyor': 'Carl Fitzgerald - The Warmfront Team',
'survey_date': datetime.strptime('2024-04-11', '%Y-%m-%d'),
'Wall Insulation': 'Cavity wall, retro drilled, containing loose fibre insulation. Consider getting a '
'CIGA check and extracting the cavity, replacing with bead insulation.',
'Wall Render': None,
'Existing solar PV': 'No existing solar',
'Orientation': 'Front house direction: East, Back house direction: West',
'Access to mains?': 'Property has access to the mains',
},
{
'uprn': 90060989,
# 'address': '53 Bromley',
# 'postcode': 'DY5 4PJ',
'surveyor': 'Carl Fitzgerald - The Warmfront Team',
'survey_date': datetime.strptime('2024-04-11', '%Y-%m-%d'),
'Wall Insulation': "Filled at build, partially filled - celotex/king board, 50mm cavity remaining - "
"recommends a cavity wall fill",
"Roof": "Hipped roof",
'Existing solar PV': 'No existing solar',
'Orientation': "Front house direction: North, Back house direction: South, Side house direction: West",
'Access to mains?': 'Property has access to the mains',
},
{
'uprn': 90048026,
# 'address': '91 Osprey Drive',
# 'postcode': 'DY1 2JS',
'surveyor': 'Carl Fitzgerald - The Warmfront Team',
'survey_date': datetime.strptime('2024-04-11', '%Y-%m-%d'),
'Wall Insulation': 'Cavity wall, retro drilled, containing loose fibre insulation. Consider getting a '
'CIGA check and extracting the cavity, replacing with bead insulation.',
'Wall Render': 'Tile hung front and rear of property',
'Existing solar PV': 'No existing solar',
'Orientation': 'Side house direction: East',
'Access to mains?': 'Property has access to the mains',
},
{
'uprn': 90093693,
# 'address': '150 Huntingtree Road',
# 'postcode': 'B63 4HP',
'surveyor': 'Carl Fitzgerald - The Warmfront Team',
'survey_date': datetime.strptime('2024-04-11', '%Y-%m-%d'),
'Heating': 'Electric (storage heaters)',
'Wall Insulation': 'Cavity wall, retro drilled, containing loose fibre insulation. Consider getting a '
'CIGA check and extracting the cavity, replacing with bead insulation.',
"Roof": "Hipped roof",
'Existing solar PV': 'No existing solar',
'Orientation': "Front house direction: North West, Back house direction: South East, Side house direction: "
"North East",
},
{
'uprn': 90051858,
# 'address': '195 Ashenhurst Road',
# 'postcode': 'DY1 2JB',
'surveyor': 'Carl Fitzgerald - The Warmfront Team',
'survey_date': datetime.strptime('2024-04-11', '%Y-%m-%d'),
'Wall Insulation': 'Cavity wall, retro drilled, containing loose fibre insulation. Consider getting a '
'CIGA check and extracting the cavity, replacing with bead insulation.',
'Wall Render': "Solid render front and rear of property",
'Existing solar PV': 'No existing solar',
'Orientation': 'Front house direction: South, Back house direction: North',
'Access to mains?': 'Property has access to the mains',
},
{
'uprn': 90106884,
# 'address': '27 Milton Road',
# 'postcode': 'WV14 8HZ',
'surveyor': 'Carl Fitzgerald - The Warmfront Team',
'survey_date': datetime.strptime('2024-04-11', '%Y-%m-%d'),
'Wall Insulation': 'Cavity wall, retro drilled, containing loose fibre insulation. Consider getting a '
'CIGA check and extracting the cavity, replacing with bead insulation.',
'Wall Render': "Solid render front and rear of property",
'Existing solar PV': 'No existing solar',
'Orientation': 'Front house direction: South East, Back house direction: North West',
'Access to mains?': 'Property has access to the mains',
},
]
session = sessionmaker(bind=db_engine)()
upload_non_intrusive_survey_notes(session=session, non_invasive_notes=non_invasive_notes, batch_size=500)

1
etl/customers/immo/pilot/requirements.txt Normal file
View file

@ -0,0 +1 @@
extract-msg

12
infrastructure/terraform/main.tf
View file

@ -181,4 +181,16 @@ module "lambda_carbon_prediction_ecr" {
module "lambda_heat_prediction_ecr" {
ecr_name = "lambda-heat-prediction-${var.stage}"
source = "./modules/ecr"
}
##############################################
# CDN - Cloudfront
##############################################
module "cloudfront_distribution" {
source = "./modules/cloudfront"
bucket_name = module.s3.bucket_name
bucket_id = module.s3.bucket_id
bucket_arn = module.s3.bucket_arn
bucket_domain_name = module.s3.bucket_domain_name
stage = var.stage
}

65
infrastructure/terraform/modules/cloudfront/main.tf Normal file
View file

@ -0,0 +1,65 @@
resource "aws_cloudfront_distribution" "s3_distribution" {
origin {
domain_name = var.bucket_domain_name
origin_id = "S3-${var.bucket_name}"
s3_origin_config {
origin_access_identity = aws_cloudfront_origin_access_identity.oai.cloudfront_access_identity_path
}
}
enabled = true
default_cache_behavior {
allowed_methods = ["GET", "HEAD"]
cached_methods = ["GET", "HEAD"]
target_origin_id = "S3-${var.bucket_name}"
viewer_protocol_policy = "redirect-to-https"
compress = true
forwarded_values {
query_string = false
cookies {
forward = "none"
}
}
min_ttl = 0
default_ttl = 86400
max_ttl = 31536000
}
price_class = "PriceClass_All"
restrictions {
geo_restriction {
restriction_type = "none"
}
}
viewer_certificate {
cloudfront_default_certificate = true
}
}
resource "aws_cloudfront_origin_access_identity" "oai" {
comment = "OAI for ${var.bucket_name}"
}
resource "aws_s3_bucket_policy" "bucket_policy" {
bucket = var.bucket_id
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Effect = "Allow"
Principal = {
AWS = "arn:aws:iam::cloudfront:user/CloudFront Origin Access Identity ${aws_cloudfront_origin_access_identity.oai.id}"
}
Action = "s3:GetObject"
Resource = "${var.bucket_arn}/*"
},
]
})
}

24
infrastructure/terraform/modules/cloudfront/variables.tf Normal file
View file

@ -0,0 +1,24 @@
variable "bucket_name" {
description = "The name of the bucket"
type = string
}
variable "stage" {
description = "The deployment stage"
type = string
}
variable "bucket_id" {
description = "The ID of the S3 bucket"
type = string
}
variable "bucket_arn" {
description = "The ARN of the S3 bucket"
type = string
}
variable "bucket_domain_name" {
description = "The regional domain name of the S3 bucket"
type = string
}

12
infrastructure/terraform/modules/s3/outputs.tf
View file

@ -2,3 +2,15 @@ output "bucket_name" {
description = "The name of the S3 bucket"
value = aws_s3_bucket.bucket.bucket
}
output "bucket_id" {
value = aws_s3_bucket.bucket.id
}
output "bucket_arn" {
value = aws_s3_bucket.bucket.arn
}
output "bucket_domain_name" {
value = aws_s3_bucket.bucket.bucket_regional_domain_name
}

113
recommendations/Costs.py
View file

@ -79,6 +79,18 @@ CONVENTIONAL_BOILER_COSTS = {
"40kw": 1776
}
# Assumes 3 hours to remove each heater (including re-decorating)
ROOM_HEATER_REMOVAL_COST = 120
ROOM_HEATER_REMOVAL_LABOUR_HOURS = 3
# This is a cost quoted by Jim for a system flush - existig system will run more efficiently
SYSTEM_FLUSH_COST = 250
SINGLE_RADIATOR_COST = 150
DOUBLE_RADIATOR_COST = 300
FLUE_COST = 600
PIPEWORK_COST = 750 # Min cost is £500
class Costs:
"""
@ -1100,9 +1112,67 @@ class Costs:
"labour_days": labour_days,
}
def low_carbon_boiler(self, is_combi, size):
def heater_removal(self, n_rooms):
"""
Estimates the costs of removal of heaters, including the redecoration costs of the space behind the heater
:return:
"""
removal_cost = ROOM_HEATER_REMOVAL_COST * n_rooms
removal_labour_hours = ROOM_HEATER_REMOVAL_LABOUR_HOURS * n_rooms
vat = removal_cost * self.VAT_RATE
subtotal_before_vat = removal_cost
total_cost = subtotal_before_vat + vat
return {
"total": total_cost,
"subtotal": subtotal_before_vat,
"vat": vat,
"labour_hours": removal_labour_hours,
"labour_days": np.ceil(removal_labour_hours / 8),
}
@staticmethod
def _estimate_n_radiators(number_habitable_rooms, total_floor_area, property_type, built_form):
# Base number of radiators: one per habitable room
base_radiators = number_habitable_rooms
# Additional radiators for non-habitable essential areas (e.g., kitchens, hallways)
additional_radiators = 3 # Initial assumption
# Adjust additional radiators based on property type
if property_type == 'Flat':
additional_radiators -= 1 # Flats may need fewer radiators due to less exposure
elif property_type in ['House', 'Bungalow', 'Maisonette']:
# Multiple floors in Maisonette may require additional heating points
additional_radiators += 2 # Houses and bungalows might need more due to greater exposure
else:
raise Exception("Invalid property type")
# Adjust total radiator needs based on built form
form_factor = {
'Mid-Terrace': 0.95,
'Semi-Detached': 1.05,
'Detached': 1.25,
'End-Terrace': 1.05
}
# Calculate total heating power needed and number of radiators based on standard output
total_heating_power_required = total_floor_area * 80 # Watts per square meter
radiator_output = 1000 # Average wattage per radiator
total_radiators_based_on_power = (total_heating_power_required / radiator_output) * form_factor[built_form]
# Final estimation taking the higher of calculated needs or base room count
estimated_radiators = max(total_radiators_based_on_power, base_radiators + additional_radiators)
return round(estimated_radiators)
def boiler(self, is_combi, size, exising_room_heaters, system_change, n_heated_rooms, n_rooms):
"""
Based on a basic estimate of median value £2600 to install a low carbon combi boiler
First time central heating vosts can als be found here:
https://www.checkatrade.com/blog/cost-guides/central-heating-installation-cost/
:return:
"""
@ -1110,23 +1180,58 @@ class Costs:
# The unit cost is the cost without VAT
# We now need to estimate the cost of the works
labour_days = 2
labour_rate = 500
labour_hours = labour_days * 8
labour_rate = 300
# Average cost of installation is 1 (maybe 2days) at £300 per day
# https://www.checkatrade.com/blog/cost-guides/new-boiler-cost/
# To be pessimistic, assume 2 days work and £500 day rate
# To be pessimistic, assume 2 days work
labour_cost = labour_rate * self.labour_adjustment_factor * labour_days
# Add contingency and preliminaries
labour_cost = labour_cost * (1 + self.CONTINGENCY + self.PRELIMINARIES)
# labour_days = labour_days + (removal_labour_hours / 8)
vat = labour_cost * self.VAT_RATE
subtotal_before_vat = unit_cost + labour_cost
total_cost = subtotal_before_vat + vat
# if there are existing room heaters, we need to add the cost of removing them
if exising_room_heaters:
removal_costing = self.heater_removal(n_rooms=n_heated_rooms)
# Add the totals to the existing totals
total_cost += removal_costing["total"]
subtotal_before_vat += removal_costing["subtotal"]
labour_hours += removal_costing["labour_hours"]
labour_days += removal_costing["labour_days"]
vat += removal_costing["vat"]
if system_change:
# We need the cost of radiators
n_radiators = self._estimate_n_radiators(
number_habitable_rooms=n_rooms,
total_floor_area=self.property.floor_area,
property_type=self.property.data["property-type"],
built_form=self.property.data["built-form"]
)
additionals_labour_cost = labour_rate * self.labour_adjustment_factor
radiator_cost = DOUBLE_RADIATOR_COST * n_radiators
system_change_cost = radiator_cost + FLUE_COST + PIPEWORK_COST + additionals_labour_cost
system_change_cost_before_vat = system_change_cost / (1 + self.VAT_RATE)
system_change_vat = system_change_cost - system_change_cost_before_vat
# We add an extra labour day for the system change
labour_days += 1
labour_hours += 8
total_cost += system_change_cost
subtotal_before_vat += system_change_cost_before_vat
vat += system_change_vat
return {
"total": total_cost,
"subtotal": subtotal_before_vat,
"vat": vat,
"labour_hours": labour_days * 8,
"labour_hours": labour_hours,
"labour_days": labour_days,
}

4
recommendations/FireplaceRecommendations.py
View file

@ -32,7 +32,8 @@ class FireplaceRecommendations(Definitions):
if number_open_fireplaces == 0:
return
estimated_cost = number_open_fireplaces * self.COST_OF_WORK
already_installed = "sealing_open_fireplace" in self.property.already_installed
estimated_cost = number_open_fireplaces * self.COST_OF_WORK if not already_installed else 0
# We recommend installing two mechanical ventilation systems
self.recommendation = [
@ -44,6 +45,7 @@ class FireplaceRecommendations(Definitions):
"starting_u_value": None,
"new_u_value": None,
"sap_points": None,
"already_installed": already_installed,
"total": estimated_cost,
# Take a very basic estimate of 6 hours, multipled by the number of open fireplaces to seal
"labour_hours": 6 * number_open_fireplaces,

12
recommendations/FloorRecommendations.py
View file

@ -8,7 +8,7 @@ from datatypes.enums import QuantityUnits
from backend.Property import Property
from recommendations.recommendation_utils import (
r_value_per_mm_to_u_value, calculate_u_value_uplift, is_diminishing_returns, update_lowest_selected_u_value,
get_recommended_part, get_floor_u_value
get_recommended_part, get_floor_u_value, override_costs
)
from recommendations.Costs import Costs
@ -192,12 +192,21 @@ class FloorRecommendations(Definitions):
material=material.to_dict(),
non_insulation_materials=non_insulation_materials
)
already_installed = "suspended_floor_insulation" in self.property.already_installed
if already_installed:
cost_result = override_costs(cost_result)
elif material["type"] == "solid_floor_insulation":
cost_result = self.costs.solid_floor_insulation(
insulation_floor_area=self.property.insulation_floor_area,
material=material.to_dict(),
non_insulation_materials=non_insulation_materials
)
already_installed = "solid_floor_insulation" in self.property.already_installed
if already_installed:
cost_result = override_costs(cost_result)
else:
raise NotImplementedError("Implement me!")
@ -217,6 +226,7 @@ class FloorRecommendations(Definitions):
"starting_u_value": u_value,
"new_u_value": new_u_value,
"sap_points": None,
"already_installed": already_installed,
**cost_result
}
)

45
recommendations/HeatingControlRecommender.py
View file

@ -1,5 +1,5 @@
from recommendations.Costs import Costs
from recommendations.recommendation_utils import check_simulation_difference
from recommendations.recommendation_utils import check_simulation_difference, override_costs
from backend.Property import Property
from etl.epc_clean.epc_attributes.MainheatControlAttributes import MainheatControlAttributes
@ -159,20 +159,30 @@ class HeatingControlRecommender:
has_room_thermostat = not needs_room_thermostat
has_trvs = not needs_trvs
cost_result = self.costs.roomstat_programmer_trvs(
number_heated_rooms=int(self.property.data["number-heated-rooms"]),
has_programmer=has_programmer,
has_room_thermostat=has_room_thermostat,
has_trvs=has_trvs
)
description = "upgrade heating controls to Room thermostat, programmer and TRVs"
already_installed = "heating_control" in self.property.already_installed
if already_installed:
cost_result = override_costs(cost_result)
description = "Heating controls have already been upgraded, no further action needed."
self.recommendation.append(
{
"type": "heating_control",
"parts": [],
"description": "upgrade heating controls to Room thermostat, programmer and TRVs",
**self.costs.roomstat_programmer_trvs(
number_heated_rooms=int(self.property.data["number-heated-rooms"]),
has_programmer=has_programmer,
has_room_thermostat=has_room_thermostat,
has_trvs=has_trvs
),
"description": description,
**cost_result,
"starting_u_value": None,
"new_u_value": None,
"sap_points": None,
"already_installed": already_installed,
"simulation_config": simulation_config
}
)
@ -211,17 +221,28 @@ class HeatingControlRecommender:
if self.property.data["mainheatc-energy-eff"] in ["Poor", "Very Poor", "Average", "Good"]:
simulation_config["mainheatc_energy_eff_ending"] = "Very Good"
cost_result = self.costs.time_and_temperature_zone_control(
number_heated_rooms=int(self.property.data["number-heated-rooms"])
)
description = ("Upgrade heating controls to Smart Thermostats, room sensors and smart radiator valves (time & "
"temperature zone control)")
already_installed = "heating_control" in self.property.already_installed
if already_installed:
cost_result = override_costs(cost_result)
description = "Heating controls have already been upgraded, no further action needed."
self.recommendation.append(
{
"type": "heating_control",
"parts": [],
"description": "Upgrade heating controls to Smart Thermostats, room sensors and smart radiator valves",
**self.costs.time_and_temperature_zone_control(
number_heated_rooms=int(self.property.data["number-heated-rooms"])
),
"description": description,
**cost_result,
"starting_u_value": None,
"new_u_value": None,
"sap_points": None,
"already_installed": already_installed,
"simulation_config": simulation_config
}
)

141
recommendations/HeatingRecommender.py
View file

@ -1,7 +1,7 @@
import pandas as pd
from recommendations.Costs import Costs
from recommendations.recommendation_utils import check_simulation_difference
from recommendations.recommendation_utils import check_simulation_difference, override_costs
from backend.Property import Property
from etl.epc_clean.epc_attributes.MainheatAttributes import MainHeatAttributes
from etl.epc_clean.epc_attributes.HotWaterAttributes import HotWaterAttributes
@ -18,6 +18,11 @@ class HeatingRecommender:
self.recommendations = []
def recommend(self, phase=0):
# TODO: We could have a system flush recommendation for an existing boiler, where there is no need to replace
# the boiler, but instead flushing the system will make it run more efficiently. There is a cost for this
# in the Costs class, stored as SYSTEM_FLUSH_COST
self.recommendations = []
# This first iteration of the recommender will provide very basic recommendation
# We recommend heating controls based on the main heating system
@ -33,8 +38,7 @@ class HeatingRecommender:
if has_electric_heating_description or no_heating_no_mains:
# Recommend high heat retention storage heaters
self.recommend_electric_storage_heaters(phase=phase, system_change=True, heating_controls_only=False)
return
self.recommend_hhr_storage_heaters(phase=phase, system_change=True, heating_controls_only=False)
# if the property has mains heating with boiler and radiators, we recommend optimal heating controls
has_boiler = self.property.main_heating["clean_description"] in ["Boiler and radiators, mains gas"]
@ -44,9 +48,38 @@ class HeatingRecommender:
'No system present, electric heaters assumed'
] and self.property.data["mains-gas-flag"]
if has_boiler or no_heating_has_mains:
self.recommend_boiler_upgrades(phase=phase, no_heating_has_mains=no_heating_has_mains)
return
has_gas_heaters = (
self.property.main_heating["clean_description"] in ["Room heaters, mains gas"] and
self.property.data["mains-gas-flag"]
)
# We also check if the property has electric heating, but it has access to the mains gas
electic_heating_has_mains = has_electric_heating_description and self.property.data["mains-gas-flag"]
portable_heaters_has_mains = (
self.property.main_heating["clean_description"] in ["Portable electric heaters assumed for most rooms"] and
self.property.data["mains-gas-flag"]
)
if (
has_boiler or
no_heating_has_mains or
electic_heating_has_mains or
has_gas_heaters or
portable_heaters_has_mains
):
# This indicates that the home previously did not have a boiler in place and so would require
# an overhaul to the system - right now, this is all reasons, apart from if there is an existing boiler
system_change = not has_boiler
exising_room_heaters = self.property.main_heating["clean_description"] in [
"Room heaters, electric", "Room heaters, mains gas"
]
self.recommend_boiler_upgrades(
phase=phase, system_change=system_change, exising_room_heaters=exising_room_heaters
)
return
@staticmethod
def check_simulation_difference(old_config, new_config):
@ -61,9 +94,8 @@ class HeatingRecommender:
return differences
@staticmethod
def combine_heating_and_controls(
controls_recommendations, heating_simulation_config, costs, description, phase, heating_controls_only,
self, controls_recommendations, heating_simulation_config, costs, description, phase, heating_controls_only,
system_change
):
"""
@ -112,6 +144,11 @@ class HeatingRecommender:
recommendation_description = f"{description} and {controls_description}"
already_installed = "cavity_wall_insulation" in self.property.already_installed
if already_installed:
total_costs = override_costs(total_costs)
recommendation_description = "Heating system has already been upgraded, no further action needed."
recommendation = {
"phase": phase,
"parts": [
@ -122,6 +159,7 @@ class HeatingRecommender:
"starting_u_value": None,
"new_u_value": None,
"sap_points": None,
"already_installed": already_installed,
**total_costs,
"simulation_config": recommendation_simulation_config
}
@ -153,9 +191,8 @@ class HeatingRecommender:
return output
def recommend_electric_storage_heaters(self, phase, system_change, heating_controls_only):
def recommend_hhr_storage_heaters(self, phase, system_change, heating_controls_only):
"""
We recommend electric storage heaters as an upgrade to the heating system.
We will recommend upgrading to a high heat retention storage system, if the current system is not already
high heat retention storage
@ -256,12 +293,16 @@ class HeatingRecommender:
return closest_size
def recommend_boiler_upgrades(self, phase, no_heating_has_mains):
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
is generally more expensive
:param phase:
:param no_heating_has_mains: indicaes if the property has no heating system, but has access to the mains gas
:param system_change: Indicates if the property would be undergoing a heating system change. This could be true
if the home didn't have a heating system in place, or if the home had electric heating
previously
:param exising_room_heaters: Indicates if the property had room heaters previously - if so, a boiler
recommendation will need to be accompanied by removal of the room heaters
:return:
"""
@ -270,6 +311,7 @@ class HeatingRecommender:
# We now recommend boiler upgrades, if applicable
simulation_config = {}
boiler_costs = {}
boiler_recommendation = {}
if self.property.data["mainheat-energy-eff"] in ["Very Poor", "Poor", "Average"]:
boiler_size = self.estimate_boiler_size(
property_type=self.property.data["property-type"],
@ -279,17 +321,23 @@ class HeatingRecommender:
num_heated_rooms=self.property.data["number-heated-rooms"],
)
# If heating and hot water come from the mains, we need a combi boiler, otherwise we need a regular boiler
hotwater_from_mains = self.property.hotwater["clean_description"] in ["From main system"]
is_combi = hotwater_from_mains or no_heating_has_mains
# We recommend a combi boiler under the following conditions
# 1) If there are 4 or fewer rooms (we don't use heqted rooms because none of the rooms could be
# heated if there is no existing heating system).
# 2) There 1 or fewer bathrooms
# Otherwise, we recommend a gas condensing boiler, which will server a larger property, that has multiple
# bathrooms
is_combi = (
(self.property.number_of_rooms <= 4) and
(self.property.n_bathrooms in [None, 0, 1])
)
if is_combi:
description = "Upgrade to a new combi boiler"
else:
description = "Upgrade to a new boiler"
description = "Upgrade to a new gas condensing boiler"
simulation_config = {"mainheat_energy_eff_ending": "Good"}
if no_heating_has_mains:
if system_change:
# Installation of a boiler improves the hot water system so we need to reflect this in
# the outcome of the recommendation
heating_ending_config = MainHeatAttributes("Boiler and radiators, mains gas").process()
@ -314,24 +362,35 @@ class HeatingRecommender:
"hot_water_energy_eff_ending": "Good"
}
boiler_costs = self.costs.low_carbon_boiler(is_combi=is_combi, size=f"{boiler_size}kw")
self.recommendations.append(
{
"phase": recommendation_phase,
"parts": [
# TODO
],
"type": "heating",
"description": description,
"starting_u_value": None,
"new_u_value": None,
"sap_points": None,
"simulation_config": simulation_config,
**boiler_costs
}
boiler_costs = self.costs.boiler(
is_combi=is_combi,
size=f"{boiler_size}kw",
exising_room_heaters=exising_room_heaters,
system_change=system_change,
n_heated_rooms=self.property.data["number-heated-rooms"],
n_rooms=self.property.number_of_rooms
)
already_installed = "heating" in self.property.already_installed
if already_installed:
boiler_costs = override_costs(boiler_costs)
description = "Heating system has already been upgraded, no further action needed."
boiler_recommendation = {
"phase": recommendation_phase,
"parts": [
# TODO
],
"type": "heating",
"description": description,
"starting_u_value": None,
"new_u_value": None,
"sap_points": None,
"already_installed": already_installed,
"simulation_config": simulation_config,
**boiler_costs
}
# We recommend the heating controls
# If the property did not previously have a boiler, we combine
controls_recommender = HeatingControlRecommender(self.property)
@ -341,9 +400,8 @@ class HeatingRecommender:
if not controls_recommender.recommendation:
return
if no_heating_has_mains:
# We combine the heating and controls recommendations
boiler_recommendation = self.recommendations[0].copy()
if system_change:
# We combine the heating and controls recommendations, in the case of a system change
combined_recommendations = []
for controls_recommendation in controls_recommender.recommendation:
combined_recommendation = self.combine_heating_and_controls(
@ -358,10 +416,15 @@ class HeatingRecommender:
combined_recommendations.extend(combined_recommendation)
# Overwrite the existing boiler recommendation
self.recommendations = combined_recommendations
self.recommendations.extend(combined_recommendations)
else:
# We increment the recommendation phase, since the heating controls are separate from the boiler upgrade
recommendation_phase += 1
# but we'll only upgrade if we have a heating recommendation
has_heating_recommendation = any(
recommendation["type"] == "heating" for recommendation in self.recommendations
)
if has_heating_recommendation:
recommendation_phase += 1
# The heating controls recommendation is distrinct from the boiler upgrade recommendation
# We insert phase into the recommendations for heating controls
for recommendation in controls_recommender.recommendation:

11
recommendations/HotwaterRecommendations.py
View file

@ -1,5 +1,6 @@
from backend.Property import Property
from recommendations.Costs import Costs
from recommendations.recommendation_utils import override_costs
class HotwaterRecommendations:
@ -41,6 +42,13 @@ class HotwaterRecommendations:
recommendation_cost = self.costs.hot_water_tank_insulation()
already_installed = "hot_water_tank_insulation" in self.property.already_installed
if already_installed:
recommendation_cost = override_costs(recommendation_cost)
description = "Insulation tank has already been insulated, no further action required"
else:
description = "Insulate hot water tank"
self.recommendations.append(
{
"phase": phase,
@ -48,10 +56,11 @@ class HotwaterRecommendations:
# TODO
],
"type": "hot_water_tank_insulation",
"description": "Insulate the hot water tank with an insulation jacket",
"description": description,
"starting_u_value": None,
"new_u_value": None,
"sap_points": None,
"already_installed": already_installed,
**recommendation_cost,
"simulation_config": {"hot_water_energy_eff_ending": "Average"}
}

7
recommendations/LightingRecommendations.py
View file

@ -1,6 +1,7 @@
from backend.Property import Property
from typing import List
from recommendations.Costs import Costs
from recommendations.recommendation_utils import override_costs
class LightingRecommendations:
@ -91,6 +92,11 @@ class LightingRecommendations:
heat_demand_change, carbon_change = self.estimate_lighting_impact(number_non_lel_outlets)
already_installed = "low_energy_lighting" in self.property.already_installed
if already_installed:
cost_result = override_costs(cost_result)
description = "Low energy lighting has already been installed, no further action required"
self.recommendation = [
{
"phase": phase,
@ -99,6 +105,7 @@ class LightingRecommendations:
"description": description,
"starting_u_value": None,
"new_u_value": None,
"already_installed": already_installed,
# For SAP points, we use the fact that lighting is usually worth 2 points and we scale this to
# the proportion of lights that will be set to low energy
"sap_points": round(2 * (number_non_lel_outlets / number_lighting_outlets), 2),

8
recommendations/Recommendations.py
View file

@ -11,6 +11,7 @@ from recommendations.SolarPvRecommendations import SolarPvRecommendations
from recommendations.WindowsRecommendations import WindowsRecommendations
from recommendations.HeatingRecommender import HeatingRecommender
from recommendations.HotwaterRecommendations import HotwaterRecommendations
from recommendations.SecondaryHeating import SecondaryHeating
from backend.ml_models.AnnualBillSavings import AnnualBillSavings
@ -46,6 +47,7 @@ class Recommendations:
self.solar_recommender = SolarPvRecommendations(property_instance=property_instance)
self.heating_recommender = HeatingRecommender(property_instance=property_instance)
self.hotwater_recommender = HotwaterRecommendations(property_instance=property_instance)
self.secondary_heating_recommender = SecondaryHeating(property_instance=property_instance)
def recommend(self):
@ -130,6 +132,12 @@ class Recommendations:
property_recommendations.append(self.lighting_recommender.recommendation)
phase += 1
if "secondary_heating" not in self.exclusions:
self.secondary_heating_recommender.recommend(phase=phase)
if self.secondary_heating_recommender.recommendation:
property_recommendations.append(self.secondary_heating_recommender.recommendation)
phase += 1
# Renewables
if "solar_pv" not in self.exclusions:
self.solar_recommender.recommend(phase=phase)

16
recommendations/RoofRecommendations.py
View file

@ -5,7 +5,7 @@ from typing import List
from datatypes.enums import QuantityUnits
from recommendations.recommendation_utils import (
get_roof_u_value, r_value_per_mm_to_u_value, calculate_u_value_uplift, is_diminishing_returns,
update_lowest_selected_u_value, get_recommended_part, convert_thickness_to_numeric
update_lowest_selected_u_value, get_recommended_part, convert_thickness_to_numeric, override_costs
)
from recommendations.Costs import Costs
@ -20,8 +20,9 @@ class RoofRecommendations:
DIMINISHING_RETURNS_U_VALUE = 0.14
# It is recommended that lofts should have at least 270mm of insulation
MINIMUM_LOFT_ISULATION_MM = 270
# It is recommended that lofts should have at least 270mm of insulation. If the property has more than 200mm of
# loft insulation in place already, we do not recommend anything for the moment
MINIMUM_LOFT_ISULATION_MM = 200
# Flat roof should have at least 100mm of insulation
MINIMUM_FLAT_ROOF_ISULATION_MM = 100
@ -71,7 +72,7 @@ class RoofRecommendations:
# Building regulations part L recommend installing at least 270mm of insulation, however generally we
# experience diminishing returns in terms of SAP once we go beyond around 150mm of insulation
# This only holds true for pitched roofs.
if (insulation_thickness >= self.MINIMUM_LOFT_ISULATION_MM) and self.property.roof["is_pitched"]:
if (insulation_thickness > self.MINIMUM_LOFT_ISULATION_MM) and self.property.roof["is_pitched"]:
return
if (insulation_thickness >= self.MINIMUM_FLAT_ROOF_ISULATION_MM) and self.property.roof["is_flat"]:
@ -206,12 +207,18 @@ class RoofRecommendations:
floor_area=self.property.insulation_floor_area,
material=material
)
already_installed = "loft_insulation" in self.property.already_installed
if already_installed:
cost_result = override_costs(cost_result)
elif material["type"] == "flat_roof_insulation":
cost_result = self.costs.flat_roof_insulation(
floor_area=self.property.insulation_floor_area,
material=material,
non_insulation_materials=non_insulation_materials
)
already_installed = "flat_roof_insulation" in self.property.already_installed
if already_installed:
cost_result = override_costs(cost_result)
else:
raise ValueError("Invalid material type")
@ -231,6 +238,7 @@ class RoofRecommendations:
"starting_u_value": u_value,
"new_u_value": new_u_value,
"sap_points": None,
"already_installed": already_installed,
**cost_result
}
)

65
recommendations/SecondaryHeating.py Normal file
View file

@ -0,0 +1,65 @@
from recommendations.Costs import Costs
from recommendations.recommendation_utils import override_costs
from backend.Property import Property
class SecondaryHeating:
"""
This class recommends the removal of the secondary heating system for properties that have a primary heating
system.
"""
# The list of existing heating systems that are accepted
ACCEPTED_MAINHEAT_DESCRIPTIONS = ["Boiler and radiators, mains gas"]
ACCEPTED_SECONDHEAT_DESCRIPTIONS = ["Room heaters, electric"]
# These are the heaters where works are required to remove them
FIXED_HEATER_DESCRIPTIONS = ["Room heaters, electric"]
def __init__(self, property_instance: Property):
self.property = property_instance
self.costs = Costs(self.property)
self.recommendation = []
def recommend(self, phase: int):
# Reset
self.recommendation = []
if self.property.main_heating["clean_description"] not in self.ACCEPTED_MAINHEAT_DESCRIPTIONS:
return
# TODO: We need to clean secondary data
if self.property.data['secondheat-description'] not in self.ACCEPTED_SECONDHEAT_DESCRIPTIONS:
return
if self.property.data['secondheat-description'] in self.FIXED_HEATER_DESCRIPTIONS:
# We have an associated cost otherwise, there is no cost
n_rooms = self.property.data['number-heated-rooms']
else:
n_rooms = 0
costs = self.costs.heater_removal(n_rooms=n_rooms)
already_installed = "secondary_heating" in self.property.already_installed
if already_installed:
costs = override_costs(costs)
description = "Secondary heating system has already been removed, no further action required"
else:
description = "Remove the secondary heating system"
self.recommendation.append(
{
"phase": phase,
"parts": [],
"type": "secondary_heating",
"description": description,
"starting_u_value": None,
"new_u_value": None,
"sap_points": None,
"already_installed": already_installed,
**costs,
"simulation_config": {
"secondheat_description_ending": "None"
}
}
)

9
recommendations/SolarPvRecommendations.py
View file

@ -1,5 +1,6 @@
import numpy as np
from recommendations.Costs import Costs
from recommendations.recommendation_utils import override_costs
class SolarPvRecommendations:
@ -110,6 +111,10 @@ class SolarPvRecommendations:
description = (f"Install a {kw} kilowatt-peak (kWp) solar photovoltaic (PV) p"
f"anel system on {round(roof_coverage_percent)}% the roof.")
already_installed = "solar_pv" in self.property.already_installed
if already_installed:
cost_result = override_costs(cost_result)
self.recommendation.append(
{
"phase": phase,
@ -119,9 +124,11 @@ class SolarPvRecommendations:
"starting_u_value": None,
"new_u_value": None,
"sap_points": None,
"already_installed": already_installed,
**cost_result,
# This is required for simulating the SAP impact. solar_pv_percentage is between 0 & 1 so we scale
# back up here
"photo_supply": 100 * roof_coverage
"photo_supply": 100 * roof_coverage,
"has_battery": has_battery
}
)

11
recommendations/VentilationRecommendations.py
View file

@ -50,7 +50,11 @@ class VentilationRecommendations(Definitions):
part = self.materials.copy()
estimated_cost = n_units * part[0]["cost"]
already_installed = "cavity_wall_insulation" in self.property.already_installed
estimated_cost = n_units * part[0]["cost"] if not already_installed else 0
labour_hours = 4 * n_units if not already_installed else 0
labour_days = 4 * n_units / 8.0 if not already_installed else 0
part[0]["total"] = estimated_cost
part[0]["quantity"] = n_units
@ -65,6 +69,7 @@ class VentilationRecommendations(Definitions):
"description": f"Install {n_units} {part[0]['description']} units",
"starting_u_value": None,
"new_u_value": None,
"already_installed": already_installed,
"sap_points": 0,
"heat_demand": 0,
"adjusted_heat_demand": 0,
@ -72,7 +77,7 @@ class VentilationRecommendations(Definitions):
"energy_cost_savings": 0,
"total": estimated_cost,
# We use a very simple and rough estimate of 4 hours per unit
"labour_hours": 4 * n_units,
"labour_days": 4 * n_units / 8.0 # Assume 8 hour day
"labour_hours": labour_hours,
"labour_days": labour_days # Assume 8 hour day
}
]

15
recommendations/WallRecommendations.py
View file

@ -8,7 +8,7 @@ from backend.Property import Property
from BaseUtility import Definitions
from recommendations.recommendation_utils import (
r_value_per_mm_to_u_value, calculate_u_value_uplift, is_diminishing_returns, update_lowest_selected_u_value,
get_recommended_part, get_wall_u_value
get_recommended_part, get_wall_u_value, override_costs
)
from recommendations.config import PARTIALLY_FILLED_PERCENTAGE_ASSUMPTION
from recommendations.Costs import Costs
@ -221,6 +221,10 @@ class WallRecommendations(Definitions):
material=material.to_dict(),
)
already_installed = "cavity_wall_insulation" in self.property.already_installed
if already_installed:
cost_result = override_costs(cost_result)
recommendations.append(
{
"phase": phase,
@ -237,6 +241,7 @@ class WallRecommendations(Definitions):
"starting_u_value": u_value,
"new_u_value": new_u_value,
"sap_points": None,
"already_installed": already_installed,
**cost_result
}
)
@ -277,12 +282,19 @@ class WallRecommendations(Definitions):
material=material.to_dict(),
non_insulation_materials=non_insulation_materials
)
already_installed = "internal_wall_insulation" in self.property.already_installed
if already_installed:
cost_result = override_costs(cost_result)
elif material["type"] == "external_wall_insulation":
cost_result = self.costs.external_wall_insulation(
wall_area=self.property.insulation_wall_area,
material=material.to_dict(),
non_insulation_materials=non_insulation_materials
)
already_installed = "external_wall_insulation" in self.property.already_installed
if already_installed:
cost_result = override_costs(cost_result)
else:
raise ValueError("Invalid material type")
@ -301,6 +313,7 @@ class WallRecommendations(Definitions):
"description": self._make_description(material),
"starting_u_value": u_value,
"new_u_value": new_u_value,
"already_installed": already_installed,
"sap_points": None,
**cost_result
}

27
recommendations/WindowsRecommendations.py
View file

@ -4,6 +4,7 @@ import numpy as np
from backend.Property import Property
from recommendations.Costs import Costs
from recommendation_utils import override_costs
class WindowsRecommendations:
@ -70,18 +71,23 @@ class WindowsRecommendations:
is_secondary_glazing=is_secondary_glazing
)
glazing_type = "secondary glazing" if is_secondary_glazing else "double glazing"
if self.property.windows["glazing_coverage"] in ["partial", "most"]:
description = f"Install {glazing_type} to the remaining windows"
already_installed = "windows_glazing" in self.property.already_installed
if already_installed:
cost_result = override_costs(cost_result)
description = "The property already has double glazing installed. No further action is required."
else:
description = f"Install {glazing_type} to all windows"
glazing_type = "secondary glazing" if is_secondary_glazing else "double glazing"
if self.property.windows["glazing_coverage"] in ["partial", "most"]:
description = f"Install {glazing_type} to the remaining windows"
else:
description = f"Install {glazing_type} to all windows"
if self.property.is_listed:
description += ". Secondary glazing recommended due to listed building status"
elif self.property.is_heritage:
description += ". Secondary glazing recommended due to herigate building status"
elif self.property.in_conservation_area:
description += ". Secondary glazing recommended due to conservation area status"
if self.property.is_listed:
description += ". Secondary glazing recommended due to listed building status"
elif self.property.is_heritage:
description += ". Secondary glazing recommended due to herigate building status"
elif self.property.in_conservation_area:
description += ". Secondary glazing recommended due to conservation area status"
self.recommendation = [
{
@ -92,6 +98,7 @@ class WindowsRecommendations:
"starting_u_value": None,
"new_u_value": None,
"sap_points": None,
"already_installed": already_installed,
**cost_result,
"is_secondary_glazing": is_secondary_glazing
}

18
recommendations/optimiser/optimiser_functions.py
View file

@ -1,17 +1,13 @@
def prepare_input_measures(property_recommendations, goal, housing_type):
def prepare_input_measures(property_recommendations, goal):
"""
Basic function to convert recommendations_to_upload to a format that is
suitable for the optimiser - large
:param property_recommendations: object containing the recommendations, created in the plan trigger api
:param goal: goal to be optimised for, should be one of the keys in gain_map. E.g. if the gain is SAP points,
the goal should reflect that desired gain
:param housing_type: type of housing the recommendations are for - should be one of "Social" or "Private"
:return: Nested list of input measures
"""
if housing_type not in ["Social", "Private"]:
raise ValueError("Invalid housing type - investigate me")
goal_map = {
"Increase EPC": "sap_points"
}
@ -20,12 +16,14 @@ def prepare_input_measures(property_recommendations, goal, housing_type):
if not goal_key:
raise NotImplementedError("Not implemented this gain type - investigate me")
# We don't include suspended and solid floor insulation as possible measures in private housing, because
# of the need to decant the tenant
ignored_measures = ["suspended_floor_insulation", "solid_floor_insulation"] if housing_type == "Private" else []
input_measures = []
for recs in property_recommendations:
if recs[0]["type"] == "solar_pv":
# if the recommendation is a solar recommendation without a battery, we exclude it from the optimisation.
# That will ensure that the optimiser only considers solar recommendations with batteries, so we don't
# under-report the potential cost
recs = [r for r in recs if r["has_battery"]]
input_measures.append(
[
{
@ -34,7 +32,7 @@ def prepare_input_measures(property_recommendations, goal, housing_type):
"gain": rec[goal_key],
"type": rec["type"]
}
for rec in recs if rec["type"] not in ignored_measures
for rec in recs
]
)

12
recommendations/recommendation_utils.py
View file

@ -767,3 +767,15 @@ def check_simulation_difference(old_config, new_config):
differences = {key + "_ending": new_config[key] for key in new_config if old_config[key] != new_config[key]}
return differences
def override_costs(costs):
"""
If the method is overridden, we want to make sure that the costs are zero. This function sets the costs to zero
:param costs: Dictionary of costing, as returned by the Costs class
:return:
"""
for k in costs:
costs[k] = 0
return costs