mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
94 lines
3 KiB
Python
94 lines
3 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 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]
|
|
}
|
|
|
|
# We update the description to indicate it's insulated
|
|
if recommendation["type"] == "wall_insulation":
|
|
scoring_dict["WALLS_DESCRIPTION_ENDING"] = UPGRADES_MAP[property.walls["clean_description"]]
|
|
elif recommendation["type"] == "floor_insulation":
|
|
scoring_dict["FLOOR_DESCRIPTION_ENDING"] = UPGRADES_MAP[property.floor["clean_description"]]
|
|
else:
|
|
raise NotImplementedError("Implement me")
|
|
|
|
return scoring_dict
|