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