Merge pull request #535 from Hestia-Homes/main

Fixing fundign packages where all measures are fixed. Addition EPC estimation with pre-defined heating system

backend/Property.py

@@ -1302,7 +1302,8 @@ class Property:
         # If there is no existing solar PV, the photo-supply field will be None or a missing value
 
         # We use inspections data to tell us this
-        if self.inspections:
+
+        if getattr(self.inspections, "roof_orientation", None):
             has_no_existing_solar_pv = self.inspections.roof_orientation.value not in [
                 "already has solar pv", "roof too small", "no roof"
             ]

backend/SearchEpc.py

@@ -156,6 +156,7 @@ class SearchEpc:
         size=None,
         property_type=None,
         fast=False,
+        heating_system: [str, None] = None
     ):
         """
         Address lines 1 and postcode are mandatory fields. The other address lines are optional
@@ -180,6 +181,9 @@ class SearchEpc:
         self.house_number = self.get_house_number(self.address1)
         self.numeric_house_number = self.extract_numeric_housenumber_part(self.house_number)
 
+        # property attributes
+        self.heating_system = heating_system
+
         self.max_retries = max_retries if max_retries is not None else self.MAX_RETRIES
 
         self.client = EpcClient(auth_token=auth_token)
@@ -571,7 +575,8 @@ class SearchEpc:
         lmks_to_drop: list[str] | None = None,
         built_form: str = "",
         property_type: str = "",
-        exclude_old: bool = False
+        exclude_old: bool = False,
+        heating_system: [str, None] = None
     ):
         """
         Fetches and processes EPC data for a given initial postcode, applying successive trimming
@@ -591,6 +596,7 @@ class SearchEpc:
         :param built_form: The 'built-form' value to be used for filtering the EPC data.
         :param property_type: The 'property-type' value to be used for filtering the EPC data.
         :param exclude_old: Flag to exclude EPC data older than 10 years.
+        :param heating_system: Optional heating system type for additional filtering.
         :return:
         """
 
@@ -703,6 +709,11 @@ class SearchEpc:
                                 epc_data["property-type"] == estimation_property_type)
                             ]
 
+                    if heating_system is not None:
+                        epc_data = epc_data[
+                            epc_data["mainheat-description"] == heating_system
+                            ]
+
                     if not epc_data.empty:
                         return epc_data  # Return the filtered data if it's not empty
 
@@ -712,7 +723,7 @@ class SearchEpc:
         # If loop finishes without a valid response, raise an exception
         raise Exception("Unable to find postcode data after trimming - investigate me")
 
-    def estimate_epc(self, property_type, built_form, lmks_to_drop=None, exclude_old=False):
+    def estimate_epc(self, property_type, built_form, lmks_to_drop=None, exclude_old=False, heating_system=None):
         """
         For a property that does not have an EPC, we retrieve the EPC data for the closest properties
         and estimate the EPC for the property in question.
@@ -726,6 +737,8 @@ class SearchEpc:
         :param lmks_to_drop:    This is a list of LMK keys that should be dropped from the estimation process. This
                                 is used as an override for testing, to drop EPCs for the property we are testing
         :param exclude_old:     Used to drop any expired EPCs (more than 10 years old)
+        :param heating_system:  The heating system of the property we are estimating, if known. Will aim to filter EPCs
+                                to matching heating systems
         :return:
         """
 
@@ -736,7 +749,8 @@ class SearchEpc:
             lmks_to_drop=lmks_to_drop,
             built_form=built_form,
             property_type=property_type,
-            exclude_old=exclude_old
+            exclude_old=exclude_old,
+            heating_system=heating_system
         )
 
         # Check if it's a new build EPC. A property that doesn't have an EPC is not going to be a new build
@@ -906,7 +920,8 @@ class SearchEpc:
                 # We can try and estimate
                 estimated_epc = self.estimate_epc(
                     property_type=self.ordnance_survey_client.property_type,
-                    built_form=self.ordnance_survey_client.built_form
+                    built_form=self.ordnance_survey_client.built_form,
+                    heating_system=self.heating_system
                 )
                 self.newest_epc = estimated_epc
                 self.older_epcs = []

backend/app/assumptions.py

@@ -77,7 +77,8 @@ DESCRIPTIONS_TO_FUEL_TYPES = {
     "Electric ceiling heating, electric": {"fuel": "Electricity", "cop": 1},
     "Air source heat pump, warm air, electric": {
         "fuel": "Electricity", "cop": AVERAGE_ASHP_EFFICIENCY / 100
-    }
+    },
+    "Electric heat pump for water heating only": {"fuel": "Electricity", "cop": 1},
 }
 
 # These are the measure types where if there is a ventilation recommendation, we force the inclusion of it

backend/engine/engine.py

@@ -3,6 +3,7 @@ import json
 from copy import deepcopy
 from datetime import datetime
 
+from sqlalchemy import Nullable
 from tqdm import tqdm
 import pandas as pd
 import numpy as np
@@ -59,7 +60,7 @@ from recommendations.recommendation_utils import convert_thickness_to_numeric, g
 logger = setup_logger()
 
 BATCH_SIZE = 5
-SCORING_BATCH_SIZE = 100
+SCORING_BATCH_SIZE = 300
 
 
 def extract_portfolio_aggregation_data(
@@ -373,6 +374,24 @@ def get_funding_data():
     return project_scores_matrix, partial_project_scores_matrix, whlg_eligible_postcodes
 
 
+def parse_heating_system(config):
+    """
+    Helper function to extract a heating system, which can be used to estimate EPC. This is a very limited,
+    placeholder function to cover some initial immediate cases.
+    :return:
+    """
+
+    ll_heating = config.get("landlord_heating_system", None)
+    if not ll_heating:
+        return None
+
+    if ll_heating == "electric storage heaters":
+        # Return with the same format at the EPC
+        return "Electric storage heaters"
+
+    return None
+
+
 async def model_engine(body: PlanTriggerRequest):
     logger.info("Model Engine triggered with body: %s", json.loads(body.model_dump_json()))
 
@@ -502,8 +521,8 @@ async def model_engine(body: PlanTriggerRequest):
                 address1 = config.get("domna_full_address", None)
 
             address1 = str(int(address1)) if isinstance(address1, float) else str(address1)
-
             full_address = config["domna_full_address"] if body.file_format == "domna_asset_list" else None
+            heating_system = parse_heating_system(config)
 
             epc_searcher = SearchEpc(
                 address1=address1,
@@ -511,7 +530,8 @@ async def model_engine(body: PlanTriggerRequest):
                 uprn=uprn,
                 auth_token=get_settings().EPC_AUTH_TOKEN,
                 os_api_key="",
-                full_address=full_address
+                full_address=full_address,
+                heating_system=heating_system
             )
             epc_searcher.ordnance_survey_client.built_form = config.get("built_form", None)
             epc_searcher.ordnance_survey_client.property_type = config.get("property_type", None)
@@ -911,25 +931,10 @@ async def model_engine(body: PlanTriggerRequest):
                     housing_type=body.housing_type,
                     budget=body.budget,
                     target_gain=gain,
-                    funding=funding
+                    funding=funding,
+                    work_package=eco_packages[p.id][2]
                 )
 
-                # Given the solutions we select the optimal one
-                # 1) If the scheme is ECO4, the full project funding and uplift are deducted from the cost
-                # 2) If the sheme is GBIS, the partial project funding and uplift are deducted from the cost
-                # 3) Otherwise, no funding is deducted from the cost
-                solutions["cost_less_full_project_funding"] = np.where(
-                    solutions["scheme"] == "none",
-                    solutions["total_cost"],
-                    np.where(
-                        solutions["scheme"] == "eco4",
-                        solutions["total_cost"] - solutions["full_project_funding"] - solutions["total_uplift"],
-                        solutions["total_cost"] - solutions["partial_project_funding"] - solutions["total_uplift"]
-                    )
-                )
-
-                solutions = solutions.sort_values("cost_less_full_project_funding", ascending=True)
-
                 # If the solution isn't eligible, we can't really consider it
                 solutions = solutions[
                     (solutions["is_eligible"] & (solutions["scheme"] != "none")) | (solutions["scheme"] == "none")
@@ -944,8 +949,12 @@ async def model_engine(body: PlanTriggerRequest):
 
                 # This is the list of measures that we will recommend
                 scheme = optimal_solution["scheme"]
-                funded_measures = optimal_solution["items"] if scheme != "none" else []
+
-                solution = optimal_solution["items"] + optimal_solution["unfunded_items"]
+                # We create this full list of selected measures, which is used in the next section for setting
+                # default measures
+                solution = deepcopy(optimal_solution["items"]) + deepcopy(optimal_solution["unfunded_items"])
+                funded_measures = deepcopy(optimal_solution["items"]) if scheme != "none" else []
+
                 # This is the total amount of funding that the project will produce (EXCLUDING uplifts) (£)
                 project_funding = optimal_solution["full_project_funding"] if scheme == "eco4" else \
                     optimal_solution["partial_project_funding"]

recommendations/optimiser/funding_optimiser.py

@@ -9,6 +9,7 @@ In the future, we will adapt this into a class-based structure to allow for more
 
 from copy import deepcopy
 import pandas as pd
+import numpy as np
 
 from backend.app.plan.schemas import (
     WALL_INSULATION_MEASURES, ROOF_INSULATION_MEASURES, ECO4_ELIGIBILE_FABRIC_MEASURES
@@ -198,7 +199,49 @@ def _ensure_unfunded_costs(groups):
     return groups
 
 
-def optimise_with_funding_paths(p, input_measures, housing_type, funding: Funding, budget=None, target_gain=None):
+def _get_already_installed_gain(selected_measures, needs_pre_eco_hhrsh_upgrade):
+    """
+    Calculate already installed gain, with special case for pre-ECO4 HHRSH upgrade.
+    :param selected_measures: List of selected measures
+    :param needs_pre_eco_hhrsh_upgrade: Boolean indicating if pre-ECO4 HHRSH upgrade is needed
+    :return:
+    """
+    if needs_pre_eco_hhrsh_upgrade:
+        return sum(
+            [x["gain"] for x in selected_measures if
+             x["already_installed"] or x["type"] == "high_heat_retention_storage_heaters"]
+        )
+
+    return sum([x["gain"] for x in selected_measures if x["already_installed"]])
+
+
+def _move_hhrsh_to_unfunded(picked, unfunded_picked, needs_pre_eco_hhrsh_upgrade):
+    """
+    This function handles the case of moving HHRSH to unfunded picks if needed, where we have an ECO4 project
+    where an unfunded measure needs to be installed first.
+    :param picked: List of picked measures
+    :param unfunded_picked: List of unfunded picked measures
+    :param needs_pre_eco_hhrsh_upgrade: Boolean indicating if pre-ECO4 HHRSH upgrade is needed
+    :return:
+    """
+
+    if not needs_pre_eco_hhrsh_upgrade:
+        return picked, unfunded_picked
+
+    # We append HHRSH to unfunded items
+    hhrsh_measure = [x for x in picked if x["type"] == "high_heat_retention_storage_heaters"]
+    if not hhrsh_measure:
+        raise ValueError("Expected HHRSH measure to be in total picks")
+    unfunded_picked += hhrsh_measure
+    # Remove from total picks
+    picked = [x for x in picked if x["type"] != "high_heat_retention_storage_heaters"]
+
+    return picked, unfunded_picked
+
+
+def optimise_with_funding_paths(
+    p, input_measures, housing_type, funding: Funding, budget=None, target_gain=None, work_package=None
+):
     """
     run_optimizer(sub_measures, budget, target_gain) -> (picked_options, sub_cost, sub_gain)
     """
@@ -227,7 +270,9 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
         })
 
     # This function will filter down on innovation measures if we are social EPC D
-    funding_paths, optimisation_input_measures = make_funding_paths(p, input_measures, housing_type, funding)
+    funding_paths, optimisation_input_measures = make_funding_paths(
+        p, input_measures, housing_type, funding, work_package
+    )
 
     # We now produce a fabric only path for ECO4
     # We add in generic insulation funding paths (where there is no fixed measure)
@@ -244,6 +289,10 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
             ] + funding_paths
         )
 
+    needs_pre_eco_hhrsh_upgrade = (
+        (p.data["current-energy-rating"] == "D") and work_package == "solar_hhrsh_eco4"
+    )
+
     for path_spec in funding_paths:
 
         # ECO4 fabric only path = special case
@@ -281,8 +330,13 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
 
             scheme = _path_scheme([path_spec])
 
-            # We sum of gain, for already installed measures
+            # We sum of gain, for already installed measures. In this, we also include HHRSH, when we have
-            already_installed_gain = sum([x["gain"] for x in picked if x["already_installed"]])
+            # an EPC D property that needs HHRSH but HHRSH isn't an eligible measure
+            already_installed_gain = _get_already_installed_gain(
+                picked, needs_pre_eco_hhrsh_upgrade
+            )
+            # If we need a pre-eco4 HHRSH upgrade, we move HHRSH to unfunded items
+            picked, unfunded_picked = _move_hhrsh_to_unfunded(picked, [], needs_pre_eco_hhrsh_upgrade)
 
             solutions.append(
                 {
@@ -295,7 +349,7 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
                     "is_eligible": _is_eligible_funding_package(
                         scheme, float(p.data["current-energy-efficiency"]), sub_gain
                     ),
-                    "unfunded_items": [],
+                    "unfunded_items": unfunded_picked,
                     "already_installed_gain": already_installed_gain
                 }
             )
@@ -348,9 +402,7 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
             # If we have a budget, we need to ensure the subproblem respects it so we remove the fixed cost (which
             # may already be over budget) and the fixed gain (which may not be achievable)
 
-            if fixed_gain > target_gain:
+            if (fixed_gain > target_gain) or (fixed_gain <= target_gain and not sub_measures):
-                picked, sub_cost, sub_gain = ([], 0.0, 0.0)
-            elif fixed_gain <= target_gain and not sub_measures:
                 picked, sub_cost, sub_gain = ([], 0.0, 0.0)
             else:
                 picked, sub_cost, sub_gain = run_optimizer(
@@ -359,8 +411,9 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
                     sub_target_gain=target_gain - fixed_gain if target_gain is not None else None
                 )
 
-            if picked is None:
+            # if picked is None:
-                continue
+            #     # If we have something in sub_measures, then we have a partial solution, just not enough to
+            #     continue
 
             scheme = _path_scheme(path_spec)
 
@@ -422,7 +475,16 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
                     total_gain += unfunded_gain
 
             # We now grab the "already installed gain"
-            already_installed_gain = sum([x["gain"] for x in total_picks if x["already_installed"]])
+            # We sum of gain, for already installed measures. In this, we also include HHRSH, when we have
+            # an EPC D property that needs HHRSH but HHRSH isn't an eligible measure
+            already_installed_gain = _get_already_installed_gain(
+                total_picks, needs_pre_eco_hhrsh_upgrade
+            )
+
+            # If we need a pre-eco4 HHRSH upgrade, we move HHRSH to unfunded items
+            total_picks, unfunded_picked = _move_hhrsh_to_unfunded(
+                total_picks, unfunded_picked, needs_pre_eco_hhrsh_upgrade
+            )
 
             solutions.append({
                 "fixed_ids": fixed_ids,
@@ -479,6 +541,22 @@ def optimise_with_funding_paths(p, input_measures, housing_type, funding: Fundin
     solutions["total_uplift"] = solutions.apply(lambda x: get_total_uplift(x), axis=1)
     solutions["total_uplift_score"] = solutions.apply(lambda x: get_total_innovation_score(x), axis=1)
 
+    # Given the solutions we select the optimal one
+    # 1) If the scheme is ECO4, the full project funding and uplift are deducted from the cost
+    # 2) If the sheme is GBIS, the partial project funding and uplift are deducted from the cost
+    # 3) Otherwise, no funding is deducted from the cost
+    solutions["cost_less_full_project_funding"] = np.where(
+        solutions["scheme"] == "none",
+        solutions["total_cost"],
+        np.where(
+            solutions["scheme"] == "eco4",
+            solutions["total_cost"] - solutions["full_project_funding"] - solutions["total_uplift"],
+            solutions["total_cost"] - solutions["partial_project_funding"] - solutions["total_uplift"]
+        )
+    )
+
+    solutions = solutions.sort_values("cost_less_full_project_funding", ascending=True)
+
     return solutions
 
 
@@ -837,7 +915,7 @@ def _make_generic_gbis_funding_paths(input_gbis_measures, funding_paths):
     return funding_paths + gbis_funding_paths
 
 
-def make_funding_paths(p, input_measures, housing_type, funding: Funding):
+def make_funding_paths(p, input_measures, housing_type, funding: Funding, work_package=None):
     """
     This function generates funding paths based on the input measures and the tenure of the property.
     It checks for the presence of specific measures and creates paths that include necessary insulation measures
@@ -848,6 +926,8 @@ def make_funding_paths(p, input_measures, housing_type, funding: Funding):
     :param input_measures:
     :param housing_type:
     :param funding: The funding object that provides methods to check eligibility and calculate funding.
+    :param work_package: Optional work package information. We handle the case of an EPC D property needing a heating
+                         upgrade, where the heating upgrade needs to be conducted before the solar PV work
     :return:
     """
 
@@ -890,6 +970,12 @@ def make_funding_paths(p, input_measures, housing_type, funding: Funding):
             group_of_innovation_measures = []
             group_of_gbis_innovation_measures = []
             for measure in measures:
+
+                if measure["type"] == "high_heat_retention_storage_heaters" and work_package == "solar_hhrsh_eco4":
+                    # With this work type, if the property is EPC D and doesn't have an eligible heating system
+                    # we install HHRSH as a pre-requisite measure, before the ECO4 project if complete.
+                    group_of_innovation_measures.append(measure)
+
                 if measure["innovation_uplift"] or measure["type"] in remaining_insulation_type or measure[
                     "already_installed"]:
                     group_of_innovation_measures.append(measure)
@@ -906,7 +992,7 @@ def make_funding_paths(p, input_measures, housing_type, funding: Funding):
                 input_gbis_measures_innovation.extend(group_of_gbis_innovation_measures)
 
         funding_paths = _make_solar_heating_funding_paths(
-            p, input_measures_innovation, funding_paths, remaining_insulation_type, housing_type, funding
+            p, input_measures_innovation, funding_paths, remaining_insulation_type, housing_type, funding,
         )
 
         # Can only be innovation GBIS measures