Adding funding class

.idea/Model.iml

@@ -7,7 +7,7 @@
       <sourceFolder url="file://$MODULE_DIR$/open_uprn" isTestSource="false" />
       <sourceFolder url="file://$MODULE_DIR$/recommendations" isTestSource="false" />
     </content>
-    <orderEntry type="jdk" jdkName="Stonewater-wave-3" jdkType="Python SDK" />
+    <orderEntry type="jdk" jdkName="Fastapi-backend" jdkType="Python SDK" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>
   <component name="PyNamespacePackagesService">

.idea/misc.xml

@@ -3,7 +3,7 @@
   <component name="Black">
     <option name="sdkName" value="Python 3.10 (backend)" />
   </component>
-  <component name="ProjectRootManager" version="2" project-jdk-name="Stonewater-wave-3" project-jdk-type="Python SDK" />
+  <component name="ProjectRootManager" version="2" project-jdk-name="Fastapi-backend" project-jdk-type="Python SDK" />
   <component name="PyCharmProfessionalAdvertiser">
     <option name="shown" value="true" />
   </component>

backend/Funding.py

@@ -0,0 +1,297 @@
+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
+    """
+
+    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,
+        floor_area,
+        council_tax_band,
+        property_recommendations,
+        project_scores_matrix,
+        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 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.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 = []
+        for recs in self.recommendations:
+            self.measure_types.extend([r["measure_type"] for r in recs])
+
+        # 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)
+            ]
+
+        # Store the final outputs
+        self.gbis_eligibiltiy = {}
+        self.eco4_eligibility = {}
+        self.whlg_eligibility = {}
+
+    def output(
+        self,
+        measure_types: List[str],
+        estimated_funding: float,
+        notify_tenant_benefits_requirements: bool,
+        notify_council_tax_band_requirements: bool,
+        notify_tenant_low_income_requirements: bool,
+    ):
+        """"
+        """
+        return {
+            "measure_types": measure_types,
+            "estimated_funding": estimated_funding,
+            "notify_tenant_benefits_requirements": notify_tenant_benefits_requirements,
+            "notify_council_tax_band_requirements": notify_council_tax_band_requirements,
+            "notify_tenant_low_income_requirements": notify_tenant_low_income_requirements
+        }
+
+    @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_best_gbis_measure(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[0] for m in self.recommendations if m[0]["measure_type"] in measures
+        ])
+
+        measure_table["post_install_sap"] = measure_table["sap_points"] + self.starting_sap
+        # We classify the movement
+        measure_table["Finishing Band"] = measure_table["sap_points"].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])
+        # Recommend the measure, with estimated funding amount
+        recommended_measure = measure_table.head(1)
+
+        return {
+            "measure_type": recommended_measure["measure_type"],
+            "estimated_funding": recommended_measure["estimated_funding"]
+        }
+
+    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
+            len(
+                [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"])
+        ):
+            # We find the best measure for GBIS
+            recommended_measure = self.find_best_gbis_measure(
+                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(
+                measure_types=[recommended_measure["measure_type"]],
+                estimated_funding=recommended_measure["estimated_funding"],
+                notify_tenant_benefits_requirements=False,
+                notify_council_tax_band_requirements=self.council_tax_band is None,
+                notify_tenant_low_income_requirements=False,
+            )
+
+        # Low income/flex
+        if (
+            (self.starting_sap in ["G", "D", "E", "F"]) and
+            len([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_measure = self.find_best_gbis_measure(measures=valid_measures)
+            return self.output(
+                measure_types=[recommended_measure["measure_type"]],
+                estimated_funding=recommended_measure["estimated_funding"],
+                notify_tenant_benefits_requirements=True,
+                notify_council_tax_band_requirements=False,
+                notify_tenant_low_income_requirements=True,
+            )
+
+        # Otherwise, no funding availability
+        return self.output(
+            measure_types=[],
+            estimated_funding=0,
+            notify_tenant_benefits_requirements=False,
+            notify_council_tax_band_requirements=False,
+            notify_tenant_low_income_requirements=False
+        )
+
+    def gbis(self):
+        """
+        Check if a property is eligible for GBIS
+        :return:
+        """
+
+        if self.tenure == "Private":
+            self.gbis_eligibiltiy = self.gbis_prs()
+            return
+
+        raise NotImplementedError("Implement social/oo")
+
+    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()

backend/app/plan/router.py

@@ -373,6 +373,16 @@ def extract_property_request_data(
     return patch, property_already_installed, property_non_invasive_recommendations, property_valution
 
 
+def get_eco_project_scores_matrix():
+    data = read_csv_from_s3(
+        bucket_name=get_settings().DATA_BUCKET,
+        filepath="funding/ECO4 Full Project Scores Matrix.csv",
+    )
+    df = pd.DataFrame(data)
+    df.columns = ['Floor Area Segment', 'Starting Band', 'Finishing Band', 'Cost Savings']
+    return df
+
+
 router = APIRouter(
     prefix="/plan",
     tags=["plan"],
@@ -438,6 +448,12 @@ async def trigger_plan(body: PlanTriggerRequest):
             if not is_new and not body.multi_plan:
                 continue
 
+            if epc_searcher.newest_epc is None:
+                raise ValueError(
+                    "No EPCs found for this property and did not estimate - likely need to provide a"
+                    "property type and built form"
+                )
+
             if is_new:
                 create_property_targets(
                     session,
@@ -508,6 +524,7 @@ async def trigger_plan(body: PlanTriggerRequest):
         logger.info("Reading in materials and cleaned datasets")
         materials = get_materials(session)
         cleaned = get_cleaned()
+        eco_project_scores_matrix = get_eco_project_scores_matrix()
 
         kwh_client = KwhData(bucket=get_settings().DATA_BUCKET, read_consumption_data=True)
 
@@ -730,6 +747,23 @@ async def trigger_plan(body: PlanTriggerRequest):
             ]
             recommendations[p.id] = final_recommendations
 
+        # ~~~~~~~~~~~~~~~~
+        # Funding
+        # ~~~~~~~~~~~~~~~~
+        from backend.Funding import Funding
+        for p in input_properties:
+            funding_calulator = Funding(
+                tenure=body.housing_type,
+                starting_epc=p.data["current-energy-rating"],
+                starting_sap=p.data["current-energy-efficiency"],
+                floor_area=p.floor_area,
+                council_tax_band=None,  # This is seemingly always None at the moment
+                property_recommendations=recommendations[p.id],
+                project_scores_matrix=eco_project_scores_matrix,
+                gbis_abs_rate=20,
+                eco4_abs_rate=20,
+            )
+
         logger.info("Uploading recommendations to the database")
         # If we have any work to do, we create a new scenario
         engine_scenario = create_scenario(

etl/customers/cambridge/remote_assessment.py

@@ -0,0 +1,138 @@
+import os
+import time
+
+from tqdm import tqdm
+import pandas as pd
+from dotenv import load_dotenv
+from etl.find_my_epc.RetrieveFindMyEpc import RetrieveFindMyEpc
+from backend.SearchEpc import SearchEpc
+from utils.s3 import save_csv_to_s3
+
+load_dotenv(dotenv_path="backend/.env")
+EPC_AUTH_TOKEN = os.getenv("EPC_AUTH_TOKEN")
+USER_ID = 8
+PORTFOLIO_ID = 122
+
+
+def app():
+    asset_list = [
+        {
+            "address": "12 Church Lane", "postcode": "CB23 8AF", "uprn": 100090136018,
+            "property_type": "House", "built-form": "Semi-Detached"
+        },
+        {
+            "address": "21 High Street", "postcode": "CB23 8AB", "uprn": 100090136026
+        },
+        {
+            "address": "22 High Street", "postcode": "CB23 8AB", "uprn": 100090136027
+        },
+        {
+            "address": "5 Bunkers Hill", "postcode": "CB3 0LY", "uprn": 10008078615
+        },
+        {
+            "address": "6 Bunkers Hill", "postcode": "CB3 0LY", "uprn": 10008078616
+        },
+        {
+            "address": "7 Bunkers Hill", "postcode": "CB3 0LY", "uprn": 10008078617
+        },
+        {
+            "address": "32 George Nuttall Close", "postcode": "CB4 1YE", "uprn": 200004200075
+        },
+        {
+            "address": "33 George Nuttall Close", "postcode": "CB4 1YE", "uprn": 200004200076
+        },
+        {
+            "address": "35 George Nuttall Close", "postcode": "CB4 1YE", "uprn": 200004200078
+        },
+        {
+            "address": "36 George Nuttall Close", "postcode": "CB4 1YE", "uprn": 200004200079
+        }
+    ]
+    asset_list = pd.DataFrame(asset_list)
+
+    valuations_data = [
+        {'uprn': 100090136018, "valuation": 586_000},
+        {'uprn': 100090136026, "valuation": 551_000},
+        {'uprn': 100090136027, "valuation": 844_000},
+        {'uprn': 10008078615, "valuation": 763_000},
+        {'uprn': 10008078616, "valuation": 616_000},
+        {'uprn': 10008078617, "valuation": 593_000},
+        {'uprn': 200004200075, "valuation": 450_000},
+        {'uprn': 200004200076, "valuation": 457_000},
+        {'uprn': 200004200078, "valuation": 304_000},
+        {'uprn': 200004200079, "valuation": 313_000}
+    ]
+
+    # Pull the additional data
+    extracted_data = []
+    for _, home in tqdm(asset_list.iterrows(), total=len(asset_list)):
+        add1 = home["address"]
+        pc = home["postcode"]
+        # Retrieve the EPC data
+        epc_searcher = SearchEpc(
+            address1=add1,
+            postcode=pc, uprn=home["uprn"], auth_token=EPC_AUTH_TOKEN, os_api_key=""
+        )
+        epc_searcher.find_property(skip_os=True)
+        if epc_searcher.newest_epc is None:
+            continue
+
+        find_epc_searcher = RetrieveFindMyEpc(address=epc_searcher.newest_epc["address1"],
+                                              postcode=epc_searcher.newest_epc["postcode"])
+        find_epc_data = find_epc_searcher.retrieve_newest_find_my_epc_data()
+        time.sleep(0.5)
+        # We need uprn
+
+        extracted_data.append(
+            {
+                "uprn": home["uprn"],
+                **find_epc_data,
+            }
+        )
+
+    non_invasive_recommendations = [
+        {
+            "uprn": r["uprn"],
+            "recommendations": r["recommendations"]
+        } for r in extracted_data
+    ]
+
+    filename = f"{USER_ID}/{PORTFOLIO_ID}/asset_list.csv"
+    save_csv_to_s3(
+        dataframe=pd.DataFrame(asset_list),
+        bucket_name="retrofit-plan-inputs-dev",
+        file_name=filename
+    )
+
+    # Store the non-invasive recommendations in s3
+    non_invasive_recommendations_filename = f"{USER_ID}/{PORTFOLIO_ID}/non_invasive_recommendations.csv"
+    save_csv_to_s3(
+        dataframe=pd.DataFrame(non_invasive_recommendations),
+        bucket_name="retrofit-plan-inputs-dev",
+        file_name=non_invasive_recommendations_filename
+    )
+
+    # Store the valuations data in s3
+    valuations_filename = f"{USER_ID}/{PORTFOLIO_ID}/valuations.csv"
+    save_csv_to_s3(
+        dataframe=pd.DataFrame(valuations_data),
+        bucket_name="retrofit-plan-inputs-dev",
+        file_name=valuations_filename
+    )
+
+    body = {
+        "portfolio_id": str(PORTFOLIO_ID),
+        "housing_type": "Private",
+        "goal": "Increasing EPC",
+        "goal_value": "B",
+        "trigger_file_path": filename,
+        "already_installed_file_path": "",
+        "patches_file_path": "",
+        "non_invasive_recommendations_file_path": non_invasive_recommendations_filename,
+        "valuation_file_path": valuations_filename,
+        "scenario_name": "Wave 3 Packages",
+        "multi_plan": True,
+        "budget": None,
+        "exclusions": []
+    }
+    print(body)

etl/customers/stonewater/Wave 3 Preparation.py

@@ -2826,9 +2826,10 @@ def identify_incorrect_packages():
             "estimated": "EPC Estimated based on Nearby Properties"
         }
     )
-    # Find entries where the SAP score is not an integer
+    # Take non-estimated EPCs?
-    non_integer_sap = epc_data_to_append[~epc_data_to_append["EPC: SAP Score"].astype(str).str.isnumeric()]
+    # epc_data_to_append = epc_data_to_append[epc_data_to_append["EPC Estimated based on Nearby Properties"] != True]
-    non_integer_sap["UPRN"].values[0]
+    # Take the newest EPC per UPRN, based on lodgement date
+    epc_data_to_append = epc_data_to_append.sort_values("EPC: Date of EPC", ascending=False).drop_duplicates("UPRN")
 
     epc_data_to_append["EPC: Date of EPC"] = pd.to_datetime(epc_data_to_append["EPC: Date of EPC"])
     # Years since the EPC was lodged

recommendations/RoofRecommendations.py

@@ -496,6 +496,7 @@ class RoofRecommendations:
         roof_roof_insulation_materials = [
             {
                 "type": "room_roof_insulation",
+                "measure_type": "room_roof_insulation",
                 "description": "Insulating the ceiling of the roof roof and re-decorate",
                 "depths": [100],
                 "depth_unit": "mm",