mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
229 lines
8.4 KiB
Python
229 lines
8.4 KiB
Python
"""
|
|
Simple script to take a standardised asset list and calculate the abs. We'll use this code to estimate
|
|
the ABS for properties, going forward
|
|
"""
|
|
import os
|
|
import pandas as pd
|
|
import numpy as np
|
|
from dotenv import load_dotenv
|
|
from etl.find_my_epc.AssetListEpcData import AssetListEpcData
|
|
from backend.Funding import Funding
|
|
from backend.app.utils import sap_to_epc
|
|
|
|
load_dotenv(dotenv_path="backend/.env")
|
|
EPC_AUTH_TOKEN = os.getenv("EPC_AUTH_TOKEN")
|
|
|
|
asset_list = pd.read_excel(
|
|
"/Users/khalimconn-kowlessar/Documents/hestia/Instagroup Review/Livewest South-West - Standardised V2.xlsx",
|
|
sheet_name="Cavity Route (Insta Review)"
|
|
)
|
|
|
|
abs_matrix = pd.read_csv(
|
|
"/Users/khalimconn-kowlessar/Downloads/ECO4 Full Project Scores Matrix.csv"
|
|
)
|
|
pps_matrix = pd.read_excel(
|
|
"/Users/khalimconn-kowlessar/Downloads/ECO4 Partial Project Scores Matrix v5.xlsx",
|
|
header=1
|
|
)
|
|
pps_matrix.columns = [c.strip() for c in pps_matrix.columns]
|
|
|
|
# We need to estimate the number of points the work will produce and the finishing band. For this, we assume 7 for
|
|
# cavity and 15 for solar. We'll be more specific in the future, but for now, this is a good enough estimate.
|
|
route = asset_list[["domna_address_1", "domna_postcode", "epc_os_uprn"]].rename(
|
|
columns={"domna_address_1": "address", "domna_postcode": "postcode", "epc_os_uprn": "upr"}
|
|
)
|
|
route["address"] = route["address"].astype(str)
|
|
|
|
asset_list_epc_client = AssetListEpcData(
|
|
asset_list=route,
|
|
epc_auth_token=EPC_AUTH_TOKEN
|
|
)
|
|
|
|
asset_list_epc_client.get_data()
|
|
asset_list_epc_client.get_non_invasive_recommendations()
|
|
|
|
solar_sap_points = []
|
|
for r in asset_list_epc_client.non_invasive_recommendations:
|
|
if not r.get("recommendations"):
|
|
continue
|
|
solar_recommendations = [
|
|
x for x in r["recommendations"] if "solar_pv" in x["type"]
|
|
]
|
|
if solar_recommendations:
|
|
solar_recommendations = solar_recommendations[0]
|
|
else:
|
|
continue
|
|
|
|
address = r["address"]
|
|
postcode = r["postcode"]
|
|
|
|
solar_sap_points.append(
|
|
{
|
|
"address": address,
|
|
"postcode": postcode,
|
|
"sap_points": solar_recommendations["sap_points"]
|
|
}
|
|
)
|
|
|
|
solar_sap_points = pd.DataFrame(solar_sap_points)
|
|
solar_sap_points.drop_duplicates(subset=["address", "postcode"], inplace=True)
|
|
# Store the sap points in the cavity route to csv
|
|
# cwi_sap_points.to_csv(
|
|
# "/Users/khalimconn-kowlessar/Documents/hestia/Instagroup Review/cwi_sap_points_livewest_sw.csv",
|
|
# index=False
|
|
# )
|
|
|
|
avg_solar_points_by_postcode = solar_sap_points.groupby(["postcode"]).agg({"sap_points": "mean"}).reset_index()
|
|
avg_solar_points = solar_sap_points["sap_points"].median()
|
|
asset_list["domna_address_1"] = asset_list["domna_address_1"].astype(str)
|
|
asset_list = asset_list.merge(
|
|
solar_sap_points, how="left", left_on=["domna_address_1", "domna_postcode"], right_on=["address", "postcode"]
|
|
).drop(
|
|
columns=["address", "postcode"]
|
|
)
|
|
|
|
# Fill the sap points with the average cwi points
|
|
asset_list = asset_list.merge(
|
|
avg_solar_points_by_postcode.rename(columns={"postcode": "domna_postcode"}),
|
|
how="left", on=["domna_postcode"], suffixes=("", "_avg")
|
|
)
|
|
asset_list["sap_points"] = asset_list["sap_points"].fillna(asset_list["sap_points_avg"])
|
|
asset_list.drop(columns=["sap_points_avg"], inplace=True)
|
|
|
|
asset_list["sap_points"] = asset_list["sap_points"].fillna(avg_solar_points)
|
|
asset_list["post_works_sap"] = asset_list["epc_sap_score_on_register"] + asset_list["sap_points"]
|
|
asset_list["post_works_epc"] = asset_list["post_works_sap"].apply(lambda x: sap_to_epc(x))
|
|
asset_list["starting_half_band"] = asset_list["epc_sap_score_on_register"].apply(lambda x: Funding.get_sap_band(x))
|
|
asset_list["ending_half_band"] = asset_list["post_works_sap"].apply(lambda x: Funding.get_sap_band(x))
|
|
asset_list["floor_area_band"] = asset_list["epc_total_floor_area"].apply(lambda x: Funding.get_floor_area_band(x))
|
|
|
|
asset_list["ending_half_band"] = np.where(
|
|
(asset_list["post_works_epc"] == asset_list["epc_rating_on_register"]),
|
|
"Low_C",
|
|
asset_list["ending_half_band"]
|
|
)
|
|
# Realistically, we'll take the properties to a low C at worst
|
|
asset_list["ending_half_band"] = np.where(
|
|
(asset_list["post_works_sap"] < 69),
|
|
"Low_C",
|
|
asset_list["ending_half_band"]
|
|
)
|
|
|
|
asset_list = asset_list.merge(
|
|
abs_matrix, how="left", left_on=["starting_half_band", "ending_half_band", "floor_area_band"],
|
|
right_on=['Starting Band', 'Finishing Band', 'Floor Area Segment', ]
|
|
)
|
|
asset_list = asset_list.drop(columns=['Starting Band', 'Finishing Band', 'Floor Area Segment'])
|
|
|
|
asset_list = asset_list.rename(
|
|
columns={"Cost Savings": "funding_abs"}
|
|
)
|
|
|
|
print(asset_list["domna_property_id"].duplicated().sum())
|
|
|
|
# Store this data
|
|
asset_list.to_csv(
|
|
"/Users/khalimconn-kowlessar/Documents/hestia/Instagroup Review/livewest_sw_solar_abs_estimates-solar.csv",
|
|
index=False
|
|
)
|
|
|
|
# Cavity process!
|
|
# cwi_sap_points = []
|
|
# for r in asset_list_epc_client.non_invasive_recommendations:
|
|
# if not r.get("recommendations"):
|
|
# continue
|
|
# cwi_recommendations = [
|
|
# x for x in r["recommendations"] if "cavity_wall_insulation" in x["type"]
|
|
# ]
|
|
# if cwi_recommendations:
|
|
# cwi_recommendations = cwi_recommendations[0]
|
|
# else:
|
|
# continue
|
|
#
|
|
# address = r["address"]
|
|
# postcode = r["postcode"]
|
|
#
|
|
# cwi_sap_points.append(
|
|
# {
|
|
# "address": address,
|
|
# "postcode": postcode,
|
|
# "sap_points": cwi_recommendations["sap_points"]
|
|
# }
|
|
# )
|
|
#
|
|
# cwi_sap_points = pd.DataFrame(cwi_sap_points)
|
|
# cwi_sap_points = pd.read_csv(
|
|
# "/Users/khalimconn-kowlessar/Documents/hestia/Instagroup Review/cwi_sap_points_livewest_sw.csv"
|
|
# )
|
|
# cwi_sap_points.drop_duplicates(subset=["address", "postcode"], inplace=True)
|
|
avg_cwi_points_by_postcode = cwi_sap_points.groupby(["postcode"]).agg({"sap_points": "mean"}).reset_index()
|
|
avg_cwi_points = cwi_sap_points["sap_points"].median()
|
|
asset_list = asset_list.merge(
|
|
cwi_sap_points, how="left", left_on=["domna_address_1", "domna_postcode"], right_on=["address", "postcode"]
|
|
).drop(
|
|
columns=["address", "postcode"]
|
|
)
|
|
|
|
# Fill the sap points with the average cwi points
|
|
asset_list = asset_list.merge(
|
|
avg_cwi_points_by_postcode.rename(columns={"postcode": "domna_postcode"}),
|
|
how="left", on=["domna_postcode"], suffixes=("", "_avg")
|
|
)
|
|
asset_list["sap_points"] = asset_list["sap_points"].fillna(asset_list["sap_points_avg"])
|
|
asset_list.drop(columns=["sap_points_avg"], inplace=True)
|
|
|
|
asset_list["sap_points"] = asset_list["sap_points"].fillna(avg_cwi_points)
|
|
asset_list["post_works_sap"] = asset_list["epc_sap_score_on_register"] + asset_list["sap_points"]
|
|
asset_list["post_works_epc"] = asset_list["post_works_sap"].apply(lambda x: sap_to_epc(x))
|
|
asset_list["starting_half_band"] = asset_list["epc_sap_score_on_register"].apply(lambda x: Funding.get_sap_band(x))
|
|
asset_list["ending_half_band"] = asset_list["post_works_sap"].apply(lambda x: Funding.get_sap_band(x))
|
|
asset_list["floor_area_band"] = asset_list["epc_total_floor_area"].apply(lambda x: Funding.get_floor_area_band(x))
|
|
|
|
asset_list["funding_scheme"] = np.where(
|
|
(
|
|
(asset_list["post_works_epc"] == asset_list["epc_rating_on_register"])
|
|
),
|
|
"GBIS",
|
|
"ECO4"
|
|
)
|
|
asset_list = asset_list.merge(
|
|
abs_matrix, how="left", left_on=["starting_half_band", "ending_half_band", "floor_area_band"],
|
|
right_on=['Starting Band', 'Finishing Band', 'Floor Area Segment', ]
|
|
)
|
|
asset_list = asset_list.drop(columns=['Starting Band', 'Finishing Band', 'Floor Area Segment'])
|
|
|
|
# Using CWI solid 1.7 -> 0.3 rates
|
|
cwi_pps_matrix = pps_matrix[
|
|
pps_matrix["Measure_Type"].isin(["CWI_0.033"])
|
|
]
|
|
# Merge on
|
|
asset_list = asset_list.merge(
|
|
cwi_pps_matrix[['Starting Band', 'Total Floor Area Band', 'Cost Savings']].rename(
|
|
columns={
|
|
"Cost Savings": "partial_project_score",
|
|
"Starting Band": "starting_half_band",
|
|
"Total Floor Area Band": "floor_area_band"
|
|
}
|
|
),
|
|
how="left",
|
|
on=["starting_half_band", "floor_area_band"],
|
|
)
|
|
asset_list["partial_project_score"] = np.where(
|
|
(asset_list["epc_sap_score_on_register"] > 69),
|
|
None,
|
|
asset_list["partial_project_score"]
|
|
)
|
|
|
|
asset_list["funding_abs"] = np.where(
|
|
asset_list["funding_scheme"] == "GBIS",
|
|
asset_list["partial_project_score"],
|
|
asset_list["Cost Savings"]
|
|
)
|
|
|
|
asset_list["domna_property_id"].duplicated().sum()
|
|
|
|
# Store this data
|
|
asset_list.to_csv(
|
|
"/Users/khalimconn-kowlessar/Documents/hestia/Instagroup Review/livewest_sw_abs_estimates.csv",
|
|
index=False
|
|
)
|