from tqdm import tqdm import os from model_data.BoreholeClient import BoreholeClient from model_data.LandRegistryClient import LandRegistryClient from model_data.ConservationAreaClient import ConservationAreaClient from model_data.temp_inputs import input_data from model_data.Property import Property from model_data.config import EPC_AUTH_TOKEN from epc_api.client import EpcClient from model_data.downloader import pagenated_epc_download from model_data.EpcClean import EpcClean from model_data.OpenUprnClient import OpenUprnClient from model_data.analysis.UvalueEstimations import UvalueEstimations LAND_REGISTRY_PATHS = [ os.path.abspath(os.path.dirname(__file__)) + "/model_data/local_data/pp-monthly-update-new-version.csv", os.path.abspath(os.path.dirname(__file__)) + "/model_data/local_data/pp-2022 (1).csv", os.path.abspath(os.path.dirname(__file__)) + "/model_data/local_data/pp-2021.csv", os.path.abspath(os.path.dirname(__file__)) + "/model_data/local_data/pp-2020.csv", os.path.abspath(os.path.dirname(__file__)) + "/model_data/local_data/pp-2019.csv", os.path.abspath(os.path.dirname(__file__)) + "/model_data/local_data/pp-2018.csv", os.path.abspath(os.path.dirname(__file__)) + "/model_data/local_data/pp-2017-part1.csv", os.path.abspath(os.path.dirname(__file__)) + "/model_data/local_data/pp-2017-part2.csv", ] def handler(): # To begin with, the input data is a list of dictionaries, however we would read this file in epc_client = EpcClient(auth_token=EPC_AUTH_TOKEN) input_properties = [ Property(postcode=config['postcode'], address1=config['address1'], epc_client=epc_client) for config in input_data ] for p in input_properties: p.search_address_epc() p.set_year_built() uprns = [p.data['uprn'] for p in input_properties] open_uprn_client = OpenUprnClient( path=os.path.abspath( os.path.dirname(__file__) ) + "/model_data/local_data/osopenuprn_202306_csv/osopenuprn_202305.csv", uprns=uprns ) open_uprn_client.read() # We're using Ordinance Survey Open Uprn data # to find the coordinates of each address, which we will then be able to use at a later stage for p in input_properties: p.get_coordinates(open_uprn_client) conservation_area_client = ConservationAreaClient( historic_england_path=os.path.abspath( os.path.dirname(__file__) ) + "/model_data/local_data/Historic_Eng_Conservation_Areas/Conservation_Areas.shp", gov_path=os.path.abspath( os.path.dirname(__file__) ) + "/model_data/local_data/gov-conservation-area.geojson" ) conservation_area_client.read() # Check if the property is in a conversation area for p in input_properties: p.set_is_in_conservation_area(conservation_area_client) local_authorities = {p.data['local-authority'] for p in input_properties} data = [] for la in tqdm(local_authorities): data.extend( pagenated_epc_download( client=epc_client, params={"local-authority": la}, page_size=5000, n_pages=10, ) ) # Incorporate input data into cleaning cleaner = EpcClean(data + [p.data for p in input_properties]) cleaner.clean() address_meta = [ { "postcode": x["postcode"].upper(), "address1": x["address1"].upper(), "address2": x["address2"].upper(), "address3": x["address3"].upper(), "address": x["address"], "uprn": x["uprn"] } for x in data ] # Land registry land_registry_client = LandRegistryClient( paths=LAND_REGISTRY_PATHS, addresses=address_meta ) land_registry_client.read() # Borehole borehole_client = BoreholeClient( path=os.path.abspath(os.path.dirname(__file__)) + "/model_data/local_data/borehole/borehole.dbf" ) borehole_client.read() # Now, for our input properties, we need to identify the components of the building, based # on the cleaning we've done for p in input_properties: p.get_components(cleaner) uvalue_estimates = UvalueEstimations(data=data) uvalue_estimates.get_estimates(cleaner=cleaner) # Now, given the components, we want to idenfity upgrade options import pandas as pd floors_df = pd.DataFrame( [{"address1": p.address1, **p.floor} for p in input_properties] ) input_properties[2].data["address1"] input_properties[2].data["postcode"] floors_df["address1"].values[2] floors_df["original_description"].values[2] from model_data.recommendations.FloorRecommendations import FloorRecommendations self = FloorRecommendations(property_instance=input_properties[2], uvalue_estimates=uvalue_estimates) # We need to deduce a U-value for "Good" energy effieciency mainheating = pd.DataFrame( [{"address1": p.address1, "postcode": p.postcode, **p.main_heating} for p in input_properties]) hotwater = pd.DataFrame([{"address1": p.address1, **p.hotwater} for p in input_properties]) mainheating[["address1", "postcode"]] # TODO: I want to knwo what "Good" efficiency means for the description # 'Flat 28, 22 Adelina Grove' 'Solid brick, as built, insulated (assumed)' # so to do this, filter on the local authority code and property type, where we have U # values for the wall and take a median! p = input_properties[6] df = pd.DataFrame(data) res = [] for p in input_properties: distances = [] for borehole in tqdm(borehole_client.data, total=len(borehole_client.data)): dist_meeters, _ = borehole_client.distance_between_bng_coords( x1_bng=p.coordinates['x_coordinate'], y1_bng=p.coordinates['y_coordinate'], x2_bng=float(borehole['EASTING']), y2_bng=float(borehole['NORTHING']) ) distances.append(dist_meeters) res.append( { "uprn": int(p.data["uprn"]), "meters_to_nearest_borehole": min(distances) } ) res = pd.DataFrame(res) properties_dataset = [ { **p.data, "in_conservation_area": p.in_conservation_area, **p.coordinates, } for p in input_properties ] properties_dataset = pd.DataFrame(properties_dataset) properties_dataset = properties_dataset.merge(res, on="uprn", how="left") properties_dataset.to_csv("properties_dataset.csv")