mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
316 lines
12 KiB
Python
316 lines
12 KiB
Python
import pandas as pd
|
||
import backend.app.assumptions as assumptions
|
||
from backend.Property import Property
|
||
from backend.app.plan.schemas import PlanTriggerRequest
|
||
from backend.app.utils import epc_to_sap_lower_bound
|
||
from recommendations.optimiser.CostOptimiser import CostOptimiser
|
||
|
||
|
||
def prepare_input_measures(property_recommendations, goal, needs_ventilation):
|
||
"""
|
||
Prepares a nested list of measure options for optimisation.
|
||
|
||
Each sublist represents all available variants of a single measure type (e.g. all solar PV options).
|
||
Within each sublist, each measure is represented as a dictionary containing:
|
||
- id: unique recommendation identifier
|
||
- cost: total cost of the measure (including ventilation if bundled)
|
||
- gain: the relevant gain metric based on the selected goal
|
||
- type: the measure type, optionally combined with ventilation (e.g. "wall_insulation+mechanical_ventilation")
|
||
|
||
Ventilation bundling:
|
||
- If a property needs ventilation, and a measure type requires it (as defined in
|
||
assumptions.measures_needing_ventilation),
|
||
the ventilation cost and gain are added to that measure’s values.
|
||
|
||
Filtering:
|
||
- Measures with negative `energy_cost_savings` are excluded.
|
||
- Solar PV options with batteries are excluded (currently handled by a placeholder bitwise NOT).
|
||
|
||
Parameters
|
||
----------
|
||
property_recommendations : list[list[dict]]
|
||
Nested list of recommendations for a property. Each inner list represents variations of the same measure type.
|
||
goal : str
|
||
Optimisation goal, one of: "Increasing EPC", "Energy Savings", "Reducing CO2 emissions".
|
||
needs_ventilation : bool
|
||
Whether the property requires mechanical ventilation to accompany certain measures.
|
||
|
||
Returns
|
||
-------
|
||
list[list[dict]]
|
||
Nested list of prepared measure options, ready for input into the optimiser.
|
||
"""
|
||
|
||
goal_map = {
|
||
"Increasing EPC": "sap_points",
|
||
"Energy Savings": "kwh_savings",
|
||
"Reducing CO2 emissions": "co2_equivalent_savings",
|
||
}
|
||
|
||
goal_key = goal_map[goal]
|
||
if not goal_key:
|
||
raise NotImplementedError("Not implemented this gain type - investigate me")
|
||
|
||
ventilation_recommendation = next(
|
||
(measure[0] for measure in property_recommendations if measure[0]["type"] == "mechanical_ventilation"),
|
||
{}
|
||
)
|
||
|
||
input_measures = []
|
||
for recs in property_recommendations:
|
||
|
||
# Skip ventilation as a standalone optimisation option (it will be bundled)
|
||
if needs_ventilation and recs[0]["type"] == "mechanical_ventilation":
|
||
continue
|
||
|
||
# Filter out solar PV with batteries
|
||
if recs[0]["type"] == "solar_pv":
|
||
recs = [r for r in recs if ~r["has_battery"]]
|
||
|
||
# Only include measures with non-negative cost savings
|
||
recs_to_append = [rec for rec in recs if rec["energy_cost_savings"] >= 0]
|
||
if not recs_to_append:
|
||
continue
|
||
|
||
# Build enriched measure data
|
||
to_append = []
|
||
for rec in recs:
|
||
total = (
|
||
rec["total"] + ventilation_recommendation["total"]
|
||
if rec["type"] in assumptions.measures_needing_ventilation and needs_ventilation
|
||
else rec["total"]
|
||
)
|
||
gain = (
|
||
rec[goal_key] + ventilation_recommendation[goal_key]
|
||
if rec["type"] in assumptions.measures_needing_ventilation and needs_ventilation
|
||
else rec[goal_key]
|
||
)
|
||
rec_type = (
|
||
f"{rec['type']}+{ventilation_recommendation['type']}"
|
||
if rec["type"] in assumptions.measures_needing_ventilation and needs_ventilation
|
||
else rec["type"]
|
||
)
|
||
|
||
to_append.append(
|
||
{"id": rec["recommendation_id"], "cost": total, "gain": gain, "type": rec_type}
|
||
)
|
||
|
||
input_measures.append(to_append)
|
||
|
||
return input_measures
|
||
|
||
|
||
def calculate_fixed_gain(property_required_measures, recommendations, p, needs_ventilation):
|
||
"""
|
||
Calculates the total "fixed gain" from required measures for a property.
|
||
|
||
Required measures are applied regardless of optimisation. This function:
|
||
- Finds the maximum SAP points for each required measure type.
|
||
- Sums those max SAP values into a fixed gain total.
|
||
- Adds the SAP points for mechanical ventilation if:
|
||
* The property needs ventilation, and
|
||
* At least one required measure needs ventilation.
|
||
|
||
Parameters
|
||
----------
|
||
property_required_measures : list[list[dict]]
|
||
Nested list of required measures for the property.
|
||
recommendations : dict
|
||
All recommendations for all properties, keyed by property id.
|
||
p : object
|
||
Property object (must have .id).
|
||
needs_ventilation : bool
|
||
Whether ventilation should be bundled with certain measures.
|
||
|
||
Returns
|
||
-------
|
||
float
|
||
Total fixed SAP gain from required measures (and ventilation, if applicable).
|
||
"""
|
||
if not property_required_measures:
|
||
return 0
|
||
|
||
sap_by_type = [
|
||
{"type": rec["type"], "sap_points": rec["sap_points"]}
|
||
for recs in property_required_measures for rec in recs
|
||
]
|
||
|
||
max_per_type = pd.DataFrame(sap_by_type).groupby("type")["sap_points"].max().to_dict()
|
||
fixed_gain = sum(max_per_type.values())
|
||
|
||
required_types = {rec["type"] for rec in sap_by_type}
|
||
if needs_ventilation and any(r in required_types for r in assumptions.measures_needing_ventilation):
|
||
fixed_gain += next(
|
||
(r[0]["sap_points"] for r in recommendations[p.id] if r[0]["type"] == "mechanical_ventilation"),
|
||
0
|
||
)
|
||
|
||
return fixed_gain
|
||
|
||
|
||
def calculate_gain(body: PlanTriggerRequest, p: Property, fixed_gain: float) -> float | None:
|
||
"""
|
||
Calculates the target gain value for optimisation based on the goal.
|
||
|
||
- For "Increasing EPC": Computes the SAP gain needed to reach the target EPC,
|
||
applies a slack adjustment (via CostOptimiser), and subtracts fixed gains from required measures.
|
||
- For "Energy Savings" or "Reducing CO2 emissions": Returns None,
|
||
which signals the optimiser to simply maximise gain under a budget.
|
||
|
||
Parameters
|
||
----------
|
||
body : object
|
||
Request body object containing optimisation settings (goal, goal_value, simulate_sap_10, etc.)
|
||
p : object
|
||
Property object with EPC data (must have p.data["current-energy-efficiency"]).
|
||
fixed_gain : float
|
||
Total fixed gain from required measures (returned by calculate_fixed_gain).
|
||
|
||
Returns
|
||
-------
|
||
float or None
|
||
Required SAP gain for EPC, or None for non-EPC goals.
|
||
"""
|
||
if body.goal == "Increasing EPC":
|
||
current_sap = int(p.data["current-energy-efficiency"])
|
||
gain = CostOptimiser.calculate_sap_gain_with_slack(
|
||
epc_to_sap_lower_bound(body.goal_value) - current_sap
|
||
) - fixed_gain
|
||
if body.simulate_sap_10:
|
||
gain += 3
|
||
return max(gain, 0)
|
||
elif body.goal in ["Energy Savings", "Reducing CO2 emissions"]:
|
||
return None
|
||
else:
|
||
raise NotImplementedError(f"Goal {body.goal} is not supported")
|
||
|
||
|
||
def add_required_measures(property_id, property_required_measures, recommendations, selected):
|
||
"""
|
||
Ensures the cheapest variant of each required measure is added to the selected recommendations.
|
||
|
||
For each required measure type, this function:
|
||
- Finds the lowest-cost variant not already selected.
|
||
- Adds it to the selected recommendation IDs.
|
||
- Returns a flattened list of all selected measure details for final output.
|
||
|
||
Parameters
|
||
----------
|
||
property_id : int
|
||
Unique identifier for the property.
|
||
property_required_measures : list[list[dict]]
|
||
Nested list of required measures for the property.
|
||
recommendations : dict
|
||
All recommendations for all properties, keyed by property id.
|
||
selected : set
|
||
Set of already selected recommendation IDs from the optimiser.
|
||
|
||
Returns
|
||
-------
|
||
list[dict]
|
||
Flat list of selected measure details, each containing:
|
||
{"id", "cost", "gain", "type"}
|
||
"""
|
||
for recs in property_required_measures:
|
||
cheapest = min(
|
||
(rec for rec in recs if rec["recommendation_id"] not in selected),
|
||
key=lambda rec: rec["total"],
|
||
)
|
||
selected.add(cheapest["recommendation_id"])
|
||
|
||
return [
|
||
{"id": rec["recommendation_id"], "cost": rec["total"], "gain": rec["sap_points"], "type": rec["type"]}
|
||
for recs in recommendations[property_id] for rec in recs
|
||
if rec["recommendation_id"] in selected
|
||
]
|
||
|
||
|
||
def add_best_practice_measures(property_id, solution, recommendations, selected):
|
||
"""
|
||
Ensures best-practice measures like ventilation and trickle vents are included
|
||
in the selected recommendations when appropriate.
|
||
|
||
Rules:
|
||
- If a measure requiring ventilation is selected AND ventilation is not already present,
|
||
add the corresponding mechanical ventilation recommendation.
|
||
- Always add trickle vents if they exist in the recommendations.
|
||
|
||
Parameters
|
||
----------
|
||
property_id : int
|
||
The unique identifier for the property.
|
||
solution : list[dict]
|
||
The current list of selected measures (each containing id, type, gain, cost).
|
||
recommendations : dict
|
||
All recommendations for all properties, keyed by property id.
|
||
selected : set
|
||
Set of already selected recommendation IDs.
|
||
|
||
Returns
|
||
-------
|
||
set
|
||
Updated set of selected recommendation IDs, including ventilation and trickle vents if applicable.
|
||
"""
|
||
# Check if any selected measure requires ventilation
|
||
ventilation_selected = [r for r in solution if "+mechanical_ventilation" in r["type"]]
|
||
|
||
# If ventilation has been selected, or one of the measures needs ventilation, we need to ensure ventilation is
|
||
# included
|
||
needs_ventilation = any(
|
||
x in [r["type"] for r in solution] for x in assumptions.measures_needing_ventilation
|
||
) or len(ventilation_selected) > 0
|
||
|
||
if needs_ventilation:
|
||
ventilation_rec = next(
|
||
(r[0] for r in recommendations[property_id] if r[0]["type"] == "mechanical_ventilation"),
|
||
None
|
||
)
|
||
if ventilation_rec:
|
||
selected.add(ventilation_rec["recommendation_id"])
|
||
|
||
# Always add trickle vents if available
|
||
trickle_vents_rec = next(
|
||
(r[0] for r in recommendations[property_id] if r[0]["type"] == "trickle_vents"),
|
||
None
|
||
)
|
||
if trickle_vents_rec:
|
||
selected.add(trickle_vents_rec["recommendation_id"])
|
||
|
||
return selected
|
||
|
||
|
||
def flatten_recommendations_with_defaults(property_id, recommendations, selected):
|
||
"""
|
||
Flattens nested recommendation lists for a property and marks which
|
||
recommendations were selected.
|
||
|
||
Each recommendation dict is copied and an extra key `default` is added:
|
||
- True if the recommendation ID is in `selected`
|
||
- False otherwise
|
||
|
||
Parameters
|
||
----------
|
||
property_id : int
|
||
The unique identifier for the property.
|
||
recommendations : dict
|
||
All recommendations for all properties, keyed by property id.
|
||
Each value is a list of lists (grouped by measure type).
|
||
selected : set
|
||
Set of selected recommendation IDs.
|
||
|
||
Returns
|
||
-------
|
||
list[dict]
|
||
A flattened list of recommendation dicts for the given property,
|
||
each with an added `default` field.
|
||
"""
|
||
final_recommendations = [
|
||
[
|
||
{**rec, "default": rec["recommendation_id"] in selected}
|
||
for rec in recommendations_by_type
|
||
]
|
||
for recommendations_by_type in recommendations[property_id]
|
||
]
|
||
|
||
# Flatten the nested list of lists into a single list
|
||
return [rec for recommendations_by_type in final_recommendations for rec in recommendations_by_type]
|