junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-27 23:35:01 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Map the default plan's selected measures onto the export row 🟩

Browse source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-07-13 23:30:11 +00:00
parent 5dab7beae5
commit 432898444c
1 changed files with 47 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
repositories/scenario_export/scenario_export_postgres_repository.py
View file

@ -5,7 +5,7 @@ from typing import Any, Optional, Sequence
from sqlalchemy import text
from sqlmodel import Session
from domain.scenario_export.scenario_sheet import PropertyScenarioData
from domain.scenario_export.scenario_sheet import ExportMeasure, PropertyScenarioData
from repositories.scenario_export.scenario_export_repository import (
ScenarioExportRepository,
)
@ -27,6 +27,25 @@ _ROWS_SQL = text(
" ORDER BY p.id"
)
# The selected measures on each Property's default Plan: the recommendations
# kept in the Optimised Package (``default`` and not ``already_installed``),
# with the installed material's battery flag for the solar pivot (ADR-0065).
# ``default`` is a reserved word, hence the quoting.
_MEASURES_SQL = text(
"SELECT r.property_id, r.measure_type, r.estimated_cost, r.sap_points,"
" r.co2_equivalent_savings, r.kwh_savings, r.energy_cost_savings,"
" COALESCE(m.includes_battery, FALSE) AS includes_battery"
" FROM recommendation r"
" JOIN plan pl ON pl.id = r.plan_id"
" LEFT JOIN material m ON m.id = r.material_id"
" WHERE pl.portfolio_id = :portfolio_id"
" AND pl.scenario_id = :scenario_id"
" AND pl.is_default = TRUE"
" AND pl.property_id = ANY(:property_ids)"
' AND r."default" = TRUE'
" AND r.already_installed = FALSE"
)
def _str(value: Any) -> Optional[str]:
"""Render a persisted scalar (int UPRN, Epc enum band) as its string form,
@ -57,14 +76,13 @@ class ScenarioExportPostgresRepository(ScenarioExportRepository):
) -> list[PropertyScenarioData]:
if not property_ids:
return []
result = self._session.connection().execute(
_ROWS_SQL,
{
"portfolio_id": portfolio_id,
"scenario_id": scenario_id,
"property_ids": list(property_ids),
},
)
params: dict[str, Any] = {
"portfolio_id": portfolio_id,
"scenario_id": scenario_id,
"property_ids": list(property_ids),
}
measures = self._measures_by_property(params)
result = self._session.connection().execute(_ROWS_SQL, params)
return [
PropertyScenarioData(
property_id=row[0],
@ -80,6 +98,26 @@ class ScenarioExportPostgresRepository(ScenarioExportRepository):
energy_bill_savings=_float(row[10]),
energy_consumption_savings=_float(row[11]),
valuation_increase=_float(row[12]),
measures=measures.get(row[0], ()),
)
for row in result.all()
]
def _measures_by_property(
self, params: dict[str, Any]
) -> dict[int, list[ExportMeasure]]:
result = self._session.connection().execute(_MEASURES_SQL, params)
by_property: dict[int, list[ExportMeasure]] = {}
for row in result.all():
by_property.setdefault(row[0], []).append(
ExportMeasure(
measure_type=_str(row[1]) or "",
estimated_cost=_float(row[2]),
sap_points=_float(row[3]),
co2_equivalent_savings=_float(row[4]),
kwh_savings=_float(row[5]),
energy_cost_savings=_float(row[6]),
includes_battery=bool(row[7]),
)
)
return by_property