Model/model_data/Property.py
Khalim Conn-Kowlessar 35a9679220 Added property tests
2023-06-19 19:42:08 +01:00

99 lines
3.6 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
class Property:
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")
for description, attribute in cleaner.cleaned.items():
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])