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

big commit

Browse source
This commit is contained in:
Khalim Conn-Kowlessar 2025-03-03 14:38:01 +00:00
parent ddfbf33494
commit bb8070967b
9 changed files with 159 additions and 65 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.idea/Model.iml generated
View file

@ -7,7 +7,7 @@
<sourceFolder url="file://$MODULE_DIR$/open_uprn" isTestSource="false" />
<sourceFolder url="file://$MODULE_DIR$/recommendations" isTestSource="false" />
</content>
<orderEntry type="jdk" jdkName="AssetList" jdkType="Python SDK" />
<orderEntry type="jdk" jdkName="Stonewater-wave-3" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
<component name="PyNamespacePackagesService">

2
.idea/misc.xml generated
View file

@ -3,7 +3,7 @@
<component name="Black">
<option name="sdkName" value="Python 3.10 (backend)" />
</component>
<component name="ProjectRootManager" version="2" project-jdk-name="AssetList" project-jdk-type="Python SDK" />
<component name="ProjectRootManager" version="2" project-jdk-name="Stonewater-wave-3" project-jdk-type="Python SDK" />
<component name="PyCharmProfessionalAdvertiser">
<option name="shown" value="true" />
</component>

76
asset_list/AssetList.py
View file

@ -344,6 +344,7 @@ class AssetList:
self.standardised_asset_list = self.raw_asset_list.copy()
# Will be used to store aggregated figures against the various work types
self.work_type_figures = {}
self.work_type_breakdowns = {}
self.flat_data = None
self.duplicated_addresses = None
@ -577,7 +578,7 @@ class AssetList:
self.standardised_asset_list[self.landlord_wall_construction] = np.where(
self.standardised_asset_list[self.landlord_wall_construction].str.lower().str.contains(
"average thermal transmittance"
),
) == True,
"new build - average thermal transmittance",
self.standardised_asset_list[self.landlord_wall_construction]
)
@ -1019,6 +1020,23 @@ class AssetList:
)
)
# Also include work without the SAP filter as optimistic
self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"] = (
(self.standardised_asset_list["non-intrusives: Construction"] == "CAVITY") &
(self.standardised_asset_list["non-intrusives: Insulated"].isin(["RETRO DRILLED", "FILLED AT BUILD"])) &
(~self.standardised_asset_list['non-intrusives: Material'].isin(
["GREY LOOSE BEAD", "COMPACTED BEAD", "FIBRE BATT NO CAVITY", "EMPTY NARROW BELOW 30mm"]
)
)
)
# Adjust
self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"] = np.where(
self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"],
False,
self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"]
)
######################################################
# Solar
######################################################
@ -1109,8 +1127,7 @@ class AssetList:
) | (
self.standardised_asset_list[
"walls_u_value"].apply(
lambda x: x <= 0.7 if not pd.isnull(
x) else False
lambda x: x <= 0.7 if not pd.isnull(x) else False
)
)
)
@ -1322,26 +1339,58 @@ class AssetList:
# ~self.standardised_asset_list["solar_eligible_other_floor_needs_loft"]
)
blocks_of_flats = self.standardised_asset_list[
self.standardised_asset_list[self.STANDARD_PROPERTY_TYPE] == "block of flats"
]
non_blocks_of_flats = self.standardised_asset_list[
self.standardised_asset_list[self.STANDARD_PROPERTY_TYPE] != "block of flats"
]
# Produce some aggregate figures
self.work_type_figures = {
# Empty cavity from non-intrusives
"Empty Cavity (non-intrusives)": (
self.standardised_asset_list["non_intrusive_indicates_empty_cavity"].sum()
"Empty Cavity (non-intrusives)": non_blocks_of_flats["non_intrusive_indicates_empty_cavity"].sum(),
"Empty Cavity (non-intrusives, blocks of flats)": (
blocks_of_flats["non_intrusive_indicates_empty_cavity"].sum()
),
"Empty Cavity (non-intrusives, no SAP filter)": (
self.standardised_asset_list["non_intrusive_indicates_empty_cavity_no_sap_filter"].sum()
non_blocks_of_flats["non_intrusive_indicates_empty_cavity_no_sap_filter"].sum()
),
"Empty Cavity (non-intrusives, no SAP filter, blocks of flats)": (
blocks_of_flats["non_intrusive_indicates_empty_cavity_no_sap_filter"].sum()
),
"Empty Cavity (EPC)": (
(
self.standardised_asset_list["epc_indicates_empty_cavity"] &
~self.standardised_asset_list["non_intrusive_indicates_empty_cavity"]
non_blocks_of_flats["epc_indicates_empty_cavity"] &
~non_blocks_of_flats["non_intrusive_indicates_empty_cavity"]
).sum()
),
"Empty Cavity (EPC, blocks of flat)": (
(
blocks_of_flats["epc_indicates_empty_cavity"] &
~blocks_of_flats["non_intrusive_indicates_empty_cavity"]
).sum()
),
"Cavity Extraction": (
(
~non_blocks_of_flats["non_intrusive_indicates_empty_cavity"] &
~non_blocks_of_flats["epc_indicates_empty_cavity"] &
non_blocks_of_flats["non_intrusive_indicates_cavity_extraction"]
).sum()
),
"Cavity Extraction (blocks of flats)": (
(
~blocks_of_flats["non_intrusive_indicates_empty_cavity"] &
~blocks_of_flats["epc_indicates_empty_cavity"] &
blocks_of_flats["non_intrusive_indicates_cavity_extraction"]
).sum()
),
"Cavity Extraction (no SAP filter)": (
(
~self.standardised_asset_list["non_intrusive_indicates_empty_cavity"] &
~self.standardised_asset_list["epc_indicates_empty_cavity"] &
self.standardised_asset_list["non_intrusive_indicates_cavity_extraction"]
self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"]
).sum()
),
"Solar PV (Solid Floor)": (
@ -1398,6 +1447,15 @@ class AssetList:
"Non-Intrusive Data Showed Cavity Extraction",
self.standardised_asset_list["cavity_reason"]
)
# extraction no sap filter
self.standardised_asset_list["cavity_reason"] = np.where(
(
self.standardised_asset_list["non_intrusive_indicates_cavity_extraction_no_sap_filter"] &
pd.isnull(self.standardised_asset_list["cavity_reason"])
),
"Non-Intrusive Data Showed Cavity Extraction but all SAP scores allowed",
self.standardised_asset_list["cavity_reason"]
)
# Flag solar
self.standardised_asset_list["solar_reason"] = None

67
asset_list/app.py
View file

@ -246,43 +246,43 @@ def app():
# - We want: fully insulated property (all wall types), EPC D or below (floors should be solid)
# - Or the insulation required is loft/cavity (floors should be solid)
# data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Colchester"
# data_filename = "Warmfront data- Colchester Borough Homes (Complete).xlsx"
# sheet_name = "Sheet1"
# postcode_column = 'Full Address.1'
# fulladdress_column = "Full Address"
# address1_column = None
# address1_method = "first_word"
# address_cols_to_concat = []
# missing_postcodes_method = None
# landlord_year_built = "Build Date"
# landlord_os_uprn = None
# landlord_property_type = "Property Type"
# landlord_wall_construction = "Wallinsul"
# landlord_heating_system = "HeatSorc"
# landlord_existing_pv = None
# landlord_property_id = "Property Reference"
# For Westward
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Westward"
data_filename = "WESTWARD - completed list..xlsx"
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Colchester"
data_filename = "Warmfront data- Colchester Borough Homes (Complete).xlsx"
sheet_name = "Sheet1"
postcode_column = "WFT EDIT Postcode"
fulladdress_column = "Address"
postcode_column = 'Full Address.1'
fulladdress_column = "Full Address"
address1_column = None
address1_method = "house_number_extraction"
address1_method = "first_word"
address_cols_to_concat = []
missing_postcodes_method = None
landlord_year_built = "Build date"
landlord_os_uprn = "UPRN"
landlord_property_type = "Location type"
landlord_wall_construction = "Wall Construction (EPC)"
landlord_heating_system = "Heat Source"
landlord_existing_pv = "PV (Y/N)"
landlord_property_id = "Place ref"
landlord_year_built = "Build Date"
landlord_os_uprn = None
landlord_property_type = "Property Type"
landlord_wall_construction = "Wallinsul"
landlord_heating_system = "HeatSorc"
landlord_existing_pv = None
landlord_property_id = "Property Reference"
# For Westward
# data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Westward"
# data_filename = "WESTWARD - completed list..xlsx"
# sheet_name = "Sheet1"
# postcode_column = "WFT EDIT Postcode"
# fulladdress_column = "Address"
# address1_column = None
# address1_method = "house_number_extraction"
# address_cols_to_concat = []
# missing_postcodes_method = None
# landlord_year_built = "Build date"
# landlord_os_uprn = "UPRN"
# landlord_property_type = "Location type"
# landlord_wall_construction = "Wall Construction (EPC)"
# landlord_heating_system = "Heat Source"
# landlord_existing_pv = "PV (Y/N)"
# landlord_property_id = "Place ref"
# Maps addresses to uprn in problematic cases
MANUAL_UPRN_MAP = {}
manual_uprn_map = {}
asset_list = AssetList(
local_filepath=os.path.join(data_folder, data_filename),
@ -352,7 +352,7 @@ def app():
epc_data_chunk, errors_chunk, no_epc_chunk = get_data(
df=chunk,
row_id_name=asset_list.DOMNA_PROPERTY_ID,
manual_uprn_map=MANUAL_UPRN_MAP,
manual_uprn_map=manual_uprn_map,
)
# We now retrieve any failed properties
@ -360,7 +360,7 @@ def app():
epc_data_failed, _, _ = get_data(
df=chunk_failed,
row_id_name=asset_list.DOMNA_PROPERTY_ID,
manual_uprn_map=MANUAL_UPRN_MAP,
manual_uprn_map=manual_uprn_map,
epc_api_only=False
)
@ -464,6 +464,7 @@ def app():
)
cleaned = msgpack.unpackb(cleaned, raw=False)
# TODO: We should break out the identification of work types to flag blocks of flats specifically
asset_list.identify_worktypes(cleaned)
pprint(asset_list.work_type_figures)

12
backend/Funding.py
View file

@ -149,7 +149,8 @@ class Funding:
:return:
"""
measure_table = pd.DataFrame([
m for m in self.recommendations if m in measures and m["default"]
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
@ -180,13 +181,10 @@ class Funding:
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])
# Recommend the measure, with estimated funding amount
recommended_measure = measure_table.head(1)
return {
"measure_type": recommended_measure["measure_type"],
"estimated_funding": recommended_measure["estimated_funding"]
}
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):
"""

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

@ -825,7 +825,7 @@ async def trigger_plan(body: PlanTriggerRequest):
property_recommendations=recommendations[p.id],
project_scores_matrix=eco_project_scores_matrix,
whlg_eligible_postcodes=whlg_eligible_postcodes,
gbis_abs_rate=20,
gbis_abs_rate=15,
eco4_abs_rate=15,
)
funding_calulator.check_eligibiltiy()

41
etl/customers/remote_assessments/app.py
View file

@ -4,7 +4,7 @@ from dotenv import load_dotenv
from utils.s3 import save_csv_to_s3
from etl.find_my_epc.AssetListEpcData import AssetListEpcData
PORTFOLIO_ID = 137
PORTFOLIO_ID = 134
USER_ID = 8
load_dotenv(dotenv_path="backend/.env")
@ -19,10 +19,25 @@ def app():
asset_list = [
{
"address": "41 Gainsborough Way",
"postcode": "BA21 5XU",
"uprn": 30016708,
"address": "Flat 2, 42 Malden Road, London NW5 3HG",
"postcode": "NW5 3HG",
"uprn": 5117165,
},
{
"address": "15 Bournville Lane",
"postcode": "B30 2JY",
"uprn": 100070301128
},
{
"address": "34 Bournville Lane",
"postcode": "B30 2LN",
"uprn": 100070301140
},
{
"address": "36 Bournville Lane",
"postcode": "B30 2LN",
"uprn": 100070301142
}
]
asset_list = pd.DataFrame(asset_list)
@ -52,9 +67,21 @@ def app():
valuation_data = [
{
"uprn": 30016708,
"valuation": 189000
}
"uprn": 5117165,
"valuation": 467_000
},
{
"uprn": 100070301128,
"valuation": 335_000
},
{
"uprn": 100070301140,
"valuation": 276_000
},
{
"uprn": 100070301142,
"valuation": 276_000
},
]
# Store valuation data to s3
valuation_filename = f"{USER_ID}/{PORTFOLIO_ID}/valuation.csv"

20
etl/find_my_epc/AssetListEpcData.py
View file

@ -72,12 +72,20 @@ class AssetListEpcData:
epc_searcher.find_property(skip_os=True)
if epc_searcher.newest_epc is None:
continue
find_epc_searcher = RetrieveFindMyEpc(
address=epc_searcher.newest_epc["address1"],
postcode=epc_searcher.newest_epc["postcode"]
)
find_epc_data = find_epc_searcher.retrieve_newest_find_my_epc_data()
# Attempt both methods:
try:
find_epc_searcher = RetrieveFindMyEpc(
address=epc_searcher.newest_epc["address1"] + ", " + epc_searcher.newest_epc["address2"],
postcode=epc_searcher.newest_epc["postcode"]
)
find_epc_data = find_epc_searcher.retrieve_newest_find_my_epc_data()
except Exception as e:
logger.error(f"Error retrieving find my epc data: {e}")
find_epc_searcher = RetrieveFindMyEpc(
address=epc_searcher.newest_epc["address1"],
postcode=epc_searcher.newest_epc["postcode"]
)
find_epc_data = find_epc_searcher.retrieve_newest_find_my_epc_data()
time.sleep(0.5)
# We need uprn

2
recommendations/HeatingRecommender.py
View file

@ -852,6 +852,8 @@ class HeatingRecommender:
else:
heating_simulation_config["mainheat_energy_eff_ending"] = self.property.data["mainheat-energy-eff"]
# TODO:We possibly shouldn't touch the hot water energy efficiency if we aren't recommending dual immersion
# we'll keep this for the moment though
if self.property.data["hot-water-energy-eff"] in ["Very Poor", "Poor"]:
heating_simulation_config["hot_water_energy_eff_ending"] = "Average"
else: