Model/etl/sfr/midlands_portfolio_est_funding.py
2024-09-17 15:26:38 +01:00

209 lines
6.8 KiB
Python

import msgpack
import pandas as pd
from utils.s3 import read_from_s3
from recommendations.recommendation_utils import (
estimate_number_of_floors, esimtate_pitched_roof_area, estimate_external_wall_area, estimate_perimeter
)
def app():
"""
Aims to estimate the amount of GBIS funding eligible
:return:
"""
cleaned = read_from_s3(
s3_file_name="cleaned_epc_data/cleaned.bson",
bucket_name="retrofit-data-dev"
)
cleaned = msgpack.unpackb(cleaned, raw=False)
epc_data = pd.read_excel(
"/Users/khalimconn-kowlessar/Downloads/20240820 portfolio_epc_data.xlsx"
)
# For simplicity, get roofs or cavities
epc_data = epc_data.merge(
pd.DataFrame(cleaned["roof-description"]),
how="left",
left_on="ROOF_DESCRIPTION",
right_on="original_description"
)
epc_data["needs_roof_work"] = epc_data["insulation_thickness"].isin(
[
None,
"100",
'150',
'50',
'75',
'below average',
'25',
'12'
]
) & (epc_data["is_flat"] | epc_data["is_pitched"])
epc_data = epc_data.merge(
pd.DataFrame(cleaned["walls-description"]),
how="left",
left_on="WALLS_DESCRIPTION",
right_on="original_description",
suffixes=("", "_wall")
)
epc_data["needs_cavity_done"] = epc_data["is_cavity_wall"] & epc_data["insulation_thickness_wall"].isin(
['none', "below average"]
)
epc_data["needs_solid_wall"] = (epc_data["is_solid_brick"] | epc_data["is_system_built"]) & epc_data[
"insulation_thickness_wall"].isin(['none', "below average"])
epc_data["could_take_solar"] = (epc_data["is_flat"] | epc_data["is_pitched"])
loft_insulation_per_m2 = 16.07
flat_roof_insulation_per_m2 = 195
cwi_per_m2 = 14.21
ewi_per_m2 = 200
gbis_abs = 30
eco4_abs = 24
solar_pv_cost = 4009
# We assume the work will take the home from a high D to a low D
def get_abs(floor_area):
if floor_area <= 72:
return 155
if floor_area <= 97:
return 169
if floor_area <= 199:
return 196.4
return 350.1
# We assume the work will take the home from a high E to a high C
def get_eco4_abs(floor_area):
if floor_area <= 72:
return 596.6
if floor_area <= 97:
return 650.2
if floor_area <= 199:
return 755.8
return 1347.1
estimated_costs = []
for _, home in epc_data.iterrows():
to_append = {
"uprn": home["UPRN"],
"address": home["ADDRESS"],
"postcode": home["POSTCODE"],
}
project_abs = get_abs(home["TOTAL_FLOOR_AREA"])
available_funding = project_abs * gbis_abs
n_floors = estimate_number_of_floors(home["PROPERTY_TYPE"])
floor_height = float(home["FLOOR_HEIGHT"]) if not pd.isnull(home["FLOOR_HEIGHT"]) else 2.5
# We estimate the amount of insulation required
est_perimeter = estimate_perimeter(
floor_area=float(home["TOTAL_FLOOR_AREA"]) / n_floors,
num_rooms=float(home["NUMBER_HABITABLE_ROOMS"]) / n_floors
)
insulation_needed = estimate_external_wall_area(
num_floors=n_floors,
floor_height=floor_height,
perimeter=est_perimeter,
built_form=home["BUILT_FORM"],
)
# At the very least we'll need solid wall + solar
if home["needs_solid_wall"] and home["could_take_solar"]:
measure = "EWI + Solar"
total_cost = insulation_needed * ewi_per_m2 + solar_pv_cost
eco4_project_abs = get_eco4_abs(home["TOTAL_FLOOR_AREA"])
eco4_available_funding = eco4_project_abs * eco4_abs
cost_of_work_after_funding = total_cost - eco4_available_funding
cost_of_work_after_funding = 0 if cost_of_work_after_funding < 0 else cost_of_work_after_funding
to_append = {
**to_append,
"scheme": "eco4",
"available_funding": eco4_available_funding,
"measure": measure,
"project_abs": eco4_project_abs,
"cost_of_work": total_cost,
"cost_of_work_after_funding": cost_of_work_after_funding,
}
estimated_costs.append(to_append)
continue
# Check if it needs the walls done
if home["needs_cavity_done"]:
cost_of_insulation = insulation_needed * cwi_per_m2
cost_of_work_after_funding = cost_of_insulation - available_funding
cost_of_work_after_funding = 0 if cost_of_work_after_funding < 0 else cost_of_work_after_funding
to_append = {
**to_append,
"scheme": "gbis",
"available_funding": available_funding,
"measure": "Cavity Wall Insulation",
"project_abs": project_abs,
"cost_of_work": cost_of_insulation,
"cost_of_work_after_funding": cost_of_work_after_funding
}
estimated_costs.append(to_append)
continue
if home["needs_roof_work"]:
# We estimate how much the cost of insulation would be
if home["is_pitched"]:
measure = "Loft Insulation"
roof_area = float(home["TOTAL_FLOOR_AREA"]) / n_floors
cost_of_insulation = roof_area * loft_insulation_per_m2
else:
measure = "Flat Roof Insulation"
roof_area = float(home["TOTAL_FLOOR_AREA"]) / n_floors
cost_of_insulation = roof_area * flat_roof_insulation_per_m2
cost_of_work_after_funding = cost_of_insulation - available_funding
cost_of_work_after_funding = 0 if cost_of_work_after_funding < 0 else cost_of_work_after_funding
to_append = {
**to_append,
"scheme": "gbis",
"available_funding": available_funding,
"measure": measure,
"project_abs": project_abs,
"cost_of_work": cost_of_insulation,
"cost_of_work_after_funding": cost_of_work_after_funding
}
estimated_costs.append(to_append)
continue
estimated_costs = pd.DataFrame(estimated_costs)
estimated_costs.to_csv("/Users/khalimconn-kowlessar/Documents/hestia/Customers/sfr/estimated_costs_gbis.csv")
# epc_data[["UPRN", "ADDRESS", "POSTCODE"]].to_csv(
# "/Users/khalimconn-kowlessar/Documents/hestia/sfr/council_tax_bands_sample.csv")
n_properties_for_ashp = epc_data[
(epc_data["PROPERTY_TYPE"] == "House") &
(epc_data["BUILT_FORM"].isin(["Detached", "Semi-Detached"]))
].shape[0]