mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
198 lines
8 KiB
Python
198 lines
8 KiB
Python
import pandas as pd
|
|
from backend.Property import Property
|
|
from utils.s3 import read_from_s3
|
|
|
|
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 prepare_materials(materials):
|
|
"""
|
|
This function will prepare the materials for recommendations
|
|
:param materials: list of materials, as retrieved from the database
|
|
:return:
|
|
"""
|
|
return [row2dict(material) for material in materials]
|
|
|
|
|
|
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 scoring_dict["walls_thermal_transmittance"] is None:
|
|
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 scoring_dict["floor_thermal_transmittance"] is None:
|
|
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 scoring_dict["roof_thermal_transmittance"] is None:
|
|
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"],
|
|
)
|
|
|
|
for col in [
|
|
"walls_insulation_thickness", "floor_insulation_thickness", "roof_insulation_thickness"
|
|
]:
|
|
if scoring_dict[col] is None:
|
|
scoring_dict[col] = "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
|
|
scoring_dict["walls_thermal_transmittance_ENDING"] = recommendation["new_u_value"]
|
|
scoring_dict["walls_insulation_thickness_ENDING"] = "above average"
|
|
scoring_dict["WALLS_ENERGY_EFF_ENDING"] = "Good"
|
|
else:
|
|
if scoring_dict["walls_thermal_transmittance_ENDING"] is None:
|
|
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"] = recommendation["new_u_value"]
|
|
# We don't really see above average for this in the training data
|
|
scoring_dict["floor_insulation_thickness_ENDING"] = "average"
|
|
scoring_dict["FLOOR_ENERGY_EFF_ENDING"] = "Good"
|
|
else:
|
|
if scoring_dict["floor_thermal_transmittance_ENDING"] is None:
|
|
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"] == "roof_insulation":
|
|
scoring_dict["roof_thermal_transmittance_ENDING"] = recommendation["new_u_value"]
|
|
|
|
parts = recommendation["parts"]
|
|
if len(parts) != 1:
|
|
raise ValueError("More than one part for roof insulation - investiage me")
|
|
|
|
scoring_dict["roof_insulation_thickness_ENDING"] = str(parts[0]["depths"][0])
|
|
scoring_dict["ROOF_ENERGY_EFF_ENDING"] = "Very Good"
|
|
else:
|
|
# Fill missing roof u-values - this fill is not based on recommended upgrades
|
|
if scoring_dict["roof_thermal_transmittance_ENDING"] is None:
|
|
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"
|
|
|
|
if recommendation["type"] == "mechanical_ventilation":
|
|
scoring_dict["MECHANICAL_VENTILATION_ENDING"] = 'mechanical, extract only'
|
|
|
|
if recommendation["type"] == "sealing_open_fireplace":
|
|
scoring_dict["NUMBER_OPEN_FIREPLACES_ENDING"] = 0
|
|
|
|
if recommendation["type"] not in [
|
|
"wall_insulation", "floor_insulation", "roof_insulation", "mechanical_ventilation", "sealing_open_fireplace",
|
|
]:
|
|
raise NotImplementedError("Implement me")
|
|
|
|
return scoring_dict
|