diff --git a/etl/eligibility/Eligibility.py b/etl/eligibility/Eligibility.py index 730ff6e1..8b205c79 100644 --- a/etl/eligibility/Eligibility.py +++ b/etl/eligibility/Eligibility.py @@ -1,6 +1,7 @@ from recommendations.recommendation_utils import convert_thickness_to_numeric from etl.epc_clean.epc_attributes.RoofAttributes import RoofAttributes from etl.epc_clean.epc_attributes.WallAttributes import WallAttributes +from etl.epc_clean.epc_attributes.FloorAttributes import FloorAttributes class Eligibility: @@ -17,20 +18,43 @@ class Eligibility: loft = None cavity = None + solid_wall = None + room_roof = None + flat_roof = None + suspended_floor = None + solid_floor = None - # schemes + # schemes based on Warmfront now + gbis_warmfront = None + eco4_warmfront = None + # Schemes based on full eligibility gbis = None eco4 = None # If the loft has less than 100mm of insulation, we classify the home has needing loft insulation LOFT_INSULATION_THRESHOLD = 100 + # Because EPCS have different values for tenure, we need to remap them to a common set of values + tenure_remap = { + 'NO DATA!': "unknown", + 'Not defined - use in the case of a new dwelling for which the intended tenure in not known. It is no': + "unknown", + 'Owner-occupied': 'Owner-occupied', + 'Rented (private)': 'Rented (private)', + 'Rented (social)': 'Rented (social)', + 'owner-occupied': 'Owner-occupied', + 'rental (private)': 'Rented (private)', + 'rental (social)': 'Rented (social)', + 'unknown': "unknown", + } + def __init__(self, epc, cleaned): self.epc = epc self.cleaned = cleaned self.walls = self.parse_fabric("walls-description") self.roof = self.parse_fabric("roof-description") + self.floor = self.parse_fabric("floor-description") self.loft_insulation() self.cavity_insulation() @@ -51,6 +75,9 @@ class Eligibility: elif key == "roof-description": cleaner_cls = RoofAttributes(self.epc["roof-description"]) + elif key == "floor-description": + cleaner_cls = FloorAttributes(self.epc["floor-description"]) + else: raise ValueError("Invalid key") output = cleaner_cls.process() @@ -144,7 +171,72 @@ class Eligibility: "type": "full" } - def check_gbis(self): + def solid_wall_insulation(self): + """ + Given the description of the walls, this function determines if the property is suitable for solid wall + insulation + :return: + """ + + is_solid = self.walls["is_solid_brick"] + is_insulated = self.walls["insulation_thickness"] in ["average", "above average"] + + if is_solid and is_insulated: + self.solid_wall = { + "suitability": True, + } + return + + self.solid_wall = { + "suitability": False, + } + + def room_roof_insulation(self): + is_room_roof = self.roof["is_roof_room"] + + insulation_thickness = convert_thickness_to_numeric( + self.roof["insulation_thickness"], + self.roof["is_pitched"], + self.roof["is_flat"] + ) + + self.room_roof = { + "suitability": is_room_roof and insulation_thickness == 0, + "thickness": insulation_thickness + } + + def flat_roof_insulation(self): + is_flat = self.roof["is_flat"] + insulation_thickness = convert_thickness_to_numeric( + self.roof["insulation_thickness"], + self.roof["is_pitched"], + self.roof["is_flat"] + ) + + self.flat_roof = { + "suitability": is_flat and insulation_thickness <= 100, + "thickness": insulation_thickness + } + + def suspended_floor_insulation(self): + is_suspended = self.floor["is_suspended"] + is_insulated = self.floor["insulation_thickness"] in ["average", "above average"] + + self.suspended_floor = { + "suitability": is_suspended and (not is_insulated), + } + return + + def solid_floor_insulation(self): + is_solid = self.floor["is_solid"] + is_insulated = self.floor["insulation_thickness"] in ["average", "above average"] + + self.solid_floor = { + "suitability": is_solid and (not is_insulated), + } + return + + def check_gbis_warmfront(self): """ The Eligibility criteria for the Great British Insulation Scheme (GBIS) can be found here: https://www.ofgem.gov.uk/environmental-and-social-schemes/great-british-insulation-scheme/homeowners-and-tenants @@ -176,15 +268,11 @@ class Eligibility: self.cavity_insulation() self.loft_insulation() - # self.gbis = (self.cavity["suitability"] or self.loft["suitability"]) and ( - # int(self.epc["current-energy-efficiency"]) <= 68 - # ) - - self.gbis = (self.cavity["suitability"]) and ( + self.gbis_warmfront = (self.cavity["suitability"]) and ( int(self.epc["current-energy-efficiency"]) <= 68 ) - def check_eco4(self, post_retrofit_sap=None): + def check_eco4_warmfront(self, post_retrofit_sap=None): """ This funciton will check if the property is eligible for funding under the ECO4 scheme @@ -214,7 +302,7 @@ class Eligibility: current_sap = int(self.epc["current-energy-efficiency"]) if current_sap > 54: - self.eco4 = { + self.eco4_warmfront = { "eligible": False, "message": "sap too high" } @@ -227,7 +315,7 @@ class Eligibility: is_eligible = self.cavity["suitability"] & self.loft["suitability"] if post_retrofit_sap is None: - self.eco4 = { + self.eco4_warmfront = { "eligible": is_eligible, "message": "subject to post retrofit sap" } @@ -235,8 +323,132 @@ class Eligibility: is_eligible = is_eligible & (post_retrofit_sap >= 69) - self.eco4 = { + self.eco4_warmfront = { "eligible": is_eligible, "message": None } return + + def check_gbis(self): + + """ + The Eligibility criteria for the Great British Insulation Scheme (GBIS) can be found here: + https://www.ofgem.gov.uk/environmental-and-social-schemes/great-british-insulation-scheme/homeowners-and-tenants + + Full delivery guidance and be downloaded here: + https://www.ofgem.gov.uk/sites/default/files/2023-08/Great%20British%20Insulation%20Scheme%20Delivery + %20Guidance%20V101693416860968.pdf + + For social housing, the criteria is the following: + + If the property is currently an EPC D: + - It's valid for innovation measures only but not a heating control measure + - The property must be rented at below the market rate. All eligible social housing is treated based on the + low income group, therefore the tennant must be in receipt of one the eligible benefits + + If the property is currently an EPC E or below: + - It's valid for all eligible insulation measures + - The property must be rented at below the market rate. All eligible social housing is treated based on the + low income group, therefore the tennant must be in receipt of one the eligible benefits + + From GBIS guidance document: + Determining whether the premises are let below market rate + + 3.101 Social housing under this provision will only be eligible where the housing is let below + the market rate. The supplier must produce a declaration signed by a social landlord + providing confirmation that the social housing premises are let below the market rate, + or where the premises are currently void, have previously and will be let below the + market rate. The declaration to be signed by a social landlord is included within the + Eligibility and Pre-Retrofit Declaration form. This declaration form must be retained by + suppliers and be available on request for audit purposes. + + 3.102 Where social housing is let at or above the market rate, the property can be treated as + a private domestic premises, where the occupant meets the eligibility requirements. + See section on PRS from paragraph 1.13 for more information. + + This method searches ALL of the possible measures that can be implemented under GBIS. This includes: + - cavity wall (including party wall) + - loft + - solid wall + - pitched roof + - flat roof + - under-floor + - solid floor + - park home + - room-in-roof + + :return: + """ + + self.cavity_insulation() + self.loft_insulation() + self.solid_wall_insulation() + self.room_roof_insulation() + self.flat_roof_insulation() + self.suspended_floor_insulation() + self.solid_floor_insulation() + + tenure = self.tenure_remap.get(self.epc["tenure"], None) + current_sap = int(self.epc["current-energy-efficiency"]) + is_below_e = current_sap <= 54 + is_below_c = current_sap <= 68 + + needs_measure = ( + self.cavity["suitability"] or + self.loft["suitability"] or + self.solid_wall["suitability"] or + self.room_roof["suitability"] or + self.flat_roof["suitability"] or + self.suspended_floor["suitability"] or + self.solid_floor["suitability"] + ) + + if tenure == "Rented (social)": + + if is_below_c and (not is_below_e): + # this is a placeholder methodology + self.gbis = { + "eligible": int(self.epc["potential-energy-efficiency"]) > current_sap, + "message": "proxy methodology until we complete innovation measure recommendations" + } + return + elif (not is_below_c) and is_below_e: + self.gbis = { + "eligible": needs_measure, + "message": "proxy methodology until we complete innovation measure recommendations" + } + return + else: + self.gbis = { + "eligible": False, + "message": "not eligible" + } + return + + elif tenure == "Rented (private)": + self.gbis = { + "eligible": is_below_c and needs_measure, + "message": "conditional tenant occupancy requirements and coucil tax band" + } + return + elif tenure == "Owner-occupied": + self.gbis = { + "eligible": False, + "message": "Out-of-scope" + } + return + + elif (tenure is None) or tenure == "unknown": + self.gbis = { + "eligible": needs_measure, + "message": "unknown tenure" + } + return + else: + raise ValueError("Implement me other tenure types") + + def check_eco4_potential(self): + """ + Because ECO4 supports nearly all measures, if we have commercial agreements in place then we + :return: + """ diff --git a/etl/eligibility/ha_15_32/app.py b/etl/eligibility/ha_15_32/app.py index 10c46f12..027b2c53 100644 --- a/etl/eligibility/ha_15_32/app.py +++ b/etl/eligibility/ha_15_32/app.py @@ -470,6 +470,8 @@ def get_ha_32data(ha_data, cleaned, cleaning_data, created_at): "walls": None, "date_epc": None, "message": "No EPC found", + "gbis_eligible_future": None, + "gbis_eligible_future_message": None, } ) continue @@ -483,8 +485,8 @@ def get_ha_32data(ha_data, cleaned, cleaning_data, created_at): penultimate_epc = newest_epc eligibility = Eligibility(epc=newest_epc, cleaned=cleaned) - eligibility.check_gbis() - eligibility.check_eco4() + eligibility.check_gbis_warmfront() + eligibility.check_eco4_warmfront() # If there is no eligibility, we need to check the penultimate epc # However, we only check the penultimate epc if the property is identified @@ -493,12 +495,17 @@ def get_ha_32data(ha_data, cleaned, cleaning_data, created_at): # However, if the property HAS been identified, we don't want to check the penultimate EPC since # The newest EPC will reflect the current state of the home and therefore we determine if there is a new # opportunity for retrofit - if (not eligibility.eco4["eligible"]) and (not eligibility.gbis) and (house["identified"]): + if (not eligibility.eco4_warmfront["eligible"]) and (not eligibility.gbis_warmfront) and (house["identified"]): eligibility = Eligibility(epc=penultimate_epc, cleaned=cleaned) - eligibility.check_gbis() - eligibility.check_eco4() + eligibility.check_gbis_warmfront() + eligibility.check_eco4_warmfront() - if eligibility.eco4["eligible"]: + # If the house is not identified, we do a full gbis and eco4 check + # TODO: Add in ECO4 check + eligibility.check_gbis() + # eligibility.check_eco4() + + if eligibility.eco4_warmfront["eligible"]: scoring_dictionary = prepare_model_data_row( property_id=house["row_id"], modelling_epc=eligibility.epc, @@ -511,33 +518,33 @@ def get_ha_32data(ha_data, cleaned, cleaning_data, created_at): { "row_id": house["row_id"], "warmfront_identified": house["identified"], - "gbis_eligible": eligibility.gbis, - "eco4_eligible": eligibility.eco4["eligible"], + "gbis_eligible": eligibility.gbis_warmfront, + "eco4_eligible": eligibility.eco4_warmfront["eligible"], "sap": float(eligibility.epc["current-energy-efficiency"]), "roof": eligibility.roof["clean_description"], "walls": eligibility.walls["clean_description"], "date_epc": eligibility.epc["lodgement-date"], "message": "eco4 conditional on post sap", + "gbis_eligible_future": eligibility.gbis["eligible"], + "gbis_eligible_future_message": eligibility.gbis["message"], } ) continue - # if (house["identified"] and not eligibility.gbis) and ( - # house["identified"] and not eligibility.eco4["eligible"]): - # raise NotImplementedError("Investigate ms") - # If nothing is eligible or gbis is eligible, then we make a record this results.append( { "row_id": house["row_id"], "warmfront_identified": house["identified"], - "gbis_eligible": eligibility.gbis, - "eco4_eligible": eligibility.eco4["eligible"], + "gbis_eligible": eligibility.gbis_warmfront, + "eco4_eligible": eligibility.eco4_warmfront["eligible"], "sap": float(eligibility.epc["current-energy-efficiency"]), "roof": eligibility.roof["clean_description"], "walls": eligibility.walls["clean_description"], "date_epc": eligibility.epc["lodgement-date"], - "message": None + "message": None, + "gbis_eligible_future": eligibility.gbis["eligible"], + "gbis_eligible_future_message": eligibility.gbis["message"], } ) @@ -613,8 +620,8 @@ def get_ha_15data(ha_data, cleaned, cleaning_data, created_at): penultimate_epc = newest_epc eligibility = Eligibility(epc=newest_epc, cleaned=cleaned) - eligibility.check_gbis() - eligibility.check_eco4() + eligibility.check_gbis_warmfront() + eligibility.check_eco4_warmfront() # If there is no eligibility, we need to check the penultimate epc # However, we only check the penultimate epc if the property is identified @@ -623,12 +630,12 @@ def get_ha_15data(ha_data, cleaned, cleaning_data, created_at): # However, if the property HAS been identified, we don't want to check the penultimate EPC since # The newest EPC will reflect the current state of the home and therefore we determine if there is a new # opportunity for retrofit - if (not eligibility.eco4["eligible"]) and (not eligibility.gbis) and (house["identified"]): + if (not eligibility.eco4_warmfront["eligible"]) and (not eligibility.gbis_warmfront) and (house["identified"]): eligibility = Eligibility(epc=penultimate_epc, cleaned=cleaned) - eligibility.check_gbis() - eligibility.check_eco4() + eligibility.check_gbis_warmfront() + eligibility.check_eco4_warmfront() - if eligibility.eco4["eligible"]: + if eligibility.eco4_warmfront["eligible"]: scoring_dictionary = prepare_model_data_row( property_id=house["row_id"], modelling_epc=eligibility.epc, @@ -641,8 +648,8 @@ def get_ha_15data(ha_data, cleaned, cleaning_data, created_at): { "row_id": house["row_id"], "warmfront_identified": house["identified"], - "gbis_eligible": eligibility.gbis, - "eco4_eligible": eligibility.eco4["eligible"], + "gbis_eligible": eligibility.gbis_warmfront, + "eco4_eligible": eligibility.eco4_warmfront["eligible"], "sap": float(eligibility.epc["current-energy-efficiency"]), "roof": eligibility.roof["clean_description"], "walls": eligibility.walls["clean_description"], @@ -652,17 +659,13 @@ def get_ha_15data(ha_data, cleaned, cleaning_data, created_at): ) continue - # if (house["identified"] and not eligibility.gbis) and ( - # house["identified"] and not eligibility.eco4["eligible"]): - # raise NotImplementedError("Investigate ms") - # If nothing is eligible or gbis is eligible, then we make a record this results.append( { "row_id": house["row_id"], "warmfront_identified": house["identified"], - "gbis_eligible": eligibility.gbis, - "eco4_eligible": eligibility.eco4["eligible"], + "gbis_eligible": eligibility.gbis_warmfront, + "eco4_eligible": eligibility.eco4_warmfront["eligible"], "sap": float(eligibility.epc["current-energy-efficiency"]), "roof": eligibility.roof["clean_description"], "walls": eligibility.walls["clean_description"],