from datetime import datetime import re from epc_api.client import EpcClient from model_data.config import EPC_AUTH_TOKEN from model_data.BaseUtility import BaseUtility 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 lighting = None coordinates = None def __init__(self, id, postcode, address1, epc_client=None, data=None): self.id = id self.postcode = postcode self.address1 = address1 self.data = data self.full_sap_epc = None self.in_conservation_area = None self.year_built = None self.energy = None self.ventilation = None self.solar_pv = None self.solar_hot_water = None self.wind_turbine = None self.number_of_open_fireplaces = None self.number_of_extensions = None self.number_of_storeys = 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 set_coordinates(self, coordinates): """ This method sets the coordinates of the property, given the open uprn data :param coordinates: dictionary """ self.coordinates = {key.lower(): value for key, value in coordinates.items()} def set_energy(self): """ Extracts and formats data about the home's energy and co2 consumption To being with, this is just formatting epc data Data: - primary_energy_consumption This is based on the "energy-consumption-current" field in the EPC data. Current estimated total energy consumption for the property in a 12 month period (kWh/m2). Displayed on EPC as the current primary energy use per square metre of floor area. - co2_emissions This is based on the "co2-emissions-current" field in the EPC data. CO₂ emissions per year in tonnes/year. """ self.energy = { "primary_energy_consumption": float(self.data["energy-consumption-current"]), "co2_emissions": float(self.data["co2-emissions-current"]), } def set_ventilation(self): """ Extracts and formats data about the home's ventilation To being with, this is just formatting epc data Data: - ventilation This is based on the "ventilation-type" field in the EPC data. Ventilation type of the property. """ ventilation = self.data["mechanical-ventilation"] # perform some simple cleaning - when checking 300k properties, the only unique values were # {'', 'mechanical, supply and extract', 'NO DATA!', 'natural', 'mechanical, extract only'} if ventilation in self.DATA_ANOMALY_MATCHES or ventilation in [""]: ventilation = None self.ventilation = { "ventilation": ventilation, } def set_solar_pv(self): """ Extracts and formats data about the home's solar pv To being with, this is just formatting epc data Data: - solar_pv This is based on the "photo-supply" field in the EPC data. When checking 100k properties, either the value was "" or a stringified number """ solar_pv = self.data["photo-supply"] if solar_pv == "": solar_pv = None else: solar_pv = float(solar_pv) self.solar_pv = { "solar_pv": solar_pv, } def set_solar_hot_water(self): """ Extracts and formats data about the home's solar hot water We are just formatting the solar-water-heating-flag in the epc data :return: """ value_map = { "Y": True, "N": False, "": None, } self.solar_hot_water = { "solar_hot_water": value_map[self.data["solar-water-heating-flag"]], } def set_wind_turbine(self): """ Extracts and formats data about the home's wind turbine We are just formatting the wind-turbine-flag in the epc data :return: """ wind_turbine_count = self.data["wind-turbine-count"] if wind_turbine_count == "": wind_turbine_count = None else: wind_turbine_count = int(wind_turbine_count) self.wind_turbine = { "wind_turbine": wind_turbine_count, } def set_property_counts(self): """ For EPC fields that are just counts, we'll set them here These are fields that are integers but may contain additional values such as "" so we can't do a direct conversion straight to an integer :return: """ fields = { "number_of_open_fireplaces": "number-open-fireplaces", "number_of_extensions": "extension-count", "number_of_storeys": "flat-storey-count", } for attribute, epc_field in fields.items(): value = self.data["extension-count"] if value == "" or value in self.DATA_ANOMALY_MATCHES: value = 0 else: value = int(value) setattr(self, attribute, value) def get_components(self, cleaned): """ 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 cleaned: This is the dictionary of components found in cleaner.cleaned :return: """ if not cleaned: raise ValueError("Cleaner does not contain cleaned data") if not self.data: raise ValueError("Property does not contain data") self.set_energy() self.set_ventilation() self.set_solar_pv() self.set_solar_hot_water() self.set_wind_turbine() self.set_property_counts() for description, attribute in cleaned.items(): if self.data[description] in self.DATA_ANOMALY_MATCHES: setattr( self, self.ATTRIBUTE_MAP[description], {"original_description": self.data[description], "clean_description": self.data[description]} ) continue attributes = [ x for x in 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, in_conservation_area): """ Sets whether the property is in a conservation area given the output of the ConservationAreaClient :param in_conservation_area: string value, indicating whether the property is in a conservation area """ self.in_conservation_area = in_conservation_area 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 = datetime.strptime(self.full_sap_epc["lodgement-date"], '%Y-%m-%d').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