import msgpack from pathlib import Path from datetime import datetime import pandas as pd from utils.s3 import read_from_s3 from utils.logger import setup_logger from dotenv import load_dotenv from backend.app.utils import read_parquet_from_s3 from tqdm import tqdm from backend.SearchEpc import SearchEpc from etl.eligibility.Eligibility import Eligibility from etl.eligibility.ha_15_32.app import prepare_model_data_row from etl.epc.DataProcessor import DataProcessor from etl.epc.settings import COLUMNS_TO_MERGE_ON from backend.ml_models.api import ModelApi import re ENV_FILE = Path(__file__).parent / "etl" / "eligibility" / "ha_15_32" / ".env" logger = setup_logger() load_dotenv(ENV_FILE) def load_ha_33(): """ Load HA33 data :return: """ pd.set_option('display.max_rows', 500) pd.set_option('display.max_columns', 500) pd.set_option('display.width', 1000) files = [ "HA 33 Assets 1 of 4.csv", "HA 33 Assets 2 of 4.csv", "HA 33 Assets 3 of 4.csv", "HA 33 Assets 4 of 4.csv" ] data = [] for file in files: part = pd.read_csv(f"etl/eligibility/ha_15_32/{file}", low_memory=False) cols_to_top = [c for c in part.columns if "Unnamed:" in c] part = part.drop(columns=cols_to_top) data.append(part) data = pd.concat(data) return data def standardise_ha33(data): data = data[~pd.isnull(data["ADDRESS"])] split_addresses = data['ADDRESS'].str.split(',', expand=True) split_addresses.columns = ['address1', 'address2', 'address3', 'address4', 'address5'] data = pd.concat([data, split_addresses], axis=1) del split_addresses # Using regex to replace 'FT {number}' or 'FT{number}', with '{number}' data['address1'] = data['address1'].str.replace(r'FT\s*(\d+)', r'\1', regex=True) data.columns = [col.strip() for col in data.columns] # TODO: we have 23 THIRTY SEVENTH AVENUE, can we replace THIRTY SEVENTH with 37TH return data def get_ha_33data(data, cleaned, cleaning_data, created_at): house_type_lookup = { "Bungalow": "Bungalow", "Flat": "Flat", 'House': "House", 'Maisonette': "Maisonette", 'Flalolflfp mujjjjunjimj': "Flat", 'STUDIO': "Flat", } # house = data[data["row_id"] == "h3390"].squeeze() flat_pattern = r'flat\s+(\d+)' # data = data[data["row_id"].isin(eco_row_ids)] scoring_data = [] results = [] nodata = [] for _, house in tqdm(data.iterrows(), total=len(data)): # Check if we gave a flat in address 3 if re.search(flat_pattern, house["address2"].lower(), re.IGNORECASE): address1 = house["address2"].strip() else: address1 = house["address1"].strip() # I.e. just a number if len(address1) <= 3: address1 = address1 + " " + house["address2"].strip() searcher = SearchEpc( address1=address1, postcode=house["POST CODE"] ) response = searcher.search() if response["status"] == 204: nodata.append(house["row_id"]) continue newest_epc, older_epcs, _ = searcher.retrieve( property_type=house_type_lookup.get(house["PROPERTY TYPE"], None), address=house["ADDRESS"], ) eligibility = Eligibility(epc=newest_epc, cleaned=cleaned) eligibility.check_gbis_warmfront() eligibility.check_eco4_warmfront() # If the house is not identified, we do a full gbis and eco4 check eligibility.check_gbis() eligibility.check_eco4() if eligibility.eco4_warmfront["eligible"]: scoring_dictionary = prepare_model_data_row( property_id=house["row_id"], modelling_epc=eligibility.epc, cleaned=cleaned, cleaning_data=cleaning_data, created_at=created_at ) scoring_data.extend(scoring_dictionary) # If nothing is eligible or gbis is eligible, then we make a record this results.append( { "row_id": house["row_id"], "gbis_eligible": eligibility.gbis_warmfront, "eco4_eligible": eligibility.eco4_warmfront["eligible"], "eco4_message": eligibility.eco4_warmfront["message"], "sap": float(eligibility.epc["current-energy-efficiency"]), "gbis_eligible_future": eligibility.gbis["eligible"], "gbis_eligible_future_message": eligibility.gbis["message"], "eco4_eligible_future": eligibility.eco4["eligible"], "eco4_eligible_future_message": eligibility.eco4["message"], # Property components "roof": eligibility.roof["clean_description"], "walls": eligibility.walls["clean_description"], "heating": eligibility.epc["mainheat-description"], "tenure": eligibility.tenure, "date_epc": eligibility.epc["lodgement-date"], } ) # import pickle # with open("ha33_results.pickle", "wb") as f: # pickle.dump({ # "results": results, # "scoring_data": scoring_data, # "nodata": nodata # }, f) # with open("ha33_results.pickle", "rb") as f: # data = pickle.load(f) # results = data["results"] # scoring_data = data["scoring_data"] # nodata = data["nodata"] scoring_df = pd.DataFrame(scoring_data) # Implement the same process that is being used in the recommendation engine to cleaning scoring_df # Perform the same cleaning as in the model - first clean number of room variables though scoring_df = DataProcessor.apply_averages_cleaning( data_to_clean=scoring_df, cleaning_data=cleaning_data, cols_to_merge_on=['PROPERTY_TYPE', 'BUILT_FORM', 'CONSTRUCTION_AGE_BAND', 'LOCAL_AUTHORITY'], colnames=["NUMBER_HABITABLE_ROOMS", "NUMBER_HEATED_ROOMS"], ) scoring_df = DataProcessor.apply_averages_cleaning( data_to_clean=scoring_df, cleaning_data=cleaning_data, cols_to_merge_on=COLUMNS_TO_MERGE_ON + ["LOCAL_AUTHORITY"], ).drop(columns=["LOCAL_AUTHORITY"]) scoring_df = DataProcessor.clean_missings_after_description_process( scoring_df, ignore_cols=[c for c in scoring_df.columns if ("thermal_transmittance" in c) or ( "insulation_thickness" in c) or ("ENERGY_EFF" in c)] ) scoring_df = DataProcessor.clean_efficiency_variables(scoring_df) model_api = ModelApi(portfolio_id="ha33-eligibility", timestamp=created_at) 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" } ) # merge the predictions onto the scoring_df predictions = all_predictions["sap_change_predictions"].copy() results_df = pd.DataFrame(results) 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" ) return results_df, scoring_data, nodata def analyse_ha_33(results_df, data): # results_df_social = results_df[results_df["tenure"] == "Rented (social)"] # # results_df_social["tenure"].value_counts() data[data["row_id"].isin(results_df["row_id"].values)]["PROPERTY TYPE"].value_counts() n_identified = (results_df["gbis_eligible"] | results_df["eco4_eligible"]).sum() n_eco4 = results_df["eco4_eligible"].sum() n_gbis = results_df[~results_df["eco4_eligible"]]["gbis_eligible"].sum() eco_eligibile = results_df[results_df["eco4_eligible"]] eco_eligibile["walls"].value_counts() eco_eligibile["roof"].value_counts() results_df[results_df["gbis_eligible"] | results_df["eco4_eligible"]]["tenure"].value_counts() results_df_social["eligibility_classification"].value_counts() future_possibilities_eco = results_df[ (results_df["eco4_eligible_future"] == True) & (~(results_df["gbis_eligible"] | results_df["eco4_eligible"])) ].copy() future_possibilities_gbis = results_df[ (results_df["gbis_eligible_future"] == True) & (results_df["eco4_eligible_future"] == False) & ( ~(results_df["gbis_eligible"] | results_df["eco4_eligible"])) ].copy() def app(): """ Because HA33 is large, we deal with it separately :return: """ data = load_ha_33() data = standardise_ha33(data) data["row_id"] = ["h33" + str(i) for i in range(0, len(data))] cleaned = read_from_s3( s3_file_name="cleaned_epc_data/cleaned.bson", bucket_name="retrofit-data-dev" ) cleaned = msgpack.unpackb(cleaned, raw=False) cleaning_data = read_parquet_from_s3( bucket_name="retrofit-data-dev", file_key="sap_change_model/cleaning_dataset.parquet", ) created_at = datetime.now().isoformat() results_df, _, _ = get_ha_33data(data, cleaned, cleaning_data, created_at) # Read in import pickle with open("ha33_results.pickle", "rb") as f: data = pickle.load(f) results_df = pd.DataFrame(data["results"]) scoring_data = data["scoring_data"] nodata = data["nodata"]