use sqlalchemy 2.0 typing in recommendations , and write processing logic

backend/app/db/functions/recommendations_functions.py

@@ -1,3 +1,4 @@
+from typing import List
 from sqlalchemy import text
 from sqlalchemy import insert, delete
 from sqlalchemy.orm import Session
@@ -610,11 +611,11 @@ def clear_portfolio_in_batches(
     print("Portfolio cleared in batches.")
 
 
-def get_plans_by_portfolio_id(portfolio_id: int) -> list[Plan]:
+def get_plans_by_portfolio_id(portfolio_id: int) -> List[Plan]:
     raise NotImplementedError
 
 
-def get_scenario(scenario_id: int) -> list[Scenario]:
+def get_scenario(scenario_id: int) -> List[Scenario]:
     raise NotImplementedError
 
 

backend/app/db/models/recommendations.py

@@ -1,5 +1,15 @@
-from sqlalchemy import Column, BigInteger, String, Float, Boolean, TIMESTAMP, ForeignKey, Enum
-from sqlalchemy.orm import declarative_base
+from typing import Iterable, Optional
+from sqlalchemy import (
+    Column,
+    BigInteger,
+    String,
+    Float,
+    Boolean,
+    TIMESTAMP,
+    ForeignKey,
+    Enum,
+)
+from sqlalchemy.orm import declarative_base, Mapped, mapped_column
 from sqlalchemy.sql import func
 from backend.app.db.models.portfolio import Portfolio, PropertyModel
 from backend.app.db.models.materials import Material
@@ -11,7 +21,7 @@ Base = declarative_base()
 
 
 class Recommendation(Base):
-    __tablename__ = 'recommendation'
+    __tablename__ = "recommendation"
 
     id = Column(BigInteger, primary_key=True, autoincrement=True)
     property_id = Column(BigInteger, ForeignKey(PropertyModel.id), nullable=False)
@@ -37,15 +47,20 @@ class Recommendation(Base):
 
 
 class RecommendationMaterials(Base):
-    __tablename__ = 'recommendation_materials'
+    __tablename__ = "recommendation_materials"
 
     id = Column(BigInteger, primary_key=True, autoincrement=True)
-    recommendation_id = Column(BigInteger, ForeignKey('recommendation.id'), nullable=False)
+    recommendation_id = Column(
+        BigInteger, ForeignKey("recommendation.id"), nullable=False
+    )
     material_id = Column(BigInteger, ForeignKey(Material.id), nullable=False)
     created_at = Column(TIMESTAMP, nullable=False, server_default=func.now())
     depth = Column(Float, nullable=False)
     quantity = Column(Float, nullable=False)
-    quantity_unit = Column(Enum(QuantityUnits, values_callable=lambda x: [e.value for e in x]), nullable=False)
+    quantity_unit = Column(
+        Enum(QuantityUnits, values_callable=lambda x: [e.value for e in x]),
+        nullable=False,
+    )
     estimated_cost = Column(Float, nullable=False)
 
 
@@ -58,19 +73,35 @@ class PlanTypeEnum(enum.Enum):
 
 
 class Plan(Base):
-    __tablename__ = 'plan'
+    __tablename__ = "plan"
 
-    id = Column(BigInteger, primary_key=True, autoincrement=True)
-    name = Column(String, nullable=True, default="")
-    portfolio_id = Column(BigInteger, ForeignKey(Portfolio.id), nullable=False)
-    property_id = Column(BigInteger, ForeignKey(PropertyModel.id), nullable=False)
-    scenario_id = Column(BigInteger, ForeignKey('scenario.id'))  # Doesn't have to be linked to a scenario
-    created_at = Column(TIMESTAMP, nullable=False, server_default=func.now())
-    is_default = Column(Boolean, nullable=False)
-    valuation_increase_lower_bound = Column(Float)
-    valuation_increase_upper_bound = Column(Float)
-    valuation_increase_average = Column(Float)
-    plan_type = Column(
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+
+    name: Mapped[Optional[str]] = mapped_column(String, nullable=True, default="")
+
+    portfolio_id: Mapped[int] = mapped_column(
+        BigInteger, ForeignKey(Portfolio.id), nullable=False
+    )
+
+    property_id: Mapped[int] = mapped_column(
+        BigInteger, ForeignKey(PropertyModel.id), nullable=False
+    )
+
+    scenario_id: Mapped[Optional[int]] = mapped_column(
+        BigInteger, ForeignKey("scenario.id")
+    )
+
+    created_at: Mapped = mapped_column(  # type: ignore
+        TIMESTAMP, nullable=False, server_default=func.now()
+    )
+
+    is_default: Mapped[bool] = mapped_column(Boolean, nullable=False)
+
+    valuation_increase_lower_bound: Mapped[Optional[float]] = mapped_column(Float)
+    valuation_increase_upper_bound: Mapped[Optional[float]] = mapped_column(Float)
+    valuation_increase_average: Mapped[Optional[float]] = mapped_column(Float)
+
+    plan_type: Mapped[Optional[PlanTypeEnum]] = mapped_column(
         Enum(
             PlanTypeEnum,
             name="plan_type",
@@ -79,31 +110,35 @@ class Plan(Base):
         ),
         nullable=True,
     )
-    post_sap_points = Column(Float)
-    post_epc_rating = Column(Enum(Epc))
-    post_co2_emissions = Column(Float)
-    co2_savings = Column(Float)
-    post_energy_bill = Column(Float)
-    energy_bill_savings = Column(Float)
-    post_energy_consumption = Column(Float)  # energy demand in kWh/year
-    energy_consumption_savings = Column(Float)
-    valuation_post_retrofit = Column(Float)
-    valuation_increase = Column(Float)
+
+    post_sap_points: Mapped[Optional[float]] = mapped_column(Float)
+    post_epc_rating: Mapped[Optional[Epc]] = mapped_column(Enum(Epc))
+    post_co2_emissions: Mapped[Optional[float]] = mapped_column(Float)
+    co2_savings: Mapped[Optional[float]] = mapped_column(Float)
+    post_energy_bill: Mapped[Optional[float]] = mapped_column(Float)
+    energy_bill_savings: Mapped[Optional[float]] = mapped_column(Float)
+    post_energy_consumption: Mapped[Optional[float]] = mapped_column(Float)
+    energy_consumption_savings: Mapped[Optional[float]] = mapped_column(Float)
+    valuation_post_retrofit: Mapped[Optional[float]] = mapped_column(Float)
+    valuation_increase: Mapped[Optional[float]] = mapped_column(Float)
+
     # Financial metrics, excluding funding
-    cost_of_works = Column(Float)
-    contingency_cost = Column(Float)
+    cost_of_works: Mapped[Optional[float]] = mapped_column(Float)
+    contingency_cost: Mapped[Optional[float]] = mapped_column(Float)
 
 
 class PlanRecommendations(Base):
-    __tablename__ = 'plan_recommendations'
+    __tablename__ = "plan_recommendations"
 
     id = Column(BigInteger, primary_key=True, autoincrement=True)
-    plan_id = Column(BigInteger, ForeignKey('plan.id'), nullable=False)
-    recommendation_id = Column(BigInteger, ForeignKey('recommendation.id'), nullable=False)
+    plan_id = Column(BigInteger, ForeignKey("plan.id"), nullable=False)
+    recommendation_id = Column(
+        BigInteger, ForeignKey("recommendation.id"), nullable=False
+    )
 
 
 class Scenario(Base):
-    __tablename__ = 'scenario'
+    __tablename__ = "scenario"
 
     id = Column(BigInteger, primary_key=True, autoincrement=True)
     name = Column(String, nullable=False)
@@ -201,3 +236,7 @@ class InstalledMeasure(Base):
     heat_demand_savings = Column(Float)
     source = Column(String)
     is_active = Column(Boolean, nullable=False, default=True)
+
+
+def enum_values(e: Iterable[PlanTypeEnum]) -> list[str]:
+    return [m.value for m in e]

backend/categorisation/categorisation_logic.py

@@ -0,0 +1,12 @@
+from typing import List
+from backend.app.db.models.recommendations import Plan
+
+
+class CategorisationLogic:
+    @staticmethod
+    def get_compliant_plans(plans: List[Plan]) -> List[Plan]:
+        raise NotImplementedError
+
+    @staticmethod
+    def get_cheapest_plan(plans: List[Plan]) -> Plan:
+        raise NotImplementedError

backend/categorisation/processor.py

@@ -1,10 +1,35 @@
+from typing import List
+
+from backend.app.db.functions.recommendations_functions import (
+    get_plans_by_portfolio_id,
+    get_property_ids,
+    set_plan_default,
+)
+from backend.app.db.models.recommendations import Plan
+from backend.categorisation.categorisation_logic import CategorisationLogic
+
+
 def process_portfolio(portfolio_id: int) -> None:
     # Get all plans (including scenarios) for all properties in the portfolio
+    plans: List[Plan] = get_plans_by_portfolio_id(portfolio_id)
 
     # For each property, get all compliant plans
+    property_ids: List[int] = get_property_ids(portfolio_id)
 
     # For each property, find the cheapest compliant plan
+    for id in property_ids:
+        plans_for_property: List[Plan] = [
+            plan for plan in plans if plan.property_id == id
+        ]
 
-    # For each property, set is_default for cheapest compliant plan
-    # If no compliant plans, set it to the cheapest plan
-    pass
+        compliant_plans_for_property: List[Plan] = (
+            CategorisationLogic.get_compliant_plans(plans_for_property)
+        )
+
+        # Choose cheapest compliant plan, or fallback to cheapest overall plan
+        plans_to_consider = compliant_plans_for_property or plans_for_property
+        cheapest_plan = CategorisationLogic.get_cheapest_plan(plans_to_consider)
+
+        # Update DB: set is_default = True for cheapest plan, False for others
+        for plan in plans_for_property:
+            set_plan_default(plan.id, plan.id == cheapest_plan.id)