From 816a1fa565116fd92e02cf426d29c8d2f68727d1 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Sat, 8 Mar 2025 21:58:46 +0000 Subject: [PATCH] matching for acis done --- asset_list/AssetList.py | 777 +++++++++++++++++++++++----------------- 1 file changed, 457 insertions(+), 320 deletions(-) diff --git a/asset_list/AssetList.py b/asset_list/AssetList.py index d9922a97..689e752b 100644 --- a/asset_list/AssetList.py +++ b/asset_list/AssetList.py @@ -800,7 +800,7 @@ class AssetList: self.standardised_asset_list[self.ATTRIBUTE_HAS_SOLAR] = ( self.standardised_asset_list[self.FIND_EPC_DATA_NAMES["Solar photovoltaics"]] | - ~self.standardised_asset_list[self.EPC_API_DATA_NAMES["photo-supply"]].isin(["0.0", 0, None, ""]) + ~self.standardised_asset_list[self.EPC_API_DATA_NAMES["photo-supply"]].isin(["0.0", 0, None, "", np.nan]) ) accepted_epc_property_types = ["House", "Flat", "Bungalow", "Maisonette"] @@ -1007,10 +1007,8 @@ class AssetList: ) elif self.old_format_non_intrusives_present: non_intrusives_wall_filter = ( - self.standardised_asset_list['non-intrusives: WFT Findings'].isin( - [ - "EMPTY CAVITY", "Partial fill" - ] + self.standardised_asset_list['non-intrusives: WFT Findings'].str.lower().isin( + ["empty cavity", "partial fill"] ) ) else: @@ -1018,7 +1016,7 @@ class AssetList: self.standardised_asset_list["non_intrusive_indicates_empty_cavity"] = ( (~self.standardised_asset_list[self.STANDARD_PROPERTY_TYPE].isin(["bedsit"])) & - non_intrusives_wall_filter & + non_intrusives_wall_filter & (self.standardised_asset_list[self.STANDARD_YEAR_BUILT] <= 2002) & ( self.standardised_asset_list[ @@ -1066,39 +1064,14 @@ class AssetList: self.standardised_asset_list["epc_indicates_empty_cavity"] ) + ###################################################### + # Extraction + ###################################################### + # as needing a CIGA check. What is the logic we should be applying here? if self.non_intrusives_present: - - if self.non_intrusives_present: - - - - - - - - - - - ###################################################### - # Extraction - ###################################################### - - # as needing a CIGA check. What is the logic we should be applying here? - self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"] = ( - (self.standardised_asset_list["non-intrusives: Construction"] == "CAVITY") & - (self.standardised_asset_list["non-intrusives: Insulated"].isin(["RETRO DRILLED", "FILLED AT BUILD"])) & - (~self.standardised_asset_list['non-intrusives: Material'].isin( - ["GREY LOOSE BEAD", "COMPACTED BEAD", "FIBRE BATT NO CAVITY", "EMPTY NARROW BELOW 30mm"] - ) - ) & ( - self.standardised_asset_list[self.ATTRIBUTE_SAP_THRESHOLD_AND_BELOW] - ) - ) - - # Also include work without the SAP filter as optimistic - self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"] = ( + extraction_wall_filter = ( (self.standardised_asset_list["non-intrusives: Construction"] == "CAVITY") & (self.standardised_asset_list["non-intrusives: Insulated"].isin(["RETRO DRILLED", "FILLED AT BUILD"])) & (~self.standardised_asset_list['non-intrusives: Material'].isin( @@ -1107,314 +1080,446 @@ class AssetList: ) ) - # Adjust - self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"] = np.where( - self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"], - False, - self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"] + self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"] = ( + extraction_wall_filter & ( + self.standardised_asset_list[self.ATTRIBUTE_SAP_THRESHOLD_AND_BELOW] + ) ) - ###################################################### - # Solar - ###################################################### - # Criteria: - # Check 1: Does the property have a valid heating system? - self.standardised_asset_list["solar_landlord_data_indicates_correct_heating_system"] = ( - self.standardised_asset_list[self.STANDARD_HEATING_SYSTEM].isin( - ["air source heat pump", "ground source heat pump", "high heat retention storage heaters"] + # Also include work without the SAP filter as optimistic + self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"] = ( + extraction_wall_filter + ) + + elif self.old_format_non_intrusives_present: + print("Review these categories with Kieran") + extraction_wall_filter = ( + self.standardised_asset_list['non-intrusives: WFT Findings'].str.lower().str.strip().isin( + ["retro drilled", "retro filled", "fibre from build", "polybead"] ) ) - self.standardised_asset_list["solar_epc_data_indicates_correct_heating_system"] = ( - ( - self.standardised_asset_list[self.EPC_API_DATA_NAMES["mainheat-description"]] - .str.lower().str.contains("air source heat pump|ground source heat pump") - ) | ( - self.standardised_asset_list[ - self.EPC_API_DATA_NAMES["mainheat-description"]].str.lower().str.contains( - "electric storage heaters" - ) & ( - self.standardised_asset_list[self.EPC_API_DATA_NAMES[ - "mainheatcont-description"]] == "Controls for high heat retention storage heaters" - ) + self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"] = ( + extraction_wall_filter & ( + self.standardised_asset_list[self.ATTRIBUTE_SAP_THRESHOLD_AND_BELOW] + ) + ) + + self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"] = ( + extraction_wall_filter + ) + + else: + raise NotImplementedError("need to implement the case for non-intrusives") + + # Adjust + self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"] = np.where( + self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"], + False, + self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"] + ) + + ###################################################### + # Solar + ###################################################### + # Criteria: + # Check 1: Does the property have a valid heating system? + self.standardised_asset_list["solar_landlord_data_indicates_correct_heating_system"] = ( + self.standardised_asset_list[self.STANDARD_HEATING_SYSTEM].isin( + ["air source heat pump", "ground source heat pump", "high heat retention storage heaters"] + ) + ) + self.standardised_asset_list["solar_landlord_data_indicates_needs_heating_upgrade"] = ( + self.standardised_asset_list[self.STANDARD_HEATING_SYSTEM].isin( + ["electric storage heaters", "room heaters"] + ) + ) + + self.standardised_asset_list["solar_epc_data_indicates_correct_heating_system"] = ( + ( + self.standardised_asset_list[self.EPC_API_DATA_NAMES["mainheat-description"]] + .str.lower().str.contains("air source heat pump|ground source heat pump") + ) | ( + self.standardised_asset_list[ + self.EPC_API_DATA_NAMES["mainheat-description"]].str.lower().str.contains( + "electric storage heaters" + ) & ( + self.standardised_asset_list[self.EPC_API_DATA_NAMES[ + "mainheatcont-description"]] == "Controls for high heat retention storage heaters" ) ) + ) - # Check 2: Does the property have solar already - self.standardised_asset_list["property_has_solar"] = ( - (self.standardised_asset_list[self.STANDARD_EXISTING_PV] == "already has PV") | - (self.standardised_asset_list["non-intrusives: PV, ACCESS ISSUE, SEE NOTES"] == "SOLAR PV ON ROOF") | - (self.standardised_asset_list[self.ATTRIBUTE_HAS_SOLAR]) + self.standardised_asset_list["solar_epc_data_indicates_requires_heating_upgrade"] = ( + self.standardised_asset_list[self.EPC_API_DATA_NAMES["mainheat-description"]].str.lower().str.contains( + "electric storage heaters|room heaters" + ) & ( + self.standardised_asset_list[ + self.EPC_API_DATA_NAMES["mainheatcont-description"] + ] != "Controls for high heat retention storage heaters" ) + ) - # Check 3: Does the property meet the fabric condition - # Solar PV installs are subject to the minimum insulation requirements which means: - # 1) one of the following insulation measures must be installed as part of the same - # ECO4 project: - # • roof insulation (flat roof, pitched roof, room-in-roof) - # • exterior facing wall insulation (cavity wall, solid wall) - # • party cavity wall insulation - # • floor insulation (solid and underfloor) - # - # OR - # - # all measures (except any exempted measure referred to in paragraph 4.28) - # listed in paragraph a) must already be installed - # - # With this in mind, we look for 2 clases - # 1) The property is fully insulated apart from the loft (<200mm insulation) - # 2) THe property is fully insulated + # Basic check - both of the previous two shouldn't be true simultaneously + if ( + self.standardised_asset_list["solar_epc_data_indicates_correct_heating_system"] & + self.standardised_asset_list["solar_epc_data_indicates_requires_heating_upgrade"] + ).sum(): + raise ValueError("Both heating system checks are true - this should not be possible") - self.standardised_asset_list["solar_landlord_walls_insulated"] = ( - self.standardised_asset_list[self.STANDARD_WALL_CONSTRUCTION].isin( - ["filled cavity", "insulated solid brick"] + # Check 2: Does the property have solar already + if self.non_intrusives_present: + existing_solar_non_intrusives_check = ( + self.standardised_asset_list["non-intrusives: PV, ACCESS ISSUE, SEE NOTES"] == "SOLAR PV ON ROOF" + ) + elif self.old_format_non_intrusives_present: + existing_solar_non_intrusives_check = ( + self.standardised_asset_list["non-intrusives: WFT Findings"].str.lower().str.strip().isin( + ["solar pv on roof"] ) ) + else: + raise NotImplementedError("need to implement the case for non-intrusives") + self.standardised_asset_list["property_has_solar"] = ( + (self.standardised_asset_list[self.STANDARD_EXISTING_PV] == "already has PV") | + existing_solar_non_intrusives_check | + (self.standardised_asset_list[self.ATTRIBUTE_HAS_SOLAR]) + ) + + # Check 3: Does the property meet the fabric condition + # Solar PV installs are subject to the minimum insulation requirements which means: + # 1) one of the following insulation measures must be installed as part of the same + # ECO4 project: + # • roof insulation (flat roof, pitched roof, room-in-roof) + # • exterior facing wall insulation (cavity wall, solid wall) + # • party cavity wall insulation + # • floor insulation (solid and underfloor) + # + # OR + # + # all measures (except any exempted measure referred to in paragraph 4.28) + # listed in paragraph a) must already be installed + # + # With this in mind, we look for 2 clases + # 1) The property is fully insulated apart from the loft (<200mm insulation) + # 2) THe property is fully insulated + + print("Should we include cavity properties where they might be uninsulated?") + self.standardised_asset_list["solar_landlord_walls_insulated"] = ( + self.standardised_asset_list[self.STANDARD_WALL_CONSTRUCTION].isin( + ["filled cavity", "insulated solid brick"] + ) + ) + + if self.non_intrusives_present: self.standardised_asset_list["solar_non_intrusives_walls_insulated"] = ( self.standardised_asset_list["non-intrusives: Insulated"].isin( ["EWI", "RETRO DRILLED", "FILLED AT BUILD"] ) ) - - # TODO: We don't have information about the roof from this landlord - - # We merge on the u-value for average thermal transmittance - walls_uvalue_data = pd.DataFrame(cleaned["walls-description"]) - walls_uvalue_data = walls_uvalue_data[ - ~pd.isnull(walls_uvalue_data["thermal_transmittance"]) - ][["original_description", "thermal_transmittance"]].rename( - columns={ - "original_description": self.EPC_API_DATA_NAMES["walls-description"], - "thermal_transmittance": "walls_u_value" - } - ) - self.standardised_asset_list = self.standardised_asset_list.merge( - walls_uvalue_data, how="left", on=self.EPC_API_DATA_NAMES["walls-description"] + elif self.old_format_non_intrusives_present: + self.standardised_asset_list["solar_non_intrusives_walls_insulated"] = ( + self.standardised_asset_list["non-intrusives: WFT Findings"].str.lower().str.strip().isin( + ["retro drilled", "retro filled", "ewi", "retro drilled/ solid"] + ) ) + else: + raise NotImplementedError("need to implement the case for non-intrusives") - self.standardised_asset_list["solar_epc_walls_insulated"] = ( + # TODO: We don't have information about the roof from this landlord + + # We merge on the u-value for average thermal transmittance + walls_uvalue_data = pd.DataFrame(cleaned["walls-description"]) + walls_uvalue_data = walls_uvalue_data[ + ~pd.isnull(walls_uvalue_data["thermal_transmittance"]) + ][["original_description", "thermal_transmittance"]].rename( + columns={ + "original_description": self.EPC_API_DATA_NAMES["walls-description"], + "thermal_transmittance": "walls_u_value" + } + ) + self.standardised_asset_list = self.standardised_asset_list.merge( + walls_uvalue_data, how="left", on=self.EPC_API_DATA_NAMES["walls-description"] + ) + + self.standardised_asset_list["solar_epc_walls_insulated"] = ( + ( + self.standardised_asset_list[ + self.EPC_API_DATA_NAMES[ + "walls-description"]].str.lower().str.contains( + "|".join( + self.EPC_INSULATED_WALLS_SUBSTRINGS) + ) + ) | ( + self.standardised_asset_list[ + "walls_u_value"].apply( + lambda x: x <= 0.7 if not pd.isnull(x) else False + ) + ) + ) + + # We merge on the u-value for average thermal transmittance + roof_uvalue_data = pd.DataFrame(cleaned["roof-description"]) + roof_uvalue_data = roof_uvalue_data[ + ~pd.isnull(roof_uvalue_data["thermal_transmittance"]) + ][["original_description", "thermal_transmittance"]].rename( + columns={ + "original_description": self.EPC_API_DATA_NAMES["roof-description"], + "thermal_transmittance": "roof_u_value" + } + ) + + self.standardised_asset_list = self.standardised_asset_list.merge( + roof_uvalue_data, how="left", on=self.EPC_API_DATA_NAMES["roof-description"] + ) + + # If the u-value of a roof is less than 0.7 we consider it insulated + self.standardised_asset_list["solar_epc_roof_insulated"] = ( + self.standardised_asset_list[self.EPC_API_DATA_NAMES["roof-description"]].str.lower().str.contains( + "|".join(self.EPC_INSULATED_ROOF_SUBSTRINGS), regex=False + ) | ( + self.standardised_asset_list[self.ATTRIBUTE_EPC_ROOF_INSULATION_THICKNESS].apply( + lambda x: int(x) >= 200 if str(x).isdigit() else False + ) + ) | ( + self.standardised_asset_list["roof_u_value"].apply( + lambda x: x <= 0.7 if not pd.isnull(x) else False + ) + ) + ) + + self.standardised_asset_list["solar_epc_loft_needs_topup"] = self.standardised_asset_list[ + self.ATTRIBUTE_EPC_ROOF_INSULATION_THICKNESS].apply( + lambda x: int(x) < 200 if str(x).isdigit() else False + ) + + # TODO: Fill with False - should be temp! + self.standardised_asset_list["epc_has_floor_recommendation"] = ( + self.standardised_asset_list["epc_has_floor_recommendation"].fillna(False) + ) + + # We merge on the u-value for average thermal transmittance + floors_uvalue_data = pd.DataFrame(cleaned["floor-description"]) + floors_uvalue_data = floors_uvalue_data[ + ~pd.isnull(floors_uvalue_data["thermal_transmittance"]) + ][["original_description", "thermal_transmittance"]].rename( + columns={ + "original_description": self.EPC_API_DATA_NAMES["floor-description"], + "thermal_transmittance": "floor_u_value" + } + ) + + # Merge on + self.standardised_asset_list = self.standardised_asset_list.merge( + floors_uvalue_data, how="left", on=self.EPC_API_DATA_NAMES["floor-description"] + ) + + # We assume that a U-value of 0.5 or below is indicative of an insulated floor + self.standardised_asset_list["solar_epc_floor_is_solid_no_recommendation"] = ( + ( + ( + self.standardised_asset_list[self.EPC_API_DATA_NAMES["floor-description"]].str + .lower().str.contains("solid") + ) & ( + ~self.standardised_asset_list["epc_has_floor_recommendation"] + ) & ( + # We do not utilise estimated EPCs for this method because we will always find that + # "epc_has_floor_recommendation" is False + (self.standardised_asset_list["estimated"] == False) + ) + ) | ( ( self.standardised_asset_list[ - self.EPC_API_DATA_NAMES[ - "walls-description"]].str.lower().str.contains( - "|".join( - self.EPC_INSULATED_WALLS_SUBSTRINGS) - ) - ) | ( + self.EPC_API_DATA_NAMES["floor-description"]].str.lower().str.contains("solid") + ) & ( + self.standardised_asset_list[self.EPC_API_DATA_NAMES["floor-description"]].str.lower() + .str.contains(", insulated") + ) + ) + ) + + # Check for other floor types, insulated + self.standardised_asset_list["solar_epc_floor_is_other_insulated"] = ( + # The floor is suspended and insulated + ( + ( + self.standardised_asset_list[self.EPC_API_DATA_NAMES["floor-description"]].str + .lower().str.contains("suspended") + ) & ( + ~self.standardised_asset_list["epc_has_floor_recommendation"] + ) & ( + # We do not utilise estimated EPCs for this method because we will always find that + # "epc_has_floor_recommendation" is False + self.standardised_asset_list["estimated"] == False + ) + ) | ( + ( self.standardised_asset_list[ - "walls_u_value"].apply( - lambda x: x <= 0.7 if not pd.isnull(x) else False - ) + self.EPC_API_DATA_NAMES["floor-description"] + ].str.lower().str.contains("suspended") + ) & ( + self.standardised_asset_list[ + self.EPC_API_DATA_NAMES["floor-description"] + ].str.lower().str.contains(", insulated") + ) + ) | ( + self.standardised_asset_list["floor_u_value"].apply( + lambda x: x <= 0.5 if not pd.isnull(x) else False ) ) + ) + #################################### + # Check solar eligibility + #################################### - # We merge on the u-value for average thermal transmittance - roof_uvalue_data = pd.DataFrame(cleaned["roof-description"]) - roof_uvalue_data = roof_uvalue_data[ - ~pd.isnull(roof_uvalue_data["thermal_transmittance"]) - ][["original_description", "thermal_transmittance"]].rename( - columns={ - "original_description": self.EPC_API_DATA_NAMES["roof-description"], - "thermal_transmittance": "roof_u_value" - } - ) + # Set up the filters to stop repetition + correct_heating_system = ( + self.standardised_asset_list["solar_landlord_data_indicates_correct_heating_system"] | + self.standardised_asset_list["solar_epc_data_indicates_correct_heating_system"] + ) - self.standardised_asset_list = self.standardised_asset_list.merge( - roof_uvalue_data, how="left", on=self.EPC_API_DATA_NAMES["roof-description"] - ) + needs_heating_upgrade = ( + self.standardised_asset_list["solar_landlord_data_indicates_needs_heating_upgrade"] | + self.standardised_asset_list["solar_epc_data_indicates_requires_heating_upgrade"] + ) - # If the u-value of a roof is less than 0.7 we consider it insulated - self.standardised_asset_list["solar_epc_roof_insulated"] = ( - self.standardised_asset_list[self.EPC_API_DATA_NAMES["roof-description"]].str.lower().str.contains( - "|".join(self.EPC_INSULATED_ROOF_SUBSTRINGS), regex=False - ) | ( - self.standardised_asset_list[self.ATTRIBUTE_EPC_ROOF_INSULATION_THICKNESS].apply( - lambda x: int(x) >= 200 if str(x).isdigit() else False - ) - ) | ( - self.standardised_asset_list["roof_u_value"].apply( - lambda x: x <= 0.7 if not pd.isnull(x) else False - ) - ) - ) + walls_are_insulated = ( + self.standardised_asset_list["solar_landlord_walls_insulated"] | + self.standardised_asset_list["solar_epc_walls_insulated"] | + self.standardised_asset_list["solar_non_intrusives_walls_insulated"] + ) - self.standardised_asset_list["solar_epc_loft_needs_topup"] = self.standardised_asset_list[ - self.ATTRIBUTE_EPC_ROOF_INSULATION_THICKNESS].apply( - lambda x: int(x) < 200 if str(x).isdigit() else False - ) + self.standardised_asset_list["solar_eligible_solid_floor"] = ( + # Landlord data or EPC data indicates the heating system is appropriate + correct_heating_system & + # The property doesn't currently have solar + ~self.standardised_asset_list["property_has_solar"] & + # The walls are insulated + walls_are_insulated & + # Roof is insulated + self.standardised_asset_list["solar_epc_roof_insulated"] & + self.standardised_asset_list["solar_epc_floor_is_solid_no_recommendation"] + ) - # TODO: Fill with False - should be temp! - self.standardised_asset_list["epc_has_floor_recommendation"] = ( - self.standardised_asset_list["epc_has_floor_recommendation"].fillna(False) - ) + # With heating upgrade + self.standardised_asset_list["solar_eligible_solid_floor_needs_heating_upgrade"] = ( + # Needs heating upgrade + needs_heating_upgrade & + # The property doesn't currently have solar + ~self.standardised_asset_list["property_has_solar"] & + # The walls are insulated + walls_are_insulated & + # Roof is insulated + self.standardised_asset_list["solar_epc_roof_insulated"] & + self.standardised_asset_list["solar_epc_floor_is_solid_no_recommendation"] + ) + # Because the EPC data can be contradictrory, we remove any overlap + self.standardised_asset_list["solar_eligible_solid_floor_needs_heating_upgrade"] = np.where( + self.standardised_asset_list["solar_eligible_solid_floor"], + False, + self.standardised_asset_list["solar_eligible_solid_floor_needs_heating_upgrade"] + ) - # We merge on the u-value for average thermal transmittance - floors_uvalue_data = pd.DataFrame(cleaned["floor-description"]) - floors_uvalue_data = floors_uvalue_data[ - ~pd.isnull(floors_uvalue_data["thermal_transmittance"]) - ][["original_description", "thermal_transmittance"]].rename( - columns={ - "original_description": self.EPC_API_DATA_NAMES["floor-description"], - "thermal_transmittance": "floor_u_value" - } - ) + # We shouldn't have an overlap + if ( + self.standardised_asset_list["solar_eligible_solid_floor"] & + self.standardised_asset_list["solar_eligible_solid_floor_needs_heating_upgrade"] + ).sum(): + raise ValueError("Both heating upgrade and no heating upgrade are true - this should not be possible") - # Merge on - self.standardised_asset_list = self.standardised_asset_list.merge( - floors_uvalue_data, how="left", on=self.EPC_API_DATA_NAMES["floor-description"] - ) + # Solid floor but needs a loft top-up + self.standardised_asset_list["solar_eligible_solid_floor_needs_loft"] = ( + # Landlord data or EPC data indicates the heating system is appropriate + correct_heating_system & + # The property doesn't currently have solar + ~self.standardised_asset_list["property_has_solar"] & + # The walls are insulated + walls_are_insulated & + # Roof is insulated + self.standardised_asset_list["solar_epc_loft_needs_topup"] & + self.standardised_asset_list["solar_epc_floor_is_solid_no_recommendation"] + ) - # We assume that a U-value of 0.5 or below is indicative of an insulated floor - self.standardised_asset_list["solar_epc_floor_is_solid_no_recommendation"] = ( - ( - ( - self.standardised_asset_list[self.EPC_API_DATA_NAMES["floor-description"]].str - .lower().str.contains("solid") - ) & ( - ~self.standardised_asset_list["epc_has_floor_recommendation"] - ) & ( - # We do not utilise estimated EPCs for this method because we will always find that - # "epc_has_floor_recommendation" is False - (self.standardised_asset_list["estimated"] == False) - ) - ) | ( - ( - self.standardised_asset_list[ - self.EPC_API_DATA_NAMES["floor-description"]].str.lower().str.contains("solid") - ) & ( - self.standardised_asset_list[self.EPC_API_DATA_NAMES["floor-description"]].str.lower() - .str.contains(", insulated") - ) - ) - ) + self.standardised_asset_list["solar_eligible_solid_floor_needs_loft_needs_heating_upgrade"] = ( + # Needs heating upgrade + needs_heating_upgrade & + # The property doesn't currently have solar + ~self.standardised_asset_list["property_has_solar"] & + # The walls are insulated + walls_are_insulated & + # Roof is insulated + self.standardised_asset_list["solar_epc_loft_needs_topup"] & + self.standardised_asset_list["solar_epc_floor_is_solid_no_recommendation"] + ) - # Check for other floor types, insulated - self.standardised_asset_list["solar_epc_floor_is_other_insulated"] = ( - # The floor is suspended and insulated - ( - ( - self.standardised_asset_list[self.EPC_API_DATA_NAMES["floor-description"]].str - .lower().str.contains("suspended") - ) & ( - ~self.standardised_asset_list["epc_has_floor_recommendation"] - ) & ( - # We do not utilise estimated EPCs for this method because we will always find that - # "epc_has_floor_recommendation" is False - self.standardised_asset_list["estimated"] == False - ) - ) | ( - ( - self.standardised_asset_list[ - self.EPC_API_DATA_NAMES["floor-description"] - ].str.lower().str.contains("suspended") - ) & ( - self.standardised_asset_list[ - self.EPC_API_DATA_NAMES["floor-description"] - ].str.lower().str.contains(", insulated") - ) - ) | ( - self.standardised_asset_list["floor_u_value"].apply( - lambda x: x <= 0.5 if not pd.isnull(x) else False - ) - ) - ) + # Other floor type, fully insulated + self.standardised_asset_list["solar_eligible_other_floor"] = ( + # Landlord data or EPC data indicates the heating system is appropriate + correct_heating_system & + # The property doesn't currently have solar + ~self.standardised_asset_list["property_has_solar"] & + # The walls are insulated + walls_are_insulated & + # Roof is insulated + self.standardised_asset_list["solar_epc_roof_insulated"] & + self.standardised_asset_list["solar_epc_floor_is_other_insulated"] + ) - # We now put together the criteria: - # Flag properties that look eligible for solar, that have solid floors - # TODO: We'll need to revise this - self.standardised_asset_list["solar_eligible_solid_floor"] = ( - # Landlord data or EPC data indicates the heating system is appropriate - ( - self.standardised_asset_list["solar_landlord_data_indicates_correct_heating_system"] | - self.standardised_asset_list["solar_epc_data_indicates_correct_heating_system"] - ) & - # The property doesn't currently have solar - ~self.standardised_asset_list["property_has_solar"] & - # The walls are insulated - ( - self.standardised_asset_list["solar_landlord_walls_insulated"] | - self.standardised_asset_list["solar_epc_walls_insulated"] | - self.standardised_asset_list["solar_non_intrusives_walls_insulated"] - ) & - # Roof is insulated - self.standardised_asset_list["solar_epc_roof_insulated"] & - self.standardised_asset_list["solar_epc_floor_is_solid_no_recommendation"] - ) + # With heating upgrade + self.standardised_asset_list["solar_eligible_other_floor_needs_heating_upgrade"] = ( + # Needs heating upgrade + needs_heating_upgrade & + # The property doesn't currently have solar + ~self.standardised_asset_list["property_has_solar"] & + # The walls are insulated + walls_are_insulated & + # Roof is insulated + self.standardised_asset_list["solar_epc_roof_insulated"] & + self.standardised_asset_list["solar_epc_floor_is_other_insulated"] + ) - # Solid floor but needs a loft top-up - self.standardised_asset_list["solar_eligible_solid_floor_needs_loft"] = ( - # Landlord data or EPC data indicates the heating system is appropriate - ( - self.standardised_asset_list["solar_landlord_data_indicates_correct_heating_system"] | - self.standardised_asset_list["solar_epc_data_indicates_correct_heating_system"] - ) & - # The property doesn't currently have solar - ~self.standardised_asset_list["property_has_solar"] & - # The walls are insulated - ( - self.standardised_asset_list["solar_landlord_walls_insulated"] | - self.standardised_asset_list["solar_epc_walls_insulated"] | - self.standardised_asset_list["solar_non_intrusives_walls_insulated"] - ) & - # Roof is insulated - self.standardised_asset_list["solar_epc_loft_needs_topup"] & - self.standardised_asset_list["solar_epc_floor_is_solid_no_recommendation"] - ) + # Other floor type, needs loft top-up + self.standardised_asset_list["solar_eligible_other_floor_needs_loft"] = ( + # Landlord data or EPC data indicates the heating system is appropriate + correct_heating_system & + # The property doesn't currently have solar + ~self.standardised_asset_list["property_has_solar"] & + # The walls are insulated + walls_are_insulated & + # Roof need loft top-up + self.standardised_asset_list["solar_epc_loft_needs_topup"] & + # Floor is not solid, but is insulated + self.standardised_asset_list["solar_epc_floor_is_other_insulated"] + ) - # Other floor type, fully insulated + # With heating upgrade + self.standardised_asset_list["solar_eligible_other_floor_needs_loft_needs_heating_upgrade"] = ( + # Landlord data or EPC data indicates the heating system is appropriate + needs_heating_upgrade & + # The property doesn't currently have solar + ~self.standardised_asset_list["property_has_solar"] & + # The walls are insulated + walls_are_insulated & + # Roof need loft top-up + self.standardised_asset_list["solar_epc_loft_needs_topup"] & + # Floor is not solid, but is insulated + self.standardised_asset_list["solar_epc_floor_is_other_insulated"] + ) - self.standardised_asset_list["solar_eligible_other_floor"] = ( - # Landlord data or EPC data indicates the heating system is appropriate - ( - self.standardised_asset_list["solar_landlord_data_indicates_correct_heating_system"] | - self.standardised_asset_list["solar_epc_data_indicates_correct_heating_system"] - ) & - # The property doesn't currently have solar - ~self.standardised_asset_list["property_has_solar"] & - # The walls are insulated - ( - self.standardised_asset_list["solar_landlord_walls_insulated"] | - self.standardised_asset_list["solar_epc_walls_insulated"] - ) & - # Roof is insulated - self.standardised_asset_list["solar_epc_roof_insulated"] & - self.standardised_asset_list["solar_epc_floor_is_other_insulated"] - ) + # Drop anything we don't need + self.standardised_asset_list = self.standardised_asset_list.drop( + columns=["walls_u_value", "roof_u_value", "floor_u_value"] + ) - # Other floor type, needs loft top-up - self.standardised_asset_list["solar_eligible_other_floor_needs_loft"] = ( - # Landlord data or EPC data indicates the heating system is appropriate - ( - self.standardised_asset_list["solar_landlord_data_indicates_correct_heating_system"] | - self.standardised_asset_list["solar_epc_data_indicates_correct_heating_system"] - ) & - # The property doesn't currently have solar - ~self.standardised_asset_list["property_has_solar"] & - # The walls are insulated - ( - self.standardised_asset_list["solar_landlord_walls_insulated"] | - self.standardised_asset_list["solar_epc_walls_insulated"] - ) & - # Roof need loft top-up - self.standardised_asset_list["solar_epc_loft_needs_topup"] & - # Floor is not solid, but is insulated - self.standardised_asset_list["solar_epc_floor_is_other_insulated"] - ) - - # Drop anything we don't need - self.standardised_asset_list = self.standardised_asset_list.drop( - columns=["walls_u_value", "roof_u_value", "floor_u_value"] - ) - - # Adjust flagged extraction jobs to remove anything for solar - self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"] = ( - self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"] & - ~self.standardised_asset_list["solar_eligible_solid_floor"] & - ~self.standardised_asset_list["solar_eligible_solid_floor_needs_loft"] - # ~self.standardised_asset_list["solar_eligible_other_floor"] & - # ~self.standardised_asset_list["solar_eligible_other_floor_needs_loft"] - ) + # Adjust flagged extraction jobs to remove anything for solar + self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"] = ( + self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"] & + ~self.standardised_asset_list["solar_eligible_solid_floor"] & + ~self.standardised_asset_list["solar_eligible_solid_floor_needs_loft"] + # ~self.standardised_asset_list["solar_eligible_other_floor"] & + # ~self.standardised_asset_list["solar_eligible_other_floor_needs_loft"] + ) blocks_of_flats = self.standardised_asset_list[ self.standardised_asset_list[self.STANDARD_PROPERTY_TYPE] == "block of flats" @@ -1484,17 +1589,6 @@ class AssetList: ) } - # We produce a breakdown of the property types, for cavity fills - cavity_fills = self.standardised_asset_list[ - self.standardised_asset_list["non_intrusive_indicates_empty_cavity"] | ( - self.standardised_asset_list["epc_indicates_empty_cavity"] - ) - ] - - self.work_type_breakdowns = { - "empty_cavity": cavity_fills[self.STANDARD_PROPERTY_TYPE].value_counts() - } - # Finally, we note why each property has been flagged self.standardised_asset_list["cavity_reason"] = None self.standardised_asset_list["cavity_reason"] = np.where( @@ -1538,25 +1632,68 @@ class AssetList: self.standardised_asset_list["solar_reason"] = None self.standardised_asset_list["solar_reason"] = np.where( self.standardised_asset_list["solar_eligible_solid_floor"], - "Solid Floor, Insulated, No Solar", + "Solid Floor, Insulated, No Existing Solar", self.standardised_asset_list["solar_reason"] ) + self.standardised_asset_list["solar_reason"] = np.where( + self.standardised_asset_list["solar_eligible_solid_floor_needs_heating_upgrade"], + "Solid Floor, Insulated, No Existing Solar, Needs Heating Upgrade", + self.standardised_asset_list["solar_reason"] + ) + self.standardised_asset_list["solar_reason"] = np.where( self.standardised_asset_list["solar_eligible_solid_floor_needs_loft"], - "Solid Floor, Insulated, Needs Loft", + "Solid Floor, Insulated, Needs Loft, No Existing Solar", self.standardised_asset_list["solar_reason"] ) + self.standardised_asset_list["solar_reason"] = np.where( + self.standardised_asset_list["solar_eligible_solid_floor_needs_loft_needs_heating_upgrade"], + "Solid Floor, Insulated, Needs Loft, No Existing Solar, Needs Heating Upgrade", + self.standardised_asset_list["solar_reason"] + ) + self.standardised_asset_list["solar_reason"] = np.where( self.standardised_asset_list["solar_eligible_other_floor"], - "Other Floor, Insulated, No Solar", + "Other Floor, Insulated, No Existing Solar", self.standardised_asset_list["solar_reason"] ) self.standardised_asset_list["solar_reason"] = np.where( - self.standardised_asset_list["solar_eligible_other_floor_needs_loft"], - "Other Floor, Insulated, Needs Loft", + self.standardised_asset_list["solar_eligible_other_floor_needs_heating_upgrade"], + "Other Floor, Insulated, No Existing Solar, Needs Heating Upgrade", self.standardised_asset_list["solar_reason"] ) + self.standardised_asset_list["solar_reason"] = np.where( + self.standardised_asset_list["solar_eligible_other_floor_needs_loft"], + "Other Floor, Insulated, Needs Loft, No Existing Solar", + self.standardised_asset_list["solar_reason"] + ) + self.standardised_asset_list["solar_reason"] = np.where( + self.standardised_asset_list["solar_eligible_other_floor_needs_loft_needs_heating_upgrade"], + "Other Floor, Insulated, Needs Loft, No Existing Solar, Needs Heating Upgrade", + self.standardised_asset_list["solar_reason"] + ) + + # Flag anything that has existing outcomes + if self.outcomes is not None: + self.standardised_asset_list["cavity_reason"] = np.where( + ( + (self.standardised_asset_list["Surveyed"] > 0) | + (self.standardised_asset_list["Installer Refusal"] > 0) + ), + None, + self.standardised_asset_list["cavity_reason"] + ) + + if self.master_surveyed is not None: + self.standardised_asset_list["cavity_reason"] = np.where( + ( + (~pd.isnull(self.standardised_asset_list["SUBMISSION DATE"])) + ), + None, + self.standardised_asset_list["cavity_reason"] + ) + def flat_analysis(self): # We need to deduce the building name - we strip out the house number