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

preparing caha

Browse source
This commit is contained in:
Khalim Conn-Kowlessar 2024-11-12 18:21:19 +00:00
parent 2e78ba6d5d
commit 6d01490962
2 changed files with 47 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
etl/customers/ksquared/Wave3 Modelling.py
View file

@ -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()

11
etl/find_my_epc/RetrieveFindMyEpc.py
View file

@ -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 = []