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

Merge branch 'main' of github.com:Hestia-Homes/Model into mlmodel

Browse source
This commit is contained in:
Michael Duong 2023-08-19 13:10:27 +00:00
commit 6b94286e7f
8 changed files with 361 additions and 209 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
backend/app/db/functions/portfolio_functions.py Normal file
View file

@ -0,0 +1,39 @@
from sqlalchemy.orm import sessionmaker
from sqlalchemy import func
from backend.app.db.connection import db_engine
from backend.app.db.models.recommendations import Plan, PlanRecommendations, Recommendation
from backend.app.db.models.portfolio import Portfolio
def aggregate_portfolio_recommendations(portfolio_id: int):
Session = sessionmaker(bind=db_engine)
with Session() as session:
# Aggregate multiple fields
aggregates = (
session.query(
func.sum(Recommendation.estimated_cost).label("cost"),
# For future usage we will aggregate multiple fields in this step
# func.sum(Recommendation.heat_demand).label("total_heat_demand"),
# func.sum(Recommendation.energy_savings).label("total_energy_savings")
)
.join(PlanRecommendations, PlanRecommendations.recommendation_id == Recommendation.id)
.join(Plan, Plan.id == PlanRecommendations.plan_id)
.filter(Plan.portfolio_id == portfolio_id, Plan.is_default == True, Recommendation.default == True)
.one()
)
aggregates_dict = {
"cost": aggregates.cost or 0,
# "total_heat_demand": aggregates.total_heat_demand or 0,
# "total_energy_savings": aggregates.total_energy_savings or 0
}
# Get the portfolio and update the fields
portfolio = session.query(Portfolio).filter_by(id=portfolio_id).one()
# Update the data
for key, value in aggregates_dict.items():
setattr(portfolio, key, value)
# Merge the updated portfolio back into the session
session.merge(portfolio)
session.commit()

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

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

@ -22,6 +22,12 @@ from backend.app.db.functions.materials_functions import get_materials
from backend.app.db.functions.recommendations_functions import (
create_plan, create_recommendation, create_recommendation_material, create_plan_recommendations
)
from backend.app.db.functions.portfolio_functions import aggregate_portfolio_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
@ -100,6 +106,25 @@ def filter_materials(materials):
return 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 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
@router.post("/trigger")
async def trigger_plan(body: PlanTriggerRequest):
logger.info("Getting the inputs")
@ -173,6 +198,8 @@ async def trigger_plan(body: PlanTriggerRequest):
logger.info("Getting components and properties recommendations")
# TODO: Move this to a class. We probably was a Recommender class which takes the injects the optimisers
# in as a dependency and then the optimisers can take the input measures in as part of the setup() method
recommendations = {}
for p in input_properties:
property_recommendations = []
@ -207,7 +234,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 +264,49 @@ 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)
# Use the optimiser to pick the default recommendations and decide if we need certain
# recommendations to get to the goal
property_recommendations = insert_temp_recommendation_id(property_recommendations)
if not property_recommendations:
continue
input_measures = prepare_input_measures(property_recommendations, 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
property_recommendations = [
[
{**rec, "default": True if rec["recommendation_id"] in selected_recommendations else False}
for rec in recommendations_by_type
]
for recommendations_by_type in property_recommendations
]
# We'll also unlist the recommendations so they're a bit easier to handle from here onwards
property_recommendations = [
rec for recommendations_by_type in property_recommendations for rec in recommendations_by_type
]
recommendations[p.id] = property_recommendations
@ -255,10 +325,8 @@ async def trigger_plan(body: PlanTriggerRequest):
update_property_data(property_id=p.id, portfolio_id=body.portfolio_id, property_data=property_data)
# Upload recommendations
recommendations_to_upload = recommendations.get(p.id, [])
# TODO: We start off by optimising the recommendations
recommendations_to_upload = recommendations[p.id]
if not recommendations_to_upload:
continue
@ -281,7 +349,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
@ -305,4 +373,12 @@ async def trigger_plan(body: PlanTriggerRequest):
recommendation_ids=uploaded_recommendation_ids
)
logger.info("Creating portfolio aggregations")
# We implement this in the simplest way possible which will be just to query the database for all
# recommendations associated to the portfolio and then aggregate them. This is not the most efficient
# way to do this, but it's the simplest and will be a process that we can re-use since when we change a
# recommendation from being default to not default, we'll need to re-run this process to re-calculate the
# the portfolion level impact
aggregate_portfolio_recommendations(portfolio_id=body.portfolio_id)
return Response(status_code=200)

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(property_recommendations, goal):
"""
Basic function to convert recommendations_to_upload to a format that is
suitable for the optimiser - large
:param property_recommendations: 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 property_recommendations:
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")