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

Merge pull request #573 from Hestia-Homes/eco-eligiblity-bug

Browse source 
Eco eligiblity bug
This commit is contained in:
KhalimCK 2025-11-29 16:20:22 +08:00 • committed by GitHub
commit c7c3270a05
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 134 additions and 46 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
backend/Property.py
View file

@ -889,6 +889,8 @@ class Property:
"current_energy_demand": self.current_energy_consumption,
"current_energy_demand_heating_hotwater": self.current_energy_consumption_heating_hotwater,
"estimated": self.data.get("estimated", False),
# We indicate if we've overwritten a SAP 05 EPC
"sap_05_overwritten": self.data.get("sap_05_overwritten", False),
**self.current_energy_bill
}

73
backend/SearchEpc.py
View file

@ -199,7 +199,7 @@ class SearchEpc:
)
self.data = None
self.newest_epc = None
self.newest_epc = {}
self.older_epcs = None
self.full_sap_epc = None
self.metadata = None
@ -214,6 +214,9 @@ class SearchEpc:
self.property_type = property_type
self.fast = fast
# By default, this is set to false. This flag indicates whether we should overwrite SAP 2005 entires.
self.overwrite_sap05 = False
def set_strict_property_type_search(self):
"""
This method sets the strict property type search flag to True. When this flag is set, the search will
@ -531,6 +534,9 @@ class SearchEpc:
if uprns:
uprn = uprns.pop()
# Convert to int
if not pd.isnull(uprn):
uprn = int(uprn)
else:
newest_epc["uprn-source"] = self.UPRN_SOURCE_SIMULATED
uprn = hash(self.address1 + self.postcode)
@ -649,6 +655,11 @@ class SearchEpc:
epc_data["lodgement-datetime"] > (pd.Timestamp.now() - pd.DateOffset(years=10))
]
# Regardless of whether or not we exclude old, we drop any SAP05 entries, which will be problematic
# if we include them
if not epc_data.empty:
epc_data = epc_data[~epc_data["mainheat-description"].str.lower().str.contains("sap05:")]
if not epc_data.empty:
# Further processing of the EPC data
@ -694,6 +705,18 @@ class SearchEpc:
estimation_built_form = "End-Terraced"
elif (built_form == "") or (pd.isnull(built_form)):
estimation_built_form = epc_built_form
elif built_form == "Enclosed Mid-Terrace":
# We check if we have any enclosed and if not, we fall back to mid-terrace
if sum(epc_data["built-form"] == "Enclosed Mid-Terrace") > 0:
estimation_built_form = "Enclosed Mid-Terrace"
else:
estimation_built_form = "Mid-Terrace"
elif built_form == "Enclosed End-Terrace":
# An enclosed end terrace has three two external facing walls so we fall back to mid-terrace
if sum(epc_data["built-form"] == "Enclosed End-Terrace") > 0:
estimation_built_form = "Enclosed Mid-Terrace"
else:
estimation_built_form = "Mid-Terrace"
else:
estimation_built_form = built_form
@ -917,7 +940,7 @@ class SearchEpc:
return agg[key].values[0]
def find_property(self, skip_os=False, api_data=None):
def find_property(self, skip_os=False, api_data=None, overwrite_sap05=False):
"""
This method will attempt to identify a property. It will, at first, use the EPC api to try and
find the EPC for the property and the associated UPRN. If this fails, it will use the Ordnance Survey API to
@ -931,6 +954,10 @@ class SearchEpc:
:param skip_os: If True, the ordnance survey api will be skipped and only the EPC api will be used
:param api_data: If provided, this data will be used instead of querying the EPC api
:param overwrite_sap05: For extrememly old, SAP05 EPCs, we may wish to overwrite them with an estimated EPC.
This is because the SAP05 EPCs will have missing information such as the main heating
will be described as SAP05:Main-Heating, which isn't particularly useful for the
purpose of providing recommendations.
"""
# Step 1: use the epc api to find the property and uprn
@ -944,8 +971,22 @@ class SearchEpc:
(
self.newest_epc, self.older_epcs, self.full_sap_epc, self.address_clean, self.postcode_clean, self.uprn
) = self.extract_epc_data(address=self.full_address)
# Before we return, we check if we need to overwrite a SAP05 EPC
# If we have don't have SAP05 in the heating description and overwrite_sap05 is False, we return
is_sap_o5 = "SAP05:" in self.newest_epc.get("mainheat-description", "")
if (
(not is_sap_o5) and (not overwrite_sap05) and (response["status"] == 200)
):
return
# By default, we don't exclude old but we will do, when we are estimating to overwrite a SAP05 EPC
lmks_to_drop, exclude_old = [], False
if is_sap_o5:
self.overwrite_sap05 = True
lmks_to_drop = [self.newest_epc["lmk-key"]]
exclude_old = True
# Step 2: If we don't have an EPC, we use the ordnance survey api to find the uprn
if skip_os:
if self.ordnance_survey_client.property_type is not None:
@ -954,10 +995,18 @@ class SearchEpc:
property_type=self.ordnance_survey_client.property_type,
built_form=self.ordnance_survey_client.built_form,
heating_system=self.heating_system,
associated_uprns=self.associated_uprns
associated_uprns=self.associated_uprns,
lmks_to_drop=lmks_to_drop,
exclude_old=exclude_old
)
if self.overwrite_sap05:
# We keep a record of the fact that we have performed a SAP05 overwrite
estimated_epc["sap_05_overwritten"] = True
# If we have overwritten a SAP05 EPC, we need to update older_epcs too
self.older_epcs = [] if not self.overwrite_sap05 else [self.newest_epc.copy()]
self.newest_epc = estimated_epc
self.older_epcs = []
self.full_sap_epc = {}
# Finally, set a standardised address 1 and postcode
@ -1000,6 +1049,20 @@ class SearchEpc:
self.postcode_clean = self.ordnance_survey_client.postcode_os
return
def set_uprn_source(self, file_format):
"""
Utility function to set the uprn source based on the file format. Only works for domna_asset_lists
and this is very much placeholder until we standardised our input data formats
:param file_format:
:return:
"""
if not self.newest_epc:
raise ValueError("No EPC data available to set UPRN source - run find_property first")
if self.newest_epc.get("estimated") and file_format == "domna_asset_list" and (self.newest_epc["uprn"] < 0):
self.newest_epc["uprn-source"] = self.UPRN_SOURCE_SIMULATED
def check_attribute_variations(self):
attribute_map = {
"walls-description": {
@ -1057,7 +1120,7 @@ class SearchEpc:
return "ground"
def get_metadata(self):
if self.newest_epc is None:
if not self.newest_epc:
raise ValueError("No EPC data available")
# We check if the property has ever been downgraded on SAP

5
backend/app/db/functions/tasks/Tasks.py
View file

@ -194,13 +194,12 @@ class SubTaskInterface:
elif "in progress" in statuses:
task.status = "in progress"
if task.job_started is None:
task.job_started = now
# if task.job_started is None:
# task.job_started = now
else:
# All waiting
task.status = "waiting"
task.job_started = None
task.job_completed = None
task.updated_at = now

3
backend/app/db/models/materials.py
View file

@ -45,7 +45,10 @@ class MaterialType(enum.Enum):
solar_pv = "solar_pv"
solar_battery = "solar_battery"
scaffolding = "scaffolding"
# Heating systems
high_heat_retention_storage_heaters = "high_heat_retention_storage_heaters"
air_soruce_heat_pump = "air_soruce_heat_pump"
boiler_upgrade = "boiler_upgrade"
sealing_fireplace = "sealing_fireplace"
roomstat_programmer_trvs = "roomstat_programmer_trvs"
time_temperature_zone_control = "time_temperature_zone_control"

1
backend/app/db/models/portfolio.py
View file

@ -175,6 +175,7 @@ class PropertyDetailsEpcModel(Base):
current_energy_demand = Column(Float)
current_energy_demand_heating_hotwater = Column(Float)
estimated = Column(Boolean, default=False)
sap_05_overwritten = Column(Boolean, default=False)
# Include estimates for energy bills, across the different types of energy
heating_cost_current = Column(Float)
hot_water_cost_current = Column(Float)

2
backend/app/plan/router.py
View file

@ -133,7 +133,7 @@ async def trigger_plan_entrypoint(body: PlanTriggerRequest):
task_source="backend/plan/router.py:trigger_plan_entrypoint",
service="plan_engine",
inputs=data,
task_only=False
task_only=False,
)
data["task_id"] = str(task_id)
data["subtask_id"] = str(subtask_id)

49
backend/engine/engine.py
View file

@ -637,12 +637,8 @@ async def model_engine(body: PlanTriggerRequest):
epc_searcher.ordnance_survey_client.property_type = config.get("property_type", None)
# For the moment, our OS API access is unavailable, so we skip and interpolate
epc_searcher.find_property(skip_os=True, api_data=epc_api_data)
if epc_searcher.newest_epc.get("estimated") and body.file_format == "domna_asset_list" and (
epc_searcher.newest_epc["uprn"] < 0
):
epc_searcher.newest_epc["uprn-source"] = epc_searcher.UPRN_SOURCE_SIMULATED
epc_searcher.find_property(skip_os=True, api_data=epc_api_data, overwrite_sap05=True)
epc_searcher.set_uprn_source(file_format=body.file_format)
# We check for an energy assessment we have performed on this property:
energy_assessment = db_funcs.energy_assessment_functions.get_latest_assessment_by_uprn(
@ -658,12 +654,6 @@ async def model_engine(body: PlanTriggerRequest):
if not is_new and not body.multi_plan:
continue
if epc_searcher.newest_epc is None:
raise ValueError(
"No EPCs found for this property and did not estimate - likely need to provide a"
"property type and built form"
)
if is_new:
db_funcs.property_functions.create_property_targets(
session,
@ -690,35 +680,20 @@ async def model_engine(body: PlanTriggerRequest):
uprn=epc_searcher.uprn,
)
# Pull this out as it may get overwritten
property_non_invasive_recommendations = req_data.non_invasive_recommendations
patch = req_data.patch
property_non_invasive_recommendations, patch = req_data.non_invasive_recommendations, req_data.patch
# if we have a remote assment data type, we pull the additional data and include it
epc_page_source = {}
if (body.event_type == "remote_assessment") and not (epc_searcher.newest_epc.get("estimated")):
try:
property_non_invasive_recommendations, patch, epc_page_source = RetrieveFindMyEpc.get_from_epc(
epc_searcher.newest_epc, epc_page, rrn=rrn
property_non_invasive_recommendations, patch, epc_page_source = (
RetrieveFindMyEpc.get_from_epc_with_fallback(
epc=epc_searcher.newest_epc,
epc_page=epc_page,
rrn=rrn,
cleaned_address=epc_searcher.address_clean,
config_address=config["address"]
)
except Exception as e:
logger.error(f"Failed to retrieve without cleaning address {e}")
try:
epc_to_use = deepcopy(epc_searcher.newest_epc)
for k in ["address", "address1"]:
epc_to_use[k] = epc_searcher.address_clean
property_non_invasive_recommendations, patch, epc_page_source = RetrieveFindMyEpc.get_from_epc(
epc_to_use, epc_page, rrn=rrn
)
except Exception as e:
# Final attempt
logger.error(f"Failed to retrieve without cleaning address {e}")
epc_to_use = deepcopy(epc_searcher.newest_epc)
for k in ["address", "address1"]:
epc_to_use[k] = config["address"]
property_non_invasive_recommendations, patch, epc_page_source = RetrieveFindMyEpc.get_from_epc(
epc_to_use, epc_page, rrn=rrn
)
# If we have a property type, this means when we pull the epc data, we might need to make a patch
)
epc_records = patch_epc(patch, epc_records)
@ -1190,6 +1165,8 @@ async def model_engine(body: PlanTriggerRequest):
# when we have buildings, we tweak our solar PV recommendations as if one unit needs it, we apply it to all
# of them
# TODO: We can probably do better and optimise at the building level - this is temp
# Idea: - optimise all measures except solar at the unit level. Then, test with and without solar for
# all units at the same time
logger.info("Adjusting solar PV recommendations for buildings")
building_ids = set([p.building_id for p in input_properties if p.building_id is not None])

45
etl/find_my_epc/RetrieveFindMyEpc.py
View file

@ -1,7 +1,8 @@
import time
import re
import pandas as pd
import requests
import pandas as pd
from copy import deepcopy
from bs4 import BeautifulSoup
from datetime import datetime
@ -697,6 +698,7 @@ class RetrieveFindMyEpc:
"Increase loft insulation to 250mm": ["loft_insulation"],
"Solar photovoltaics panels, 25% of roof area": ["solar_pv"],
'Air or ground source heat pump': ["air_source_heat_pump"],
"Add PV Battery": ["solar_pv_battery"],
}
survey = True
@ -777,3 +779,44 @@ class RetrieveFindMyEpc:
}
return non_invasive_recommendations, patch, page_source
@classmethod
def get_from_epc_with_fallback(
cls, epc, epc_page, rrn, cleaned_address=None, config_address=None
):
"""
Attempt get_from_epc with:
1) Original EPC
2) EPC with cleaned address
3) EPC with configured address
in that order.
"""
# The data we'll use to attempt retrieval
# 1) Original
attempts = [epc]
# 2) Cleaned
if cleaned_address:
modified = deepcopy(epc)
for k in ["address", "address1"]:
modified[k] = cleaned_address
attempts.append(modified)
# 3) Config address fallback
if config_address:
modified = deepcopy(epc)
for k in ["address", "address1"]:
modified[k] = config_address
attempts.append(modified)
# Iterate attempts
last_error = None
for idx, attempt in enumerate(attempts, start=1):
try:
return cls.get_from_epc(attempt, epc_page, rrn=rrn)
except Exception as e:
last_error = e
logger.error(f"Attempt {idx} failed: {e}")
raise RuntimeError(f"All EPC retrieval attempts failed: {last_error}")