added pps matrix to test data

This commit is contained in:
Khalim Conn-Kowlessar 2025-08-06 12:23:26 +01:00
parent 30e1eca74b
commit 98d64b9430
6 changed files with 17089 additions and 82 deletions

View file

@ -1,6 +1,6 @@
from enum import IntEnum, Enum from enum import IntEnum, Enum
CRM_PIPELINE_NAME = 'Operations - Housing Associations' CRM_PIPELINE_NAME = 'Operations - Social Housing'
class HubspotProcessStatus(IntEnum): class HubspotProcessStatus(IntEnum):

View file

@ -44,26 +44,27 @@ def app():
""" """
# inputs: # inputs:
reconcile_programme = True # If True, the hubspot upload will include all properties with a project code reconcile_programme = False # If True, the hubspot upload will include all properties with a project code
customer_domain = "https://southend.gov.uk" customer_domain = "https://shgroup.org.uk"
installer_name = "J & J CRUMP" installer_name = "SCIS"
asset_list_filepath = ( asset_list_filepath = (
"/Users/khalimconn-kowlessar/Documents/hestia/Customers/Southend/July 2025 Programme/SOUTHEND - RYAN - " "/Users/khalimconn-kowlessar/Downloads/20250701 Optivo Southern - Standardised.xlsx"
"Standardised 2.xlsx"
) )
asset_list_sheet_name = "Standardised Asset List" asset_list_sheet_name = "Solar Route Revised (100)"
asset_list_header = 0 asset_list_header = 0
contact_details_filepath = None contact_details_filepath = (
contacts_sheet_name = "Sheet 1" "/Users/khalimconn-kowlessar/Downloads/southern_optivo_solar_pv.xlsx"
contacts_landlord_property_id = "UPRN" )
contacts_phone_number_column = "phone_number" contacts_sheet_name = "Sheet1"
contacts_secondary_phone_number_column = "secondary_phone_number" contacts_landlord_property_id = "landlord_property_id"
contacts_secondary_contact_full_name = "secondary_contact_full_name" contacts_phone_number_column = "Primary phone number"
contacts_email_column = "email" contacts_secondary_phone_number_column = "Secondary phone number"
contacts_fullname_column = "fullname" contacts_secondary_contact_full_name = None
contacts_firstname_column = "First Name" contacts_email_column = "Email Address"
contacts_lastname_column = "Last Name" contacts_fullname_column = None
contacts_firstname_column = "Name"
contacts_lastname_column = None
existing_programme_filepath = None existing_programme_filepath = None
@ -89,6 +90,18 @@ def app():
reconcile_programme=reconcile_programme reconcile_programme=reconcile_programme
) )
for x in asset_list.hubspot_data["Phone <CONTACT phone>"].values:
normalize_uk_phone(x)
asset_list.hubspot_data["Phone <CONTACT phone>"] = (
asset_list.hubspot_data["Phone <CONTACT phone>"].astype("Int64").astype(str).apply(normalize_uk_phone)
)
asset_list.hubspot_data["Secondary Phone <CONTACT secondary_phone_number>"] = asset_list.hubspot_data[
"Secondary Phone <CONTACT secondary_phone_number>"].astype(
"Int64").astype(
str).apply(
normalize_uk_phone)
# Remove the existing programme # Remove the existing programme
# existing_programme = pd.read_csv(existing_programme_filepath, encoding="utf-8-sig") # existing_programme = pd.read_csv(existing_programme_filepath, encoding="utf-8-sig")
# asset_list.hubspot_data = asset_list.hubspot_data[ # asset_list.hubspot_data = asset_list.hubspot_data[

View file

@ -3,13 +3,14 @@ import pandas as pd
import numpy as np import numpy as np
from typing import List from typing import List
from backend.app.plan.schemas import HousingType from backend.app.plan.schemas import HousingType, WALL_INSULATION_MEASURES, ROOF_INSULATION_MEASURES
class EligibilityCaveats(Enum): class EligibilityCaveats(Enum):
TENANT_ON_BENEFITS_OR_LOW_INCOME = "tenant_on_benefits_or_low_income" TENANT_ON_BENEFITS_OR_LOW_INCOME = "tenant_on_benefits_or_low_income"
INNOVATION_REQUIRED = "innovation_required" INNOVATION_REQUIRED = "innovation_required"
SOLAR_NEEDS_HEATING = "solar_needs_heating" SOLAR_NEEDS_HEATING = "solar_needs_heating"
NEEDS_INSULATION_TO_MINIMUM_STANDARDS = "needs_insulation_to_minimum_standards"
class Funding: class Funding:
@ -46,6 +47,12 @@ class Funding:
self.gbis_eligible = False self.gbis_eligible = False
self.gbis_eligibility_caveats = [] self.gbis_eligibility_caveats = []
# Funding calculation variables
self.full_project_abs = None
self.eco4_funding = None
self.partial_project_abs = None
# ----------------------- # -----------------------
# Utility Helpers # Utility Helpers
# ----------------------- # -----------------------
@ -111,8 +118,10 @@ class Funding:
# Private Rented Sector # Private Rented Sector
# ----------------------- # -----------------------
def eco4_prs_eligibility(self, starting_sap: int, ending_sap: int, measure_types: List, mainheat_description: str, def eco4_prs_eligibility(
heating_control_description: str): self, starting_sap: int, ending_sap: int, measure_types: List, has_solar: bool, solar_eligible: bool
):
""" """
ECO4 PRS eligibility: ECO4 PRS eligibility:
- EPC EG - EPC EG
@ -122,27 +131,30 @@ class Funding:
meets_epc = starting_sap <= 54 # EPC EG meets_epc = starting_sap <= 54 # EPC EG
meets_upgrade_target = self._meets_upgrade_target(starting_sap, ending_sap) meets_upgrade_target = self._meets_upgrade_target(starting_sap, ending_sap)
if not meets_epc or not meets_upgrade_target:
self.eco4_eligible = False
self.eco4_eligibility_caveats = []
return
if has_solar and not solar_eligible:
self.eco4_eligible = False
self.eco4_eligibility_caveats.append(EligibilityCaveats.SOLAR_NEEDS_HEATING)
return
has_swi = "internal_wall_insulation" in measure_types or "external_wall_insulation" in measure_types has_swi = "internal_wall_insulation" in measure_types or "external_wall_insulation" in measure_types
has_renewable = "air_source_heat_pump" in measure_types or "ground_source_heat_pump" in measure_types has_renewable = "air_source_heat_pump" in measure_types or "ground_source_heat_pump" in measure_types
has_ftch = "first_time_central_heating" in measure_types has_ftch = "first_time_central_heating" in measure_types
has_dhc = "district_heating_connection" in measure_types has_dhc = "district_heating_connection" in measure_types
has_eligible_electric_heating = (
any(x in mainheat_description for x in [
"air source heat pump", "ground source heat pump", "boiler and radiators, electric"
]) or (
("electric storage heaters" in mainheat_description)
and (heating_control_description.lower() == "controls for high heat retention storage heaters")
)
)
solar_counts_as_renewable = has_eligible_electric_heating and "solar_pv" in measure_types
if meets_upgrade_target and meets_epc and ( if meets_upgrade_target and meets_epc and (
has_swi or has_renewable or has_ftch or has_dhc or solar_counts_as_renewable has_swi or has_renewable or has_ftch or has_dhc or solar_eligible
): ):
self.eco4_eligible = True self.eco4_eligible = True
self.eco4_eligibility_caveats.append(EligibilityCaveats.TENANT_ON_BENEFITS_OR_LOW_INCOME) self.eco4_eligibility_caveats.append(EligibilityCaveats.TENANT_ON_BENEFITS_OR_LOW_INCOME)
return
self.eco4_eligible = False
self.eco4_eligibility_caveats = []
def gbis_prs_eligibility(self, starting_sap: int, council_tax_band: str, measure_types: List): def gbis_prs_eligibility(self, starting_sap: int, council_tax_band: str, measure_types: List):
""" """
@ -152,19 +164,26 @@ class Funding:
""" """
gbis_measures = { gbis_measures = {
"general": [ "general": [
# Cannot do CWI
"internal_wall_insulation", "external_wall_insulation", "internal_wall_insulation", "external_wall_insulation",
"flat_roof_insulation", "suspended_floor_insulation", "flat_roof_insulation", "suspended_floor_insulation",
"room_roof_insulation", "solid_floor_insulation", "park_home_insulation" "room_roof_insulation", "solid_floor_insulation"
], ],
"low_income": [ "low_income": [
"internal_wall_insulation", "external_wall_insulation", "internal_wall_insulation", "external_wall_insulation",
"flat_roof_insulation", "suspended_floor_insulation", "flat_roof_insulation", "suspended_floor_insulation",
"room_roof_insulation", "solid_floor_insulation", "room_roof_insulation", "solid_floor_insulation",
"cavity_wall_insulation", "loft_insulation", "park_home_insulation" "cavity_wall_insulation", "loft_insulation"
] ]
} }
meets_epc = starting_sap <= 69 # EPC DG meets_epc = starting_sap <= 69 # EPC DG
is_single_measure = len(measure_types) == 1
if not is_single_measure or not meets_epc:
self.gbis_eligible = False
self.gbis_eligibility_caveats = []
return
# General route # General route
if meets_epc and council_tax_band in ["A", "B", "C", "D", "E"]: if meets_epc and council_tax_band in ["A", "B", "C", "D", "E"]:
@ -180,14 +199,24 @@ class Funding:
# Social Housing # Social Housing
# ----------------------- # -----------------------
def eco4_sh_eligibility(self, starting_sap: int, ending_sap: int, measure_types: List, has_innovation: bool, def eco4_sh_eligibility(
innovation_measures: List): self,
starting_sap: int,
ending_sap: int,
has_innovation: bool,
has_solar: bool,
solar_eligible: bool
):
""" """
ECO4 Social Housing eligibility. ECO4 Social Housing eligibility.
- EPC EG: eligible, no income check. - EPC EG: eligible, no income check.
- EPC D: innovation measure required. - EPC D: innovation measure required.
If solar PV is the innovation measure, must also install ASHP or HHRSH. If solar PV is the innovation measure, must also install ASHP or HHRSH.
""" """
if has_solar and not solar_eligible:
# The package contins solar PV but it doesn't meet the eligibility requirements
return
meets_epc = starting_sap <= 69 meets_epc = starting_sap <= 69
meets_upgrade_target = self._meets_upgrade_target(starting_sap, ending_sap) meets_upgrade_target = self._meets_upgrade_target(starting_sap, ending_sap)
@ -196,43 +225,53 @@ class Funding:
# EPC D innovation rule # EPC D innovation rule
if 55 <= starting_sap <= 68: # EPC D if 55 <= starting_sap <= 68: # EPC D
# If we don't meet the innovation requirements, we're not eligible
if not has_innovation: if not has_innovation:
self.eco4_eligible = False self.eco4_eligible = False
self.eco4_eligibility_caveats.append(EligibilityCaveats.INNOVATION_REQUIRED) self.eco4_eligibility_caveats.append(EligibilityCaveats.INNOVATION_REQUIRED)
return return
if "solar_pv" in innovation_measures and not any( self.eco4_eligible = True
m in measure_types for m in ["air_source_heat_pump", "high_heat_retention_storage_heater"] self.eco4_eligibility_caveats = []
):
self.eco4_eligible = False
self.eco4_eligibility_caveats.append(EligibilityCaveats.SOLAR_NEEDS_HEATING)
return
self.eco4_eligible = True self.eco4_eligible = True
self.eco4_eligibility_caveats = []
def gbis_sh_eligibility(self, starting_sap: int, measure_types: List, has_innovation: bool, def gbis_sh_eligibility(self, starting_sap: int, measure_types: List, has_innovation: bool):
innovation_measures: List):
""" """
GBIS Social Housing eligibility. GBIS Social Housing eligibility.
- EPC EG: insulation measures OK. - EPC EG: single insulation measure
- EPC D: innovation measure required (same solar PV + heating rule). - EPC D: single insulation, innovation measure
""" """
meets_epc = starting_sap <= 69
if not meets_epc: meets_epc = starting_sap <= 69 # EPC DG
is_single_measure = len(measure_types) == 1
# Check if has a valid measure
insulation_measures = [
'internal_wall_insulation', 'external_wall_insulation', 'cavity_wall_insulation',
'loft_insulation', 'flat_roof_insulation', 'room_roof_insulation',
'suspended_floor_insulation', 'solid_floor_insulation',
]
has_valid_measures = any(m in measure_types for m in insulation_measures)
if not is_single_measure or not meets_epc or not has_valid_measures:
self.gbis_eligible = False
self.gbis_eligibility_caveats = []
return return
if 55 <= starting_sap <= 68: # EPC D if 55 <= starting_sap <= 68: # EPC D
# Since it's single measure if has_innovation is true, the single insulation measure
# must be the innovation measure
if not has_innovation: if not has_innovation:
self.gbis_eligible = False self.gbis_eligible = False
self.gbis_eligibility_caveats.append(EligibilityCaveats.INNOVATION_REQUIRED) self.gbis_eligibility_caveats.append(EligibilityCaveats.INNOVATION_REQUIRED)
return return
if "solar_pv" in innovation_measures and not any( # If we don't have an innovation measure, we're not eligible
m in measure_types for m in ["air_source_heat_pump", "high_heat_retention_storage_heater"] self.gbis_eligible = False
): self.gbis_eligibility_caveats.append(EligibilityCaveats.SOLAR_NEEDS_HEATING)
self.gbis_eligible = False return
self.gbis_eligibility_caveats.append(EligibilityCaveats.SOLAR_NEEDS_HEATING)
return
self.gbis_eligible = True self.gbis_eligible = True
@ -253,27 +292,212 @@ class Funding:
return data["Cost Savings"].values[0] return data["Cost Savings"].values[0]
@staticmethod
def get_starting_ending_uvalues(current_uvalue: float) -> tuple[str, str]:
"""
Returns the closest starting U-value and appropriate ending U-value for solid wall insulation.
- If current_uvalue is closest to 0.45, assume the improvement target is 0.21.
- Otherwise, assume the target is 0.30.
"""
possible_starting_u_values = [2.00, 1.70, 1.00, 0.60, 0.45]
closest_starting = min(possible_starting_u_values, key=lambda x: abs(x - current_uvalue))
ending_uvalue = 0.21 if closest_starting == 0.45 else 0.30
return f"{closest_starting:.2f}", f"{ending_uvalue:.2f}"
def calculate_partial_project_abs(
self,
measure_type: str,
current_wall_uvalue: float = None,
is_partial: bool = False,
existing_li_thickness: float = None,
# is_roof_insulated: bool = False
):
"""
Calculate the partial project ABS score for a single measure.
"""
df = self.partial_project_scores_matrix[
(self.partial_project_scores_matrix["Total Floor Area Band"] == self.floor_area_band) &
(self.partial_project_scores_matrix["Starting Band"] == self.starting_sap_band)
]
if measure_type == "internal_wall_insulation":
if current_wall_uvalue is None:
raise ValueError("current_wall_uvalue is required for IWI")
starting_str, ending_str = self.get_starting_ending_uvalues(current_wall_uvalue)
measure_code = f"IWI_solid_{starting_str}_{ending_str}"
pps = df[df["Measure_Type"] == measure_code]
if pps.shape[0] != 1:
raise ValueError(f"Invalid IWI category: {measure_code}")
return pps.squeeze()["Cost Savings"]
if measure_type == "external_wall_insulation":
if current_wall_uvalue is None:
raise ValueError("current_wall_uvalue is required for EWI")
starting_str, ending_str = self.get_starting_ending_uvalues(current_wall_uvalue)
measure_code = f"EWI_solid_{starting_str}_{ending_str}"
pps = df[df["Measure_Type"] == measure_code]
if pps.shape[0] != 1:
raise ValueError(f"Invalid EWI category: {measure_code}")
return pps.squeeze()["Cost Savings"]
if measure_type == "cavity_wall_insulation":
measure_code = "CWI_partial_fill" if is_partial else "CWI_0.033"
pps = df[df["Measure_Type"] == measure_code]
if pps.shape[0] != 1:
raise ValueError(f"Invalid CWI category: {measure_code}")
return pps.squeeze()["Cost Savings"]
if measure_type == "loft_insulation":
if existing_li_thickness is None:
raise ValueError("existing_li_thickness is required for LI")
measure_code = "LI_lessequal100" if existing_li_thickness <= 100 else "LI_greater100"
pps = df[df["Measure_Type"] == measure_code]
if pps.shape[0] != 1:
raise ValueError(f"Invalid LI category: {measure_code}")
return pps.squeeze()["Cost Savings"]
if measure_type == "flat_roof_insulation":
pps = df[df["Measure_Type"] == "FRI"]
if pps.shape[0] != 1:
raise ValueError("Invalid FRI category")
return pps.squeeze()["Cost Savings"]
if measure_type == "room_roof_insulation":
# Use the more conservative score (unin is usually lower)
# code = "RIRI_res_unin" if not is_roof_insulated else "RIRI_res_in"
code = "RIRI_res_unin"
pps = df[df["Measure_Type"] == code]
if pps.shape[0] != 1:
raise ValueError(f"Invalid RIRI category: {code}")
return pps.squeeze()["Cost Savings"]
if measure_type == "suspended_floor_insulation":
pps = df[df["Measure_Type"] == "UFI"]
if pps.shape[0] != 1:
raise ValueError("Invalid UFI category")
return pps.squeeze()["Cost Savings"]
if measure_type == "solid_floor_insulation":
pps = df[df["Measure_Type"] == "SFI"]
if pps.shape[0] != 1:
raise ValueError("Invalid SFI category")
return pps.squeeze()["Cost Savings"]
raise ValueError(f"Invalid measure type for partial project ABS calculation: {measure_type}")
# ----------------------- # -----------------------
# Main Entry Point # Main Entry Point
# ----------------------- # -----------------------
@staticmethod
def check_solar_eligible_heating_system(mainheat_description, heating_control_description):
"""
Checks if the main heating system is eligible for solar PV funding.
:param mainheat_description: Describes the primary heating system
:param heating_control_description: Heating controls associated to the primary heating system
:return:
"""
return (
any(x in mainheat_description.lower() for x in [
"air source heat pump", "ground source heat pump", "boiler and radiators, electric"
]) or (
("electric storage heaters" in mainheat_description)
and (heating_control_description.lower() == "controls for high heat retention storage heaters")
)
)
def check_solar_eligibility(
self,
measure_types,
mainheat_description,
heating_control_description,
has_wall_insulation_recommendation: bool = False,
has_roof_insulation_recommendation: bool = False,
):
"""
Because of the various pre-requisites for solar, we have a self-contained function to check for
eligibility
Returns a tuple of booleans (has_solar, solar_eligible)
"""
if "solar_pv" not in measure_types:
return False, False
# 1) We check if there is an eligible heating system in place
has_eligibile_heating = self.check_solar_eligible_heating_system(
mainheat_description, heating_control_description
)
if not has_eligibile_heating:
# We check if there is a recommendation for an ASHP or HHRSH
if ("air_source_heat_pump" not in measure_types) and (
"high_heat_retention_storage_heater" not in measure_types):
return True, False
# 2) We check if there is a wall insulation measure for this property. If so, we make sure
# we have a wall insulation recommendation in this package
if has_wall_insulation_recommendation:
# Make sure we have a wall insulation recommendation
if not any(m in measure_types for m in WALL_INSULATION_MEASURES):
return True, False
# 3) We check if there is a roof insulation measure for this property. If so, we make sure
# we have a roof insulation recommendation in this package
if has_roof_insulation_recommendation:
# Make sure we have a roof insulation recommendation
if not any(m in measure_types for m in ROOF_INSULATION_MEASURES):
return True, False
return True, True
def check_funding( def check_funding(
self, measures: List[dict], self,
measures: List[dict],
starting_sap: int, starting_sap: int,
ending_sap: int, ending_sap: int,
floor_area: float, floor_area: float,
mainheat_description: str, mainheat_description: str,
heating_control_description: str, heating_control_description: str,
is_cavity: bool, is_cavity: bool,
council_tax_band: str = None current_wall_uvalue: float,
is_partial: False,
existing_li_thickness: float,
council_tax_band: str = None,
has_wall_insulation_recommendation: bool = False,
has_roof_insulation_recommendation: bool = False,
): ):
""" """
Given a list of measures, check ECO4/GBIS eligibility. Given a list of measures, check ECO4/GBIS eligibility.
Because measures like solar PV are subject to the minimum insulation requirements and we can get
exemptions on floor insulation recommendations, if has_wall_insulation_recommendation or
has_roof_insulation_recommendation are true, we check that the measures package contain a wall or roof
insulation measure otherwise solar PV isn't eligible
""" """
# Normalize measures # Normalize measures
measure_types, has_innovation, innovation_measures = self._split_measures(measures) measure_types, has_innovation, innovation_measures = self._split_measures(measures)
# Determine if we have a solar eligible heating system
has_solar, solar_eligible = self.check_solar_eligibility(
measure_types,
mainheat_description,
heating_control_description,
has_wall_insulation_recommendation,
has_roof_insulation_recommendation,
)
# Track EPC bands and floor area # Track EPC bands and floor area
self.starting_sap_band = self.get_sap_band(starting_sap) self.starting_sap_band = self.get_sap_band(starting_sap)
self.ending_sap_band = self.get_sap_band(ending_sap) self.ending_sap_band = self.get_sap_band(ending_sap)
@ -281,26 +505,36 @@ class Funding:
if self.tenure == "Private": if self.tenure == "Private":
# ECO4 PRS # ECO4 PRS
self.eco4_prs_eligibility(starting_sap, ending_sap, measure_types, mainheat_description, self.eco4_prs_eligibility(starting_sap, ending_sap, measure_types, has_solar, solar_eligible)
heating_control_description)
# GBIS PRS # GBIS PRS
self.gbis_prs_eligibility(starting_sap, council_tax_band or "", measure_types) self.gbis_prs_eligibility(starting_sap, council_tax_band or "", measure_types)
if self.eco4_eligible: if self.eco4_eligible:
eco4_abs = self.calculate_full_project_abs() self.full_project_abs = self.calculate_full_project_abs()
eco4_funding = eco4_abs * (self.private_cavity_abs_rate if is_cavity else self.private_solid_abs_rate) self.eco4_funding = self.full_project_abs * (
return {"eco4_funding": eco4_funding} self.private_cavity_abs_rate if is_cavity else self.private_solid_abs_rate)
elif self.tenure == "Social": elif self.tenure == "Social":
# ECO4 Social # ECO4 Social
self.eco4_sh_eligibility(starting_sap, ending_sap, measure_types, has_innovation, innovation_measures) self.eco4_sh_eligibility(starting_sap, ending_sap, has_innovation, has_solar, solar_eligible)
# GBIS Social # GBIS Social
self.gbis_sh_eligibility(starting_sap, measure_types, has_innovation, innovation_measures) self.gbis_sh_eligibility(starting_sap, measure_types, has_innovation)
if self.eco4_eligible: if self.eco4_eligible:
eco4_abs = self.calculate_full_project_abs() # Calculate the full project ABS for ECO4
eco4_funding = eco4_abs * (self.social_cavity_abs_rate if is_cavity else self.social_solid_abs_rate) self.full_project_abs = self.calculate_full_project_abs()
return {"eco4_funding": eco4_funding} self.eco4_funding = self.full_project_abs * (
self.social_cavity_abs_rate if is_cavity else self.social_solid_abs_rate
)
if self.gbis_eligible:
# Calculate the partial project score - this is dependent on the measure
self.partial_project_abs = self.calculate_partial_project_abs(
measure_types[0], current_wall_uvalue, is_partial, existing_li_thickness,
)
else: else:
raise NotImplementedError("Only 'Private' and 'Social' tenures are supported.") raise NotImplementedError("Only 'Private' and 'Social' tenures are supported.")

View file

@ -8,9 +8,10 @@ TYPICAL_MEASURE_TYPES = [
"secondary_heating", "solar_pv" "secondary_heating", "solar_pv"
] ]
SPECIFIC_MEASURES = [ WALL_INSULATION_MEASURES = ["internal_wall_insulation", "external_wall_insulation", "cavity_wall_insulation"]
"internal_wall_insulation", "external_wall_insulation", "cavity_wall_insulation", ROOF_INSULATION_MEASURES = ["loft_insulation", "flat_roof_insulation", "room_roof_insulation"]
"loft_insulation", "flat_roof_insulation", "room_roof_insulation",
SPECIFIC_MEASURES = WALL_INSULATION_MEASURES + ROOF_INSULATION_MEASURES + [
"suspended_floor_insulation", "solid_floor_insulation", "suspended_floor_insulation", "solid_floor_insulation",
"boiler_upgrade", "high_heat_retention_storage_heater", "air_source_heat_pump", "boiler_upgrade", "high_heat_retention_storage_heater", "air_source_heat_pump",
"secondary_heating", "solar_pv", "double_glazing", "secondary_glazing", "secondary_heating", "solar_pv", "double_glazing", "secondary_glazing",

View file

@ -30,7 +30,11 @@ def mock_project_scores_matrix():
@pytest.fixture @pytest.fixture
def mock_partial_scores_matrix(): def mock_partial_scores_matrix():
return pd.DataFrame([{"dummy": "data"}]) # not used for eligibility tests yet df = pd.read_csv("recommendations/tests/test_data/ECO4_Partial_Project_Scores_Matrix_v6.csv")
df.columns = ['Measure category', 'Measure_Type', 'Pre_Main_Heating_Source',
'Post_Main_Heating_Source', 'Total Floor Area Band', 'Starting Band',
'Average Treatable Factor', 'Cost Savings', 'SAP Savings']
return df
@pytest.fixture @pytest.fixture
@ -69,7 +73,10 @@ def test_eco4_prs_eligible_with_swi(mock_project_scores_matrix, mock_partial_sco
mainheat_description="Boiler and radiators, mains gas", mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs", heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True, is_cavity=True,
council_tax_band="B" council_tax_band="B",
is_partial=False,
existing_li_thickness=0,
current_wall_uvalue=2
) )
assert funding.eco4_eligible assert funding.eco4_eligible
@ -98,7 +105,10 @@ def test_eco4_prs_not_eligible_high_epc(mock_project_scores_matrix, mock_partial
mainheat_description="Boiler and radiators, mains gas", mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs", heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True, is_cavity=True,
council_tax_band="B" council_tax_band="B",
is_partial=False,
existing_li_thickness=0,
current_wall_uvalue=2,
) )
assert not funding.eco4_eligible assert not funding.eco4_eligible
@ -126,7 +136,10 @@ def test_gbis_prs_general_eligibility(mock_project_scores_matrix, mock_partial_s
mainheat_description="Boiler and radiators, mains gas", mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs", heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True, is_cavity=True,
council_tax_band="A" council_tax_band="A",
is_partial=False,
existing_li_thickness=0,
current_wall_uvalue=2,
) )
assert funding.gbis_eligible assert funding.gbis_eligible
@ -154,7 +167,10 @@ def test_gbis_prs_low_income_caveat(mock_project_scores_matrix, mock_partial_sco
mainheat_description="Boiler and radiators, mains gas", mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs", heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True, is_cavity=True,
council_tax_band="B" council_tax_band="B",
is_partial=False,
existing_li_thickness=0,
current_wall_uvalue=2,
) )
assert funding.gbis_eligible assert funding.gbis_eligible
@ -186,7 +202,10 @@ def test_eco4_sh_epc_e_eligible(mock_project_scores_matrix, mock_partial_scores_
floor_area=80, floor_area=80,
mainheat_description="Boiler and radiators, mains gas", mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs", heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True is_cavity=True,
current_wall_uvalue=2,
is_partial=False,
existing_li_thickness=0,
) )
assert funding.eco4_eligible assert funding.eco4_eligible
@ -213,12 +232,169 @@ def test_eco4_sh_epc_d_requires_innovation(mock_project_scores_matrix, mock_part
floor_area=80, floor_area=80,
mainheat_description="Boiler and radiators, mains gas", mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs", heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True is_cavity=True,
current_wall_uvalue=2,
is_partial=False,
existing_li_thickness=0,
) )
assert not funding.eco4_eligible assert not funding.eco4_eligible
assert EligibilityCaveats.INNOVATION_REQUIRED in funding.eco4_eligibility_caveats assert EligibilityCaveats.INNOVATION_REQUIRED in funding.eco4_eligibility_caveats
# Test with an innovation measure
funding2 = Funding(
project_scores_matrix=mock_project_scores_matrix,
partial_project_scores_matrix=mock_partial_scores_matrix,
whlg_eligible_postcodes=mock_whlg_postcodes,
social_cavity_abs_rate=13.5,
social_solid_abs_rate=17,
private_cavity_abs_rate=13.5,
private_solid_abs_rate=17,
tenure="Social",
)
measures2 = [{"type": "internal_wall_insulation", "is_innovation": True}]
funding2.check_funding(
measures=measures2,
starting_sap=60, # EPC D
ending_sap=69,
floor_area=80,
mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True,
current_wall_uvalue=2,
is_partial=False,
existing_li_thickness=0,
)
assert funding2.eco4_eligible
assert not funding2.eco4_eligibility_caveats
# Test with innovation solar. If the measure is solar, we need to have an eligible heating system.
# If we don't have an eligible heating system in place, we need to have one as part of the measure
# package
# THIS SHOULD NOT BE ELIGIBLE
funding3 = Funding(
project_scores_matrix=mock_project_scores_matrix,
partial_project_scores_matrix=mock_partial_scores_matrix,
whlg_eligible_postcodes=mock_whlg_postcodes,
social_cavity_abs_rate=13.5,
social_solid_abs_rate=17,
private_cavity_abs_rate=13.5,
private_solid_abs_rate=17,
tenure="Social",
)
measures3 = [{"type": "solar_pv", "is_innovation": True}]
funding3.check_funding(
measures=measures3,
starting_sap=60, # EPC D
ending_sap=69,
floor_area=80,
mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True,
current_wall_uvalue=2,
is_partial=False,
existing_li_thickness=0,
)
assert not funding3.eco4_eligible
assert EligibilityCaveats.SOLAR_NEEDS_HEATING in funding3.eco4_eligibility_caveats
# Test with innovation solar and ASHP. This should be eligible
funding4 = Funding(
project_scores_matrix=mock_project_scores_matrix,
partial_project_scores_matrix=mock_partial_scores_matrix,
whlg_eligible_postcodes=mock_whlg_postcodes,
social_cavity_abs_rate=13.5,
social_solid_abs_rate=17,
private_cavity_abs_rate=13.5,
private_solid_abs_rate=17,
tenure="Social",
)
measures4 = [{"type": "solar_pv", "is_innovation": True}]
funding4.check_funding(
measures=measures4,
starting_sap=60, # EPC D
ending_sap=69,
floor_area=80,
mainheat_description="Air source heat pump, radiators",
heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True,
current_wall_uvalue=2,
is_partial=False,
existing_li_thickness=0,
)
assert funding4.eco4_eligible
assert not funding4.eco4_eligibility_caveats
# Test with innovation solar, a non-eligible heating system but a heating upgrade
funding5 = Funding(
project_scores_matrix=mock_project_scores_matrix,
partial_project_scores_matrix=mock_partial_scores_matrix,
whlg_eligible_postcodes=mock_whlg_postcodes,
social_cavity_abs_rate=13.5,
social_solid_abs_rate=17,
private_cavity_abs_rate=13.5,
private_solid_abs_rate=17,
tenure="Social",
)
measures5 = [
{"type": "solar_pv", "is_innovation": True},
{"type": "high_heat_retention_storage_heater", "is_innovation": False}
]
funding5.check_funding(
measures=measures5,
starting_sap=60, # EPC D
ending_sap=69,
floor_area=80,
mainheat_description="Electric storage heaters",
heating_control_description="Manual charge control",
is_cavity=True,
current_wall_uvalue=2,
is_partial=False,
existing_li_thickness=0,
)
assert funding5.eco4_eligible
assert not funding5.eco4_eligibility_caveats
# Test with innovation solar, an eligible heating system but a package that excludes the required
# fabric upgrades
funding6 = Funding(
project_scores_matrix=mock_project_scores_matrix,
partial_project_scores_matrix=mock_partial_scores_matrix,
whlg_eligible_postcodes=mock_whlg_postcodes,
social_cavity_abs_rate=13.5,
social_solid_abs_rate=17,
private_cavity_abs_rate=13.5,
private_solid_abs_rate=17,
tenure="Social",
)
measures6 = [
{"type": "solar_pv", "is_innovation": True},
]
funding6.check_funding(
measures=measures6,
starting_sap=60, # EPC D
ending_sap=69,
floor_area=80,
mainheat_description="Electric storage heaters",
heating_control_description="controls for high heat retention storage heaters",
is_cavity=True,
has_wall_insulation_recommendation=True,
has_roof_insulation_recommendation=False,
current_wall_uvalue=2,
is_partial=False,
existing_li_thickness=0,
)
assert not funding6.eco4_eligible
assert not funding6.eco4_eligibility_caveats
def test_eco4_sh_solar_pv_requires_heating(mock_project_scores_matrix, mock_partial_scores_matrix, mock_whlg_postcodes): def test_eco4_sh_solar_pv_requires_heating(mock_project_scores_matrix, mock_partial_scores_matrix, mock_whlg_postcodes):
"""Solar PV as innovation measure requires ASHP or HHRSH.""" """Solar PV as innovation measure requires ASHP or HHRSH."""
@ -241,7 +417,10 @@ def test_eco4_sh_solar_pv_requires_heating(mock_project_scores_matrix, mock_part
floor_area=80, floor_area=80,
mainheat_description="Boiler and radiators, mains gas", mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs", heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True is_cavity=True,
current_wall_uvalue=2,
is_partial=False,
existing_li_thickness=0,
) )
assert not funding.eco4_eligible assert not funding.eco4_eligible
@ -273,7 +452,10 @@ def test_eco4_sh_solar_pv_with_heating_is_ok(mock_project_scores_matrix, mock_pa
floor_area=80, floor_area=80,
mainheat_description="Boiler and radiators, mains gas", mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs", heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True is_cavity=True,
current_wall_uvalue=2,
is_partial=False,
existing_li_thickness=0,
) )
assert funding.eco4_eligible assert funding.eco4_eligible
@ -305,7 +487,10 @@ def test_eco4_upgrade_requirement_e_to_c_pass(mock_project_scores_matrix, mock_p
mainheat_description="Boiler and radiators, mains gas", mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs", heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True, is_cavity=True,
council_tax_band="B" council_tax_band="B",
current_wall_uvalue=2,
is_partial=False,
existing_li_thickness=0,
) )
assert funding.eco4_eligible assert funding.eco4_eligible
@ -336,7 +521,10 @@ def test_eco4_upgrade_requirement_e_to_d_fail(mock_project_scores_matrix, mock_p
mainheat_description="Boiler and radiators, mains gas", mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs", heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True, is_cavity=True,
council_tax_band="B" council_tax_band="B",
current_wall_uvalue=2,
is_partial=False,
existing_li_thickness=0,
) )
assert not funding.eco4_eligible assert not funding.eco4_eligible
@ -367,7 +555,10 @@ def test_eco4_upgrade_requirement_f_to_d_pass(mock_project_scores_matrix, mock_p
mainheat_description="Boiler and radiators, mains gas", mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs", heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True, is_cavity=True,
council_tax_band="B" council_tax_band="B",
current_wall_uvalue=2,
is_partial=False,
existing_li_thickness=0,
) )
assert funding.eco4_eligible assert funding.eco4_eligible
@ -398,7 +589,10 @@ def test_eco4_upgrade_requirement_f_to_e_fail(mock_project_scores_matrix, mock_p
mainheat_description="Boiler and radiators, mains gas", mainheat_description="Boiler and radiators, mains gas",
heating_control_description="Programmer, room thermostat and TRVs", heating_control_description="Programmer, room thermostat and TRVs",
is_cavity=True, is_cavity=True,
council_tax_band="B" council_tax_band="B",
current_wall_uvalue=2,
is_partial=False,
existing_li_thickness=0,
) )
assert not funding.eco4_eligible assert not funding.eco4_eligible

File diff suppressed because it is too large Load diff