matching for acis done

This commit is contained in:
Khalim Conn-Kowlessar 2025-03-08 21:58:46 +00:00
parent d06791db18
commit 816a1fa565

View file

@ -800,7 +800,7 @@ class AssetList:
self.standardised_asset_list[self.ATTRIBUTE_HAS_SOLAR] = ( self.standardised_asset_list[self.ATTRIBUTE_HAS_SOLAR] = (
self.standardised_asset_list[self.FIND_EPC_DATA_NAMES["Solar photovoltaics"]] | 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"] accepted_epc_property_types = ["House", "Flat", "Bungalow", "Maisonette"]
@ -1007,10 +1007,8 @@ class AssetList:
) )
elif self.old_format_non_intrusives_present: elif self.old_format_non_intrusives_present:
non_intrusives_wall_filter = ( non_intrusives_wall_filter = (
self.standardised_asset_list['non-intrusives: WFT Findings'].isin( self.standardised_asset_list['non-intrusives: WFT Findings'].str.lower().isin(
[ ["empty cavity", "partial fill"]
"EMPTY CAVITY", "Partial fill"
]
) )
) )
else: else:
@ -1018,7 +1016,7 @@ class AssetList:
self.standardised_asset_list["non_intrusive_indicates_empty_cavity"] = ( self.standardised_asset_list["non_intrusive_indicates_empty_cavity"] = (
(~self.standardised_asset_list[self.STANDARD_PROPERTY_TYPE].isin(["bedsit"])) & (~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[self.STANDARD_YEAR_BUILT] <= 2002) &
( (
self.standardised_asset_list[ self.standardised_asset_list[
@ -1066,39 +1064,14 @@ class AssetList:
self.standardised_asset_list["epc_indicates_empty_cavity"] 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_wall_filter = (
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"] = (
(self.standardised_asset_list["non-intrusives: Construction"] == "CAVITY") & (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: Insulated"].isin(["RETRO DRILLED", "FILLED AT BUILD"])) &
(~self.standardised_asset_list['non-intrusives: Material'].isin( (~self.standardised_asset_list['non-intrusives: Material'].isin(
@ -1107,314 +1080,446 @@ class AssetList:
) )
) )
# Adjust self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"] = (
self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"] = np.where( extraction_wall_filter & (
self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"], self.standardised_asset_list[self.ATTRIBUTE_SAP_THRESHOLD_AND_BELOW]
False, )
self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"]
) )
###################################################### # Also include work without the SAP filter as optimistic
# Solar self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"] = (
###################################################### extraction_wall_filter
# Criteria: )
# Check 1: Does the property have a valid heating system?
self.standardised_asset_list["solar_landlord_data_indicates_correct_heating_system"] = ( elif self.old_format_non_intrusives_present:
self.standardised_asset_list[self.STANDARD_HEATING_SYSTEM].isin( print("Review these categories with Kieran")
["air source heat pump", "ground source heat pump", "high heat retention storage heaters"] 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["non_intrusive_indicates_cavity_extraction"] = (
( extraction_wall_filter & (
self.standardised_asset_list[self.EPC_API_DATA_NAMES["mainheat-description"]] self.standardised_asset_list[self.ATTRIBUTE_SAP_THRESHOLD_AND_BELOW]
.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( self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"] = (
"electric storage heaters" extraction_wall_filter
) & ( )
self.standardised_asset_list[self.EPC_API_DATA_NAMES[
"mainheatcont-description"]] == "Controls for high heat retention storage heaters" 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["solar_epc_data_indicates_requires_heating_upgrade"] = (
self.standardised_asset_list["property_has_solar"] = ( self.standardised_asset_list[self.EPC_API_DATA_NAMES["mainheat-description"]].str.lower().str.contains(
(self.standardised_asset_list[self.STANDARD_EXISTING_PV] == "already has PV") | "electric storage heaters|room heaters"
(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[
self.EPC_API_DATA_NAMES["mainheatcont-description"]
] != "Controls for high heat retention storage heaters"
) )
)
# Check 3: Does the property meet the fabric condition # Basic check - both of the previous two shouldn't be true simultaneously
# Solar PV installs are subject to the minimum insulation requirements which means: if (
# 1) one of the following insulation measures must be installed as part of the same self.standardised_asset_list["solar_epc_data_indicates_correct_heating_system"] &
# ECO4 project: self.standardised_asset_list["solar_epc_data_indicates_requires_heating_upgrade"]
# • roof insulation (flat roof, pitched roof, room-in-roof) ).sum():
# • exterior facing wall insulation (cavity wall, solid wall) raise ValueError("Both heating system checks are true - this should not be possible")
# • 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
self.standardised_asset_list["solar_landlord_walls_insulated"] = ( # Check 2: Does the property have solar already
self.standardised_asset_list[self.STANDARD_WALL_CONSTRUCTION].isin( if self.non_intrusives_present:
["filled cavity", "insulated solid brick"] 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["solar_non_intrusives_walls_insulated"] = (
self.standardised_asset_list["non-intrusives: Insulated"].isin( self.standardised_asset_list["non-intrusives: Insulated"].isin(
["EWI", "RETRO DRILLED", "FILLED AT BUILD"] ["EWI", "RETRO DRILLED", "FILLED AT BUILD"]
) )
) )
elif self.old_format_non_intrusives_present:
# TODO: We don't have information about the roof from this landlord self.standardised_asset_list["solar_non_intrusives_walls_insulated"] = (
self.standardised_asset_list["non-intrusives: WFT Findings"].str.lower().str.strip().isin(
# We merge on the u-value for average thermal transmittance ["retro drilled", "retro filled", "ewi", "retro drilled/ solid"]
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"]
) )
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.standardised_asset_list[
self.EPC_API_DATA_NAMES[ self.EPC_API_DATA_NAMES["floor-description"]].str.lower().str.contains("solid")
"walls-description"]].str.lower().str.contains( ) & (
"|".join( self.standardised_asset_list[self.EPC_API_DATA_NAMES["floor-description"]].str.lower()
self.EPC_INSULATED_WALLS_SUBSTRINGS) .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[ self.standardised_asset_list[
"walls_u_value"].apply( self.EPC_API_DATA_NAMES["floor-description"]
lambda x: x <= 0.7 if not pd.isnull(x) else False ].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 # Set up the filters to stop repetition
roof_uvalue_data = pd.DataFrame(cleaned["roof-description"]) correct_heating_system = (
roof_uvalue_data = roof_uvalue_data[ self.standardised_asset_list["solar_landlord_data_indicates_correct_heating_system"] |
~pd.isnull(roof_uvalue_data["thermal_transmittance"]) self.standardised_asset_list["solar_epc_data_indicates_correct_heating_system"]
][["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( needs_heating_upgrade = (
roof_uvalue_data, how="left", on=self.EPC_API_DATA_NAMES["roof-description"] 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 walls_are_insulated = (
self.standardised_asset_list["solar_epc_roof_insulated"] = ( self.standardised_asset_list["solar_landlord_walls_insulated"] |
self.standardised_asset_list[self.EPC_API_DATA_NAMES["roof-description"]].str.lower().str.contains( self.standardised_asset_list["solar_epc_walls_insulated"] |
"|".join(self.EPC_INSULATED_ROOF_SUBSTRINGS), regex=False self.standardised_asset_list["solar_non_intrusives_walls_insulated"]
) | ( )
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.standardised_asset_list["solar_eligible_solid_floor"] = (
self.ATTRIBUTE_EPC_ROOF_INSULATION_THICKNESS].apply( # Landlord data or EPC data indicates the heating system is appropriate
lambda x: int(x) < 200 if str(x).isdigit() else False 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! # With heating upgrade
self.standardised_asset_list["epc_has_floor_recommendation"] = ( self.standardised_asset_list["solar_eligible_solid_floor_needs_heating_upgrade"] = (
self.standardised_asset_list["epc_has_floor_recommendation"].fillna(False) # 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 # We shouldn't have an overlap
floors_uvalue_data = pd.DataFrame(cleaned["floor-description"]) if (
floors_uvalue_data = floors_uvalue_data[ self.standardised_asset_list["solar_eligible_solid_floor"] &
~pd.isnull(floors_uvalue_data["thermal_transmittance"]) self.standardised_asset_list["solar_eligible_solid_floor_needs_heating_upgrade"]
][["original_description", "thermal_transmittance"]].rename( ).sum():
columns={ raise ValueError("Both heating upgrade and no heating upgrade are true - this should not be possible")
"original_description": self.EPC_API_DATA_NAMES["floor-description"],
"thermal_transmittance": "floor_u_value"
}
)
# Merge on # Solid floor but needs a loft top-up
self.standardised_asset_list = self.standardised_asset_list.merge( self.standardised_asset_list["solar_eligible_solid_floor_needs_loft"] = (
floors_uvalue_data, how="left", on=self.EPC_API_DATA_NAMES["floor-description"] # 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_eligible_solid_floor_needs_loft_needs_heating_upgrade"] = (
self.standardised_asset_list["solar_epc_floor_is_solid_no_recommendation"] = ( # Needs heating upgrade
( needs_heating_upgrade &
( # The property doesn't currently have solar
self.standardised_asset_list[self.EPC_API_DATA_NAMES["floor-description"]].str ~self.standardised_asset_list["property_has_solar"] &
.lower().str.contains("solid") # The walls are insulated
) & ( walls_are_insulated &
~self.standardised_asset_list["epc_has_floor_recommendation"] # Roof is insulated
) & ( self.standardised_asset_list["solar_epc_loft_needs_topup"] &
# We do not utilise estimated EPCs for this method because we will always find that self.standardised_asset_list["solar_epc_floor_is_solid_no_recommendation"]
# "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")
)
)
)
# Check for other floor types, insulated # Other floor type, fully insulated
self.standardised_asset_list["solar_epc_floor_is_other_insulated"] = ( self.standardised_asset_list["solar_eligible_other_floor"] = (
# The floor is suspended and insulated # 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[self.EPC_API_DATA_NAMES["floor-description"]].str ~self.standardised_asset_list["property_has_solar"] &
.lower().str.contains("suspended") # The walls are insulated
) & ( walls_are_insulated &
~self.standardised_asset_list["epc_has_floor_recommendation"] # Roof is insulated
) & ( self.standardised_asset_list["solar_epc_roof_insulated"] &
# We do not utilise estimated EPCs for this method because we will always find that self.standardised_asset_list["solar_epc_floor_is_other_insulated"]
# "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
)
)
)
# We now put together the criteria: # With heating upgrade
# Flag properties that look eligible for solar, that have solid floors self.standardised_asset_list["solar_eligible_other_floor_needs_heating_upgrade"] = (
# TODO: We'll need to revise this # Needs heating upgrade
self.standardised_asset_list["solar_eligible_solid_floor"] = ( needs_heating_upgrade &
# Landlord data or EPC data indicates the heating system is appropriate # The property doesn't currently have solar
( ~self.standardised_asset_list["property_has_solar"] &
self.standardised_asset_list["solar_landlord_data_indicates_correct_heating_system"] | # The walls are insulated
self.standardised_asset_list["solar_epc_data_indicates_correct_heating_system"] walls_are_insulated &
) & # Roof is insulated
# The property doesn't currently have solar self.standardised_asset_list["solar_epc_roof_insulated"] &
~self.standardised_asset_list["property_has_solar"] & self.standardised_asset_list["solar_epc_floor_is_other_insulated"]
# 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"]
)
# Solid floor but needs a loft top-up # Other floor type, needs loft top-up
self.standardised_asset_list["solar_eligible_solid_floor_needs_loft"] = ( self.standardised_asset_list["solar_eligible_other_floor_needs_loft"] = (
# Landlord data or EPC data indicates the heating system is appropriate # Landlord data or EPC data indicates the heating system is appropriate
( correct_heating_system &
self.standardised_asset_list["solar_landlord_data_indicates_correct_heating_system"] | # The property doesn't currently have solar
self.standardised_asset_list["solar_epc_data_indicates_correct_heating_system"] ~self.standardised_asset_list["property_has_solar"] &
) & # The walls are insulated
# The property doesn't currently have solar walls_are_insulated &
~self.standardised_asset_list["property_has_solar"] & # Roof need loft top-up
# The walls are insulated self.standardised_asset_list["solar_epc_loft_needs_topup"] &
( # Floor is not solid, but is insulated
self.standardised_asset_list["solar_landlord_walls_insulated"] | self.standardised_asset_list["solar_epc_floor_is_other_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, 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"] = ( # Drop anything we don't need
# Landlord data or EPC data indicates the heating system is appropriate self.standardised_asset_list = self.standardised_asset_list.drop(
( columns=["walls_u_value", "roof_u_value", "floor_u_value"]
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"]
)
# Other floor type, needs loft top-up # Adjust flagged extraction jobs to remove anything for solar
self.standardised_asset_list["solar_eligible_other_floor_needs_loft"] = ( self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"] = (
# Landlord data or EPC data indicates the heating system is appropriate self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"] &
( ~self.standardised_asset_list["solar_eligible_solid_floor"] &
self.standardised_asset_list["solar_landlord_data_indicates_correct_heating_system"] | ~self.standardised_asset_list["solar_eligible_solid_floor_needs_loft"]
self.standardised_asset_list["solar_epc_data_indicates_correct_heating_system"] # ~self.standardised_asset_list["solar_eligible_other_floor"] &
) & # ~self.standardised_asset_list["solar_eligible_other_floor_needs_loft"]
# 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"]
)
blocks_of_flats = self.standardised_asset_list[ blocks_of_flats = self.standardised_asset_list[
self.standardised_asset_list[self.STANDARD_PROPERTY_TYPE] == "block of flats" 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 # Finally, we note why each property has been flagged
self.standardised_asset_list["cavity_reason"] = None self.standardised_asset_list["cavity_reason"] = None
self.standardised_asset_list["cavity_reason"] = np.where( 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"] = None
self.standardised_asset_list["solar_reason"] = np.where( self.standardised_asset_list["solar_reason"] = np.where(
self.standardised_asset_list["solar_eligible_solid_floor"], 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"]
) )
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_reason"] = np.where(
self.standardised_asset_list["solar_eligible_solid_floor_needs_loft"], 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"]
) )
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_reason"] = np.where(
self.standardised_asset_list["solar_eligible_other_floor"], 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"]
) )
self.standardised_asset_list["solar_reason"] = np.where( self.standardised_asset_list["solar_reason"] = np.where(
self.standardised_asset_list["solar_eligible_other_floor_needs_loft"], self.standardised_asset_list["solar_eligible_other_floor_needs_heating_upgrade"],
"Other Floor, Insulated, Needs Loft", "Other Floor, Insulated, No Existing Solar, Needs Heating Upgrade",
self.standardised_asset_list["solar_reason"] 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): def flat_analysis(self):
# We need to deduce the building name - we strip out the house number # We need to deduce the building name - we strip out the house number