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 #308 from Hestia-Homes/vander-elliot

Browse source 
Vander elliot completed
This commit is contained in:
KhalimCK 2024-06-27 11:27:52 +01:00 • committed by GitHub
commit 0875213779
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 496 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
backend/Property.py
View file

@ -950,7 +950,6 @@ class Property:
construction_age_band=self.construction_age_band,
floor_area=self.floor_area,
number_habitable_rooms=self.number_of_rooms,
extension_count=float(self.data["extension-count"]),
)
def set_solar_panel_area(self, photo_supply_lookup, floor_area_decile_thresholds):

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

@ -88,3 +88,4 @@ class Material(Base):
plant_cost = Column(Float)
total_cost = Column(Float)
notes = Column(String)
is_installer_quote = Column(Boolean, nullable=False, default=False)

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

@ -284,16 +284,16 @@ 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 not is_new:
continue
create_property_targets(
session,
property_id=property_id,
portfolio_id=body.portfolio_id,
epc_target=body.goal_value,
heat_demand_target=None
)
# if not is_new:
# continue
#
# create_property_targets(
# session,
# property_id=property_id,
# portfolio_id=body.portfolio_id,
# epc_target=body.goal_value,
# heat_demand_target=None
# )
epc_records = {
'original_epc': epc_searcher.newest_epc.copy(),
@ -355,7 +355,8 @@ async def trigger_plan(body: PlanTriggerRequest):
p.get_components(cleaned, photo_supply_lookup, floor_area_decile_thresholds)
p.get_spatial_data(uprn_filenames)
# Call Google Solar API
solar_performance = solar_api_client.get(longitude=p.spatial["longitude"], latitude=p.spatial["latitude"])
# TODO: Complete me
# solar_performance = solar_api_client.get(longitude=p.spatial["longitude"], latitude=p.spatial["latitude"])
logger.info("Getting components and epc recommendations")
recommendations = {}

82
backend/tests/test_annual_bill_savings.py Normal file
View file

@ -0,0 +1,82 @@
import numpy as np
import pytest
from backend.ml_models.AnnualBillSavings import AnnualBillSavings
appliance_consumption_cases = [
{
"total_floor_area": 13.9,
"n_occupants": 1,
"consumption": 718.4795859263703
},
{
"total_floor_area": 20,
"n_occupants": 1.0306381042556767,
"consumption": 865.2316409517844
},
{
"total_floor_area": 30,
"n_occupants": 1.1731577598127325,
"consumption": 1113.5965321501362
},
{
"total_floor_area": 50,
"n_occupants": 1.6901008890848956,
"consumption": 1683.31305074609
},
{
"total_floor_area": 75,
"n_occupants": 2.361158387531988,
"consumption": 2386.2935599981865
},
{
"total_floor_area": 100,
"n_occupants": 2.739525875076067,
"consumption": 2931.6076153011486
},
{
"total_floor_area": 125,
"n_occupants": 2.8807344137165405,
"consumption": 3335.143110751552
},
{
"total_floor_area": 150,
"n_occupants": 2.934188599837662,
"consumption": 3666.3228057866513
},
{
"total_floor_area": 200,
"n_occupants": 3.001920087128373,
"consumption": 4244.625403339813
},
{
"total_floor_area": 300,
"n_occupants": 3.1319299999993095,
"consumption": 5243.086106676302
},
{
"total_floor_area": 500,
"n_occupants": 3.39193,
"consumption": 6927.400500420533
},
{
"total_floor_area": 1000,
"n_occupants": 4.04193,
"consumption": 10434.755635642652
}
]
class TestAnnualBillSavings:
@pytest.mark.parametrize(
"test_case",
appliance_consumption_cases
)
def test_appliance_estimation(self, test_case):
n_occupants = AnnualBillSavings.calculate_occupants(test_case["total_floor_area"])
assert np.isclose(n_occupants, test_case["n_occupants"])
appliance_consumption = AnnualBillSavings.estimate_electrical_appliances(
n_occupants, test_case["total_floor_area"]
)
assert np.isclose(appliance_consumption, test_case["consumption"])

10
etl/costs/app.py
View file

@ -7,10 +7,13 @@ from sqlalchemy.orm import Session
from sqlalchemy import create_engine
from backend.app.db.models.materials import Material
from recommendations.recommendation_utils import calculate_r_value_per_mm
import inspect
DATA_DIRECTORY = Path(__file__).parent / "local_data" / "Hestia Materials.xlsx"
src_file_path = inspect.getfile(lambda: None)
DATA_DIRECTORY = Path(src_file_path).parent / "local_data" / "20240626 Hestia Materials.xlsx"
# Environment file is at the same level as this file
ENV_FILE = Path(__file__).parent / "etl" / "costs" / ".env"
ENV_FILE = Path(src_file_path).parent / "etl" / "costs" / ".env"
dotenv.load_dotenv(ENV_FILE)
DB_USERNAME = os.getenv('DB_USERNAME')
@ -87,7 +90,8 @@ def app():
solid_floor_costs,
ewi_costs,
lel_costs,
flat_roof_costs
flat_roof_costs,
window_costs
]
)

129
etl/customers/vander_elliot/non_intrusives.py Normal file
View file

@ -0,0 +1,129 @@
import pandas as pd
from utils.s3 import save_csv_to_s3
from datetime import datetime
from etl.non_intrusive_surveys.upload.UploadNonIntrusives import UploadNonIntrusives
PORTFOLIO_ID = 82
USER_ID = 8
already_installed = [
{
'address': 'Flat 3 2 Linacre Lane',
'postcode': 'L20 5AH',
"already_installed": ["windows_glazing"]
}
]
def app():
"""
This script handles the creation of the portfolio for the non-intrusive surveys
:return:
"""
# In the future, we can just use the ordnance survey api
uprn_lookup = [
{'House Number': 79,
'Address Line 1': 'Clare Road',
'Address Line 2': 'Liverpool',
'Postcode': 'L20 9LZ',
'uprn': 41018850},
{'House Number': 'Flat 1',
'Address Line 1': '2 Linacre Lane',
'Address Line 2': 'Liverpool',
'Postcode': 'L20 5AH',
'uprn': 41052320},
{'House Number': 'Flat 2',
'Address Line 1': '2 Linacre Lane',
'Address Line 2': 'Liverpool',
'Postcode': 'L20 5AH',
'uprn': 41052321},
{'House Number': 'Flat 3',
'Address Line 1': '2 Linacre Lane',
'Address Line 2': 'Liverpool',
'Postcode': 'L20 5AH',
'uprn': 41052322},
{'House Number': 'Flat 4',
'Address Line 1': '2 Linacre Lane',
'Address Line 2': 'Liverpool',
'Postcode': 'L20 5AH',
'uprn': 41222759},
{'House Number': 'Flat 1',
'Address Line 1': '4 Linacre Lane',
'Address Line 2': 'Liverpool',
'Postcode': 'L20 5AH',
'uprn': 41222760},
{'House Number': 'Flat 2 (NO ACCESS)',
'Address Line 1': '4 Linacre Lane',
'Address Line 2': 'Liverpool',
'Postcode': 'L20 5AH',
'uprn': 41222761},
{'House Number': 'Flat 3',
'Address Line 1': '4 Linacre Lane',
'Address Line 2': 'Liverpool',
'Postcode': 'L20 5AH',
'uprn': 41212534},
{'House Number': 'Flat 1 (NO ACCESS)',
'Address Line 1': '29 Bedford Road',
'Address Line 2': 'Liverpool',
'Postcode': 'L4 5PS',
'uprn': 38237316},
{'House Number': 'Flat 2 (NO ACCESS)',
'Address Line 1': '29 Bedford Road',
'Address Line 2': 'Liverpool',
'Postcode': 'L4 5PS',
'uprn': 38237317}
]
non_intrusive_s3_filename = (
"customers/Vander Elliot/Non-intrusive survey template V2 - Amazon Management Services.xlsx"
)
non_intrusive = UploadNonIntrusives(
s3_template_location=non_intrusive_s3_filename,
s3_bucket="retrofit-datalake-dev",
uprn_lookup=uprn_lookup,
survey_date=datetime.strptime('2024-06-21', '%Y-%m-%d')
)
non_intrusive.upload()
# We can create the asset list from the uprn lookup
asset_list = [
{
"uprn": x["uprn"],
"address": f"{x['House Number']} {x['Address Line 1']}",
"postcode": x["Postcode"]
}
for x in uprn_lookup
]
asset_list = pd.DataFrame(asset_list)
# 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 the asset list in s3
filename = f"{USER_ID}/{PORTFOLIO_ID}/non_intrusives.csv"
save_csv_to_s3(
dataframe=asset_list,
bucket_name="retrofit-plan-inputs-dev",
file_name=filename
)
body = {
"portfolio_id": str(PORTFOLIO_ID),
"housing_type": "Private",
"goal": "Increase EPC",
"goal_value": "A",
"trigger_file_path": filename,
"already_installed_file_path": already_installed_filename,
"patches_file_path": "",
"non_invasive_recommendations_file_path": "",
"budget": None,
}
print(body)

2
etl/non_intrusive_surveys/photos/README.md
View file

@ -15,5 +15,5 @@ pip install -r requirements.txt
The main application is found in the app.py file. To run the application, use the following command:
```bash
python app.py
python UploadNonIntrusives.py
```

149
etl/non_intrusive_surveys/upload/UploadNonIntrusives.py Normal file
View file

@ -0,0 +1,149 @@
from utils.s3 import read_excel_from_s3
from utils.logger import setup_logger
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
logger = setup_logger()
class UploadNonIntrusives:
"""
This class handles the upload of findings from the non-intrusive surveys, to the database
"""
COLUMN_PREFIXES: dict = {
'Surveyor First Name': 'Surveyor',
'Surveyor Last Name': 'Surveyor',
'House Number': 'Property Details',
'Address Line 1': 'Property Details',
'Address Line 2': 'Property Details',
'Postcode': 'Property Details',
'Property Year Built': 'Property Details',
'Wall Construction': 'Walls',
'Wall Construction Notes': 'Walls',
'Existing insulation?': 'Walls',
'Retro Drilled?': 'Walls',
'Condition (cracks & damp)': 'Walls',
'Condition Notes': 'Walls',
'Alternative walls': 'Walls',
'Alternative walls percentage': 'Walls',
'Adequate Ventilation?': 'Walls',
'Ventilation notes': 'Walls',
'Party wall': 'Walls',
'Floor Type': 'Floor',
'Wall render': 'Wall Render',
'Wall Render Condition': 'Wall Render',
'Roof Type': 'Roof',
'Roof insulation ': 'Roof',
'Roof Condition': 'Roof',
'Obvious Roof Shading': 'Roof',
'Roof orientation - Primary': 'Roof',
'Roof orientation - Secondary': 'Roof',
'Obstructions on the roof': 'Roof',
'Flue type': 'Heating',
'Is there an extension?': 'Access',
'Are there any out-buildings?': 'Access',
'Is there a conservatory?': 'Access',
'Is the property straight onto a footpath?': 'Access',
'Is there a requirement for planning consent for works?': 'Access',
'Is there space for an external unit?': 'Air Source Heat Pump',
'Could a cylinder fit in the loft?': 'Air Source Heat Pump',
'Are there obvious areas of heat loss from the walls?': 'Thermography',
'Are there obvious areas of heat loss from the roof?': 'Thermography',
'Does the existing insulation exhibit signs of inconsistent performance or underperformance?': 'Thermography',
'Is there excessive levels of heat loss from windows?': 'Thermography',
'Is there excessive levels of heat loss from doors?': 'Thermography',
'Material inside the walls': 'Borescope Test',
'Cavity depth (mm)': 'Borescope Test',
'Is there rubble in the cavity?': 'Borescope Test',
'Wall tie type': 'Borescope Test',
'Wall tie integrity': 'Borescope Test',
'Inner block work': 'Borescope Test',
'Current glazing': 'Windows',
'Windows Age (pre/post 2002)': 'Windows',
'Glazing gap': 'Windows',
'Are there obvious trickle vents in the windows?': 'Windows',
'Is there sufficient space in the garden?': 'Ground Source Heat Pump',
'Does the property need a CIGA check?': 'Funding',
'Is the property eligible for GBIS?': 'Funding',
'Is the property eligible for ECO4?': 'Funding',
'Is the property eligible for the Local Authority Flex Scheme?': 'Funding',
'Is the property eligible for HUG?': 'Funding',
'Is the property eligible for LAD?': 'Funding',
'Other funding recommendations': 'Funding'
}
def __init__(self, s3_template_location, s3_bucket, uprn_lookup, survey_date):
self.s3_template_location = s3_template_location
self.s3_bucket = s3_bucket
self.template = self.read_template()
self.uprn_lookup = uprn_lookup
self.survey_date = survey_date
def read_template(self):
"""
This method reads the template from S3
"""
return read_excel_from_s3(file_key=self.s3_template_location, bucket_name=self.s3_bucket, header_row=2)
def upload(self):
"""
This method uploads the non-intrusive survey data to the database
"""
if self.uprn_lookup is None:
raise Exception("Implement call to ordnance survey to get uprn lookup data")
logger.info("Preparing non-intrusive notes")
non_intrusives = self.template.to_dict(orient="records")
non_invasive_notes = []
for survey in non_intrusives:
# Remove any NAN entries
survey_clean = {self.COLUMN_PREFIXES[k] + ": " + k: v for k, v in survey.items() if v == v}
uprn_data = [
x for x in self.uprn_lookup if (
str(x['House Number']).strip() == str(survey_clean['Property Details: House Number']).strip() and
x['Address Line 1'] == survey_clean['Property Details: Address Line 1'].strip() and
x['Address Line 2'] == survey_clean['Property Details: Address Line 2'].strip() and
x['Postcode'] == survey_clean['Property Details: Postcode'].strip()
)
]
if len(uprn_data) != 1:
address = (
str(survey_clean['Property Details: House Number']) + ' ' +
survey_clean['Property Details: Address Line 1'] + ' ' +
survey_clean['Property Details: Address Line 2'] + ' ' +
survey_clean['Property Details: Postcode']
)
raise Exception(f"Failed to find UPRN data for {address}")
surveyor = (
survey_clean.pop("Surveyor: Surveyor First Name") + " " +
survey_clean.pop("Surveyor: Surveyor Last Name")
)
# Include all of the information apart from data that includes the Property details prefix and the
# surveyor - we do however include Property Details: Property Year Built
notes_to_upload = {
k: v for k, v in survey_clean.items() if k == "Property Details: Property Year Built" or (
not k.startswith("Property Details") and
not k.startswith("Surveyor")
)
}
non_invasive_notes.append({
"uprn": uprn_data[0]['uprn'],
"surveyor": surveyor,
"survey_date": self.survey_date,
**notes_to_upload
})
# Implement call to upload notes_to_upload to the database
logger.info("Uploading non-intrusive notes to the database")
session = sessionmaker(bind=db_engine)()
upload_non_intrusive_survey_notes(session=session, non_invasive_notes=non_invasive_notes, batch_size=500)

0
etl/non_intrusive_surveys/upload/__init__.py Normal file
View file

121
recommendations/Costs.py
View file

@ -104,9 +104,9 @@ DOUBLE_RADIATOR_COST = 300
FLUE_COST = 600
PIPEWORK_COST = 750 # Min cost is £500
# This is the cost per meter squared for cavity extraction
# https://www.checkatrade.com/blog/cost-guides/cavity-wall-insulation-removal-cost/
CAVITY_EXTRACTION_COST = 21.5
# Based on SCIS figures
# TODO: Add this to databse
CAVITY_EXTRACTION_COST = 25
class Costs:
@ -203,6 +203,20 @@ class Costs:
:return: A dictionary containing detailed cost breakdown.
"""
# CWI usually takes 1 day
labour_hours = 8
labour_days = 1
# if the material is based on an installer cost, we return the flat price
if material["is_installer_quote"]:
total_cost = material["total_cost"] * wall_area
return {
"total": total_cost,
"labour_hours": labour_hours,
"labour_days": labour_days,
}
material_cost_per_m2 = material["material_cost"]
base_material_cost = material_cost_per_m2 * wall_area
@ -220,11 +234,6 @@ class Costs:
total_cost = subtotal_before_vat + vat_cost
labour_hours = material["labour_hours_per_unit"] * wall_area
# Assume a team of 2
labour_days = (labour_hours / 8) / 2
if is_extraction_and_refill:
# bump up the cost of the work
total_cost = total_cost + CAVITY_EXTRACTION_COST * wall_area
@ -252,6 +261,20 @@ class Costs:
:return: A dictionary containing detailed cost breakdown.
"""
labour_hours = material["labour_hours_per_unit"] * floor_area
# Assume a team of 1 person
labour_days = labour_hours / 8
if material["is_installer_quote"]:
total_cost = material["total_cost"] * floor_area
return {
"total": total_cost,
"labour_hours": labour_hours,
"labour_days": labour_days,
}
material_cost_per_m2 = material["material_cost"]
# We inflate material costs due to recent price increases
@ -273,11 +296,6 @@ class Costs:
total_cost = subtotal_before_vat + vat_cost
labour_hours = material["labour_hours_per_unit"] * floor_area
# Assume a team of 1 person
labour_days = labour_hours / 8
return {
"total": total_cost,
"subtotal": subtotal_before_vat,
@ -314,6 +332,22 @@ class Costs:
:return:
"""
# if the material is based on an installer cost, we return the flat price
if material["is_installer_quote"]:
total_cost = material["total_cost"] * wall_area
labour_hours = material["labour_hours_per_unit"] * wall_area
# To install internal wall insulation, a small to medium size project might be conducted by a team of 3-5
# people
labour_days = (labour_hours / 8) / 4
return {
"total": total_cost,
"labour_hours": labour_hours,
"labour_days": labour_days,
}
# Extract and check the different types of data we'll need
demolition_data = [x for x in non_insulation_materials if x["type"] == "iwi_wall_demolition"]
vapour_barrier_data = [x for x in non_insulation_materials if x["type"] == "iwi_vapour_barrier"]
@ -398,6 +432,21 @@ class Costs:
:return:
"""
# if the material is based on an installer cost, we return the flat price
if material["is_installer_quote"]:
total_cost = material["total_cost"] * insulation_floor_area
labour_hours = material["labour_hours_per_unit"] * insulation_floor_area
# To install suspended floor insulation, a small to medium size project might be conducted by a team of 3
# people
labour_days = (labour_hours / 8) / 3
return {
"total": total_cost,
"labour_hours": labour_hours,
"labour_days": labour_days,
}
demolition_data = [x for x in non_insulation_materials if x["type"] == "suspended_floor_demolition"]
vapour_barrier_data = [x for x in non_insulation_materials if x["type"] == "suspended_floor_vapour_barrier"]
redecoration_data = [x for x in non_insulation_materials if x["type"] == "suspended_floor_redecoration"]
@ -500,6 +549,21 @@ class Costs:
:return:
"""
# if the material is based on an installer cost, we return the flat price
if material["is_installer_quote"]:
total_cost = material["total_cost"] * insulation_floor_area
labour_hours = material["labour_hours_per_unit"] * insulation_floor_area
# To install suspended floor insulation, a small to medium size project might be conducted by a team of 3
# people
labour_days = (labour_hours / 8) / 3
return {
"total": total_cost,
"labour_hours": labour_hours,
"labour_days": labour_days,
}
demolition_data = [x for x in non_insulation_materials if x["type"] == "solid_floor_demolition"]
preparation_data = [x for x in non_insulation_materials if x["type"] == "solid_floor_preparation"]
vapour_barrier_data = [x for x in non_insulation_materials if x["type"] == "solid_floor_vapour_barrier"]
@ -619,6 +683,24 @@ class Costs:
:return:
"""
if material["is_installer_quote"]:
total_cost = material["total_cost"] * wall_area
# Add on a buffer for scaffolding
if self.property.data["property-type"] == "House":
total_cost += self.EWI_SCAFFOLDING_PRELIMINARIES * total_cost
labour_hours = material["labour_hours_per_unit"] * wall_area
# To install internal wall insulation, a small to medium size project might be conducted by a team of 3-5
# people
labour_days = (labour_hours / 8) / 4
return {
"total": total_cost,
"labour_hours": labour_hours,
"labour_days": labour_days,
}
# For semi detatched and detatched houses, as well as maisonettes, we price for scaffolding
if self.property.data["property-type"] == "House":
@ -872,6 +954,19 @@ class Costs:
"""
if material["is_installer_quote"]:
total_cost = material["total_cost"] * number_of_windows
labour_hours = material["labour_hours_per_unit"] * number_of_windows
# To install windows, a small to medium size project might be conducted by a team of 2-3 people
labour_days = (labour_hours / 8) / 2
return {
"total": total_cost,
"labour_hours": labour_hours,
"labour_days": labour_days,
}
material_cost = material["material_cost"] * number_of_windows
labour_cost = (

1
recommendations/Recommendations.py
View file

@ -316,6 +316,7 @@ class Recommendations:
expected_adjusted_energy = AnnualBillSavings.adjust_energy_to_metered(
epc_energy_consumption=expected_heat_demand,
current_epc_rating=property_instance.data["current-energy-rating"],
total_floor_area=property_instance.floor_area
)
adjusted_heat_demand_change = (

7
recommendations/optimiser/optimiser_functions.py
View file

@ -18,11 +18,10 @@ def prepare_input_measures(property_recommendations, goal):
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"]]
# if the recommendation is a solar recommendation with a battery, we exclude it from the optimisation.
recs = [r for r in recs if ~r["has_battery"]]
input_measures.append(
[

7
recommendations/recommendation_utils.py
View file

@ -673,8 +673,10 @@ def esimtate_pitched_roof_area(floor_area: float, floor_height: float) -> float:
def estimate_windows(
property_type, built_form, construction_age_band, floor_area, number_habitable_rooms, extension_count
property_type, built_form, construction_age_band, floor_area, number_habitable_rooms
):
# If there is an extension, that will boost the number of habitable rooms
# Base window count based on habitable rooms
window_count = number_habitable_rooms
@ -717,9 +719,6 @@ def estimate_windows(
# Older houses with smaller, more numerous windows
window_count += 1
# Adjust for extensions (each extension might add windows)
window_count += extension_count
# Adjustments for specific property types
if property_type in ["Flat", "Maisontte"]:
# Flats might have fewer windows due to shared walls