import pandas as pd import re from epc_api.client import EpcClient from model_data.config import EPC_AUTH_TOKEN from model_data.OpenUprnClient import OpenUprnClient from model_data.EpcClean import EpcClean from model_data.BaseUtility import BaseUtility from model_data.ConservationAreaClient import ConservationAreaClient class Property(BaseUtility): ATTRIBUTE_MAP = { "floor-description": "floor", "hotwater-description": "hotwater", "main-fuel": "main_fuel", "mainheat-description": "main_heating", "mainheatcont-description": "main_heating_controls", "roof-description": "roof", "walls-description": "walls", "windows-description": "windows", "lighting-description": "lighting" } floor = None hotwater = None main_fuel = None main_heating = None main_heating_controls = None roof = None walls = None windows = None coordinates = None def __init__(self, postcode, address1, epc_client=None, data=None): self.postcode = postcode self.address1 = address1 self.data = data self.full_sap_epc = None self.in_conservation_area = None self.year_built = None if epc_client: self.epc_client = epc_client else: self.epc_client = EpcClient(auth_token=EPC_AUTH_TOKEN) def search_address_epc(self): """ This method searches for an address in the EPC database and returns the first result :return: property data """ if self.data: return # This will fail if a property does not have an EPC - this has been documented as a case to handle response = self.epc_client.domestic.search(params={"address": self.address1, "postcode": self.postcode}) # Check if we have a full sap EPC self.full_sap_epc = [r for r in response["rows"] if r["transaction-type"] == "new dwelling"] self.full_sap_epc = self.full_sap_epc[0] if self.full_sap_epc else self.full_sap_epc if len(response["rows"]) > 1: newest_response = [ r for r in response["rows"] if r["inspection-date"] == max([x["inspection-date"] for x in response["rows"]]) ] if len(newest_response) > 1: raise Exception("More than one result found for this address - investigate me") response["rows"] = newest_response self.data = response["rows"][0] def get_coordinates(self, open_uprn_client: OpenUprnClient): """ This method utlises the OpenOprnClient to get the coordinates of the property The OpenOprnClient interfactes with the Ordinance Survey Open UPRN database to extract property coordinates. This database holds lookups between UPRN and coordinates. :param open_uprn_client: Instance of OpenOprnClient. This method expects the client to have already read the data """ if open_uprn_client.data is None: raise ValueError("OpenUprnClient has not read data") self.coordinates = ( open_uprn_client.data[open_uprn_client.data["UPRN"] == int(self.data["uprn"])] .to_dict("records")[0] ) self.coordinates = {key.lower(): value for key, value in self.coordinates.items()} def get_components(self, cleaner: EpcClean): """ Given the cleaning that has been performed, we'll use this to identify the property components, from roof to walls to windows, heating and hot water :param cleaner: :return: """ if not cleaner.cleaned: raise ValueError("Cleaner does not contain cleaned data") if not self.data: raise ValueError("Property does not contain data") for description, attribute in cleaner.cleaned.items(): if self.data[description] in self.DATA_ANOMALY_MATCHES: setattr(self, self.ATTRIBUTE_MAP[description], {"original_description": self.data[description]}) continue attributes = [ x for x in cleaner.cleaned[description] if x["original_description"] == self.data[description] ] if len(attributes) != 1: raise ValueError("Either No attributes or multiple found for %s" % description) setattr(self, self.ATTRIBUTE_MAP[description], attributes[0]) def set_is_in_conservation_area(self, conservation_area_client: ConservationAreaClient): if not self.coordinates: raise ValueError("Coordinates have not been set, run get_coordinates() first") is_in_conservation_area = conservation_area_client.is_in_conservation_area_historic_england( x_bng=self.coordinates["x_coordinate"], y_bng=self.coordinates["y_coordinate"] ) self.in_conservation_area = is_in_conservation_area if is_in_conservation_area == "unknown": # We double check the secondary data source backup = conservation_area_client.is_in_conservation_area_historic_gov( longitude=self.coordinates["longitude"], latitude=self.coordinates["latitude"] ) if backup: self.in_conservation_area = ConservationAreaClient.IN_CONSERVATION_AREA else: self.in_conservation_area = ConservationAreaClient.UNKNOWN def set_year_built(self): """ Estimates when the property was built based on as much available data as possible. """ if self.full_sap_epc: self.year_built = pd.to_datetime(self.full_sap_epc["lodgement-date"]).year return if self.data["construction-age-band"] not in self.DATA_ANOMALY_MATCHES: # Take the lower limit. If we're pessimistic about the age of the property, that at least means we have # more options for recommendations if that age falls before the year that insulation in walls became # common practice band = [int(x) for x in re.findall(r'\b\d{4}\b', self.data["construction-age-band"])] self.year_built = band[0] return # We don't know when the property was built self.year_built = None