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

working on l&g

Browse source
This commit is contained in:
Khalim Conn-Kowlessar 2025-04-29 16:47:03 +01:00
parent afabbf9228
commit b0e6526e54
7 changed files with 314 additions and 105 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
asset_list/AssetList.py
View file

@ -1342,7 +1342,7 @@ class AssetList:
)
self.standardised_asset_list["solar_landlord_data_indicates_needs_heating_upgrade"] = (
self.standardised_asset_list[self.STANDARD_HEATING_SYSTEM].isin(
["electric storage heaters", "room heaters", "electric radiators", "no heating"]
["electric storage heaters", "room heaters", "electric radiators", "no heating", "electric fuel"]
)
)
@ -1651,7 +1651,7 @@ class AssetList:
"SAP Category"],
self.standardised_asset_list["cavity_reason"]
)
else:
elif self.non_intrusives_present:
self.standardised_asset_list["cavity_reason"] = np.where(
(
self.standardised_asset_list["epc_indicates_empty_cavity"] &
@ -1675,6 +1675,16 @@ class AssetList:
"SAP Category"],
self.standardised_asset_list["cavity_reason"]
)
else:
self.standardised_asset_list["cavity_reason"] = np.where(
(
self.standardised_asset_list["epc_indicates_empty_cavity"] &
~self.standardised_asset_list["non_intrusive_indicates_empty_cavity"] &
pd.isnull(self.standardised_asset_list["cavity_reason"])
),
"EPC Shows Empty Cavity: " + self.standardised_asset_list["SAP Category"],
self.standardised_asset_list["cavity_reason"]
)
self.standardised_asset_list["cavity_reason"] = np.where(
(
@ -1716,17 +1726,18 @@ class AssetList:
self.standardised_asset_list["solar_reason"] = None
# Map of variables and fill values for the solar_reason variable
# ordering of this map is important, where we flag our prioritised work types first
solar_reason_map = {
"solar_eligible": "Solar Eligible: ",
"solar_eligible_solid_wall_uninsulated": "Solar Eligible, Solid Wall Uninsulated, EPC E or Below: ",
"solar_eligible_needs_heating_upgrade": (
"Solar Eligible, Needs Heating Upgrade: "
),
"solar_eligible_solid_wall_uninsulated": "Solar Eligible, Solid Wall Uninsulated, EPC E or Below: ",
)
}
for variable, reason in solar_reason_map.items():
self.standardised_asset_list["solar_reason"] = np.where(
self.standardised_asset_list[variable],
self.standardised_asset_list[variable] & pd.isnull(self.standardised_asset_list["solar_reason"]),
reason + self.standardised_asset_list["SAP Category"],
self.standardised_asset_list["solar_reason"]
)

138
asset_list/app.py
View file

@ -89,6 +89,37 @@ 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)
# Abri
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Abri"
data_filename = "data for domna.xlsx"
sheet_name = "Sheet1"
postcode_column = 'post_code'
fulladdress_column = None
address1_column = "address##1"
address1_method = None
address_cols_to_concat = ["address##1", "address##2", "address##3"]
missing_postcodes_method = None
landlord_year_built = "build_date"
landlord_os_uprn = None
landlord_property_type = "PropertyType"
landlord_built_form = "BuildForm"
landlord_wall_construction = "Wall Construction"
landlord_roof_construction = None
landlord_heating_system = "HeatingType"
landlord_existing_pv = None
landlord_property_id = "place_ref"
landlord_sap = None
outcomes_filename = None
outcomes_sheetname = None
outcomes_postcode = None
outcomes_houseno = None
outcomes_id = None
outcomes_address = None
master_filepaths = []
master_to_asset_list_filepath = None
phase = False
ecosurv_landlords = None
# Bromford
data_folder = ("/Users/khalimconn-kowlessar/Documents/hestia/Customers/Bromford/Apr 2025 Programme "
"Rebuild/Prepared data/")
@ -125,7 +156,7 @@ def app():
master_to_asset_list_filepath = None
phase = False
ecosurv_landlords = "paul butler|bromford"
# Torus
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Torus/Phase 1"
data_filename = "Torus Property Asset List - Phase 1.xlsx"
@ -459,6 +490,8 @@ def app():
landlord_heating_system = "Heat Source"
landlord_existing_pv = "PV (Y/N)"
landlord_property_id = "Place ref"
landlord_roof_construction = None
landlord_sap = None
outcomes_filename = None
outcomes_sheetname = None
outcomes_postcode = None
@ -466,6 +499,9 @@ def app():
master_filepaths = []
master_to_asset_list_filepath = None
outcomes_id = None
outcomes_address = None
phase = False
ecosurv_landlords = None
# For ACIS - programme re-build
# data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/ACIS/ACIS Full Programme Review March 2025"
@ -483,17 +519,23 @@ def app():
# landlord_property_type = "Property type"
# landlord_built_form = None
# landlord_wall_construction = "Wall Constuction"
# landlord_roof_construction = None
# landlord_sap = None
# landlord_heating_system = "Heating"
# landlord_existing_pv = None
# outcomes_filename = "ACIS Group - 25.11.2024 - outcomes.xlsx"
# outcomes_sheetname = "Feedback"
# outcomes_postcode = "Postcode"
# outcomes_address = "Address"
# outcomes_houseno = "No"
# outcomes_id = None
# master_filepaths = [
# os.path.join(data_folder, "ECO 3 -Table 1.csv"),
# os.path.join(data_folder, "ECO 4 -Table 1.csv"),
# ]
# master_to_asset_list_filepath = None
# phase = False
# ecosurv_landlords = None
# For plus dane
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Plus Dane"
@ -618,7 +660,7 @@ def app():
epc_api_only = False
force_retrieve_data = False
skip = None # Used to skip already completed chunks
chunk_size = 1000
chunk_size = 5000
filename = "Chunk {i}.csv"
download_folder = os.path.join(data_folder, "Chunks")
if not os.path.exists(download_folder):
@ -804,98 +846,6 @@ def app():
asset_list.flat_analysis()
################################################################
# WESTWARD - comparison between Kieran's method & automated
################################################################
# Check 1)
cavity_fills = pd.read_excel(
os.path.join(data_folder, "WESTWARD - Route March Prep.xlsx"),
sheet_name="Straight Fill"
)
cavity_fills = cavity_fills.merge(
asset_list.standardised_asset_list[
[asset_list.STANDARD_LANDLORD_PROPERTY_ID, "cavity_reason"]
],
how="left",
left_on=asset_list.landlord_property_id,
right_on=asset_list.STANDARD_LANDLORD_PROPERTY_ID
)
cavity_fills["cavity_reason"] = cavity_fills["cavity_reason"].fillna("Not identified")
print(cavity_fills["cavity_reason"].value_counts())
# Didn't identify 3 properties because they're bedsits
# 4 properties were identified, not based on the non-intrusives but instead because
# Westward said they were built in 2003/2007. Have adjusted this to use the age from the
# epc as well, as EPC says 1975 and they look like 1975 properties
# 37 properties flagged as already having solar - these are all because the landlord said they have solar
# e.g.
# https://earth.google.com/web/search/11+Winsland+Avenue+TOTNES+TQ9+5FT/@50.43354465,-3.71318276,46.57468503a,
# 59.14004365d,35y,0h,0t,
# 0r/data=CpABGmISXAolMHg0ODZkMWQxOGE4NWRiZjdkOjB4YjBhM2E5M2Q3YWVlMWEwYhlZYgp7fzdJQCHFfC9027QNwCohMTEgV2luc2xhbmQgQXZlbnVlIFRPVE5FUyBUUTkgNUZUGAIgASImCiQJbxsQEoo3SUARXQcp_HE3SUAZBmiZGJ6yDcAhCA0fqq63DcBCAggBOgMKATBCAggASg0I____________ARAA
# https://earth.google.com/web/search/15+St+Anne%27s+Ct,+Newton+Abbot+TQ12+1TL/@50.53068337,-3.61611128,
# 11.74908956a,135.73212429d,35y,0h,0t,
# 0r/data=CpUBGmcSYQolMHg0ODZkMDVkMjFhODhjZjgxOjB4MjBmMzE2Zjc3MGI2NGMwYxlCxHLw8UNJQCFZqyzALe4MwComMTUgU3QgQW5uZSdzIEN0LCBOZXd0b24gQWJib3QgVFExMiAxVEwYAiABIiYKJAm-r6U2iDdJQBHS5ICRdDdJQBmYGVpmiLINwCG8wcrtqbYNwEICCAE6AwoBMEICCABKDQj___________8BEAA
# Check 2)
cavity_fills_with_solar = pd.read_excel(
os.path.join(data_folder, "WESTWARD - Route March Prep.xlsx"),
sheet_name="Solar PV - Straight Fill"
)
cavity_fills_with_solar = cavity_fills_with_solar.merge(
asset_list.standardised_asset_list[
[asset_list.STANDARD_LANDLORD_PROPERTY_ID, "cavity_reason"]
],
how="left",
left_on=asset_list.landlord_property_id,
right_on=asset_list.STANDARD_LANDLORD_PROPERTY_ID
)
cavity_fills_with_solar["cavity_reason"] = cavity_fills_with_solar["cavity_reason"].fillna("Not identified")
print(cavity_fills_with_solar["cavity_reason"].value_counts())
# 203 properties total
# 140 properties were flagged up based on non-intrusives (Non-Intrusive Data Showed Empty Cavity)
# 63 property already has solar
# Check 3) RDF
rdf = pd.read_excel(
os.path.join(data_folder, "WESTWARD - Route March Prep.xlsx"),
sheet_name="RDF CIGA checks"
)
rdf = rdf.merge(
asset_list.standardised_asset_list[
[asset_list.STANDARD_LANDLORD_PROPERTY_ID, "cavity_reason", "solar_reason"]
],
how="left",
left_on=asset_list.landlord_property_id,
right_on=asset_list.STANDARD_LANDLORD_PROPERTY_ID
)
rdf["cavity_reason"] = rdf["cavity_reason"].fillna("Not identified")
print(rdf["cavity_reason"].value_counts())
# 264 properties are not identified, 261 of which are due to the fact they contain materials
# The other 3 were determined to be eligible for solar instead
# Many of these units that were identified for rdf works could be solar jobs
rdf_with_solar = pd.read_excel(
os.path.join(data_folder, "WESTWARD - Route March Prep.xlsx"),
sheet_name="Solar PV - RDF CIGA Checks"
)
rdf_with_solar = rdf_with_solar.merge(
asset_list.standardised_asset_list[
[asset_list.STANDARD_LANDLORD_PROPERTY_ID, "cavity_reason", "solar_reason"]
],
how="left",
left_on=asset_list.landlord_property_id,
right_on=asset_list.STANDARD_LANDLORD_PROPERTY_ID
)
rdf_with_solar["cavity_reason"] = rdf_with_solar["cavity_reason"].fillna("Not identified")
rdf_with_solar["cavity_reason"].value_counts()
# All others identified - some flagged as empties due to EPC or landlord data suggesting as much
# 5 not identified due to containing COMPACTED BEAD
asset_list.standardised_asset_list = asset_list.standardised_asset_list[
asset_list.standardised_asset_list[asset_list.landlord_property_id]
]
asset_list.load_contact_details(
local_filepath=os.path.join(data_folder, "Full property list wth D&V report V look up 12.2.25.xlsx"),
sheet_name="Report 1",

10
asset_list/mappings/built_form.py
View file

@ -143,6 +143,12 @@ BUILT_FORM_MAPPINGS = {
'Sixth Floor': 'top-floor',
'Sheltered Bung': 'semi-detached',
'Guest': 'unknown',
'Fifth Floor': 'mid-floor'
'Fifth Floor': 'mid-floor',
'Flat Within Block': 'mid-floor',
'Coach House with Garage': 'detached',
'Over Garage House': 'top-floor',
'Apartment': 'mid-floor',
'Flat Over Shop': 'top-floor',
'Flat Over Garage': 'top-floor',
'Bridge Flat': 'mid-floor'
}

12
asset_list/mappings/heating_systems.py
View file

@ -4,6 +4,7 @@ STANDARD_HEATING_SYSTEMS = {
"gas combi boiler",
"electric storage heaters",
"district heating",
"communal heating"
"gas condensing boiler",
"oil boiler",
"gas condensing combi",
@ -202,5 +203,14 @@ HEATING_MAPPINGS = {
'Wet - Underfloor Solar': 'other',
'No Heating Required Gas': 'unknown',
'Electric - Storage/Panel Heaters Gas': 'electric storage heaters',
'Electric - Storage/Panel Heaters Solid': 'electric storage heaters'
'Electric - Storage/Panel Heaters Solid': 'electric storage heaters',
'District Heat Network': 'district heating',
'Not Applicable': 'no heating',
'Not Responsible': 'unknown',
'Communal Oil': 'communal heating',
'Communal Electric': 'communal heating',
'Renewables (Air / Ground Source Pumps)': 'air source heat pump',
'Communal Renewable': 'air source heat pump',
}

8
asset_list/mappings/property_type.py
View file

@ -178,5 +178,11 @@ PROPERTY_MAPPING = {
'Parking Space': 'other',
'Community Centre': 'other',
'Communal Facility': 'other',
'Semi': 'house'
'Semi': 'house',
'House with Compulsory Garage': 'house',
'Flat with Compulsory Garage': 'flat',
'Other': 'other',
'Maisonette with Compulsory Garage': 'maisonette',
'Room in Shared Property': 'other'
}

27
asset_list/mappings/walls.py
View file

@ -158,7 +158,6 @@ WALL_CONSTRUCTION_MAPPINGS = {
'2017 onwards': 'new build - average thermal transmittance',
'ND (inferred)': 'unknown',
'Flat / maisonette': 'other',
'Other': 'other',
'Timber Frame': 'timber frame unknown insulation',
'Cavity Wall': 'cavity unknown insulation',
@ -166,5 +165,29 @@ WALL_CONSTRUCTION_MAPPINGS = {
'PRC': 'system built',
'Cross Wall': 'system built',
'Solid Wall': 'solid brick unknown insulation',
'Traditional': 'other'
'Traditional': 'other',
'Solid': 'solid brick unknown insulation',
'Wates no fines': 'system built',
'Concrete Frame': 'system built',
'PRCWATES': 'system built',
'Refurbished Cornish': 'system built',
'Bailey Stratton': 'other',
'Refurbished Reema': 'system built',
'PRCREEMA': 'system built',
'Trustsell Type': 'system built',
'Petra Nissan': 'unknown',
'Reinstated Airey': 'system built',
'Refurbished Airey': 'system built',
# From Abri- slightly unclear on types but not a large portion of the data
'No Fines Type': 'system built',
'Refurbished Unity': 'system built',
'Timber Framed': 'timber frame unknown insulation',
'Refurbished Woolaway': 'system built',
'Modern Methods of Construction': 'other',
'BISF - Brit Iron & Steel Federation': 'system built',
'Steel Framed': 'system built',
'Timber Framed with confirmed Fire Stopping': 'timber frame unknown insulation',
'Sipporex': 'system built'
}

203
etl/customers/l_and_g/risk_matrix.py Normal file
View file

@ -0,0 +1,203 @@
from itertools import product
from recommendations.recommendation_utils import estimate_external_wall_area, estimate_windows
import numpy as np
import pandas as pd
def app():
# Given a combination of variables, this code attempts to break down the costs of works to achieve upgrade
# targets
upgrade_path = [
"wall_insulation", "roof_insulation", "ventilation", "windows", "low_energy_lighting",
"heating", "solar"
]
pricing_matrix = {
"cavity_wall_insulation": 14.5,
"ventilation": 350,
"room_roof_insulation": 210,
"loft_insulation": 15,
"internal_wall_insulation": 215,
"external_wall_insulation": 298.35,
"low_energy_lighting": 35, # per light
"flat_roof_insulation": 195,
"double_glazing": 1140,
"secondary_glazing": 970,
"air_source_heat_pump": 16500,
"solar_pv": 6200,
"high_heat_retention_storage": 1000, # per heater
}
dwelling_types = [
"Semi Detached House",
"Detached House",
"Mid Terrace House",
"Mid Floor Flat",
"Top Floor Flat",
"Ground Floor Flat"
]
num_floors_map = {
"Semi Detached House": 2,
"Detached House": 2,
"Mid Terrace House": 2,
"Mid Floor Flat": 1,
"Top Floor Flat": 1,
"Ground Floor Flat": 1
}
built_form_map = {
"Semi Detached House": "Semi-Detached",
"Detached House": "Detached",
"Mid Terrace House": "Mid Terrace",
"Mid Floor Flat": "Semi-Detached",
"Top Floor Flat": "Semi-Detached",
"Ground Floor Flat": "Semi-Detached"
}
lighting_count = {
"Semi Detached House": 15,
"Detached House": 19,
"Mid Terrace House": 12,
"Mid Floor Flat": 10,
"Top Floor Flat": 10,
"Ground Floor Flat": 10
}
# If we have a flat, we won't use the 199m2 floor area
floor_areas = [73, 97, 199]
# We remove age bracket, as we ended up with 360 combinations
# age_brackets = ["1945-1970", "1971-2002", "Post 2002"]
wall_type = ["cavity", "non-cavity"]
roof_type = ["pitched", "other"]
planning_constraints = [True, False]
# This is the list of all combinations of the above variables
combinations_untrimmed = product(
*[
dwelling_types, floor_areas, wall_type, roof_type, planning_constraints
]
)
# TODO: Possibly need to add an additional cost for immersion hot water
combinations = []
for comb in combinations_untrimmed:
if "Flat" in comb[0] and comb[1] == 199:
continue
# If we have a flat, not too much difference if it's in a conservation area or not
if "Flat" in comb[0] and comb[4] is True:
continue
combinations.append(comb)
risk_matrix = []
for combination in combinations:
n_floors = num_floors_map[combination[0]]
bf = built_form_map[combination[0]]
pt = "House" if "Flat" not in combination[0] else "Flat"
# Model the home as a box
ground_floor_area = combination[1] / n_floors
perimeter = np.sqrt(ground_floor_area) * 4
# This is the amount of insulation required
external_wall_area = estimate_external_wall_area(
num_floors=n_floors,
floor_height=2.5,
perimeter=perimeter,
built_form=bf
)
n_rooms = np.floor(combination[1] / 15)
n_windows = estimate_windows(
property_type=pt,
built_form=bf,
construction_age_band="",
floor_area=combination[1],
number_habitable_rooms=n_rooms
)
# We determine the exact upgrade pathway for this combination, guided by the generic upgrade pathway
combination_upgrade_pathway = []
for upgrade in upgrade_path:
if upgrade == "wall_insulation":
if combination[2] == "cavity":
combination_upgrade_pathway.append("cavity_wall_insulation")
else:
combination_upgrade_pathway.append("solid_wall_insulation")
continue
if upgrade == "roof_insulation":
if combination[3] == "pitched":
combination_upgrade_pathway.append("loft_insulation")
else:
combination_upgrade_pathway.append("non_pitched_roof_insualtion")
continue
if upgrade == "ventilation":
combination_upgrade_pathway.append("ventilation")
continue
if upgrade == "low_energy_lighting":
combination_upgrade_pathway.append("low_energy_lighting")
continue
if upgrade == "windows":
if not combination[4]:
combination_upgrade_pathway.append("double_glazing")
else:
combination_upgrade_pathway.append("secondary_glazing")
continue
if upgrade == "heating":
if combination[0] in ["Semi Detached House", "Detached House"]:
combination_upgrade_pathway.append("high_heat_retention_storage")
else:
combination_upgrade_pathway.append("air_source_heat_pump")
continue
if upgrade == "solar":
if combination[0] in ["Semi Detached House", "Detached House", "Mid Terrace House"]:
combination_upgrade_pathway.append("solar_pv")
continue
combination_costs = []
for measure in combination_upgrade_pathway:
unit_cost = pricing_matrix[measure]
# Wall insulation
if measure in ["cavity_wall_insulation", "internal_wall_insulation", "external_wall_insulation"]:
cost = unit_cost * external_wall_area
elif measure in ["loft_insulation"]:
cost = unit_cost * ground_floor_area
elif measure == "ventilation":
if combination[1] == 73:
cost = unit_cost * 2
elif combination[1] == 97:
cost = unit_cost * 3
else:
cost = unit_cost * 4
elif measure == "low_energy_lighting":
n_lights = lighting_count[combination[0]]
if combination[1] == 73:
inflation = 1
elif combination[1] == 97:
inflation = 1.2
else:
inflation = 1.5
cost = unit_cost * n_lights * inflation
elif measure in ["double_glazing", "secondary_glazing"]:
cost = unit_cost * n_windows
elif measure == "high_heat_retention_storage":
cost = unit_cost * n_rooms
elif measure in ["air_source_heat_pump", "solar_pv"]:
cost = unit_cost
else:
raise NotImplementedError("Implement: %s" % measure)
combination_costs.append(
{
"measure": measure,
"cost": cost
}
)
combination_costs = pd.DataFrame(combination_costs)