mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
b97d06882f
Stop writing the m2m (remove create_plan_recommendations + its call, the bulk link insert and the now-dead plan_ids_by_index, and the plan_recommendations delete in delete_property_batch) and remove the PlanRecommendationRow model + its shim alias and the test_export fixture inserts. Measures now link to their Plan solely via recommendation.plan_id (writers set it, readers join on it). The live drop of the plan_recommendations table is the FE-owned Drizzle migration documented in docs/migrations/recommendation-plan-id.md, sequenced after the read-cut + backfill. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
120 lines
4.7 KiB
Python
120 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import ClassVar, Optional
|
|
|
|
from sqlalchemy import BigInteger, Column, ForeignKey, TIMESTAMP
|
|
from sqlalchemy import Enum as SAEnum
|
|
from sqlalchemy.sql import func
|
|
from sqlmodel import Field, SQLModel
|
|
|
|
from datatypes.enums import QuantityUnits
|
|
from domain.modelling.plan import PlanMeasure
|
|
|
|
# Calculator metrics are in kg CO₂/yr; the live ``recommendation`` column is
|
|
# tonnes (legacy ``emissions_kg / 1000``). Convert on the way in.
|
|
_KG_PER_TONNE = 1000.0
|
|
|
|
|
|
class RecommendationRow(SQLModel, table=True):
|
|
"""The single SQLModel definition of the live ``recommendation`` table
|
|
(ADR-0017 amendment) — one row per persisted Plan Measure.
|
|
|
|
Carries full legacy column parity (the readers iterate the columns / sum
|
|
them) **plus** ``plan_id``, the FK that links a measure to its Plan and
|
|
replaces the retired ``plan_recommendations`` m2m. Out-of-cluster columns
|
|
(``property_id``) are plain indexed ints, not FK constraints, matching the
|
|
mirror convention so ``SQLModel.metadata.create_all`` needs no foreign
|
|
table to exist (the live FKs are owned by the Drizzle schema).
|
|
"""
|
|
|
|
__tablename__: ClassVar[str] = "recommendation" # pyright: ignore[reportIncompatibleVariableOverride]
|
|
|
|
id: Optional[int] = Field(default=None, primary_key=True)
|
|
property_id: int = Field(index=True)
|
|
plan_id: Optional[int] = Field(
|
|
default=None,
|
|
sa_column=Column(
|
|
BigInteger,
|
|
ForeignKey("plan.id", ondelete="CASCADE"),
|
|
nullable=True,
|
|
index=True,
|
|
),
|
|
)
|
|
created_at: Optional[datetime] = Field(
|
|
default=None,
|
|
sa_column=Column(TIMESTAMP, nullable=False, server_default=func.now()),
|
|
)
|
|
|
|
type: str
|
|
measure_type: Optional[str] = Field(default=None)
|
|
description: str
|
|
estimated_cost: Optional[float] = Field(default=None)
|
|
starting_u_value: Optional[float] = Field(default=None)
|
|
new_u_value: Optional[float] = Field(default=None)
|
|
sap_points: Optional[float] = Field(default=None)
|
|
heat_demand: Optional[float] = Field(default=None)
|
|
kwh_savings: Optional[float] = Field(default=None) # delivered kWh/yr
|
|
co2_equivalent_savings: Optional[float] = Field(default=None) # tonnes/yr
|
|
energy_savings: Optional[float] = Field(default=None)
|
|
energy_cost_savings: Optional[float] = Field(default=None) # £/yr
|
|
property_valuation_increase: Optional[float] = Field(default=None)
|
|
rental_yield_increase: Optional[float] = Field(default=None)
|
|
total_work_hours: Optional[float] = Field(default=None)
|
|
labour_days: Optional[float] = Field(default=None)
|
|
default: bool = True
|
|
already_installed: bool = False
|
|
|
|
@classmethod
|
|
def from_domain(
|
|
cls, measure: PlanMeasure, *, property_id: int, plan_id: int
|
|
) -> "RecommendationRow":
|
|
return cls(
|
|
property_id=property_id,
|
|
plan_id=plan_id,
|
|
type=measure.measure_type,
|
|
measure_type=measure.measure_type,
|
|
description=measure.description,
|
|
estimated_cost=measure.cost.total,
|
|
sap_points=measure.impact.sap_points,
|
|
co2_equivalent_savings=(
|
|
measure.impact.co2_savings_kg_per_yr / _KG_PER_TONNE
|
|
),
|
|
kwh_savings=measure.kwh_savings,
|
|
energy_cost_savings=measure.energy_cost_savings,
|
|
default=True,
|
|
already_installed=False,
|
|
)
|
|
|
|
|
|
class RecommendationMaterialRow(SQLModel, table=True):
|
|
"""The live ``recommendation_materials`` table — one row per material used
|
|
by a Recommendation. ``recommendation_id`` is an intra-cluster FK;
|
|
``material_id`` is a plain int (out-of-cluster, mirror convention)."""
|
|
|
|
__tablename__: ClassVar[str] = "recommendation_materials" # pyright: ignore[reportIncompatibleVariableOverride]
|
|
|
|
id: Optional[int] = Field(default=None, primary_key=True)
|
|
recommendation_id: int = Field(
|
|
sa_column=Column(
|
|
BigInteger, ForeignKey("recommendation.id"), nullable=False
|
|
)
|
|
)
|
|
material_id: int = Field(index=True)
|
|
created_at: Optional[datetime] = Field(
|
|
default=None,
|
|
sa_column=Column(TIMESTAMP, nullable=False, server_default=func.now()),
|
|
)
|
|
depth: Optional[float] = Field(default=None)
|
|
quantity: Optional[float] = Field(default=None)
|
|
quantity_unit: Optional[QuantityUnits] = Field(
|
|
default=None,
|
|
sa_column=Column(
|
|
SAEnum(
|
|
QuantityUnits,
|
|
values_callable=lambda cls: [m.value for m in cls], # pyright: ignore[reportUnknownLambdaType, reportUnknownMemberType, reportUnknownVariableType]
|
|
),
|
|
nullable=True,
|
|
),
|
|
)
|
|
estimated_cost: Optional[float] = Field(default=None)
|