Model/model_data/Property.py

import pandas as pd
import re
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",
        "lighting-description": "lighting"
    }

    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
        self.full_sap_epc = None
        self.in_conservation_area = None
        self.year_built = 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 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])

    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 = pd.to_datetime(self.full_sap_epc["lodgement-date"]).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