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

adding measure map so we can specific certain measures

Browse source
This commit is contained in:
Khalim Conn-Kowlessar 2024-09-05 12:16:25 +01:00
parent 984abe7292
commit 85eaeccad8
6 changed files with 341 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -620,6 +620,13 @@ async def trigger_plan(body: PlanTriggerRequest):
if individual_units:
# Model the solar potential at the property level
for unit in tqdm(individual_units):
# TODO: Tidy up this code
# We don't need to do this if we have global inclusions that don't include solar
if body.inclusions:
if "solar_pv" not in body.inclusions:
continue
property_instance = [p for p in input_properties if p.id == unit["property_id"]][0]
# At this level, we check if the property is suitable for solar and if now, skip
if not property_instance.is_solar_pv_valid():
@ -668,7 +675,9 @@ async def trigger_plan(body: PlanTriggerRequest):
recommendations_scoring_data = []
representative_recommendations = {}
for p in tqdm(input_properties):
recommender = Recommendations(property_instance=p, materials=materials, exclusions=body.exclusions)
recommender = Recommendations(
property_instance=p, materials=materials, exclusions=body.exclusions, inclusions=body.inclusions
)
property_recommendations, property_representative_recommendations = recommender.recommend()
if not property_recommendations:

79
backend/app/plan/schemas.py
View file

@ -1,6 +1,53 @@
from pydantic import BaseModel, conlist, validator
from typing import Optional
TYPICAL_MEASURE_TYPES = [
"wall_insulation",
"roof_insulation",
"ventilation",
"floor_insulation",
"windows",
"fireplace",
"heating",
"hot_water",
"low_energy_lighting",
"secondary_heating",
"solar_pv"
]
SPECIFIC_MEASURES = [
# Specific measures
# Walls
"internal_wall_insulation",
"external_wall_insulation",
"cavity_wall_insulation"
# Roof
"loft_insulation",
"flat_roof_insulation",
"room_roof_insulation",
# Floor
"suspended_floor_insulation",
"solid_floor_insulation",
# Heating
"boiler_upgrade",
"high_heat_retention_storage_heater",
"air_source_heat_pump",
# Specific measures that will typically come from an energy assessment
"trickle_vents",
"draught_proofing",
"mixed_glazing", # This covers partial double glazing and secondary glazing
]
# This allows us to extend high level categories for measures such as "wall_insulation" to the specific measures
# such as "external_wall_insulation", "internal_wall_insulation", "cavity_wall_insulation"
MEASURE_MAP = {
"wall_insulation": ["internal_wall_insulation", "external_wall_insulation", "cavity_wall_insulation"],
"roof_insulation": ["loft_insulation", "flat_roof_insulation", "room_roof_insulation"],
"floor_insulation": ["suspended_floor_insulation", "solid_floor_insulation"],
"heating": ["boiler_upgrade", "high_heat_retention_storage_heater", "air_source_heat_pump"],
}
class PlanTriggerRequest(BaseModel):
budget: Optional[float] = None
@ -13,33 +60,13 @@ class PlanTriggerRequest(BaseModel):
patches_file_path: Optional[str] = None
non_invasive_recommendations_file_path: Optional[str] = None
exclusions: Optional[conlist(str, min_items=1)] = None
inclusions: Optional[conlist(str, min_items=1)] = None
scenario_name: Optional[str] = ""
# If true, will allow us to create multiple plans for the same portfolio, whereas if this is false, if this property
# exists in the portfolio, it will be ignored
multi_plan: Optional[bool] = False
# Pre-defined list of possibilities for exclusions
_allowed_exclusions = {
# Measure classes
"wall_insulation",
"ventilation",
"roof_insulation",
"floor_insulation",
"windows",
"fireplace",
"heating",
"hot_water",
"lighting",
"solar_pv",
# Specific measures
"air_source_heat_pump",
"internal_wall_insulation",
"external_wall_insulation",
"secondary_heating",
"boiler_upgrade",
"high_heat_retention_storage_heater",
}
_allowed_goals = {"Increasing EPC"}
_allowed_housing_types = {"Social", "Private"}
@ -47,10 +74,16 @@ class PlanTriggerRequest(BaseModel):
# Validator to ensure exclusions are within the pre-defined possibilities
@validator('exclusions', each_item=True)
def check_exclusions(cls, v):
if v not in cls._allowed_exclusions:
if v not in TYPICAL_MEASURE_TYPES + SPECIFIC_MEASURES:
raise ValueError(f"{v} is not an allowed exclusion")
return v
@validator('inclusions', each_item=True)
def check_inclusions(cls, v):
if v not in TYPICAL_MEASURE_TYPES + SPECIFIC_MEASURES:
raise ValueError(f"{v} is not an allowed inclusion")
return v
# Validator to ensure that the goal is within the pre-defined possibilities
@validator('goal')
def check_goal(cls, v):

25
etl/customers/bcc_tender/app.py
View file

@ -95,6 +95,31 @@ epc_data["eligibility_type"] = np.where(
epc_data["eligibility_type"]
)
# Example EPCS to analysis
analysis_epcs = epc_data[~pd.isnull(epc_data["eligibility_type"])].copy()
# Keep just columns we need
analysis_epcs = analysis_epcs[
[
"UPRN", "TENURE", "CURRENT_ENERGY_RATING", "WALLS_DESCRIPTION", "ROOF_DESCRIPTION",
"CONSTRUCTION_AGE_BAND", "TOTAL_FLOOR_AREA", "PROPERTY_TYPE", "BUILT_FORM", "MAINHEAT_DESCRIPTION",
"eligibility_type",
]
]
analysis_epcs["grouped_epc_band"] = np.where(
analysis_epcs["CURRENT_ENERGY_RATING"].isin(["D"]),
"EPC D",
"EPC E-G"
)
analysis_epcs.to_csv("/Users/khalimconn-kowlessar/Documents/hestia/Customers/bcc tender/analysis_epcs.csv", index=False)
# Create aggregations and we store this information
agg_cols = ["CURRENT_ENERGY_RATING", "CONSTRUCTION_AGE_BAND", "PROPERTY_TYPE", "BUILT_FORM", "grouped_epc_band"]
agg_cols = ["WALLS_DESCRIPTION", "ROOF_DESCRIPTION", "MAINHEAT_DESCRIPTION"]
for col in agg_cols:
agg_df = analysis_epcs.groupby([col]).size().reset_index(name="Number of Properties")
agg_df["Percentage of Properties"] = 100 * agg_df["Number of Properties"] / agg_df["Number of Properties"].sum()
agg_df.to_csv(f"/Users/khalimconn-kowlessar/Documents/hestia/Customers/bcc tender/{col}.csv", index=False)
# Eligibiilty 6: GBIS General Eligibility, Social - tenure is social rented and EPC rating D-G, but also the property
# should be rented out below market rate
# This is a subset of Eligibility 3 - we likely don't need to do any scaling

196
etl/customers/vectis/outputs.py Normal file
View file

@ -0,0 +1,196 @@
import pandas as pd
from utils.s3 import save_csv_to_s3
def app():
# This is the payload to be used to extract the energy assessment data from s3 and upload it to the database,
# as well as produce links to each of the uploaded documents.
portfolio_id = 101
body = {
"portfolio_id": portfolio_id,
"surveyor": "JAFFERSONS ENERGY CONSULTANTS",
"project_code": "VEC001",
}
# These are the recommendations based on the on-site survey of the property.
non_intrusive_recommendations = [
{
# 2 Grove Mansions
"uprn": 121016121,
"recommendations": [
{
"type": "draught_proofing",
"cost": 123,
"survey": True,
"sap_points": 1
},
{
"type": "mixed_glazing", "cost": 12345, "survey": True,
"description": "Install double glazing to north facing windows and secondary glazing to the "
"remaining windows at the front of the building",
"sap_points": 3
},
{"type": "trickle_vents", "cost": 500, "survey": True},
{"type": "suspended_floor_insulation", "cost": None, "survey": True, "sap_points": 2},
{"type": "internal_wall_insulation", "cost": None, "survey": True, "sap_points": 5},
]
},
{
# 8 Grove Mansions
"uprn": 10024087855,
"recommendations": [
{"type": "draught_proofing", "cost": 123, "survey": True, "sap_points": 2},
{
"type": "mixed_glazing", "cost": 12345, "survey": True,
"description": "Install double glazing to north facing windows and secondary glazing to the "
"remaining windows at the front of the building",
"sap_points": 4
},
{"type": "trickle_vents", "cost": 500, "survey": True},
{"type": "low_energy_lighting", "cost": None, "survey": True, "sap_points": 0},
{"type": "internal_wall_insulation", "cost": None, "survey": True, 'sap_points': 5},
]
},
{
# 9 Grove Mansions
"uprn": 121016128,
"recommendations": [
{"type": "draught_proofing", "cost": 123, "survey": True, "sap_points": 1},
{
"type": "mixed_glazing", "cost": 12345, "survey": True,
"description": "Install double glazing to north facing windows and secondary glazing to the "
"remaining windows at the front of the building",
"sap_points": 3
},
{"type": "trickle_vents", "cost": 500, "survey": True},
{"type": "low_energy_lighting", "cost": None, "survey": True, "sap_points": 1},
{"type": "suspended_floor_insulation", "cost": None, "sap_points": 1},
{"type": "internal_wall_insulation", "cost": None, "survey": True, "sap_points": 6},
]
},
{
# 5 Grove Mansions
"uprn": 121016124,
"recommendations": [
{
"type": "mixed_glazing", "cost": 12345, "survey": True,
"description": "Install double glazing to north facing windows and secondary glazing to the "
"remaining windows at the front of the building",
"sap_points": 5
},
{"type": "trickle_vents", "cost": 500, "survey": True},
{"type": "low_energy_lighting", "cost": None, "survey": True, "sap_points": 2},
{"type": "internal_wall_insulation", "cost": None, "survey": True, "sap_points": 8},
]
},
{
# 14 Grove Mansions
"uprn": 121016117,
"recommendations": [
{"type": "draught_proofing", "cost": 123, "survey": True, "sap_points": 1},
{
"type": "mixed_glazing", "cost": 12345, "survey": True,
"description": "Install double glazing to north facing windows and secondary glazing to the "
"remaining windows at the front of the building",
"sap_points": 4
},
{"type": "trickle_vents", "cost": 500, "survey": True},
{"type": "low_energy_lighting", "cost": None, "survey": True, "sap_points": 1},
{"type": "internal_wall_insulation", "cost": None, "survey": True, "sap_points": 6},
]
},
{
# 19 Grove Mansions
"uprn": 10024087902,
"recommendations": [
{"type": "low_energy_lighting", "cost": None, "survey": True, "sap_points": 0},
{"type": "internal_wall_insulation", "cost": None, "survey": True, "sap_points": 2},
{"type": "room_roof_insulation", "cost": None, "survey": True, "sap_points": 16},
]
},
]
asset_list = [
{
"uprn": 121016121, "address": "", "postcode": ""
},
{
"uprn": 10024087855, "address": "", "postcode": ""
},
{
"uprn": 121016128, "address": "", "postcode": ""
},
{
"uprn": 121016124, "address": "", "postcode": ""
},
{
"uprn": 121016117, "address": "", "postcode": ""
},
{
"uprn": 10024087902, "address": "", "postcode": ""
},
]
asset_list = pd.DataFrame(asset_list)
filename = f"{8}/{portfolio_id}/asset_list.csv"
save_csv_to_s3(
dataframe=asset_list,
bucket_name="retrofit-plan-inputs-dev",
file_name=filename
)
# TODO Create asset list
# TODO: Store asset list & non_intrusive_recommendations
# Store non-invasive recommendations in S3
non_invasive_recommendations_filename = f"{8}/{portfolio_id}/non_invasive_recommendations.json"
save_csv_to_s3(
dataframe=pd.DataFrame(non_intrusive_recommendations),
bucket_name="retrofit-plan-inputs-dev",
file_name=non_invasive_recommendations_filename
)
# This is the first scenario which includes the first batch of recommendations
body1 = {
"portfolio_id": str(portfolio_id),
"housing_type": "Private",
"goal": "Increasing EPC",
"goal_value": "A",
"trigger_file_path": filename,
"already_installed_file_path": "",
"patches_file_path": "",
"non_invasive_recommendations_file_path": non_invasive_recommendations_filename,
"inclusions": [
"draught_proofing", "mixed_glazing", "trickle_vents", "low_energy_lighting",
],
"budget": None,
"scenario_name": "Quick wins - do now while tenanted",
"multi_plan": True,
}
# This is the second scenario which includes the second batch of recommendations
body2 = {
"portfolio_id": str(portfolio_id),
"housing_type": "Private",
"goal": "Increasing EPC",
"goal_value": "A",
"trigger_file_path": filename,
"already_installed_file_path": "",
"patches_file_path": "",
"non_invasive_recommendations_file_path": non_invasive_recommendations_filename,
"inclusions": [
"draught_proofing",
"mixed_glazing",
"trickle_vents",
"low_energy_lighting",
"suspended_floor_insulation",
"internal_wall_insulation"
],
"budget": None,
"scenario_name": "Do when void",
"multi_plan": True,
}
print(body1)
print(body2)

57
recommendations/Recommendations.py
View file

@ -17,6 +17,7 @@ from recommendations.SecondaryHeating import SecondaryHeating
from backend.ml_models.AnnualBillSavings import AnnualBillSavings
from backend.apis.GoogleSolarApi import GoogleSolarApi
import backend.app.assumptions as assumptions
from backend.app.plan.schemas import TYPICAL_MEASURE_TYPES, SPECIFIC_MEASURES, MEASURE_MAP
ASHP_COP = 3
STARTING_DUMMY_ID_VALUE = -9999
@ -32,15 +33,24 @@ class Recommendations:
property_instance: Property,
materials: List,
exclusions: List[str] = None,
inclusions: List[str] = None,
):
"""
:param property_instance: Instance of the Property class, for the home associated to property_id
:param materials: List of materials to be used in the recommendations
:param exclusions: List of specific measures or measure types to exclude from recommendations. Defaulted to
None, meaning no exclusions to be applied
:param inclusions: List of specific measures of measure types to include. Defaulted to None, meaning all
measures are included
"""
self.property_instance = property_instance
self.materials = materials
self.exclusions = exclusions if exclusions else []
self.inclusions = inclusions if inclusions else []
self.all_typical_measures = TYPICAL_MEASURE_TYPES
self.all_specific_measures = SPECIFIC_MEASURES
self.floor_recommender = FloorRecommendations(property_instance=property_instance, materials=materials)
self.wall_recomender = WallRecommendations(property_instance=property_instance, materials=materials)
@ -56,6 +66,24 @@ class Recommendations:
self.hotwater_recommender = HotwaterRecommendations(property_instance=property_instance)
self.secondary_heating_recommender = SecondaryHeating(property_instance=property_instance)
def find_included_measures(self):
"""
Determines the set of measures to be included in recommendations
"""
inclusions_full = [MEASURE_MAP[x] if x in MEASURE_MAP else x for x in self.inclusions]
exclusions_full = [MEASURE_MAP[x] if x in MEASURE_MAP else x for x in self.exclusions]
if inclusions_full and exclusions_full:
# All typical measures
return self.all_specific_measures
if inclusions_full:
return inclusions_full
if exclusions_full:
return [m for m in self.all_specific_measures if m not in exclusions_full]
def recommend(self):
"""
@ -68,15 +96,20 @@ class Recommendations:
property_recommendations = []
phase = 0
measures = self.find_included_measures()
# Building Fabric
if "wall_insulation" not in self.exclusions:
self.wall_recomender.recommend(phase=phase, exclusions=self.exclusions)
if (
("wall_insulation" in measures) or
("internal_wall_insulation" in measures) or
("external_wall_insulation" in measures)
):
self.wall_recomender.recommend(phase=phase, measures=measures)
if self.wall_recomender.recommendations:
property_recommendations.append(self.wall_recomender.recommendations)
phase += 1
if "roof_insulation" not in self.exclusions:
if "roof_insulation" in measures:
self.roof_recommender.recommend(phase=phase)
if self.roof_recommender.recommendations:
property_recommendations.append(self.roof_recommender.recommendations)
@ -90,32 +123,32 @@ class Recommendations:
# real impact on the SAP score. Therefore, we don't need to include phasing for ventilation. If we
# have any
# wall or roof recommendations, we will ensure that ventilation is included in the simulation
if "ventilation" not in self.exclusions:
if "ventilation" in measures:
if self.wall_recomender.recommendations or self.roof_recommender.recommendations:
self.ventilation_recomender.recommend()
if self.ventilation_recomender.recommendation:
property_recommendations.append(self.ventilation_recomender.recommendation)
if "floor_insulation" not in self.exclusions:
if "floor_insulation" in measures:
self.floor_recommender.recommend(phase=phase)
if self.floor_recommender.recommendations:
property_recommendations.append(self.floor_recommender.recommendations)
phase += 1
if "windows" not in self.exclusions:
if "windows" in measures:
self.windows_recommender.recommend(phase=phase)
if self.windows_recommender.recommendation:
property_recommendations.append(self.windows_recommender.recommendation)
phase += 1
if "fireplace" not in self.exclusions:
if "fireplace" in measures:
self.fireplace_recommender.recommend(phase=phase)
if self.fireplace_recommender.recommendation:
property_recommendations.append(self.fireplace_recommender.recommendation)
phase += 1
# Heating and Electical systems
if "heating" not in self.exclusions:
if "heating" in measures:
cavity_or_loft_recommendations = [
r for r in self.wall_recomender.recommendations + self.roof_recommender.recommendations
@ -167,26 +200,26 @@ class Recommendations:
phase += amount_to_increment
# Hot water
if "hot_water" not in self.exclusions:
if "hot_water" in measures:
self.hotwater_recommender.recommend(phase=phase)
if self.hotwater_recommender.recommendations:
property_recommendations.append(self.hotwater_recommender.recommendations)
phase += 1
if "lighting" not in self.exclusions:
if "low_energy_lighting" in measures:
self.lighting_recommender.recommend(phase=phase)
if self.lighting_recommender.recommendation:
property_recommendations.append(self.lighting_recommender.recommendation)
phase += 1
if "secondary_heating" not in self.exclusions:
if "secondary_heating" in measures:
self.secondary_heating_recommender.recommend(phase=phase)
if self.secondary_heating_recommender.recommendation:
property_recommendations.append(self.secondary_heating_recommender.recommendation)
phase += 1
# Renewables
if "solar_pv" not in self.exclusions:
if "solar_pv" in measures:
self.solar_recommender.recommend(phase=phase)
if self.solar_recommender.recommendation:
property_recommendations.append(self.solar_recommender.recommendation)

15
recommendations/WallRecommendations.py
View file

@ -190,7 +190,7 @@ class WallRecommendations(Definitions):
return ewi_recommendations
def recommend(self, phase=0, exclusions=None):
def recommend(self, phase=0, measures=None):
# if building built after 1990 + we're able to identify U-value +
# U-value less than 0.18 and if in or close to a conversation area,
# recommend internal wall insulation as a possible measure
@ -268,7 +268,7 @@ class WallRecommendations(Definitions):
# Remaining wall types are treated with IWI or EWI
if (u_value >= self.BUILDING_REGULATIONS_PART_L_MAX_U_VALUE) and self.is_suitable_for_solid_insulation():
self.find_insulation(u_value, phase, exclusions=exclusions)
self.find_insulation(u_value, phase, measures=measures)
return
# If the u-value is within regulations, we don't do anything
@ -558,7 +558,7 @@ class WallRecommendations(Definitions):
return recommendations
def find_insulation(self, u_value, phase, exclusions=None):
def find_insulation(self, u_value, phase, measures=None):
"""
This function contains the logic for finding potential insulation measures for a property, depending
on the parts available and whether the property can have external wall insulation installed
@ -570,10 +570,13 @@ class WallRecommendations(Definitions):
# we separate the logic for for recommending them, therefore we don't
# consider diminishing returns between the two as they are considered to be separate measures
exclusions = [] if exclusions is None else exclusions
if measures is None:
ewi_valid = self.ewi_valid()
else:
ewi_valid = self.ewi_valid() and "external_wall_insulation" in measures
ewi_recommendations = []
if self.ewi_valid() and "external_wall_insulation" not in exclusions:
if ewi_valid:
ewi_recommendations = self._find_insulation(
u_value=u_value,
insulation_materials=pd.DataFrame(
@ -584,7 +587,7 @@ class WallRecommendations(Definitions):
)
iwi_recommendations = []
if "internal_wall_insulation" not in exclusions:
if "internal_wall_insulation" in measures:
iwi_recommendations = self._find_insulation(
u_value=u_value,
insulation_materials=pd.DataFrame(self.internal_wall_insulation_materials),