mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
209 lines
9.2 KiB
Python
209 lines
9.2 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.config import get_settings
|
|
import msgpack
|
|
|
|
|
|
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"] in ["internal_wall_insulation", "external_wall_insulation", "cavity_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"] in ["solid_floor_insulation", "suspended_floor_insulation", "exposed_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"] in ["loft_insulation", "room_roof_insulation", "flat_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")
|
|
|
|
# This is based on the values we have in the training data
|
|
valid_numeric_values = [
|
|
12, 25, 50, 75, 100, 150, 200, 250, 270, 300, 350, 400
|
|
]
|
|
|
|
proposed_depth = int(parts[0]["depth"])
|
|
if proposed_depth not in valid_numeric_values:
|
|
# Take the nearest value for scoring
|
|
proposed_depth = min(valid_numeric_values, key=lambda x: abs(x - proposed_depth))
|
|
|
|
scoring_dict["roof_insulation_thickness_ENDING"] = str(proposed_depth)
|
|
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"] == "low_energy_lighting":
|
|
scoring_dict["LOW_ENERGY_LIGHTING_ENDING"] = 100
|
|
scoring_dict["LIGHTING_ENERGY_EFF_STARTING"] = "Very Good"
|
|
|
|
if recommendation["type"] == "windows_glazing":
|
|
scoring_dict["MULTI_GLAZE_PROPORTION_ENDING"] = 100
|
|
scoring_dict["WINDOWS_ENERGY_EFF_ENDING"] = "Average"
|
|
|
|
is_secondary_glazing = recommendation["is_secondary_glazing"]
|
|
|
|
if scoring_dict["glazing_type_ENDING"] == "multiple":
|
|
pass
|
|
elif scoring_dict["glazing_type_ENDING"] == "single":
|
|
scoring_dict["glazing_type_ENDING"] = "secondary" if is_secondary_glazing else "double"
|
|
elif scoring_dict["glazing_type_ENDING"] == "double":
|
|
scoring_dict["glazing_type_ENDING"] = "multiple" if is_secondary_glazing else "double"
|
|
elif scoring_dict["glazing_type_ENDING"] == "secondary":
|
|
scoring_dict["glazing_type_ENDING"] = "secondary" if is_secondary_glazing else "multiple"
|
|
elif scoring_dict["glazing_type_ENDING"] in ["triple", "high performance"]:
|
|
scoring_dict["glazing_type_ENDING"] = "multiple"
|
|
else:
|
|
raise ValueError("Invalid glazing type - implement me")
|
|
|
|
if recommendation["type"] == "solar_pv":
|
|
scoring_dict["PHOTO_SUPPLY_ENDING"] = recommendation["photo_supply"]
|
|
|
|
if recommendation["type"] not in [
|
|
"mechanical_ventilation", "sealing_open_fireplace", "low_energy_lighting",
|
|
"internal_wall_insulation", "external_wall_insulation", "cavity_wall_insulation",
|
|
"loft_insulation", "room_roof_insulation", "flat_roof_insulation",
|
|
"solid_floor_insulation", "suspended_floor_insulation", "exposed_floor_insulation",
|
|
"windows_glazing", "solar_pv"
|
|
]:
|
|
raise NotImplementedError("Implement me")
|
|
|
|
return scoring_dict
|