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.number_of_rooms = 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 self.heat_loss_corridor = None self.mains_gas = None self.floor_height = None self.insulation_wall_area = None self.floor_area = 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_count_variables(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", "number_of_rooms": "number-habitable-rooms", } null_attributes = ["number_of_storeys", "number_of_rooms"] for attribute, epc_field in fields.items(): value = self.data["extension-count"] if value == "" or value in self.DATA_ANOMALY_MATCHES: if attribute in null_attributes: value = None else: 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_count_variables() self.set_heat_loss_corridor() self.set_mains_gas() self.set_floor_height() self.set_wall_area() self.set_floor_area() 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 def set_heat_loss_corridor(self): """ cleans the heat-loss-corridor :return: """ map = { "no corridor": False, "unheated corridor": True, "heated corridor": False } if self.data["heat-loss-corridor"] in self.DATA_ANOMALY_MATCHES: has_heat_loss_corridor = False else: has_heat_loss_corridor = map[self.data["heat-loss-corridor"]] length = self.data["unheated-corridor-length"] if length == "": length = None else: length = float(length) self.heat_loss_corridor = { "heat_loss_corridor": has_heat_loss_corridor, "length": length } def set_mains_gas(self): """ Sets whether the property has mains gas :return: """ map = { "Y": True, "N": False, } if self.data["mains-gas-flag"] == "" or self.data["mains-gas-flag"] in self.DATA_ANOMALY_MATCHES: self.mains_gas = None else: self.mains_gas = map[self.data["mains-gas-flag"]] def set_floor_height(self): """ Sets the floor height of the property :return: """ if self.data["floor-height"] == "" or self.data["floor-height"] in self.DATA_ANOMALY_MATCHES: self.floor_height = None else: self.floor_height = float(self.data["floor-height"]) def _clean_upload_data(self, to_update): for k, v in to_update.items(): if v in self.DATA_ANOMALY_MATCHES: to_update[k] = None return to_update def get_full_property_data(self): """ This method extracts the data which is pushed to the database, containing core information, from the EPC about a property :return: """ property_data = { "creation_status": "READY", "uprn": int(self.data["uprn"]), "building_reference_number": int(self.data["building-reference-number"]), "has_pre_condition_report": True, "has_recommendations": True, "property_type": self.data["property-type"], "built_form": self.data["built-form"], "local_authority": self.data["local-authority-label"], "constituency": self.data["constituency-label"], "number_of_rooms": self.number_of_rooms, "year_built": self.year_built, "tenure": self.data["tenure"], "current_epc_rating": self.data["current-energy-rating"], "current_sap_points": self.data["current-energy-efficiency"] } property_data = self._clean_upload_data(property_data) return property_data @classmethod def _prepare_rating_field(cls, field, rating_lookup): """ Utility function for usage in the lambda, for preparing the _rating fields """ return rating_lookup[field].value if field not in cls.DATA_ANOMALY_MATCHES else None def get_property_details_epc(self, portfolio_id: int, rating_lookup): property_details_epc = { "property_id": self.id, "portfolio_id": portfolio_id, "full_address": self.data["address"], "total_floor_area": float(self.data["total-floor-area"]), "walls": self.walls["clean_description"], "walls_rating": self._prepare_rating_field(self.data["walls-energy-eff"], rating_lookup), "roof": self.roof["clean_description"], "roof_rating": self._prepare_rating_field(self.data["roof-energy-eff"], rating_lookup), "floor": self.floor["clean_description"], "floor_rating": self._prepare_rating_field(self.data["floor-energy-eff"], rating_lookup), "windows": self.windows["clean_description"], "windows_rating": self._prepare_rating_field(self.data["windows-energy-eff"], rating_lookup), "heating": self.main_heating["clean_description"], "heating_rating": self._prepare_rating_field(self.data["mainheat-energy-eff"], rating_lookup), "heating_controls": self.main_heating_controls["clean_description"], "heating_controls_rating": self._prepare_rating_field(self.data["mainheatc-energy-eff"], rating_lookup), "hot_water": self.hotwater["clean_description"], "hot_water_rating": self._prepare_rating_field(self.data["hot-water-energy-eff"], rating_lookup), "lighting": self.lighting["clean_description"], "lighting_rating": self._prepare_rating_field(self.data["lighting-energy-eff"], rating_lookup), "mainfuel": self.main_fuel["clean_description"], "ventilation": self.ventilation["ventilation"], "solar_pv": self.solar_pv["solar_pv"], "solar_hot_water": self.solar_hot_water["solar_hot_water"], "wind_turbine": self.wind_turbine["wind_turbine"], "floor_height": self.floor_height, "heat_loss_corridor": self.heat_loss_corridor["heat_loss_corridor"], "unheated_corridor_length": self.heat_loss_corridor["length"], "number_of_open_fireplaces": self.number_of_open_fireplaces, "number_of_extensions": self.number_of_extensions, "number_of_storeys": self.number_of_storeys, "mains_gas": self.mains_gas, "energy_tariff": self.data["energy-tariff"], "primary_energy_consumption": self.energy["primary_energy_consumption"], "co2_emissions": self.energy["co2_emissions"], } return property_details_epc def set_wall_area(self): """ This method is placeholder It implements our floor area model to produce an estimate of the property's insulatable wall area """ import random self.insulation_wall_area = random.uniform(60, 100) def set_floor_area(self): """ Sets the floor area based on the EPC data """ # We don't know the number of floors at the moment so we're going to assume 1 # however this is something we'll need to use Verisk data for self.floor_area = float(self.data["total-floor-area"])