diff --git a/etl/customers/ksquared/Wave3 Modelling.py b/etl/customers/ksquared/Wave3 Modelling.py index 6a507728..159fb20b 100644 --- a/etl/customers/ksquared/Wave3 Modelling.py +++ b/etl/customers/ksquared/Wave3 Modelling.py @@ -1,5 +1,6 @@ import os import time +import re from dotenv import load_dotenv from tqdm import tqdm @@ -187,16 +188,52 @@ def caha(): remap = { "Flat A, 50 Talbot Road N6 4QP": "50a Talbot Road", "Flat A, 51 First Avenue EN1 1BN": "51a, First Avenue", + "Flat B, 51 First Avenue EN1 1BN": "51b, First Avenue" } + + def remap_address(address): + # Match patterns like 'Flat A, 30 Grove Park Road' + match = re.match(r'Flat (\w), (\d+) (.+)', address) + if match: + flat_letter = match.group(1) # e.g., 'A' + number = match.group(2) # e.g., '30' + rest_of_address = match.group(3) # e.g., 'Grove Park Road' + + # Format the new address as '30A Grove Park Road' + return f"{number}{flat_letter} {rest_of_address}" + + # If pattern doesn't match, return original address + return address + extracted_data = [] asset_list = [] for _, home in tqdm(caha_asset_list.iterrows(), total=len(caha_asset_list)): + if home["Street address"] == "35 Stanford road N11 3HY" and home["Address letter or number"] == "": + continue + + if home["Street address"] == "29 Victoria Avenue N3 1BD" and home["Address letter or number"] == "": + continue + + if home["Street address"] == "11 Victoria Avenue N3 1BD" and home["Address letter or number"] == "Flat A": + continue + + if home["Street address"] == "11 Victoria Avenue N3 1BD" and home["Address letter or number"] == "Flat C": + continue + + if home["Street address"] == "10 Forest Gardens N17 6XA" and home["Address letter or number"] == "Flat C": + continue + + if home["Street address"] == "219 Cann Hall Road E11 3NJ" and home["Address letter or number"] == "Flat B": + continue + unit_number = home["Address letter or number"] street = home["Street address"] postcode = home["Postcode"] address = ", ".join([x for x in [unit_number, street] if x]) address = remap.get(address, address) address = address.replace(postcode, "").strip() + if "Victoria Avenue" not in address: + address = remap_address(address) find_epc_searcher = RetrieveFindMyEpc(address=address, postcode=postcode) find_epc_data = find_epc_searcher.retrieve_newest_find_my_epc_data() diff --git a/etl/find_my_epc/RetrieveFindMyEpc.py b/etl/find_my_epc/RetrieveFindMyEpc.py index bcd3c356..e8e1ff1d 100644 --- a/etl/find_my_epc/RetrieveFindMyEpc.py +++ b/etl/find_my_epc/RetrieveFindMyEpc.py @@ -204,7 +204,8 @@ class RetrieveFindMyEpc: return resulting_data - def format_recommendations(self, recommendations): + @staticmethod + def format_recommendations(recommendations): """ This function converts the recommendations to a format that we can use in the engine as a non-intrusive survey :param recommendations: @@ -217,6 +218,7 @@ class RetrieveFindMyEpc: "Hot water cylinder thermostat": ["cylinder_thermostat"], "High performance external doors": ["insulated_doors"], "Floor insulation (solid floor)": ["solid_floor_insulation"], + "Floor insulation (suspended floor)": ["suspended_floor_insulation"], "Double glazed windows": ["double_glazing"], "Cavity wall insulation": ["cavity_wall_insulation"], "Replace boiler with new condensing boiler": ["boiler_upgrade"], @@ -235,6 +237,13 @@ class RetrieveFindMyEpc: "roomstat_programmer_trvs", "time_temperature_zone_control" ], "Change heating to gas condensing boiler": ["boiler_upgrade"], + "Fan assisted storage heaters and dual immersion cylinder": ["high_heat_retention_storage_heaters"], + "Flat roof or sloping ceiling insulation": ["flat_roof_insulation"], + "Heating controls (room thermostat)": [ + "roomstat_programmer_trvs", "time_temperature_zone_control" + ], + "Band A condensing boiler": ["boiler_upgrade"], + "Double glazing": ["double_glazing"], } formatted_recommendations = []