mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
from collections import defaultdict
|
|
from typing import List, Tuple, cast
|
|
|
|
from backend.app.db.functions.recommendations_functions import (
|
|
get_plans_by_portfolio_id,
|
|
get_scenario,
|
|
update_plan,
|
|
)
|
|
from backend.app.db.models.recommendations import PlanModel, ScenarioModel
|
|
from backend.app.domain.classes.plan import Plan
|
|
from backend.categorisation.categorisation_logic import CategorisationLogic
|
|
from utils.logger import setup_logger
|
|
|
|
logger = setup_logger()
|
|
|
|
|
|
def process_portfolio(portfolio_id: int) -> None:
|
|
plans = _load_plans_for_portfolio(portfolio_id)
|
|
plans_by_property = _group_plans_by_property(plans)
|
|
|
|
for property_plans in plans_by_property.values():
|
|
cheapest_plan = _choose_cheapest_relevant_plan(property_plans)
|
|
_update_default_flags(property_plans, cheapest_plan)
|
|
|
|
|
|
def _load_plans_for_portfolio(portfolio_id: int) -> List[Plan]:
|
|
plan_models = get_plans_by_portfolio_id(portfolio_id)
|
|
plans: List[Plan] = []
|
|
|
|
for model in plan_models:
|
|
if not model.scenario_id:
|
|
logger.info(f"No Scenario associated with Plan of ID {model.id}")
|
|
continue
|
|
|
|
scenario_model = get_scenario(model.scenario_id)
|
|
plans.append(Plan.from_sqlalchemy(model, scenario_model))
|
|
|
|
return plans
|
|
|
|
|
|
def _group_plans_by_property(plans: List[Plan]) -> dict[int, List[Plan]]:
|
|
grouped: dict[int, List[Plan]] = defaultdict(list)
|
|
|
|
for plan in plans:
|
|
grouped[plan.record.property_id].append(plan)
|
|
|
|
return grouped
|
|
|
|
|
|
def _choose_cheapest_relevant_plan(plans: List[Plan]) -> Plan:
|
|
compliant_plans = CategorisationLogic.get_compliant_plans(plans)
|
|
|
|
plans_to_consider = compliant_plans or plans
|
|
return CategorisationLogic.get_cheapest_plan(plans_to_consider)
|
|
|
|
|
|
def _update_default_flags(plans: List[Plan], cheapest_plan: Plan) -> None:
|
|
for plan in plans:
|
|
if plan.id is None:
|
|
raise ValueError("Cannot update Plan with missing ID")
|
|
|
|
plan.set_default(plan.id == cheapest_plan.id)
|
|
|
|
plan_model, scenario_model = cast(
|
|
Tuple[PlanModel, ScenarioModel],
|
|
plan.to_sqlalchemy(),
|
|
)
|
|
|
|
update_plan(plan_model, scenario_model)
|