mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
7e79c30af1
`PlanMeasure` grows optional `kwh_savings` (delivered energy) and `energy_cost_savings` (£) — its slice of the telescoping bill cascade, signed so positive is a saving and `None` until billing runs. `RecommendationRow` declares the matching live `recommendation.kwh_savings` / `energy_cost_savings` columns and maps them in `from_domain` (None → NULL). The vestigial `recommendation.energy_savings` stays undeclared (legacy = 0). No FE migration — the columns already exist on the live table (ADR-0014 / 0017). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
130 lines
5 KiB
Python
130 lines
5 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import ClassVar, Optional
|
|
|
|
from sqlalchemy import BigInteger, Column, ForeignKey
|
|
from sqlalchemy import Enum as SAEnum
|
|
from sqlmodel import Field, SQLModel
|
|
|
|
from datatypes.epc.domain.epc import Epc
|
|
from domain.modelling.plan import Plan, PlanMeasure
|
|
|
|
# Calculator metrics are in kg CO₂/yr; the live `plan` / `recommendation`
|
|
# columns are tonnes (legacy `emissions_kg / 1000`). Convert on the way in.
|
|
_KG_PER_TONNE = 1000.0
|
|
|
|
|
|
class PlanRow(SQLModel, table=True):
|
|
"""SQLModel mirror of the live ``plan`` table (ADR-0017).
|
|
|
|
Declares only the columns the rebuild writes — identity, the flat
|
|
post-retrofit headline figures, and the cost aggregates. The legacy
|
|
SQLAlchemy model owns the live reads and the columns left for later
|
|
slices (valuation, plan_type, the energy/bill cluster). The physical
|
|
table is the shared contract.
|
|
"""
|
|
|
|
__tablename__: ClassVar[str] = "plan" # pyright: ignore[reportIncompatibleVariableOverride]
|
|
|
|
id: Optional[int] = Field(default=None, primary_key=True)
|
|
portfolio_id: int
|
|
property_id: int = Field(index=True)
|
|
scenario_id: Optional[int] = Field(default=None)
|
|
is_default: bool = False
|
|
|
|
post_sap_points: Optional[float] = Field(default=None)
|
|
post_epc_rating: Optional[Epc] = Field(
|
|
default=None,
|
|
sa_column=Column(SAEnum(Epc, name="epc"), nullable=True),
|
|
)
|
|
post_co2_emissions: Optional[float] = Field(default=None) # tonnes/yr
|
|
co2_savings: Optional[float] = Field(default=None) # tonnes/yr
|
|
cost_of_works: Optional[float] = Field(default=None)
|
|
contingency_cost: Optional[float] = Field(default=None)
|
|
post_energy_bill: Optional[float] = Field(default=None) # £/yr
|
|
energy_bill_savings: Optional[float] = Field(default=None) # £/yr
|
|
post_energy_consumption: Optional[float] = Field(default=None) # delivered kWh/yr
|
|
energy_consumption_savings: Optional[float] = Field(default=None) # kWh/yr
|
|
|
|
@classmethod
|
|
def from_domain(
|
|
cls,
|
|
plan: Plan,
|
|
*,
|
|
property_id: int,
|
|
scenario_id: int,
|
|
portfolio_id: int,
|
|
is_default: bool,
|
|
) -> "PlanRow":
|
|
return cls(
|
|
portfolio_id=portfolio_id,
|
|
property_id=property_id,
|
|
scenario_id=scenario_id,
|
|
is_default=is_default,
|
|
post_sap_points=plan.post_sap_continuous,
|
|
post_epc_rating=plan.post_epc_rating,
|
|
post_co2_emissions=plan.post_retrofit.co2_kg_per_yr / _KG_PER_TONNE,
|
|
co2_savings=plan.co2_savings_kg_per_yr / _KG_PER_TONNE,
|
|
cost_of_works=plan.cost_of_works,
|
|
contingency_cost=plan.contingency_cost,
|
|
post_energy_bill=plan.post_energy_bill,
|
|
energy_bill_savings=plan.energy_bill_savings,
|
|
post_energy_consumption=plan.post_energy_consumption,
|
|
energy_consumption_savings=plan.energy_consumption_savings,
|
|
)
|
|
|
|
|
|
class RecommendationRow(SQLModel, table=True):
|
|
"""SQLModel mirror of the live ``recommendation`` table — one row per
|
|
persisted Plan Measure (ADR-0017). Adds the new ``plan_id`` FK linking the
|
|
measure to its Plan (ON DELETE CASCADE), replacing the ``plan_recommendations``
|
|
m2m for new writes. Only the impact + cost columns the tracer fills are
|
|
declared; the energy/bill, U-value, valuation and labour columns are left
|
|
to later slices.
|
|
"""
|
|
|
|
__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,
|
|
),
|
|
)
|
|
|
|
type: str
|
|
measure_type: Optional[str] = Field(default=None)
|
|
description: str
|
|
estimated_cost: Optional[float] = Field(default=None)
|
|
sap_points: Optional[float] = Field(default=None)
|
|
co2_equivalent_savings: Optional[float] = Field(default=None) # tonnes/yr
|
|
kwh_savings: Optional[float] = Field(default=None) # delivered kWh/yr
|
|
energy_cost_savings: Optional[float] = Field(default=None) # £/yr
|
|
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,
|
|
)
|