diff --git a/.idea/Model.iml b/.idea/Model.iml index 762580d9..df6c4faa 100644 --- a/.idea/Model.iml +++ b/.idea/Model.iml @@ -7,7 +7,7 @@ - + diff --git a/.idea/misc.xml b/.idea/misc.xml index c916a158..50cad4ca 100644 --- a/.idea/misc.xml +++ b/.idea/misc.xml @@ -3,7 +3,7 @@ - + diff --git a/backend/Funding.py b/backend/Funding.py new file mode 100644 index 00000000..21430f35 --- /dev/null +++ b/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() diff --git a/backend/app/plan/router.py b/backend/app/plan/router.py index dbef6435..056f7f1c 100644 --- a/backend/app/plan/router.py +++ b/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( diff --git a/etl/customers/cambridge/remote_assessment.py b/etl/customers/cambridge/remote_assessment.py new file mode 100644 index 00000000..3f152e79 --- /dev/null +++ b/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) diff --git a/etl/customers/stonewater/Wave 3 Preparation.py b/etl/customers/stonewater/Wave 3 Preparation.py index d2232f40..0f757f7b 100644 --- a/etl/customers/stonewater/Wave 3 Preparation.py +++ b/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 - non_integer_sap = epc_data_to_append[~epc_data_to_append["EPC: SAP Score"].astype(str).str.isnumeric()] - non_integer_sap["UPRN"].values[0] + # Take non-estimated EPCs? + # epc_data_to_append = epc_data_to_append[epc_data_to_append["EPC Estimated based on Nearby Properties"] != True] + # 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 diff --git a/recommendations/RoofRecommendations.py b/recommendations/RoofRecommendations.py index 4e29083f..6778e886 100644 --- a/recommendations/RoofRecommendations.py +++ b/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",