Model/backend/Property.py

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

        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()

        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:
            self.heat_loss_corridor = False
        else:
            self.heat_loss_corridor = map[self.data["heat-loss-corridor"]]

    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] 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.data["floor-height"],
            "heat_loss_corridor": self.data["heat-loss-corridor"],
            "unheated_corridor_length": self.data["unheated-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.data["mains-gas-flag"],
            "energy_tariff": self.data["energy-tariff"],
            "primary_energy_consumption": self.energy["primary_energy_consumption"],
            "co2_emissions": self.energy["co2_emissions"],
        }

        return property_details_epc