mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
286 lines
11 KiB
Python
286 lines
11 KiB
Python
from enum import Enum
|
||
import pandas as pd
|
||
import numpy as np
|
||
from typing import List
|
||
|
||
from backend.app.plan.schemas import HousingType
|
||
|
||
|
||
class EligibilityCaveats(Enum):
|
||
TENANT_ON_BENEFITS_OR_LOW_INCOME = "tenant_on_benefits_or_low_income"
|
||
INNOVATION_REQUIRED = "innovation_required"
|
||
SOLAR_NEEDS_HEATING = "solar_needs_heating"
|
||
|
||
|
||
class Funding:
|
||
"""
|
||
Handles eligibility and funding calculations for ECO4 & GBIS (PRS + Social Housing).
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
tenure: str, # 'Private' or 'Social'
|
||
social_cavity_abs_rate: float,
|
||
social_solid_abs_rate: float,
|
||
private_cavity_abs_rate: float,
|
||
private_solid_abs_rate: float,
|
||
project_scores_matrix,
|
||
partial_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.partial_project_scores_matrix = partial_project_scores_matrix
|
||
self.whlg_eligible_postcodes = whlg_eligible_postcodes
|
||
|
||
self.eco4_eligible = False
|
||
self.eco4_eligibility_caveats = []
|
||
self.gbis_eligible = False
|
||
self.gbis_eligibility_caveats = []
|
||
|
||
# -----------------------
|
||
# Utility Helpers
|
||
# -----------------------
|
||
|
||
@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"
|
||
|
||
def _split_measures(self, measures: List[dict]):
|
||
"""
|
||
Extracts measure types and flags innovation.
|
||
measures: list of dicts like {"type": "solar_pv", "is_innovation": True}
|
||
"""
|
||
measure_types = [m["type"] for m in measures]
|
||
has_innovation = any(m.get("is_innovation", False) for m in measures)
|
||
innovation_measures = [m["type"] for m in measures if m.get("is_innovation", False)]
|
||
return measure_types, has_innovation, innovation_measures
|
||
|
||
# -----------------------
|
||
# Private Rented Sector
|
||
# -----------------------
|
||
|
||
def eco4_prs_eligibility(self, starting_sap: int, measure_types: List, mainheat_description: str,
|
||
heating_control_description: str):
|
||
"""
|
||
ECO4 PRS eligibility:
|
||
- EPC E–G
|
||
- Must include SWI, FTCH, renewable heating, or DHC
|
||
- Tenant must be on benefits (flagged)
|
||
"""
|
||
meets_epc = starting_sap <= 54 # EPC E–G
|
||
|
||
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_ftch = "first_time_central_heating" 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_epc and (has_swi or has_renewable or has_ftch or has_dhc or solar_counts_as_renewable):
|
||
self.eco4_eligible = True
|
||
self.eco4_eligibility_caveats.append(EligibilityCaveats.TENANT_ON_BENEFITS_OR_LOW_INCOME)
|
||
|
||
def gbis_prs_eligibility(self, starting_sap: int, council_tax_band: str, measure_types: List):
|
||
"""
|
||
GBIS PRS eligibility:
|
||
- General route: Council Tax Band & EPC D–G
|
||
- Low-income route: tenant on benefits (flagged)
|
||
"""
|
||
gbis_measures = {
|
||
"general": [
|
||
"internal_wall_insulation", "external_wall_insulation",
|
||
"flat_roof_insulation", "suspended_floor_insulation",
|
||
"room_roof_insulation", "solid_floor_insulation", "park_home_insulation"
|
||
],
|
||
"low_income": [
|
||
"internal_wall_insulation", "external_wall_insulation",
|
||
"flat_roof_insulation", "suspended_floor_insulation",
|
||
"room_roof_insulation", "solid_floor_insulation",
|
||
"cavity_wall_insulation", "loft_insulation", "park_home_insulation"
|
||
]
|
||
}
|
||
|
||
meets_epc = starting_sap <= 69 # EPC D–G
|
||
|
||
# General route
|
||
if meets_epc and council_tax_band in ["A", "B", "C", "D", "E"]:
|
||
if any(m in gbis_measures["general"] for m in measure_types):
|
||
self.gbis_eligible = True
|
||
|
||
# Low-income route
|
||
if meets_epc and any(m in gbis_measures["low_income"] for m in measure_types):
|
||
self.gbis_eligible = True
|
||
self.gbis_eligibility_caveats.append(EligibilityCaveats.TENANT_ON_BENEFITS_OR_LOW_INCOME)
|
||
|
||
# -----------------------
|
||
# Social Housing
|
||
# -----------------------
|
||
|
||
def eco4_sh_eligibility(self, starting_sap: int, measure_types: List, has_innovation: bool,
|
||
innovation_measures: List):
|
||
"""
|
||
ECO4 Social Housing eligibility.
|
||
- EPC E–G: eligible, no income check.
|
||
- EPC D: innovation measure required.
|
||
If solar PV is the innovation measure, must also install ASHP or HHRSH.
|
||
"""
|
||
meets_epc = starting_sap <= 69
|
||
if not meets_epc:
|
||
return
|
||
|
||
# EPC D innovation rule
|
||
if 55 <= starting_sap <= 68: # EPC D
|
||
if not has_innovation:
|
||
self.eco4_eligible = False
|
||
self.eco4_eligibility_caveats.append(EligibilityCaveats.INNOVATION_REQUIRED)
|
||
return
|
||
|
||
if "solar_pv" in innovation_measures and not any(
|
||
m in measure_types for m in ["air_source_heat_pump", "high_heat_retention_storage_heater"]
|
||
):
|
||
self.eco4_eligible = False
|
||
self.eco4_eligibility_caveats.append(EligibilityCaveats.SOLAR_NEEDS_HEATING)
|
||
return
|
||
|
||
self.eco4_eligible = True
|
||
|
||
def gbis_sh_eligibility(self, starting_sap: int, measure_types: List, has_innovation: bool,
|
||
innovation_measures: List):
|
||
"""
|
||
GBIS Social Housing eligibility.
|
||
- EPC E–G: insulation measures OK.
|
||
- EPC D: innovation measure required (same solar PV + heating rule).
|
||
"""
|
||
meets_epc = starting_sap <= 69
|
||
if not meets_epc:
|
||
return
|
||
|
||
if 55 <= starting_sap <= 68: # EPC D
|
||
if not has_innovation:
|
||
self.gbis_eligible = False
|
||
self.gbis_eligibility_caveats.append(EligibilityCaveats.INNOVATION_REQUIRED)
|
||
return
|
||
|
||
if "solar_pv" in innovation_measures and not any(
|
||
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)
|
||
return
|
||
|
||
self.gbis_eligible = True
|
||
|
||
# -----------------------
|
||
# Score Lookup
|
||
# -----------------------
|
||
|
||
def calculate_full_project_abs(self):
|
||
"""Look up ABS score for full projects."""
|
||
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.empty:
|
||
raise ValueError("Missing ABS rate, check the project scores matrix")
|
||
|
||
return data["Cost Savings"].values[0]
|
||
|
||
# -----------------------
|
||
# Main Entry Point
|
||
# -----------------------
|
||
|
||
def check_funding(
|
||
self, measures: List[dict],
|
||
starting_sap: int,
|
||
ending_sap: int,
|
||
floor_area: float,
|
||
mainheat_description: str,
|
||
heating_control_description: str,
|
||
is_cavity: bool,
|
||
council_tax_band: str = None
|
||
):
|
||
"""
|
||
Given a list of measures, check ECO4/GBIS eligibility.
|
||
"""
|
||
|
||
# Normalize measures
|
||
measure_types, has_innovation, innovation_measures = self._split_measures(measures)
|
||
|
||
# Track EPC bands and floor area
|
||
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)
|
||
|
||
if self.tenure == "Private":
|
||
# ECO4 PRS
|
||
self.eco4_prs_eligibility(starting_sap, measure_types, mainheat_description, heating_control_description)
|
||
# GBIS PRS
|
||
self.gbis_prs_eligibility(starting_sap, council_tax_band or "", measure_types)
|
||
|
||
if self.eco4_eligible:
|
||
eco4_abs = self.calculate_full_project_abs()
|
||
eco4_funding = eco4_abs * (self.private_cavity_abs_rate if is_cavity else self.private_solid_abs_rate)
|
||
return {"eco4_funding": eco4_funding}
|
||
|
||
elif self.tenure == "Social":
|
||
# ECO4 Social
|
||
self.eco4_sh_eligibility(starting_sap, measure_types, has_innovation, innovation_measures)
|
||
# GBIS Social
|
||
self.gbis_sh_eligibility(starting_sap, measure_types, has_innovation, innovation_measures)
|
||
|
||
if self.eco4_eligible:
|
||
eco4_abs = self.calculate_full_project_abs()
|
||
eco4_funding = eco4_abs * (self.social_cavity_abs_rate if is_cavity else self.social_solid_abs_rate)
|
||
return {"eco4_funding": eco4_funding}
|
||
|
||
else:
|
||
raise NotImplementedError("Only 'Private' and 'Social' tenures are supported.")
|