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

Resolve Effective-EPC fields by override then lodged then predicted 🟩

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:40:53 +00:00
parent 642789db01
commit 183d14f5f4
1 changed files with 41 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

47
repositories/scenario_export/scenario_export_postgres_repository.py
View file

@ -54,20 +54,29 @@ _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'"
" WHERE property_id = ANY(:property_ids) AND source = :source"
)
_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'"
" WHERE ep.property_id = ANY(:property_ids) AND ep.source = :source"
)
_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'"
" WHERE ep.property_id = ANY(:property_ids) AND ep.source = :source"
)
# A landlord property_type override wins over any EPC-derived value (ADR-0065);
# the descriptive prose fields have no override prose, so only property_type is
# resolved this way (the override enum carries coded fabric facts, not prose).
_OVERRIDE_PROPERTY_TYPE_SQL = text(
"SELECT property_id, override_value FROM property_overrides"
" WHERE property_id = ANY(:property_ids) AND building_part = 0"
" AND override_component = 'property_type'"
)
# EPC energy-element type → export descriptive column.
@ -190,11 +199,37 @@ class ScenarioExportPostgresRepository(ScenarioExportRepository):
def _effective_epc_by_property(
self, params: dict[str, Any]
) -> dict[int, dict[str, Any]]:
"""The override-resolved Effective-EPC descriptive picture per Property:
override → lodged → predicted (ADR-0065)."""
lodged = self._epc_facts_for_source(params, "lodged")
predicted = self._epc_facts_for_source(params, "predicted")
facts: dict[int, dict[str, Any]] = {}
for property_id in set(lodged) | set(predicted):
merged = dict(predicted.get(property_id, {}))
# A lodged cert, where present, wins field by field (a null lodged
# field falls through to the predicted value).
for key, value in lodged.get(property_id, {}).items():
if value is not None:
merged[key] = value
facts[property_id] = merged
for row in self._session.connection().execute(
_OVERRIDE_PROPERTY_TYPE_SQL, params
).all():
facts.setdefault(row[0], {})["property_type"] = _str(row[1])
return facts
def _epc_facts_for_source(
self, params: dict[str, Any], source: str
) -> dict[int, dict[str, Any]]:
connection = self._session.connection()
source_params: dict[str, Any] = {**params, "source": source}
facts: dict[int, dict[str, Any]] = {}
for row in connection.execute(_EPC_HEADER_SQL, params).all():
for row in connection.execute(_EPC_HEADER_SQL, source_params).all():
registration_date = _str(row[3])
facts.setdefault(row[0], {}).update(
{
@ -209,7 +244,7 @@ class ScenarioExportPostgresRepository(ScenarioExportRepository):
# 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():
for row in connection.execute(_EPC_ELEMENTS_SQL, source_params).all():
element_type = _str(row[1]) or ""
descriptions.setdefault((row[0], element_type), []).append(
_str(row[2]) or ""
@ -221,7 +256,7 @@ class ScenarioExportPostgresRepository(ScenarioExportRepository):
d for d in descs if d
)
for row in connection.execute(_EPC_PERF_SQL, params).all():
for row in connection.execute(_EPC_PERF_SQL, source_params).all():
facts.setdefault(row[0], {}).update(
{
"current_sap_points": row[1],