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

implemented optimiser into recommendation api

Browse source
This commit is contained in:
Khalim Conn-Kowlessar 2023-08-18 16:49:03 +01:00
parent 20aa23efa0
commit f37f6ac029
7 changed files with 309 additions and 206 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
backend/app/db/functions/property_functions.py
View file

@ -109,14 +109,26 @@ def update_property_data(property_id: int, portfolio_id: int, property_data: dic
def create_property_details_epc(property_details_epc: dict):
"""
This function will create a record for the property details EPC in the database.
This function will create or update a record for the property details EPC in the database.
:param property_details_epc: A dictionary containing details about the property EPC.
:return: True if successful, False otherwise.
"""
Session = sessionmaker(bind=db_engine)
with Session() as session:
new_property_details_epc = PropertyDetailsEpcModel(**property_details_epc)
session.add(new_property_details_epc)
existing_record = session.query(PropertyDetailsEpcModel).filter_by(
portfolio_id=property_details_epc["portfolio_id"],
property_id=property_details_epc["property_id"]
).first()
if existing_record:
# If the record exists, update its fields
for key, value in property_details_epc.items():
setattr(existing_record, key, value)
else:
# If the record doesn't exist, create a new one
new_property_details_epc = PropertyDetailsEpcModel(**property_details_epc)
session.add(new_property_details_epc)
session.commit()
return True

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

@ -23,6 +23,11 @@ from backend.app.db.functions.recommendations_functions import (
create_plan, create_recommendation, create_recommendation_material, create_plan_recommendations
)
from model_data.optimiser.GainOptimiser import GainOptimiser
from model_data.optimiser.CostOptimiser import CostOptimiser
from model_data.utils import epc_to_sap_lower_bound
from model_data.optimiser.optimiser_functions import prepare_input_measures
# TODO: This is placeholder until data is stored in DB
from backend.app.plan.uvalue_estimates_walls import uvalue_estimates_walls
from backend.app.plan.uvalue_estimates_floors import uvalue_estimates_floors
@ -100,6 +105,25 @@ def filter_materials(materials):
return materials_by_type
def insert_temp_recommendation_id(recommendations_to_upload):
"""
Creates a temporary recommendation id which is needed for
filtering recommendations between default and no, after the optimiser has been
run
:param recommendations_to_upload: nested list of recommendations, grouped by types
:return: Updated recommendations_to_upload, where where recommendation has a "recommendation_id"
integer inserted
"""
idx = 0
for recs in recommendations_to_upload:
for rec in recs:
rec["recommendation_id"] = idx
idx += 1
return recommendations_to_upload
@router.post("/trigger")
async def trigger_plan(body: PlanTriggerRequest):
logger.info("Getting the inputs")
@ -207,7 +231,8 @@ async def trigger_plan(body: PlanTriggerRequest):
)
floor_recommender.recommend()
property_recommendations.extend(floor_recommender.recommendations)
if floor_recommender.recommendations:
property_recommendations.append(floor_recommender.recommendations)
# Wall recommendations
# We would make this u-value query directly to the database
@ -236,7 +261,8 @@ async def trigger_plan(body: PlanTriggerRequest):
)
wall_recomender.recommend()
property_recommendations.extend(wall_recomender.recommendations)
if wall_recomender.recommendations:
property_recommendations.append(wall_recomender.recommendations)
recommendations[p.id] = property_recommendations
@ -259,9 +285,49 @@ async def trigger_plan(body: PlanTriggerRequest):
# TODO: We start off by optimising the recommendations
recommendations_to_upload = recommendations[p.id]
if not recommendations_to_upload:
continue
recommendations_to_upload = insert_temp_recommendation_id(recommendations_to_upload)
# Optimise the recommendations
# We need to format the recommendations for the optimiser
input_measures = prepare_input_measures(recommendations_to_upload, body.goal)
if body.budget:
optimiser = GainOptimiser(input_measures, max_cost=body.budget)
else:
# The minimum gain is the minimum number of SAP points required to get to the target SAP band
current_sap_points = int(p.data["current-energy-efficiency"])
target_sap_points = epc_to_sap_lower_bound(body.goal_value)
# If the gain is negative, the optimiser will return an empty solution
optimiser = CostOptimiser(
input_measures, min_gain=target_sap_points - current_sap_points
)
optimiser.setup()
optimiser.solve()
solution = optimiser.solution
selected_recommendations = {r["id"] for r in solution}
# We'll use the set of selected recommendations to filter the recommendations to upload
recommendations_to_upload = [
[
{**rec, "default": True if rec["recommendation_id"] in selected_recommendations else False}
for rec in recommendations_by_type
]
for recommendations_by_type in recommendations_to_upload
]
# We'll also unlist the recommendations so they're a bit easier to handle from here onwards
recommendations_to_upload = [
rec for recommendations_by_type in recommendations_to_upload for rec in recommendations_by_type
]
# Create a plan
new_plan_id = create_plan(
{
@ -281,7 +347,7 @@ async def trigger_plan(body: PlanTriggerRequest):
"type": rec["type"],
"description": rec["description"],
"estimated_cost": rec["cost"],
"default": True,
"default": rec["default"],
"starting_u_value": rec.get("starting_u_value"),
"new_u_value": rec.get("new_u_value"),
# TODO: Placeholder for SAP points in place

68
model_data/optimiser/CostOptimiser.py Normal file
View file

@ -0,0 +1,68 @@
from mip import Model, xsum, minimize, BINARY
class CostOptimiser:
"""
This class is used to minimise cost, given a constrained minimum gain
"""
def __init__(self, components, min_gain):
self.components = components
self.min_gain = min_gain
self.m = None
self.variables = []
self.solution = []
self.solution_cost = None
self.solution_gain = None
def setup(self):
# Initialize Model
self.m = Model("knapsack")
# Create variables
self.variables = [
[self.m.add_var(var_type=BINARY, name=str(component["id"])) for component in group] for group in
self.components
]
# Set objective
# This objective is to minimize
# cost_ig * x_ig, where cost_ig represents the cost for ith part in group g
# and x_ig is the binary decision variable for the ith part in group g
self.m.objective = minimize(
xsum(
component['cost'] * var for group, group_vars in zip(self.components, self.variables) for component, var
in
zip(group, group_vars)
)
)
# Add constraints
# This constrain ensures that sum of gain_ig * x_ig >= min_gain, where gain_ig represents the gain for the ith
# component
# in group g, and x_ig is the binary decision variable for the ith component in group g
self.m += xsum(
item['gain'] * var for group, group_vars in zip(self.components, self.variables) for item, var in
zip(group, group_vars)
) >= self.min_gain
# At most one item from each group
# This constraint ensures that at most one item from each group is selected
# This is expressed by summing up the decision variables for each group and ensuring that the sum is <= 1
for group_vars in self.variables:
self.m += xsum(var for var in group_vars) <= 1
def solve(self):
# Solve the problem
self.m.optimize()
self.solution = [
item for group, group_vars in zip(self.components, self.variables) for item, var in zip(group, group_vars)
if
var.x >= 0.99
]
# Get the selected items
self.solution_cost = self.m.objective.x
self.solution_gain = sum([component['gain'] for component in self.solution])

70
model_data/optimiser/GainOptimiser.py Normal file
View file

@ -0,0 +1,70 @@
from mip import Model, xsum, maximize, BINARY
class GainOptimiser:
"""
This class is used maximise gain, given a constrained cost
"""
def __init__(self, components, max_cost):
self.components = components
self.max_cost = max_cost
self.m = None
self.variables = []
self.solution = []
self.solution_gain = None
self.solution_cost = None
def setup(self):
# Initialize Model
self.m = Model("knapsack")
# Create variables
self.variables = [
[self.m.add_var(var_type=BINARY, name=str(component["id"])) for component in group] for group in
self.components
]
# Set objective
# This objective is the sum
# gain_ig * x_ig, where gain_ig represents the gain for ith part in group g
# and x_ig is the binary decision variable for the ith part in group g
self.m.objective = maximize(
xsum(
component['gain'] * var for group, group_vars in zip(self.components, self.variables) for component, var
in
zip(group, group_vars)
)
)
# Add constraints
# This constrain ensures that sum of cost_ig * x_ig <= C, where cost_ig represents the cost for the ith
# component
# in group g, and x_ig is the binary decision variable for the ith component in group g
self.m += xsum(
item['cost'] * var for group, group_vars in zip(self.components, self.variables) for item, var in
zip(group, group_vars)
) <= self.max_cost
# At most one item from each group
# This constraint ensures that at most one item from each group is selected
# This is expressed by summing up the decision variables for each group and ensuring that the sum is <= 1
for group_vars in self.variables:
self.m += xsum(var for var in group_vars) <= 1
def solve(self):
# Solve the problem
self.m.optimize()
self.solution = [
item for group, group_vars in zip(self.components, self.variables) for item, var in zip(group, group_vars)
if
var.x >= 0.99
]
# Get the selected items
self.solution_gain = self.m.objective.x
self.solution_cost = sum([component['cost'] for component in self.solution])

200
model_data/optimiser/Optimiser.py
View file

@ -1,200 +0,0 @@
from mip import Model, xsum, maximize, BINARY
from pprint import pprint
# Example parts
wall = [
{"id": 1, "cost": 2000, "gain": 5, "type": "wall"},
{"id": 2, "cost": 2300, "gain": 6, "type": "wall"}
]
floor = [
{"id": 1, "cost": 1500, "gain": 3, "type": "floor"},
{"id": 2, "cost": 1600, "gain": 3.1, "type": "floor"}
]
roof = [
{"id": 1, "cost": 1000, "gain": 2, "type": "roof"},
{"id": 2, "cost": 1100, "gain": 2.3, "type": "roof"}
]
# To solve this, we are solving a constrained Knapsack problem
# Maximize sum(gain_g . x_g) for g in groups
# subject to sum(cost_g . x_g) <= C
# subject to sum(x_g) <= 1 for g in groups
# x_g in {0, 1} for g in groups
#
# The first sum, which is the objective of the optimisation provlem, ensures that we are maximising the gain
# for the selected parts
# The second sum (and the first constraint) ensures that the cost of the selected parts is less than or equal to C
# The third sum (and the second constraint) ensures that at most one part from each group is selected
# The last constraint ensures that the decision variables are binary
# group all the parts
components = [wall, floor, roof]
class GainOptimiser:
"""
This class is used maximise gain, given a constrained cost
"""
def __init__(self, components, max_cost):
self.components = components
self.max_cost = max_cost
self.m = None
self.variables = []
self.solution = []
self.solution_gain = None
self.solution_cost = None
def setup(self):
# Initialize Model
self.m = Model("knapsack")
# Create variables
self.variables = [
[self.m.add_var(var_type=BINARY, name=str(component["id"])) for component in group] for group in
self.components
]
# Set objective
# This objective is the sum
# gain_ig * x_ig, where gain_ig represents the gain for ith part in group g
# and x_ig is the binary decision variable for the ith part in group g
self.m.objective = maximize(
xsum(
component['gain'] * var for group, group_vars in zip(self.components, self.variables) for component, var
in
zip(group, group_vars)
)
)
# Add constraints
# This constrain ensures that sum of cost_ig * x_ig <= C, where cost_ig represents the cost for the ith
# component
# in group g, and x_ig is the binary decision variable for the ith component in group g
self.m += xsum(
item['cost'] * var for group, group_vars in zip(self.components, self.variables) for item, var in
zip(group, group_vars)
) <= self.max_cost
# At most one item from each group
# This constraint ensures that at most one item from each group is selected
# This is expressed by summing up the decision variables for each group and ensuring that the sum is <= 1
for group_vars in self.variables:
self.m += xsum(var for var in group_vars) <= 1
def solve(self):
# Solve the problem
self.m.optimize()
self.solution = [
item for group, group_vars in zip(self.components, self.variables) for item, var in zip(group, group_vars)
if
var.x >= 0.99
]
# Get the selected items
self.solution_gain = self.m.objective.x
self.solution_cost = sum([component['cost'] for component in self.solution])
opt = GainOptimiser(components, max_cost=4000)
# Setup the knackpack problem
# This sets the objective & contraints
opt.setup()
# Solve the problem
opt.solve()
pprint(opt.solution)
print("total cost:", opt.solution_cost)
print("total gain:", opt.solution_gain)
# A bigger problem:
wall = [
{"id": 1, "cost": 2000, "gain": 5, "type": "wall"},
{"id": 2, "cost": 2300, "gain": 6, "type": "wall"},
{"id": 3, "cost": 2200, "gain": 5.5, "type": "wall"},
{"id": 4, "cost": 2500, "gain": 6.2, "type": "wall"},
{"id": 5, "cost": 2100, "gain": 5.1, "type": "wall"},
{"id": 6, "cost": 2400, "gain": 6.1, "type": "wall"},
{"id": 7, "cost": 2000, "gain": 5.2, "type": "wall"}
]
floor = [
{"id": 1, "cost": 1500, "gain": 3, "type": "floor"},
{"id": 2, "cost": 1600, "gain": 3.1, "type": "floor"},
{"id": 3, "cost": 1550, "gain": 3.2, "type": "floor"},
{"id": 4, "cost": 1650, "gain": 3.3, "type": "floor"},
{"id": 5, "cost": 1500, "gain": 3.4, "type": "floor"},
{"id": 6, "cost": 1550, "gain": 3.5, "type": "floor"},
{"id": 7, "cost": 1600, "gain": 3.6, "type": "floor"}
]
roof = [
{"id": 1, "cost": 1000, "gain": 2, "type": "roof"},
{"id": 2, "cost": 1100, "gain": 2.3, "type": "roof"},
{"id": 3, "cost": 1200, "gain": 2.6, "type": "roof"},
{"id": 4, "cost": 1300, "gain": 2.9, "type": "roof"},
{"id": 5, "cost": 1100, "gain": 2.5, "type": "roof"},
{"id": 6, "cost": 1200, "gain": 2.7, "type": "roof"},
{"id": 7, "cost": 1300, "gain": 2.8, "type": "roof"}
]
heating = [
{"id": 1, "cost": 3000, "gain": 7, "type": "heating"},
{"id": 2, "cost": 3200, "gain": 7.2, "type": "heating"},
{"id": 3, "cost": 3100, "gain": 7.1, "type": "heating"},
{"id": 4, "cost": 3300, "gain": 7.3, "type": "heating"},
{"id": 5, "cost": 3000, "gain": 7.4, "type": "heating"}
]
hot_water = [
{"id": 1, "cost": 2500, "gain": 6.5, "type": "hot water"},
{"id": 2, "cost": 2600, "gain": 6.6, "type": "hot water"},
{"id": 3, "cost": 2500, "gain": 6.7, "type": "hot water"},
{"id": 4, "cost": 2700, "gain": 6.8, "type": "hot water"},
{"id": 5, "cost": 2500, "gain": 6.9, "type": "hot water"}
]
solar = [
{"id": 1, "cost": 5000, "gain": 10, "type": "solar"},
{"id": 2, "cost": 5500, "gain": 11, "type": "solar"},
{"id": 3, "cost": 5300, "gain": 10.5, "type": "solar"},
{"id": 4, "cost": 5200, "gain": 10.2, "type": "solar"},
{"id": 5, "cost": 5400, "gain": 10.8, "type": "solar"}
]
heat_pumps = [
{"id": 1, "cost": 4000, "gain": 9, "type": "heat pumps"},
{"id": 2, "cost": 4200, "gain": 9.2, "type": "heat pumps"},
{"id": 3, "cost": 4100, "gain": 9.1, "type": "heat pumps"},
{"id": 4, "cost": 4300, "gain": 9.3, "type": "heat pumps"},
{"id": 5, "cost": 4000, "gain": 9.4, "type": "heat pumps"}
]
components2 = [
wall,
floor,
roof,
heating,
hot_water,
solar,
heat_pumps
]
opt2 = GainOptimiser(components2, max_cost=15000)
# Setup
opt2.setup()
# Solve the problem
opt2.solve()
pprint(opt2.solution)
print("total cost:", opt2.solution_cost)
print("total gain:", opt2.solution_gain)

33
model_data/optimiser/optimiser_functions.py Normal file
View file

@ -0,0 +1,33 @@
def prepare_input_measures(recommendations_to_upload, goal):
"""
Basic function to convert recommendations_to_upload to a format that is
suitable for the optimiser - large
:param recommendations_to_upload: object containing the recommendations, created in the plan trigger api
:param goal: goal to be optimised for, should be one of the keys in gain_map. E.g. if the gain is SAP points,
the goal should reflect that desired gain
:return: Nested list of input measures
"""
goal_map = {
"Increase EPC": "sap_points"
}
goal_key = goal_map[goal]
if not goal_key:
raise NotImplementedError("Not implemented this gain type - investigate me")
input_measures = []
for recs in recommendations_to_upload:
input_measures.append(
[
{
"id": rec["recommendation_id"],
"cost": rec["cost"],
"gain": rec[goal_key],
"type": rec["type"]
}
for rec in recs
]
)
return input_measures

54
model_data/utils.py
View file

@ -24,3 +24,57 @@ def correct_spelling(text):
corrected_text = ' '.join(corrected_words)
return corrected_text
def sap_to_epc(sap_points: int):
"""
Simple utility function to convert SAP points to EPC rating.
:param sapPoints: numerical value of SAP points, typically between 0 and 100
:return:
"""
if sap_points <= 0 or sap_points > 100:
raise ValueError("SAP points should be between 1 and 100.")
if sap_points > 91:
return "A"
elif sap_points > 80:
return "B"
elif sap_points > 69:
return "C"
elif sap_points > 55:
return "D"
elif sap_points > 39:
return "E"
elif sap_points > 21:
return "F"
else:
return "G"
def epc_to_sap_lower_bound(epc: str):
"""
Given an EPC rating, returns the lower bound SAP score required
to hit that EPC rating
:param epc: EPC rating, between A and G
:return:
"""
if epc == "A":
return 92
elif epc == "B":
return 81
elif epc == "C":
return 70
elif epc == "D":
return 56
elif epc == "E":
return 40
elif epc == "F":
return 22
elif epc == "G":
return 1
else:
raise ValueError("EPC rating should be between A and G")