mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
204 lines
8.1 KiB
Python
204 lines
8.1 KiB
Python
import pandas as pd
|
|
from backend.Property import Property
|
|
from collections import defaultdict
|
|
from utils.s3 import read_from_s3
|
|
|
|
from recommendations.config import UPGRADES_MAP
|
|
from recommendations.recommendation_utils import get_wall_u_value, get_floor_u_value, get_roof_u_value
|
|
|
|
from backend.app.db.utils import row2dict
|
|
from backend.app.config import get_settings
|
|
import msgpack
|
|
|
|
|
|
def filter_materials(materials):
|
|
materials_by_type = defaultdict(list)
|
|
|
|
mapping = {
|
|
"walls": ["internal_wall_insulation", "external_wall_insulation", "cavity_wall_insulation"],
|
|
"floor": ["suspended_floor_insulation", "solid_floor_insulation"]
|
|
}
|
|
|
|
materials = [row2dict(material) for material in materials]
|
|
|
|
for component, types in mapping.items():
|
|
materials_by_type[component] = [part for part in materials if part["type"] in types]
|
|
|
|
return dict(materials_by_type)
|
|
|
|
|
|
def insert_temp_recommendation_id(property_recommendations):
|
|
"""
|
|
Creates a temporary recommendation id which is needed for
|
|
filtering recommendations between default and no, after the optimiser has been
|
|
run
|
|
:param property_recommendations: nested list of recommendations, grouped by data_types
|
|
:return: Updated recommendations_to_upload, where where recommendation has a "recommendation_id"
|
|
integer inserted
|
|
"""
|
|
idx = 0
|
|
|
|
for recs in property_recommendations:
|
|
for rec in recs:
|
|
rec["recommendation_id"] = idx
|
|
idx += 1
|
|
|
|
return property_recommendations
|
|
|
|
|
|
def get_cleaned():
|
|
"""
|
|
This function will retrieve the cleaned dataset from s3 which has the cleaned
|
|
descriptions for the epc dataset
|
|
|
|
This data is stored in MessagePack format and therefore needs to be decoded
|
|
:return:
|
|
"""
|
|
|
|
cleaned = read_from_s3(
|
|
s3_file_name="cleaned_epc_data/cleaned.bson",
|
|
bucket_name="retrofit-data-{environment}".format(environment=get_settings().ENVIRONMENT)
|
|
)
|
|
|
|
cleaned = msgpack.unpackb(cleaned, raw=False)
|
|
|
|
return cleaned
|
|
|
|
|
|
def create_recommendation_scoring_data(
|
|
property: Property,
|
|
recommendation: dict,
|
|
starting_epc_data: pd.DataFrame,
|
|
ending_epc_data: pd.DataFrame,
|
|
fixed_data: pd.DataFrame,
|
|
):
|
|
"""
|
|
This wrapper function prepares data to be passed to the sap model api
|
|
:return:
|
|
"""
|
|
|
|
scoring_dict = {
|
|
"UPRN": property.data["uprn"],
|
|
"id": "+".join([str(property.id), str(recommendation["recommendation_id"])]),
|
|
"LOCAL_AUTHORITY": property.data["local-authority"],
|
|
**starting_epc_data.to_dict("records")[0],
|
|
**ending_epc_data.to_dict("records")[0],
|
|
**fixed_data.to_dict("records")[0]
|
|
}
|
|
|
|
# Set staring u-values if we don't have them
|
|
if not scoring_dict["walls_thermal_transmittance"]:
|
|
scoring_dict["walls_thermal_transmittance"] = get_wall_u_value(
|
|
clean_description=property.walls["clean_description"],
|
|
age_band=property.age_band,
|
|
is_granite_or_whinstone=property.walls["is_granite_or_whinstone"],
|
|
is_sandstone_or_limestone=property.walls["is_sandstone_or_limestone"]
|
|
)
|
|
|
|
if not scoring_dict["floor_thermal_transmittance"]:
|
|
scoring_dict["floor_thermal_transmittance"] = get_floor_u_value(
|
|
floor_type=property.floor_type,
|
|
area=property.floor_area,
|
|
perimeter=property.perimeter,
|
|
wall_type=property.wall_type,
|
|
insulation_thickness=property.floor["insulation_thickness"],
|
|
age_band=property.age_band,
|
|
)
|
|
|
|
if not scoring_dict["roof_thermal_transmittance"]:
|
|
scoring_dict["roof_thermal_transmittance"] = get_roof_u_value(
|
|
insulation_thickness=property.roof["insulation_thickness"],
|
|
has_dwelling_above=property.roof["has_dwelling_above"],
|
|
is_loft=property.roof["is_loft"],
|
|
is_roof_room=property.roof["is_roof_room"],
|
|
is_thatched=property.roof["is_thatched"],
|
|
age_band=property.age_band,
|
|
is_flat=property.roof["is_flat"],
|
|
is_pitched=property.roof["is_pitched"],
|
|
is_at_rafters=property.roof["is_at_rafters"],
|
|
)
|
|
|
|
# Tidy up insulation thicknesses, making sure it isn't None
|
|
if scoring_dict["walls_insulation_thickness"] is None:
|
|
scoring_dict["walls_insulation_thickness"] = "none"
|
|
|
|
if scoring_dict["floor_insulation_thickness"] is None:
|
|
scoring_dict["floor_insulation_thickness"] = "none"
|
|
|
|
if scoring_dict["roof_insulation_thickness"] is None:
|
|
scoring_dict["roof_insulation_thickness"] = "none"
|
|
|
|
# We update the description to indicate it's insulated
|
|
if recommendation["type"] == "wall_insulation":
|
|
# The upgrade made here is to the u-value of the walls and the description of the
|
|
# insulation thickness
|
|
# We may not have the u-value initially, so we calculate it
|
|
scoring_dict["walls_thermal_transmittance_ENDING"] = get_wall_u_value(
|
|
clean_description=UPGRADES_MAP[property.walls["clean_description"]],
|
|
age_band=property.age_band,
|
|
is_granite_or_whinstone=property.walls["is_granite_or_whinstone"],
|
|
is_sandstone_or_limestone=property.walls["is_sandstone_or_limestone"]
|
|
)
|
|
scoring_dict["walls_insulation_thickness_ENDING"] = "above average"
|
|
else:
|
|
if not scoring_dict["walls_thermal_transmittance_ENDING"]:
|
|
scoring_dict["walls_thermal_transmittance_ENDING"] = get_wall_u_value(
|
|
clean_description=property.walls["clean_description"],
|
|
age_band=property.age_band,
|
|
is_granite_or_whinstone=property.walls["is_granite_or_whinstone"],
|
|
is_sandstone_or_limestone=property.walls["is_sandstone_or_limestone"]
|
|
)
|
|
|
|
if scoring_dict["walls_insulation_thickness_ENDING"] is None:
|
|
scoring_dict["walls_insulation_thickness_ENDING"] = "none"
|
|
|
|
# Update description to indicate it's insulate
|
|
if recommendation["type"] == "floor_insulation":
|
|
|
|
if len(recommendation["parts"]) > 1:
|
|
raise NotImplementedError("Have more than 1 floor insulation part - handle this case")
|
|
|
|
scoring_dict["floor_thermal_transmittance_ENDING"] = get_floor_u_value(
|
|
floor_type=property.floor_type,
|
|
area=property.floor_area,
|
|
perimeter=property.perimeter,
|
|
wall_type=property.wall_type,
|
|
insulation_thickness=recommendation["parts"][0]["depths"][0],
|
|
age_band=property.age_band,
|
|
)
|
|
# We don't really see above average for this in the training data
|
|
scoring_dict["floor_insulation_thickness_ENDING"] = "average"
|
|
else:
|
|
if not scoring_dict["floor_thermal_transmittance_ENDING"]:
|
|
scoring_dict["floor_thermal_transmittance_ENDING"] = get_floor_u_value(
|
|
floor_type=property.floor_type,
|
|
area=property.floor_area,
|
|
perimeter=property.perimeter,
|
|
wall_type=property.wall_type,
|
|
insulation_thickness=property.floor["insulation_thickness"],
|
|
age_band=property.age_band,
|
|
)
|
|
|
|
if scoring_dict["floor_insulation_thickness_ENDING"] is None:
|
|
scoring_dict["floor_insulation_thickness_ENDING"] = "none"
|
|
|
|
if recommendation["type"] not in ["wall_insulation", "floor_insulation"]:
|
|
raise NotImplementedError("Implement me")
|
|
|
|
if not scoring_dict["roof_thermal_transmittance_ENDING"]:
|
|
scoring_dict["roof_thermal_transmittance_ENDING"] = get_roof_u_value(
|
|
insulation_thickness=property.roof["insulation_thickness"],
|
|
has_dwelling_above=property.roof["has_dwelling_above"],
|
|
is_loft=property.roof["is_loft"],
|
|
is_roof_room=property.roof["is_roof_room"],
|
|
is_thatched=property.roof["is_thatched"],
|
|
age_band=property.age_band,
|
|
is_flat=property.roof["is_flat"],
|
|
is_pitched=property.roof["is_pitched"],
|
|
is_at_rafters=property.roof["is_at_rafters"],
|
|
)
|
|
|
|
if scoring_dict["roof_insulation_thickness_ENDING"] is None:
|
|
scoring_dict["roof_insulation_thickness_ENDING"] = "none"
|
|
|
|
return scoring_dict
|