mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-27 23:35:01 +00:00
35a0c6e972
The "latest EPC per UPRN" selection ordered by e.id DESC, but row-id does not track recency: a gov-EPC cert re-ingested after a PasHub survey lands with a higher e.id, so the header/perf reads silently picked the stale gov cert (wrong TFA, lodgement, property_type, current band) on 3 of portfolio 838. Order by the effective lodgement date (registration -> completion -> inspection) DESC NULLS LAST, e.id DESC tiebreak. Compose the descriptive fabric from the structured SAP fields a re-survey lodges (no gov-EPC prose): walls/roof/floor from the main epc_building_part, heating and hot water from epc_main_heating_detail + the water-heating codes, decoded with the SAP engine's own code space (fabric_description.py). Structured fabric wins where present and falls back to the gov-EPC prose otherwise; windows and lighting stay on prose (their per-element codes do not reconstruct the dwelling phrase). Add built_form (epc_property, code-decoded) and property_age_band (building part construction_age_band -> RdSAP date range) columns. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
250 lines
9.4 KiB
Python
250 lines
9.4 KiB
Python
"""The Scenario Export sheet shape (ADR-0065).
|
|
|
|
Pure logic: given the persisted, override-resolved data for the Properties in one
|
|
Scenario, lay out the wide export sheet — property identity + Effective-EPC
|
|
descriptive fields, each measure's cost pivoted onto its own column against a
|
|
frozen column contract, savings summed per Property, and the Plan's post-works
|
|
figures. No I/O: the repository reads the rows and the infrastructure layer
|
|
renders the workbook; this module only shapes one sheet's rows.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Optional, Sequence
|
|
|
|
# The measure cost columns always present on every sheet, in a stable order, so
|
|
# every scenario sheet in the workbook shares one column contract regardless of
|
|
# which measures a given selection happens to contain (ADR-0065). A Property that
|
|
# lacks a measure carries a blank in that column, never a missing column.
|
|
MEASURE_COLUMNS: tuple[str, ...] = (
|
|
"suspended_floor_insulation",
|
|
"solid_floor_insulation",
|
|
"external_wall_insulation",
|
|
"internal_wall_insulation",
|
|
"cavity_wall_insulation",
|
|
"loft_insulation",
|
|
"flat_roof_insulation",
|
|
"room_roof_insulation",
|
|
"secondary_glazing",
|
|
"double_glazing",
|
|
"solar_pv",
|
|
"high_heat_retention_storage_heaters",
|
|
"air_source_heat_pump",
|
|
"boiler_upgrade",
|
|
"gas_boiler_upgrade",
|
|
"roomstat_programmer_trvs",
|
|
"time_temperature_zone_control",
|
|
"low_energy_lighting",
|
|
"mechanical_ventilation",
|
|
"system_tune_up",
|
|
"system_tune_up_zoned",
|
|
)
|
|
|
|
# A solar_pv Option that installs a battery pivots to its own column, kept in the
|
|
# frozen contract so a sheet with battery-solar shares columns with one without.
|
|
_BATTERY_MEASURE_COLUMNS: tuple[str, ...] = ("solar_pv_with_battery",)
|
|
|
|
# Every measure cost column present on a sheet: the frozen base contract plus the
|
|
# battery variant.
|
|
_ALL_MEASURE_COLUMNS: tuple[str, ...] = MEASURE_COLUMNS + _BATTERY_MEASURE_COLUMNS
|
|
|
|
# Property identity and Effective-EPC descriptive columns, leading every sheet
|
|
# before the measure columns. The descriptive fields are already override-resolved
|
|
# by the repository (property_overrides → lodged EPC → predicted, ADR-0065).
|
|
_ID_COLUMNS: tuple[str, ...] = (
|
|
"property_id",
|
|
"landlord_property_id",
|
|
"uprn",
|
|
"address",
|
|
"postcode",
|
|
"property_type",
|
|
"built_form",
|
|
"property_age_band",
|
|
"walls",
|
|
"roof",
|
|
"floor",
|
|
"windows",
|
|
"heating",
|
|
"hot_water",
|
|
"lighting",
|
|
"total_floor_area",
|
|
"number_of_rooms",
|
|
"lodgement_date",
|
|
"is_expired",
|
|
"current_epc_rating",
|
|
"current_sap_points",
|
|
)
|
|
|
|
# Per-Property roll-ups of the selected measures' attributed impact, trailing the
|
|
# measure columns (ADR-0065).
|
|
_SAVINGS_COLUMNS: tuple[str, ...] = (
|
|
"sap_points",
|
|
"co2_equivalent_savings",
|
|
"kwh_savings",
|
|
"energy_cost_savings",
|
|
)
|
|
|
|
# Package-level totals, trailing the savings roll-ups.
|
|
_TOTAL_COLUMNS: tuple[str, ...] = ("total_retrofit_cost",)
|
|
|
|
# The default Plan's post-works figures, taken straight from the persisted Plan
|
|
# (the SAP calculator's output) with no recompute (ADR-0065).
|
|
_PLAN_COLUMNS: tuple[str, ...] = (
|
|
"post_sap_points",
|
|
"post_epc_rating",
|
|
"cost_of_works",
|
|
"contingency_cost",
|
|
"co2_savings",
|
|
"energy_bill_savings",
|
|
"energy_consumption_savings",
|
|
"valuation_increase",
|
|
)
|
|
|
|
|
|
def _blank(value: Any) -> Any:
|
|
"""A missing value renders as a blank cell, not a zero or ``None``."""
|
|
return "" if value is None else value
|
|
|
|
|
|
def _sum(values: Sequence[Optional[float]]) -> float:
|
|
"""Sum optional numbers, treating an absent contribution as zero."""
|
|
return sum(value or 0.0 for value in values)
|
|
|
|
|
|
def _measure_column(measure: ExportMeasure) -> str:
|
|
"""The pivot column for a measure: a solar_pv Option carrying a battery lands
|
|
in its own ``solar_pv_with_battery`` column (ADR-0065)."""
|
|
if measure.measure_type == "solar_pv" and measure.includes_battery:
|
|
return "solar_pv_with_battery"
|
|
return measure.measure_type
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ExportMeasure:
|
|
"""One selected measure on a Property's default Plan for the Scenario: the
|
|
measure type (the pivot column), its installed cost, and the SAP / carbon /
|
|
energy / bill savings attributed to it. ``includes_battery`` distinguishes a
|
|
solar-with-battery Option so it pivots to its own column (ADR-0065)."""
|
|
|
|
measure_type: str
|
|
estimated_cost: Optional[float] = None
|
|
sap_points: Optional[float] = None
|
|
co2_equivalent_savings: Optional[float] = None
|
|
kwh_savings: Optional[float] = None
|
|
energy_cost_savings: Optional[float] = None
|
|
includes_battery: bool = False
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PropertyScenarioData:
|
|
"""One Property's persisted data for one Scenario: identity, the
|
|
Effective-EPC descriptive fields (already override-resolved by the
|
|
repository), the default Plan's post-works figures, and the selected
|
|
measures. The input row the shaper turns into one wide sheet row."""
|
|
|
|
property_id: int
|
|
landlord_property_id: Optional[str] = None
|
|
uprn: Optional[str] = None
|
|
address: Optional[str] = None
|
|
postcode: Optional[str] = None
|
|
property_type: Optional[str] = None
|
|
built_form: Optional[str] = None
|
|
property_age_band: Optional[str] = None
|
|
walls: Optional[str] = None
|
|
roof: Optional[str] = None
|
|
floor: Optional[str] = None
|
|
windows: Optional[str] = None
|
|
heating: Optional[str] = None
|
|
hot_water: Optional[str] = None
|
|
lighting: Optional[str] = None
|
|
total_floor_area: Optional[float] = None
|
|
number_of_rooms: Optional[int] = None
|
|
lodgement_date: Optional[str] = None
|
|
is_expired: Optional[bool] = None
|
|
current_epc_rating: Optional[str] = None
|
|
current_sap_points: Optional[float] = None
|
|
post_sap_points: Optional[float] = None
|
|
post_epc_rating: Optional[str] = None
|
|
cost_of_works: Optional[float] = None
|
|
contingency_cost: Optional[float] = None
|
|
co2_savings: Optional[float] = None
|
|
energy_bill_savings: Optional[float] = None
|
|
energy_consumption_savings: Optional[float] = None
|
|
valuation_increase: Optional[float] = None
|
|
measures: Sequence[ExportMeasure] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ExportSheet:
|
|
"""One laid-out scenario sheet: the ordered column contract and the rows,
|
|
each a column-keyed mapping ready for the workbook renderer."""
|
|
|
|
columns: tuple[str, ...]
|
|
rows: tuple[dict[str, Any], ...]
|
|
|
|
|
|
_COLUMNS: tuple[str, ...] = (
|
|
_ID_COLUMNS + _ALL_MEASURE_COLUMNS + _SAVINGS_COLUMNS + _TOTAL_COLUMNS + _PLAN_COLUMNS
|
|
)
|
|
|
|
|
|
def _shape_row(prop: PropertyScenarioData) -> dict[str, Any]:
|
|
"""One Property's wide row: identity + Effective-EPC descriptive fields, the
|
|
measures pivoted onto their cost columns, the savings rolled up, and the
|
|
Plan's post-works figures (ADR-0065)."""
|
|
row: dict[str, Any] = {
|
|
"property_id": prop.property_id,
|
|
"landlord_property_id": _blank(prop.landlord_property_id),
|
|
"uprn": _blank(prop.uprn),
|
|
"address": _blank(prop.address),
|
|
"postcode": _blank(prop.postcode),
|
|
"property_type": _blank(prop.property_type),
|
|
"built_form": _blank(prop.built_form),
|
|
"property_age_band": _blank(prop.property_age_band),
|
|
"walls": _blank(prop.walls),
|
|
"roof": _blank(prop.roof),
|
|
"floor": _blank(prop.floor),
|
|
"windows": _blank(prop.windows),
|
|
"heating": _blank(prop.heating),
|
|
"hot_water": _blank(prop.hot_water),
|
|
"lighting": _blank(prop.lighting),
|
|
"total_floor_area": _blank(prop.total_floor_area),
|
|
"number_of_rooms": _blank(prop.number_of_rooms),
|
|
"lodgement_date": _blank(prop.lodgement_date),
|
|
"is_expired": _blank(prop.is_expired),
|
|
"current_epc_rating": _blank(prop.current_epc_rating),
|
|
"current_sap_points": _blank(prop.current_sap_points),
|
|
}
|
|
# Every measure column is present, blank unless this Property has it, so all
|
|
# scenario sheets share one contract (ADR-0065).
|
|
for column in _ALL_MEASURE_COLUMNS:
|
|
row[column] = ""
|
|
for measure in prop.measures:
|
|
column = _measure_column(measure)
|
|
if column in _ALL_MEASURE_COLUMNS:
|
|
row[column] = measure.estimated_cost
|
|
# Roll the selected measures' attributed impact up to the Property.
|
|
row["sap_points"] = _sum([m.sap_points for m in prop.measures])
|
|
row["co2_equivalent_savings"] = _sum([m.co2_equivalent_savings for m in prop.measures])
|
|
row["kwh_savings"] = _sum([m.kwh_savings for m in prop.measures])
|
|
row["energy_cost_savings"] = _sum([m.energy_cost_savings for m in prop.measures])
|
|
row["total_retrofit_cost"] = _sum([m.estimated_cost for m in prop.measures])
|
|
# The default Plan's post-works figures, passed through as persisted.
|
|
row["post_sap_points"] = _blank(prop.post_sap_points)
|
|
row["post_epc_rating"] = _blank(prop.post_epc_rating)
|
|
row["cost_of_works"] = _blank(prop.cost_of_works)
|
|
row["contingency_cost"] = _blank(prop.contingency_cost)
|
|
row["co2_savings"] = _blank(prop.co2_savings)
|
|
row["energy_bill_savings"] = _blank(prop.energy_bill_savings)
|
|
row["energy_consumption_savings"] = _blank(prop.energy_consumption_savings)
|
|
row["valuation_increase"] = _blank(prop.valuation_increase)
|
|
return row
|
|
|
|
|
|
def shape_scenario_sheet(
|
|
properties: Sequence[PropertyScenarioData],
|
|
) -> ExportSheet:
|
|
"""Shape one Scenario's Properties into a wide export sheet (ADR-0065)."""
|
|
rows = tuple(_shape_row(prop) for prop in properties)
|
|
return ExportSheet(columns=_COLUMNS, rows=rows)
|