mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-27 23:35:01 +00:00
Populate export rows with the lodged EPC's descriptive fields 🟩
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar
parent
ed690d9e36
commit
6ea3f31d82
1 changed files with 127 additions and 18 deletions
|
|
@ -1,5 +1,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from datetime import date, datetime
|
||||
from typing import Any, Optional, Sequence
|
||||
|
||||
from sqlalchemy import text
|
||||
|
|
@ -46,6 +47,53 @@ _MEASURES_SQL = text(
|
|||
" AND r.already_installed = FALSE"
|
||||
)
|
||||
|
||||
# The Effective-EPC descriptive picture, read from the persisted lodged EPC slot
|
||||
# (ADR-0065). Header facts, descriptive energy elements, and current performance
|
||||
# are read separately and composed per Property.
|
||||
_EPC_HEADER_SQL = text(
|
||||
"SELECT property_id, property_type, total_floor_area_m2, registration_date,"
|
||||
" habitable_rooms_count"
|
||||
" FROM epc_property"
|
||||
" WHERE property_id = ANY(:property_ids) AND source = 'lodged'"
|
||||
)
|
||||
_EPC_ELEMENTS_SQL = text(
|
||||
"SELECT ep.property_id, e.element_type, e.description"
|
||||
" FROM epc_energy_element e"
|
||||
" JOIN epc_property ep ON ep.id = e.epc_property_id"
|
||||
" WHERE ep.property_id = ANY(:property_ids) AND ep.source = 'lodged'"
|
||||
)
|
||||
_EPC_PERF_SQL = text(
|
||||
"SELECT ep.property_id, perf.energy_rating_current,"
|
||||
" perf.current_energy_efficiency_band"
|
||||
" FROM epc_property_energy_performance perf"
|
||||
" JOIN epc_property ep ON ep.id = perf.epc_property_id"
|
||||
" WHERE ep.property_id = ANY(:property_ids) AND ep.source = 'lodged'"
|
||||
)
|
||||
|
||||
# EPC energy-element type → export descriptive column.
|
||||
_ELEMENT_TO_FIELD: dict[str, str] = {
|
||||
"wall": "walls",
|
||||
"roof": "roof",
|
||||
"floor": "floor",
|
||||
"window": "windows",
|
||||
"main_heating": "heating",
|
||||
"hot_water": "hot_water",
|
||||
"lighting": "lighting",
|
||||
}
|
||||
|
||||
# A cert older than this many years is flagged expired (as data_exports did).
|
||||
_EXPIRY_DAYS = 365 * 10
|
||||
|
||||
|
||||
def _is_expired(registration_date: Optional[str]) -> Optional[bool]:
|
||||
if not registration_date:
|
||||
return None
|
||||
try:
|
||||
lodged = datetime.fromisoformat(registration_date[:10]).date()
|
||||
except ValueError:
|
||||
return None
|
||||
return (date.today() - lodged).days > _EXPIRY_DAYS
|
||||
|
||||
|
||||
def _str(value: Any) -> Optional[str]:
|
||||
"""Render a persisted scalar (int UPRN, Epc enum band) as its string form,
|
||||
|
|
@ -82,26 +130,44 @@ class ScenarioExportPostgresRepository(ScenarioExportRepository):
|
|||
"property_ids": list(property_ids),
|
||||
}
|
||||
measures = self._measures_by_property(params)
|
||||
epc = self._effective_epc_by_property(params)
|
||||
result = self._session.connection().execute(_ROWS_SQL, params)
|
||||
return [
|
||||
PropertyScenarioData(
|
||||
property_id=row[0],
|
||||
landlord_property_id=_str(row[1]),
|
||||
uprn=_str(row[2]),
|
||||
address=_str(row[3]),
|
||||
postcode=_str(row[4]),
|
||||
post_sap_points=_float(row[5]),
|
||||
post_epc_rating=_str(row[6]),
|
||||
cost_of_works=_float(row[7]),
|
||||
contingency_cost=_float(row[8]),
|
||||
co2_savings=_float(row[9]),
|
||||
energy_bill_savings=_float(row[10]),
|
||||
energy_consumption_savings=_float(row[11]),
|
||||
valuation_increase=_float(row[12]),
|
||||
measures=measures.get(row[0], ()),
|
||||
rows: list[PropertyScenarioData] = []
|
||||
for row in result.all():
|
||||
facts = epc.get(row[0], {})
|
||||
rows.append(
|
||||
PropertyScenarioData(
|
||||
property_id=row[0],
|
||||
landlord_property_id=_str(row[1]),
|
||||
uprn=_str(row[2]),
|
||||
address=_str(row[3]),
|
||||
postcode=_str(row[4]),
|
||||
post_sap_points=_float(row[5]),
|
||||
post_epc_rating=_str(row[6]),
|
||||
cost_of_works=_float(row[7]),
|
||||
contingency_cost=_float(row[8]),
|
||||
co2_savings=_float(row[9]),
|
||||
energy_bill_savings=_float(row[10]),
|
||||
energy_consumption_savings=_float(row[11]),
|
||||
valuation_increase=_float(row[12]),
|
||||
property_type=facts.get("property_type"),
|
||||
walls=facts.get("walls"),
|
||||
roof=facts.get("roof"),
|
||||
floor=facts.get("floor"),
|
||||
windows=facts.get("windows"),
|
||||
heating=facts.get("heating"),
|
||||
hot_water=facts.get("hot_water"),
|
||||
lighting=facts.get("lighting"),
|
||||
total_floor_area=facts.get("total_floor_area"),
|
||||
number_of_rooms=facts.get("number_of_rooms"),
|
||||
lodgement_date=facts.get("lodgement_date"),
|
||||
is_expired=facts.get("is_expired"),
|
||||
current_epc_rating=facts.get("current_epc_rating"),
|
||||
current_sap_points=facts.get("current_sap_points"),
|
||||
measures=measures.get(row[0], ()),
|
||||
)
|
||||
)
|
||||
for row in result.all()
|
||||
]
|
||||
return rows
|
||||
|
||||
def _measures_by_property(
|
||||
self, params: dict[str, Any]
|
||||
|
|
@ -121,3 +187,46 @@ class ScenarioExportPostgresRepository(ScenarioExportRepository):
|
|||
)
|
||||
)
|
||||
return by_property
|
||||
|
||||
def _effective_epc_by_property(
|
||||
self, params: dict[str, Any]
|
||||
) -> dict[int, dict[str, Any]]:
|
||||
connection = self._session.connection()
|
||||
facts: dict[int, dict[str, Any]] = {}
|
||||
|
||||
for row in connection.execute(_EPC_HEADER_SQL, params).all():
|
||||
registration_date = _str(row[3])
|
||||
facts.setdefault(row[0], {}).update(
|
||||
{
|
||||
"property_type": _str(row[1]),
|
||||
"total_floor_area": _float(row[2]),
|
||||
"lodgement_date": registration_date,
|
||||
"is_expired": _is_expired(registration_date),
|
||||
"number_of_rooms": row[4],
|
||||
}
|
||||
)
|
||||
|
||||
# A Property can lodge several rows of one element type (e.g. two walls);
|
||||
# join their descriptions, as data_exports did.
|
||||
descriptions: dict[tuple[int, str], list[str]] = {}
|
||||
for row in connection.execute(_EPC_ELEMENTS_SQL, params).all():
|
||||
element_type = _str(row[1]) or ""
|
||||
descriptions.setdefault((row[0], element_type), []).append(
|
||||
_str(row[2]) or ""
|
||||
)
|
||||
for (property_id, element_type), descs in descriptions.items():
|
||||
field = _ELEMENT_TO_FIELD.get(element_type)
|
||||
if field is not None:
|
||||
facts.setdefault(property_id, {})[field] = "; ".join(
|
||||
d for d in descs if d
|
||||
)
|
||||
|
||||
for row in connection.execute(_EPC_PERF_SQL, params).all():
|
||||
facts.setdefault(row[0], {}).update(
|
||||
{
|
||||
"current_sap_points": row[1],
|
||||
"current_epc_rating": _str(row[2]),
|
||||
}
|
||||
)
|
||||
|
||||
return facts
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue