Model/backend/Funding.py

import pandas as pd
import numpy as np
from typing import List

from backend.app.plan.schemas import HousingType


class Funding:
    """
    Given a property, this class identifies if the home is possibly eligible for funding under
    the various funding schemes. It will also calculate the expected amount of funding available
    and flag any tenant specific requirements that need to be considered to the funding to be attained
    """

    SCHEMES = ["eco4", "gbis", "whlg"]

    ECO_SAP_SCORE_THREHOLDS = [
        {'Band': 'High_A', 'From': 96.0, 'Up to': 100.0, 'Mid-point': 98.0},
        {'Band': 'Low_A', 'From': 92.0, 'Up to': 96.0, 'Mid-point': 94.0},
        {'Band': 'High_B', 'From': 86.0, 'Up to': 91.0, 'Mid-point': 88.5},
        {'Band': 'Low_B', 'From': 81.0, 'Up to': 86.0, 'Mid-point': 83.5},
        {'Band': 'High_C', 'From': 74.5, 'Up to': 80.0, 'Mid-point': 77.25},
        {'Band': 'Low_C', 'From': 69.0, 'Up to': 74.5, 'Mid-point': 71.75},
        {'Band': 'High_D', 'From': 61.5, 'Up to': 68.0, 'Mid-point': 64.75},
        {'Band': 'Low_D', 'From': 55.0, 'Up to': 61.5, 'Mid-point': 58.25},
        {'Band': 'High_E', 'From': 46.5, 'Up to': 54.0, 'Mid-point': 50.25},
        {'Band': 'Low_E', 'From': 39.0, 'Up to': 46.5, 'Mid-point': 42.75},
        {'Band': 'High_F', 'From': 29.5, 'Up to': 38.0, 'Mid-point': 33.75},
        {'Band': 'Low_F', 'From': 21.0, 'Up to': 29.5, 'Mid-point': 25.25},
        {'Band': 'High_G', 'From': 10.5, 'Up to': 20.0, 'Mid-point': 15.25},
        {'Band': 'Low_G', 'From': 1.0, 'Up to': 10.5, 'Mid-point': 5.75}
    ]

    def __init__(
        self,
        tenure: HousingType,
        starting_epc,
        starting_sap,
        postcode,
        floor_area,
        council_tax_band,
        property_recommendations,
        project_scores_matrix,
        whlg_eligible_postcodes,
        gbis_abs_rate: int,
        eco4_abs_rate: int,
    ):
        """
        Use Pydantic to validate the parameter types
        :param tenure: Indicates if the property is a social or private home
        :param starting_epc: The current EPC rating of the property
        :param starting_sap: The current SAP score for the property
        :param floor_area: The total floor area of the property
        :param council_tax_band: The council tax band of the property
        :param property_recommendations: The recommendations for the property
        :param project_scores_matrix: The matrix of project scores for ECO4
        :param whlg_eligible_postcodes: The postcodes eligible for WHLG
        :param gbis_abs_rate: The assumed £/abs achieved by the installer for GBIS
        :param eco4_abs_rate: The assumed £/abs achieved by the installer for ECO4
        """

        # TODO: Things we need to include:
        #       1) Amount of funding
        #       2) Fundable measures, as a subset of measures may be fundable, not all

        self.tenure = tenure
        self.starting_epc = starting_epc
        self.starting_sap = starting_sap
        self.postcode = postcode
        self.starting_eco_band = self.sap_to_eco_band(self.starting_sap)
        self.floor_area_segment = self.classify_floor_area(floor_area)
        self.gbis_abs_rate = gbis_abs_rate
        self.eco4_abs_rate = eco4_abs_rate
        self.council_tax_band = council_tax_band

        self.recommendations = property_recommendations

        self.measure_types = list({r["measure_type"] for r in property_recommendations if r["default"]})

        # Load in the eco4 project scores matrix
        # Filter the matrix on scores relevant to this property
        self.project_scores_matrix = project_scores_matrix[
            (project_scores_matrix["Floor Area Segment"] == self.floor_area_segment) &
            (project_scores_matrix["Starting Band"] == self.starting_eco_band)
            ]

        # The postcode column is already lower case
        self.whlg_eligible_postcodes = whlg_eligible_postcodes[
            whlg_eligible_postcodes["Postcode"] == self.postcode.lower()
            ]

        # Store the final outputs
        self.gbis_eligibiltiy = {}
        self.eco4_eligibility = {}
        self.whlg_eligibility = {}

    def output(
        self,
        scheme: str,
        eligible: bool,
        types: List[str],
        measure_types: List[str],
        project_score: float,
        estimated_funding: float,
        notify_tenant_benefits_requirements: bool,
        notify_council_tax_band_requirements: bool,
        notify_tenant_low_income_requirements: bool,
        innovation_required: bool,
    ):
        """"
        """

        if scheme not in self.SCHEMES:
            raise ValueError("Scheme not recognised")

        return {
            "scheme": scheme,
            "eligible": eligible,
            "type": types,
            "measure_types": measure_types,
            "project_score": project_score,
            "estimated_funding": estimated_funding,
            "requires_benefits": notify_tenant_benefits_requirements,
            "requires_council_tax_band": notify_council_tax_band_requirements,
            "requires_low_income": notify_tenant_low_income_requirements,
            "innovation_required": innovation_required,
        }

    @staticmethod
    def classify_floor_area(floor_area):
        if floor_area <= 72:
            return "0-72"

        if floor_area <= 97:
            return "73-97"

        if floor_area <= 199:
            return "98-199"

        return "200"

    def eco4(self):
        """
        Checks if a property is eligible for ECO4
        :return:
        """
        pass

    def find_gbis_measures(self, measures):
        """
        The best measure is one that:
        1) Creates some SAP movement, therefore enables eligiblity
        2) Generates the most funding
        3) Has a reasonable ROI
        :return:
        """
        measure_table = pd.DataFrame([
            m for m in self.recommendations if
            (m["type"] in measures) or (m["measure_type"] in measures) and m["default"]
        ])

        measure_table["post_install_sap"] = measure_table["sap_points"] + self.starting_sap
        # We classify the movement
        measure_table["Finishing Band"] = np.floor(measure_table["post_install_sap"]).apply(
            lambda points: self.sap_to_eco_band(points)
        )
        # Remove any measures that generate zero SAP movement
        measure_table = measure_table[measure_table["Finishing Band"] != self.starting_eco_band]

        if measure_table.empty:
            raise NotImplementedError("No measures available, handle me!")

        # We merge on the project matrix, on post install band
        measure_table = measure_table.merge(
            self.project_scores_matrix, how="left", on="Finishing Band"
        )
        # Cost Savings is the abs
        measure_table["estimated_funding"] = measure_table["Cost Savings"] * self.gbis_abs_rate
        # We cap any estimated funding at the install cost
        measure_table["estimated_funding"] = np.where(
            measure_table["estimated_funding"] >= measure_table["total"],
            measure_table["total"],
            measure_table["estimated_funding"]
        )

        # Sort by the measure that will cost the client the least, per sap point
        measure_table["cost_minus_funding"] = measure_table["total"] - measure_table["estimated_funding"]
        measure_table["cost_minus_funding_per_sap"] = measure_table["cost_minus_funding"] / measure_table["sap_points"]
        measure_table = measure_table.sort_values(["cost_minus_funding_per_sap", "total"], ascending=[True, False])

        return measure_table[
            ["type", "measure_type", "Cost Savings", "estimated_funding"]
        ].rename(columns={"Cost Savings": "project_score"}).to_dict("records")

    def sap_to_eco_band(self, sap_points):
        """
        Giuven a sap point score, this function will classify the points into the SAP half-band
        :param sap_points:
        :return:
        """

        if sap_points > 100:
            return "High_A"

        classification = [
            x for x in self.ECO_SAP_SCORE_THREHOLDS if (x["From"] <= sap_points) and (sap_points <= x["Up to"])
        ]

        if len(classification) != 1:
            raise Exception("We should have a single classifcation for SAP points to half band")

        return classification[0]['Band']

    def gbis_prs(self):
        """
        Checks if a private rental is eligible for GBIS. There are the following possible options
        1) General Eligibilty, contigent on EPC D-G and council tax band A-D. Excludes CWI, LI and heating
           controls
        2) Low income group - contigent on EPC D-G and tenant must receive benefits. Excludes heating controls
        3) GBIS Flex route 1, 3 - Great British Insulation Scheme Routes 1 and 3 are for pre-installation
        SAP bands D-G for owner-occupied households, D-E for private rented sector households
        (Including F & G if exempt from MEES). If houseold is low income. Excludes heating controls
        4) GBIS Flex route 2 - EPC E - G and low income household. Excludes heating controls

        Eligible measures:
        • Solid wall
        • pitched roof
        • flat roof
        • under floor
        • solid floor park home and
        • room in-roof insulation

        :return:
        """

        valid_measures = [
            "internal_wall_insulation",
            "external_wall_insulation",
            "flat_roof_insulation",
            "suspended_floor_insulation",
            "room_roof_insulation",
            # Not available for every eligiblity type
            "cavity_wall_insulation",
            "loft_insulation",
        ]

        # General Eligibility
        if (
            (self.starting_epc in ["G", "D", "E", "F"]) and
            any(
                [measure in valid_measures for measure in self.measure_types
                 if measure not in ["cavity_wall_insulation", "loft_insulation"]]
            ) and
            (self.council_tax_band in [None, "A", "B", "C", "D"])
        ):
            # This function pulls out the various measures that can provide funding under GBIS
            recommended_measures = self.find_gbis_measures(
                measures=[m for m in valid_measures if m not in ["cavity_wall_insulation", "loft_insulation"]]
            )
            # If the council tax band is missing, we nofify the customer that this is a requirement that
            # should be checked
            return [
                self.output(
                    scheme="gbis",
                    eligible=True,
                    types=[m["type"]],  # This is single measure so we only have one type
                    measure_types=[m["measure_type"]],
                    project_score=m["project_score"],
                    estimated_funding=m["estimated_funding"],
                    notify_tenant_benefits_requirements=False,
                    notify_council_tax_band_requirements=self.council_tax_band is None,
                    notify_tenant_low_income_requirements=False,
                    innovation_required=False
                ) for m in recommended_measures
            ]

        # Low income/flex
        if (
            (self.starting_sap in ["G", "D", "E", "F"]) and
            any([measure in valid_measures for measure in self.measure_types])
        ):
            # Find the best measure, and can also include CWI/LI but requires the tenant to be
            # low inome or on benefits
            # We find the best measure for GBIS
            recommended_measures = self.find_gbis_measures(measures=valid_measures)
            return [
                self.output(
                    scheme="gbis",
                    eligible=True,
                    types=[m["type"]],  # This is single measure so we only have one type
                    measure_types=[m["measure_type"]],
                    project_score=m["project_score"],
                    estimated_funding=m["estimated_funding"],
                    notify_tenant_benefits_requirements=True,
                    notify_council_tax_band_requirements=False,
                    notify_tenant_low_income_requirements=True,
                    innovation_required=False
                ) for m in recommended_measures
            ]

        # Otherwise, no funding availability
        return []

    def gbis_social(self):
        """
        Because this is social housing, we have two typical means for eligibility
        1) EPC D, where an innovation measure is required
        2) EPC G-E, where an innovation measure isn't required
        :return:
        """
        valid_measures = [
            "internal_wall_insulation",
            "external_wall_insulation",
            "flat_roof_insulation",
            "suspended_floor_insulation",
            "room_roof_insulation",
            # Not available for every eligiblity type
            "cavity_wall_insulation",
            "loft_insulation",
            "heating_control"
        ]

        recommended_measures = self.find_gbis_measures(
            measures=valid_measures
        )

        # All measures are available
        if self.starting_sap == "D":
            return [
                self.output(
                    scheme="gbis",
                    eligible=True,
                    types=[m["type"]],  # This is single measure so we only have one type
                    measure_types=[m["measure_type"]],
                    project_score=m["project_score"],
                    estimated_funding=m["estimated_funding"],
                    notify_tenant_benefits_requirements=False,
                    notify_council_tax_band_requirements=False,
                    notify_tenant_low_income_requirements=False,
                    innovation_required=True
                ) for m in recommended_measures
            ]

        if self.starting_sap in ["G", "F", "E"]:
            return [
                self.output(
                    scheme="gbis",
                    eligible=True,
                    types=[m["type"]],  # This is single measure so we only have one type
                    measure_types=[m["measure_type"]],
                    project_score=m["project_score"],
                    estimated_funding=m["estimated_funding"],
                    notify_tenant_benefits_requirements=False,
                    notify_council_tax_band_requirements=False,
                    notify_tenant_low_income_requirements=False,
                    innovation_required=False
                ) for m in recommended_measures
            ]

        return []

    def gbis(self):
        """
        Check if a property is eligible for GBIS
        :return:
        """

        if self.tenure == "Private":
            self.gbis_eligibiltiy = self.gbis_prs()
            return

        if self.tenure == "Social":
            self.gbis_eligibiltiy = self.gbis_social()

        raise NotImplementedError("Implement social/oo")

    def whlg(self):
        if self.tenure == "Social":
            # We can't do anything for social housing
            self.whlg_eligibility = []
            return

        if not self.whlg_eligible_postcodes.empty:
            raise Exception("Implement me")
            # self.whlg_eligibility = [
            #     self.output(
            #         scheme,
            #         eligible,
            #         types,
            #         measure_types,
            #         project_score: float,
            #         estimated_funding: float,
            #         notify_tenant_benefits_requirements: bool,
            #         notify_council_tax_band_requirements: bool,
            #         notify_tenant_low_income_requirements: bool,
            #         innovation_required: bool,
            #     )
            # ]

    def eco4(self):
        if self.tenure == "Private":
            self.eco4_eligibiltiy = self.eco4_prs()
            return

    def check_eligibiltiy(self):
        """
        This function instigates the checking process
        :return:
        """

        self.gbis()
        # self.eco4()
        self.whlg()