mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
244 lines
8.3 KiB
Python
244 lines
8.3 KiB
Python
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
|
|
|
|
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 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()
|
|
|
|
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]})
|
|
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
|