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

working on eligibility

Browse source
This commit is contained in:
Khalim Conn-Kowlessar 2024-02-26 23:23:29 +00:00
parent b80ffda392
commit 281c6f626c
4 changed files with 167 additions and 100 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
backend/Property.py
View file

@ -147,7 +147,8 @@ class Property:
# self.base_difference_record.df
def adjust_difference_record_with_recommendations(
self, property_recommendations,
self,
property_recommendations,
property_representative_recommendations
):
"""

90
etl/eligibility/Eligibility.py
View file

@ -145,6 +145,7 @@ class Eligibility:
"reason": None,
"thickness_classification": thickness_classification
}
return
# Insulation is already thick enough
self.loft = {
@ -164,8 +165,10 @@ class Eligibility:
"""
is_cavity = self.walls["is_cavity_wall"]
is_empty = (not self.walls["is_filled_cavity"]) or (
is_empty = (not self.walls["is_filled_cavity"])
is_as_built = (
self.walls["is_as_built"] and self.walls["insulation_thickness"] not in ["average", "above average"]
and self.walls["is_assumed"]
)
is_partial_filled = "partial" in self.walls["clean_description"].lower()
# We look for potentially under performing cavities - anything that is assumed, as built and insulated
@ -175,6 +178,7 @@ class Eligibility:
is_unfilled_cavity = is_cavity and (is_empty and not is_partial_filled)
is_partial_filled_cavity = is_cavity and is_partial_filled
is_assumed_filled_cavity = is_cavity and is_as_built
is_underperforming_cavity = is_cavity and is_underperforming
# Check if it has internal or external wall insulation
@ -195,6 +199,13 @@ class Eligibility:
}
return
if is_assumed_filled_cavity:
self.cavity = {
"suitability": True,
"type": "as built assumed",
}
return
if is_partial_filled_cavity:
self.cavity = {
"suitability": True,
@ -345,7 +356,7 @@ class Eligibility:
int(self.epc["current-energy-efficiency"]) <= 68
)
def check_eco4_warmfront(self, post_retrofit_sap=None):
def check_eco4_warmfront(self):
"""
This funciton will check if the property is eligible for funding under the ECO4 scheme
@ -377,49 +388,100 @@ class Eligibility:
self.cavity_insulation()
self.loft_insulation()
# make sure conditions 2 and 3 are true
is_eligible = self.cavity["suitability"] & self.loft["suitability"]
if current_sap >= 69:
# Case 1: No conditions meet
if not self.cavity["suitability"] and (self.loft["thickness"] > 100) and current_sap >= 55:
self.eco4_warmfront = {
"eligible": False,
"message": "SAP too high",
"strict": False,
"message": "All conditions fail",
"cavity_type": self.cavity["type"],
"loft_type": self.loft["thickness_classification"]
}
return
if not is_eligible and current_sap >= 55:
# Case 2 - perfect match
if (self.cavity["type"] == "empty") and (self.loft["thickness"] <= 100) and (current_sap < 55):
self.eco4_warmfront = {
"eligible": False,
"message": "failed fabric and SAP check",
"eligible": True,
"strict": True,
"message": "Perfect suitability",
"cavity_type": self.cavity["type"],
"loft_type": self.loft["thickness_classification"]
}
return
if not is_eligible and current_sap < 55:
# Case 2.5 - near perfect match - but we would not recommend this using the model
if self.cavity["suitability"] and (self.loft["thickness"] <= 100) and (current_sap < 55):
self.eco4_warmfront = {
"eligible": True,
"strict": True,
"message": "Perfect suitability",
"cavity_type": self.cavity["type"],
"loft_type": self.loft["thickness_classification"]
}
return
# Case 3 - cavity is suitable, loft is not, sap is good
if self.cavity["suitability"] and (self.loft["thickness"] > 100) and (current_sap < 55):
self.eco4_warmfront = {
"eligible": True,
"strict": False,
"message": "Meets cavity and sap",
"cavity_type": self.cavity["type"],
"loft_type": self.loft["thickness_classification"]
}
return
# Case 4 - cavity is not suitable, loft is, sap is not - we say this is not elifible
if not self.cavity["suitability"] and (self.loft["thickness"] <= 100) and (current_sap < 55):
self.eco4_warmfront = {
"eligible": False,
"strict": False,
"message": "failed fabric check",
"cavity_type": self.cavity["type"],
"loft_type": self.loft["thickness_classification"]
}
return
if is_eligible and current_sap >= 55:
# Case 5 - cavity and loft suitable, sap too high
if self.cavity["suitability"] and (self.loft["thickness"] <= 100) and (current_sap >= 55):
self.eco4_warmfront = {
"eligible": True,
"strict": False,
"message": "Meets fabric, fails SAP check",
"cavity_type": self.cavity["type"],
"loft_type": self.loft["thickness_classification"]
}
return
if is_eligible and current_sap < 55:
# Case 6 - meets just cavity
if self.cavity["suitability"] and (self.loft["thickness"] > 100) and (current_sap >= 55):
self.eco4_warmfront = {
"eligible": True,
"message": "Meets fabric and SAP check",
"strict": False,
"message": "Meets just cavity",
"cavity_type": self.cavity["type"],
"loft_type": self.loft["thickness_classification"]
}
return
# Case 7 - fails cavity, loft but meets sap
if not self.cavity["suitability"] and (self.loft["thickness"] > 100) and (current_sap < 55):
self.eco4_warmfront = {
"eligible": False,
"strict": False,
"message": "Fails cavity nd lodt, meets SAP",
"cavity_type": self.cavity["type"],
"loft_type": self.loft["thickness_classification"]
}
return
# Case 8 - fails cavity, meets loft, fails sap
if not self.cavity["suitability"] and (self.loft["thickness"] <= 100) and (current_sap >= 55):
self.eco4_warmfront = {
"eligible": False,
"strict": False,
"message": "Fails cavity, meets loft, fails SAP",
"cavity_type": self.cavity["type"],
"loft_type": self.loft["thickness_classification"]
}

18
etl/eligibility/ha_15_32/app.py
View file

@ -387,17 +387,19 @@ def prepare_model_data_row(
}
simulations = [
[cavity_simulation],
[loft_simulation]
cavity_simulation,
loft_simulation
]
p.adjust_difference_record_with_recommendations(simulations)
recommendation_record = p.base_difference_record.df.to_dict("records")[0].copy()
scoring_dict = p.create_recommendation_scoring_data(
property_id=p.id,
recommendation_record=recommendation_record,
recommendations=simulations,
primary_recommendation_id=cavity_simulation["recommendation_id"]
)
# Make sure we definitely have the correct data
cavity_scoring = [x for x in p.recommendations_scoring_data if "cavity" in x["id"]][0]
loft_scoring = [x for x in p.recommendations_scoring_data if "loft" in x["id"]][0]
return [cavity_scoring, loft_scoring]
return [scoring_dict]
def get_ha_32data(ha_data, cleaned, cleaning_data, created_at):

156
etl/eligibility/ha_15_32/ha_analysis_batch_3.py
View file

@ -1114,7 +1114,7 @@ def get_epc_data(
results = []
scoring_data = []
nodata = []
for index, property_meta in tqdm(asset_list.iterrows(), total=len(asset_list)):
for index, property_meta in tqdm(eco4.iterrows(), total=len(eco4)):
if property_meta["matching_postcode"] is None:
continue
@ -1226,10 +1226,6 @@ def get_epc_data(
# We check the age of the cavity and if it's particularly old, we flag it
cavity_age = calculate_cavity_age(newest_epc, older_epcs, cleaned)
# Full checks
eligibility.check_gbis()
eligibility.check_eco4()
if eligibility.eco4_warmfront["eligible"]:
if eligibility.epc["uprn"] == "":
eligibility.epc["uprn"] = int(property_meta["asset_list_row_id"].split(ha_name)[1])
@ -1256,8 +1252,8 @@ def get_epc_data(
"gbis_eligible": eligibility.gbis_warmfront,
"eco4_eligible": eligibility.eco4_warmfront["eligible"],
"eco4_message": eligibility.eco4_warmfront["message"],
"eco4_strict": eligibility.eco4_warmfront["strict"],
"sap": float(eligibility.epc["current-energy-efficiency"]),
# Property components
"roof": eligibility.roof["clean_description"],
"walls": eligibility.walls["clean_description"],
@ -1267,91 +1263,97 @@ def get_epc_data(
"date_epc": eligibility.epc["lodgement-date"],
"loft_thickness": eligibility.roof["insulation_thickness"],
"cavity_age": cavity_age,
**eligibility.walls,
**eligibility.roof,
"eligibility_cavity_type": eligibility.eco4_warmfront["cavity_type"],
"eligibility_loft_type": eligibility.eco4_warmfront["loft_type"]
}
)
scoring_df = pd.DataFrame(scoring_data)
scoring_df = scoring_df.drop(
columns=[
"rdsap_change", "heat_demand_change", "carbon_change", "sap_ending", "heat_demand_ending",
"carbon_ending"
]
)
model_api = ModelApi(portfolio_id="-".join([ha_name, "eligibility"]), timestamp=created_at)
# scoring_df["is_community"].value_counts()
# scoring_df[scoring_df["is_community"] == "Unknown"]
# property_meta = asset_list[asset_list["asset_list_row_id"] == "ha_67238"].squeeze()
all_predictions = model_api.predict_all(
df=scoring_df,
bucket="retrofit-data-dev",
prediction_buckets={
"sap_change_predictions": "retrofit-sap-predictions-dev",
"heat_demand_predictions": "retrofit-heat-predictions-dev",
"carbon_change_predictions": "retrofit-carbon-predictions-dev"
}
)
results_df = pd.DataFrame(results)
scoring_df = pd.DataFrame(scoring_data)
results_df["post_install_sap"] = None
results_df["eligibility_classification"] = None
predictions = all_predictions["sap_change_predictions"].copy()
eco4 = asset_list[asset_list["ECO Eligibility"] == "eco4"]
z = results_df[results_df["row_id"].isin(eco4["asset_list_row_id"])]
z["walls"].value_counts()
z1 = z[z["walls"] == "Cavity wall, as built, no insulation"]
k = z1[z1["roof"] == "Pitched, 100 mm loft insulation"]
property_meta = asset_list[asset_list["asset_list_row_id"] == k["row_id"].values[0]].squeeze()
z[z["walls"] == "Cavity wall, as built, insulated"]["roof"].value_counts()
z[z["walls"] == "Cavity wall, as built, insulated"]["roof"].value_counts()
predictions = predictions.rename(columns={"property_id": "row_id"}).merge(
results_df[["row_id", "sap"]], how="left", on="row_id"
)
predictions["sap_uplift"] = predictions["predictions"] - predictions["sap"]
predictions = predictions.groupby("row_id")["sap_uplift"].sum().reset_index()
if not scoring_df.empty:
scoring_df = scoring_df.drop(
columns=[
"rdsap_change", "heat_demand_change", "carbon_change", "sap_ending", "heat_demand_ending",
"carbon_ending"
]
)
results_df = results_df.merge(
predictions[["sap_uplift", "row_id"]],
how="left",
on="row_id"
)
results_df["post_install_sap"] = results_df["sap"] + results_df["sap_uplift"]
model_api = ModelApi(portfolio_id="-".join([ha_name, "eligibility"]), timestamp=created_at)
eligibility_assessment = []
for _, row in results_df[results_df["eco4_eligible"] == True].iterrows():
# The upgrade requirements are dependent on the current SAP
# If the property is an F or G, it only needs to upgrade to an %
if row["sap"] <= 38:
if row["post_install_sap"] >= 57:
eligibility_classification = "highest confidence"
elif row["post_install_sap"] >= 55:
eligibility_classification = "high confidence"
elif row["post_install_sap"] >= 53:
eligibility_classification = "medium confidence"
else:
eligibility_classification = "unlikely"
else:
if row["post_install_sap"] >= 71:
eligibility_classification = "highest confidence"
elif row["post_install_sap"] >= 69:
eligibility_classification = "high confidence"
elif row["post_install_sap"] >= 67:
eligibility_classification = "medium confidence"
else:
eligibility_classification = "unlikely"
eligibility_assessment.append(
{
"row_id": row["row_id"],
"eligibility_classification": eligibility_classification
all_predictions = model_api.predict_all(
df=scoring_df,
bucket="retrofit-data-dev",
prediction_buckets={
"sap_change_predictions": "retrofit-sap-predictions-dev",
"heat_demand_predictions": "retrofit-heat-predictions-dev",
"carbon_change_predictions": "retrofit-carbon-predictions-dev"
}
)
eligibility_assessment = pd.DataFrame(eligibility_assessment)
predictions = all_predictions["sap_change_predictions"].copy()
results_df = results_df.merge(
eligibility_assessment, how="left", on="row_id"
)
predictions = predictions.rename(columns={"property_id": "row_id"}).merge(
results_df[["row_id", "sap"]], how="left", on="row_id"
)
predictions["sap_uplift"] = predictions["predictions"] - predictions["sap"]
predictions = predictions.groupby("row_id")["sap_uplift"].sum().reset_index()
results_df = results_df.merge(
predictions[["sap_uplift", "row_id"]],
how="left",
on="row_id"
)
results_df["post_install_sap"] = results_df["sap"] + results_df["sap_uplift"]
eligibility_assessment = []
for _, row in results_df[results_df["eco4_eligible"] == True].iterrows():
# The upgrade requirements are dependent on the current SAP
# If the property is an F or G, it only needs to upgrade to an %
if row["sap"] <= 38:
if row["post_install_sap"] >= 57:
eligibility_classification = "highest confidence"
elif row["post_install_sap"] >= 55:
eligibility_classification = "high confidence"
elif row["post_install_sap"] >= 53:
eligibility_classification = "medium confidence"
else:
eligibility_classification = "unlikely"
else:
if row["post_install_sap"] >= 71:
eligibility_classification = "highest confidence"
elif row["post_install_sap"] >= 69:
eligibility_classification = "high confidence"
elif row["post_install_sap"] >= 67:
eligibility_classification = "medium confidence"
else:
eligibility_classification = "unlikely"
eligibility_assessment.append(
{
"row_id": row["row_id"],
"eligibility_classification": eligibility_classification
}
)
eligibility_assessment = pd.DataFrame(eligibility_assessment)
results_df = results_df.merge(
eligibility_assessment, how="left", on="row_id"
)
# We store the results in S3 as a pickle
save_pickle_to_s3(