junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-08 11:17:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
Model/backend/Funding.py
Khalim Conn-Kowlessar 980f439f49 debugging calico epc search to handle the strict blocks
2025-06-19 22:48:25 +01:00

600 lines
22 KiB
Python
Raw Blame History

import pandas as pd
import numpy as np
from typing import List
from backend.app.plan.schemas import HousingType
class FundingOld:
"""
Given a property, this class identifies if the home is possibly eligible for funding under
the various funding schemes. It will also calculate the expected amount of funding available
and flag any tenant specific requirements that need to be considered to the funding to be attained
"""
SCHEMES = ["eco4", "gbis", "whlg"]
ECO_SAP_SCORE_THREHOLDS = [
{'Band': 'High_A', 'From': 96.0, 'Up to': 100.0, 'Mid-point': 98.0},
{'Band': 'Low_A', 'From': 92.0, 'Up to': 96.0, 'Mid-point': 94.0},
{'Band': 'High_B', 'From': 86.0, 'Up to': 91.0, 'Mid-point': 88.5},
{'Band': 'Low_B', 'From': 81.0, 'Up to': 86.0, 'Mid-point': 83.5},
{'Band': 'High_C', 'From': 74.5, 'Up to': 80.0, 'Mid-point': 77.25},
{'Band': 'Low_C', 'From': 69.0, 'Up to': 74.5, 'Mid-point': 71.75},
{'Band': 'High_D', 'From': 61.5, 'Up to': 68.0, 'Mid-point': 64.75},
{'Band': 'Low_D', 'From': 55.0, 'Up to': 61.5, 'Mid-point': 58.25},
{'Band': 'High_E', 'From': 46.5, 'Up to': 54.0, 'Mid-point': 50.25},
{'Band': 'Low_E', 'From': 39.0, 'Up to': 46.5, 'Mid-point': 42.75},
{'Band': 'High_F', 'From': 29.5, 'Up to': 38.0, 'Mid-point': 33.75},
{'Band': 'Low_F', 'From': 21.0, 'Up to': 29.5, 'Mid-point': 25.25},
{'Band': 'High_G', 'From': 10.5, 'Up to': 20.0, 'Mid-point': 15.25},
{'Band': 'Low_G', 'From': 1.0, 'Up to': 10.5, 'Mid-point': 5.75}
]
def __init__(
self,
tenure: HousingType,
starting_epc,
starting_sap,
postcode,
floor_area,
council_tax_band,
property_recommendations,
project_scores_matrix,
whlg_eligible_postcodes,
gbis_abs_rate: int,
eco4_abs_rate: int,
):
"""
Use Pydantic to validate the parameter types
:param tenure: Indicates if the property is a social or private home
:param starting_epc: The current EPC rating of the property
:param starting_sap: The current SAP score for the property
:param floor_area: The total floor area of the property
:param council_tax_band: The council tax band of the property
:param property_recommendations: The recommendations for the property
:param project_scores_matrix: The matrix of project scores for ECO4
:param whlg_eligible_postcodes: The postcodes eligible for WHLG
:param gbis_abs_rate: The assumed £/abs achieved by the installer for GBIS
:param eco4_abs_rate: The assumed £/abs achieved by the installer for ECO4
"""
# TODO: Things we need to include:
# 1) Amount of funding
# 2) Fundable measures, as a subset of measures may be fundable, not all
self.tenure = tenure
self.starting_epc = starting_epc
self.starting_sap = starting_sap
self.postcode = postcode
self.starting_eco_band = self.sap_to_eco_band(self.starting_sap)
self.floor_area_segment = self.classify_floor_area(floor_area)
self.gbis_abs_rate = gbis_abs_rate
self.eco4_abs_rate = eco4_abs_rate
self.council_tax_band = council_tax_band
self.recommendations = property_recommendations
self.measure_types = list({r["measure_type"] for r in property_recommendations if r["default"]})
# Load in the eco4 project scores matrix
# Filter the matrix on scores relevant to this property
self.project_scores_matrix = project_scores_matrix[
(project_scores_matrix["Floor Area Segment"] == self.floor_area_segment) &
(project_scores_matrix["Starting Band"] == self.starting_eco_band)
]
# The postcode column is already lower case
self.whlg_eligible_postcodes = whlg_eligible_postcodes[
whlg_eligible_postcodes["Postcode"] == self.postcode.lower()
]
# Store the final outputs
self.gbis_eligibiltiy = {}
self.eco4_eligibility = {}
self.whlg_eligibility = {}
def output(
self,
scheme: str,
eligible: bool,
types: List[str],
measure_types: List[str],
project_score: float,
estimated_funding: float,
notify_tenant_benefits_requirements: bool,
notify_council_tax_band_requirements: bool,
notify_tenant_low_income_requirements: bool,
innovation_required: bool,
):
""""
"""
if scheme not in self.SCHEMES:
raise ValueError("Scheme not recognised")
return {
"scheme": scheme,
"eligible": eligible,
"type": types,
"measure_types": measure_types,
"project_score": project_score,
"estimated_funding": estimated_funding,
"requires_benefits": notify_tenant_benefits_requirements,
"requires_council_tax_band": notify_council_tax_band_requirements,
"requires_low_income": notify_tenant_low_income_requirements,
"innovation_required": innovation_required,
}
@staticmethod
def classify_floor_area(floor_area):
if floor_area <= 72:
return "0-72"
if floor_area <= 97:
return "73-97"
if floor_area <= 199:
return "98-199"
return "200"
def eco4(self):
"""
Checks if a property is eligible for ECO4
:return:
"""
pass
def find_gbis_measures(self, measures):
"""
The best measure is one that:
1) Creates some SAP movement, therefore enables eligiblity
2) Generates the most funding
3) Has a reasonable ROI
:return:
"""
measure_table = pd.DataFrame([
m for m in self.recommendations if
(m["type"] in measures) or (m["measure_type"] in measures) and m["default"]
])
measure_table["post_install_sap"] = measure_table["sap_points"] + self.starting_sap
# We classify the movement
measure_table["Finishing Band"] = np.floor(measure_table["post_install_sap"]).apply(
lambda points: self.sap_to_eco_band(points)
)
# Remove any measures that generate zero SAP movement
measure_table = measure_table[measure_table["Finishing Band"] != self.starting_eco_band]
if measure_table.empty:
raise NotImplementedError("No measures available, handle me!")
# We merge on the project matrix, on post install band
measure_table = measure_table.merge(
self.project_scores_matrix, how="left", on="Finishing Band"
)
# Cost Savings is the abs
measure_table["estimated_funding"] = measure_table["Cost Savings"] * self.gbis_abs_rate
# We cap any estimated funding at the install cost
measure_table["estimated_funding"] = np.where(
measure_table["estimated_funding"] >= measure_table["total"],
measure_table["total"],
measure_table["estimated_funding"]
)
# Sort by the measure that will cost the client the least, per sap point
measure_table["cost_minus_funding"] = measure_table["total"] - measure_table["estimated_funding"]
measure_table["cost_minus_funding_per_sap"] = measure_table["cost_minus_funding"] / measure_table["sap_points"]
measure_table = measure_table.sort_values(["cost_minus_funding_per_sap", "total"], ascending=[True, False])
return measure_table[
["type", "measure_type", "Cost Savings", "estimated_funding"]
].rename(columns={"Cost Savings": "project_score"}).to_dict("records")
def sap_to_eco_band(self, sap_points):
"""
Giuven a sap point score, this function will classify the points into the SAP half-band
:param sap_points:
:return:
"""
if sap_points > 100:
return "High_A"
classification = [
x for x in self.ECO_SAP_SCORE_THREHOLDS if (x["From"] <= sap_points) and (sap_points <= x["Up to"])
]
if len(classification) != 1:
raise Exception("We should have a single classifcation for SAP points to half band")
return classification[0]['Band']
def gbis_prs(self):
"""
Checks if a private rental is eligible for GBIS. There are the following possible options
1) General Eligibilty, contigent on EPC D-G and council tax band A-D. Excludes CWI, LI and heating
controls
2) Low income group - contigent on EPC D-G and tenant must receive benefits. Excludes heating controls
3) GBIS Flex route 1, 3 - Great British Insulation Scheme Routes 1 and 3 are for pre-installation
SAP bands D-G for owner-occupied households, D-E for private rented sector households
(Including F & G if exempt from MEES). If houseold is low income. Excludes heating controls
4) GBIS Flex route 2 - EPC E - G and low income household. Excludes heating controls
Eligible measures:
• Solid wall
• pitched roof
• flat roof
• under floor
• solid floor park home and
• room in-roof insulation
:return:
"""
valid_measures = [
"internal_wall_insulation",
"external_wall_insulation",
"flat_roof_insulation",
"suspended_floor_insulation",
"room_roof_insulation",
# Not available for every eligiblity type
"cavity_wall_insulation",
"loft_insulation",
]
# General Eligibility
if (
(self.starting_epc in ["G", "D", "E", "F"]) and
any(
[measure in valid_measures for measure in self.measure_types
if measure not in ["cavity_wall_insulation", "loft_insulation"]]
) and
(self.council_tax_band in [None, "A", "B", "C", "D"])
):
# This function pulls out the various measures that can provide funding under GBIS
recommended_measures = self.find_gbis_measures(
measures=[m for m in valid_measures if m not in ["cavity_wall_insulation", "loft_insulation"]]
)
# If the council tax band is missing, we nofify the customer that this is a requirement that
# should be checked
return [
self.output(
scheme="gbis",
eligible=True,
types=[m["type"]], # This is single measure so we only have one type
measure_types=[m["measure_type"]],
project_score=m["project_score"],
estimated_funding=m["estimated_funding"],
notify_tenant_benefits_requirements=False,
notify_council_tax_band_requirements=self.council_tax_band is None,
notify_tenant_low_income_requirements=False,
innovation_required=False
) for m in recommended_measures
]
# Low income/flex
if (
(self.starting_sap in ["G", "D", "E", "F"]) and
any([measure in valid_measures for measure in self.measure_types])
):
# Find the best measure, and can also include CWI/LI but requires the tenant to be
# low inome or on benefits
# We find the best measure for GBIS
recommended_measures = self.find_gbis_measures(measures=valid_measures)
return [
self.output(
scheme="gbis",
eligible=True,
types=[m["type"]], # This is single measure so we only have one type
measure_types=[m["measure_type"]],
project_score=m["project_score"],
estimated_funding=m["estimated_funding"],
notify_tenant_benefits_requirements=True,
notify_council_tax_band_requirements=False,
notify_tenant_low_income_requirements=True,
innovation_required=False
) for m in recommended_measures
]
# Otherwise, no funding availability
return []
def gbis_social(self):
"""
Because this is social housing, we have two typical means for eligibility
1) EPC D, where an innovation measure is required
2) EPC G-E, where an innovation measure isn't required
:return:
"""
valid_measures = [
"internal_wall_insulation",
"external_wall_insulation",
"flat_roof_insulation",
"suspended_floor_insulation",
"room_roof_insulation",
# Not available for every eligiblity type
"cavity_wall_insulation",
"loft_insulation",
"heating_control"
]
recommended_measures = self.find_gbis_measures(
measures=valid_measures
)
# All measures are available
if self.starting_sap == "D":
return [
self.output(
scheme="gbis",
eligible=True,
types=[m["type"]], # This is single measure so we only have one type
measure_types=[m["measure_type"]],
project_score=m["project_score"],
estimated_funding=m["estimated_funding"],
notify_tenant_benefits_requirements=False,
notify_council_tax_band_requirements=False,
notify_tenant_low_income_requirements=False,
innovation_required=True
) for m in recommended_measures
]
if self.starting_sap in ["G", "F", "E"]:
return [
self.output(
scheme="gbis",
eligible=True,
types=[m["type"]], # This is single measure so we only have one type
measure_types=[m["measure_type"]],
project_score=m["project_score"],
estimated_funding=m["estimated_funding"],
notify_tenant_benefits_requirements=False,
notify_council_tax_band_requirements=False,
notify_tenant_low_income_requirements=False,
innovation_required=False
) for m in recommended_measures
]
return []
def gbis(self):
"""
Check if a property is eligible for GBIS
:return:
"""
if self.tenure == "Private":
self.gbis_eligibiltiy = self.gbis_prs()
return
if self.tenure == "Social":
self.gbis_eligibiltiy = self.gbis_social()
raise NotImplementedError("Implement social/oo")
def whlg(self):
if self.tenure == "Social":
# We can't do anything for social housing
self.whlg_eligibility = []
return
if not self.whlg_eligible_postcodes.empty:
raise Exception("Implement me")
# self.whlg_eligibility = [
# self.output(
# scheme,
# eligible,
# types,
# measure_types,
# project_score: float,
# estimated_funding: float,
# notify_tenant_benefits_requirements: bool,
# notify_council_tax_band_requirements: bool,
# notify_tenant_low_income_requirements: bool,
# innovation_required: bool,
# )
# ]
def eco4(self):
if self.tenure == "Private":
self.eco4_eligibiltiy = self.eco4_prs()
return
def check_eligibiltiy(self):
"""
This function instigates the checking process
:return:
"""
self.gbis()
# self.eco4()
self.whlg()
class Funding:
"""
New class to handle funding calculation
"""
def __init__(
self,
tenure: HousingType,
social_cavity_abs_rate: float,
social_solid_abs_rate: float,
private_cavity_abs_rate: float,
private_solid_abs_rate: float,
project_scores_matrix,
whlg_eligible_postcodes
):
self.tenure = tenure
self.social_cavity_abs_rate = social_cavity_abs_rate
self.social_solid_abs_rate = social_solid_abs_rate
self.private_cavity_abs_rate = private_cavity_abs_rate
self.private_solid_abs_rate = private_solid_abs_rate
self.starting_sap_band = None
self.ending_sap_band = None
self.floor_area_band = None
self.project_scores_matrix = project_scores_matrix
self.whlg_eligible_postcodes = whlg_eligible_postcodes
@staticmethod
def get_sap_band(sap_score_number):
bands = [
("High_A", 96, float("inf")),
("Low_A", 92, 96),
("High_B", 86, 92),
("Low_B", 81, 86),
("High_C", 74.5, 81),
("Low_C", 69, 74.5),
("High_D", 61.5, 69),
("Low_D", 55, 61.5),
("High_E", 46.5, 55),
("Low_E", 39, 46.5),
("High_F", 29.5, 39),
("Low_F", 21, 29.5),
("High_G", 10.5, 21),
("Low_G", 1, 10.5),
]
for band, lower, upper in bands:
if lower <= sap_score_number < upper:
return band
return None
@staticmethod
def get_floor_area_band(floor_area):
if floor_area <= 72:
return "0-72"
if floor_area <= 97:
return "73-97"
if floor_area <= 199:
return "98-199"
return "200"
@staticmethod
def eco4_prs_eligibility(
starting_sap: int, measures: List, mainheat_description: str, heating_control_description: str
):
"""
Handles the eligibility criteria for private rental properties under eco
:return:
"""
# Help to heat group
# 1) EPC E - G
# 2) Must receive one of SWI, FTCH, renewable heating or DHC
# 3) Tenant must be on benefits
# We don't consider the tenant being on benefits - we just notify the end user that this is a requirement
meets_epc = starting_sap <= 54
has_solid_wall = "internal_wall_insulation" in measures or "external_wall_insulation" in measures
# We check if the property has a heating system that means solar pv counts as a renewable heating system
has_eligible_electric_heating = any(x in mainheat_description for x in [
"air source heat pump", "ground source heat pump", "boiler and radiators, electric"
]) | (("electric storage heaters" in mainheat_description) and
(heating_control_description.lower() == "controls for high heat retention storage heaters")
)
# Counts as renewable heating
solar_renweable_heating = has_eligible_electric_heating & ("solar_pv" in measures)
# Is a renewable heating
ashp = "air_source_heat_pump" in measures
if meets_epc & (solar_renweable_heating or ashp or has_solid_wall):
return True
return False
def calculate_full_project_abs(self):
# Filter the project scores matrix
data = self.project_scores_matrix[
(self.project_scores_matrix["Floor Area Segment"] == self.floor_area_band) &
(self.project_scores_matrix["Starting Band"] == self.starting_sap_band) &
(self.project_scores_matrix["Finishing Band"] == self.ending_sap_band)
]
if data.emtpy:
raise ValueError("Missing abs rate, check the project scores matrix")
return data["Cost Savings"].values[0]
def check_funding(
self, measures: List,
starting_sap: int,
ending_sap: int,
floor_area: float,
mainheat_description: str,
heating_control_description: str,
is_cavity: bool
):
"""
Given a list of measures, this function will check if the package of measures is fundable
:param measures:
:param starting_sap:
:param ending_sap:
:param floor_area:
:param mainheat_description:
:param heating_control_description:
:param is_cavity: Indicates if the property has cavity wall insulation
:return:
"""
# If it's an E or D, should get to an EPC C
if starting_sap >= 55 and ending_sap < 69:
raise NotImplementedError("This property doesn't have sufficient SAP movement")
if starting_sap <= 38 & ending_sap <= 55:
# F or G should get to D
raise NotImplementedError("Implement F or G to D eligibility")
self.starting_sap_band = self.get_sap_band(starting_sap)
self.ending_sap_band = self.get_sap_band(ending_sap)
self.floor_area_band = self.get_floor_area_band(floor_area)
########################
# Private
########################
# 1) ECO4
# 2) GBIS
if self.tenure == "Private":
is_eco4_eligible = self.eco4_prs_eligibility(
starting_sap=starting_sap,
measures=measures,
mainheat_description=mainheat_description,
heating_control_description=heating_control_description
)
# Need to implement
# 1) Package has to include an insulation measure
# 2) We should use the funding for the measure that has the largest partial project score
is_gbis_eligible = ()
if not is_eco4_eligible:
return
eco4_abs = self.calculate_full_project_abs()
# We estimate rates now
eco4_funding = (
eco4_abs * self.private_cavity_abs_rate if is_cavity else eco4_abs & self.private_solid_abs_rate
)
########################
# Social
########################
# 1) ECO4
# 2) GBIS
if self.tenure == "Social":
pass
raise NotImplementedError("Only implemented for Private or Social housing")
Reference in a new issue View git blame Copy permalink