diff --git a/repositories/scenario_export/scenario_export_postgres_repository.py b/repositories/scenario_export/scenario_export_postgres_repository.py index dabbb52f6..208a7bddc 100644 --- a/repositories/scenario_export/scenario_export_postgres_repository.py +++ b/repositories/scenario_export/scenario_export_postgres_repository.py @@ -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