From 59508eb03377f67de44d665399b73c8209741301 Mon Sep 17 00:00:00 2001 From: Jun-te Kim Date: Tue, 30 Jun 2026 17:39:46 +0000 Subject: [PATCH] Add scripts/data_exports.py: principal-pitch export on the new DDD model Replaces sfr/principal_pitch/2_export_data.py, which read the retired plan_recommendations m2m and recommendation_materials table. The new model links a recommendation to its plan directly (recommendation.plan_id), keeps materials inline on the recommendation (material_id), marks the chosen plan per (scenario, property) with is_default, and stores post-works SAP/EPC and savings on the plan row (the new SAP calculator's output). Takes a portfolio id, resolves every modelled scenario (those with plans), and writes one workbook with a properties sheet per scenario. EPC descriptive fields are sourced live from the EPC service (property_details_epc is dead); property_type falls back override -> cert. Co-Authored-By: Claude Opus 4.8 (1M context) --- scripts/data_exports.py | 414 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 414 insertions(+) create mode 100644 scripts/data_exports.py diff --git a/scripts/data_exports.py b/scripts/data_exports.py new file mode 100644 index 000000000..102b1b820 --- /dev/null +++ b/scripts/data_exports.py @@ -0,0 +1,414 @@ +"""Principal-pitch data export — new DDD model edition. + +Replaces sfr/principal_pitch/2_export_data.py, which read the retired +``plan_recommendations`` m2m and ``recommendation_materials`` table. In the +current model: + * a Recommendation links to its Plan directly (``recommendation.plan_id``), + * materials are inline on the Recommendation (``material_id`` etc.), + * the chosen Plan per (scenario, property) is the one with ``is_default``, + * post-works SAP/EPC + savings live on the Plan row (the new SAP calculator's + output), so we read them directly rather than summing recommendation uplifts. + +Give it a portfolio id; it resolves every *modelled* scenario for that portfolio +(scenarios that have plans) and writes ONE workbook with a ``properties`` sheet +per scenario. EPC descriptive fields (walls/roof/heating/windows/floor area/ +lodgement) come live from the EPC service, because ``property_details_epc`` is +dead under the new backend. + + python scripts/data_exports.py --portfolio 814 + python scripts/data_exports.py --portfolio 814 --out "sfr/principal_pitch/Durkan.xlsx" + +Reads DB_* + OPEN_EPC_API_TOKEN from backend/.env. Run from the worktree root. +""" + +from __future__ import annotations + +import argparse +import re +from datetime import date, datetime +from pathlib import Path +from typing import Any, Optional + +import numpy as np +import pandas as pd +from sqlalchemy import text +from sqlalchemy.engine import Engine + +import sys + +_REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(_REPO_ROOT)) + +from backend.app.utils import sap_to_epc # noqa: E402 +from infrastructure.epc_client.epc_client_service import EpcClientService # noqa: E402 +from scripts.e2e_common import ENV_PATH, build_engine, load_env # noqa: E402 +from backend.app.config import get_settings # noqa: E402 + +# Measure columns always present in the wide sheet (stable column set across runs). +EXPECTED_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", +) + + +# --------------------------------------------------------------------------- # +# EPC descriptive fields (live from the EPC service) +# --------------------------------------------------------------------------- # +def _description_text(item: Any) -> str: + if not isinstance(item, dict): + return "" + desc = item.get("description") + if isinstance(desc, dict): + desc = desc.get("value") + return str(desc or "") + + +def _join_descriptions(value: Any) -> str: + if isinstance(value, list): + return "; ".join(t for t in (_description_text(d) for d in value) if t) + return _description_text(value) + + +# Gov RdSAP property-type codes (the raw cert stores a code, not a word). +_PROPERTY_TYPE_CODES: dict[str, str] = { + "0": "House", "1": "Bungalow", "2": "Flat", "3": "Maisonette", "4": "Park home", +} + + +def _decode_property_type(value: Any) -> Optional[str]: + if value is None: + return None + s = str(value).strip() + if s in _PROPERTY_TYPE_CODES: + return _PROPERTY_TYPE_CODES[s] + return s or None + + +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 > 365 * 10 + + +def epc_details_from_service(svc: EpcClientService, uprn: Optional[int]) -> dict[str, Any]: + """Flatten the UPRN's latest raw certificate into the descriptive fields the + export needs. Returns ``{}`` when the UPRN has no EPC (blank columns).""" + if uprn is None: + return {} + results = svc._search(uprn=uprn) # pyright: ignore[reportPrivateUsage] + if not results: + return {} + latest = max(results, key=lambda r: r.registration_date) + raw = svc._fetch_certificate(latest.certificate_number) # pyright: ignore[reportPrivateUsage] + + def _to_int(value: Any) -> Optional[int]: + try: + return int(value) + except (TypeError, ValueError): + return None + + current_sap = _to_int(raw.get("energy_rating_current")) + return { + "property_type": _decode_property_type(raw.get("property_type")), + "walls": _join_descriptions(raw.get("walls")), + "roof": _join_descriptions(raw.get("roofs")), + "floor": _join_descriptions(raw.get("floors")), + "windows": _join_descriptions(raw.get("window")), + "heating": _join_descriptions(raw.get("main_heating")), + "hot_water": _join_descriptions(raw.get("hot_water")), + "lighting": _join_descriptions(raw.get("lighting")), + "total_floor_area": raw.get("total_floor_area"), + "lodgement_date": raw.get("registration_date"), + "is_expired": _is_expired(raw.get("registration_date")), + "current_epc_rating": raw.get("current_energy_efficiency_band"), + "current_sap_points": current_sap, + "original_sap_points": current_sap, + } + + +# --------------------------------------------------------------------------- # +# DB reads (new model: scenario -> plan(is_default) -> recommendation) +# --------------------------------------------------------------------------- # +def modelled_scenarios(engine: Engine, portfolio_id: int) -> list[dict[str, Any]]: + """Scenarios for the portfolio that actually have plans, newest first.""" + with engine.connect() as conn: + rows = conn.execute( + text( + """ + SELECT s.id, s.name + FROM scenario s + WHERE s.portfolio_id = :p + AND EXISTS (SELECT 1 FROM plan pl WHERE pl.scenario_id = s.id) + ORDER BY s.id + """ + ), + {"p": portfolio_id}, + ).mappings().all() + return [dict(r) for r in rows] + + +def load_properties(engine: Engine, portfolio_id: int, svc: EpcClientService) -> pd.DataFrame: + """Base property identity (property_type falls back to the landlord override) + plus live EPC descriptive fields.""" + with engine.connect() as conn: + rows = conn.execute( + text( + """ + SELECT p.id AS property_id, p.id AS id, p.uprn, p.address, p.postcode, + p.landlord_property_id, p.number_of_rooms, + COALESCE(p.property_type, po.override_value) AS property_type + FROM property p + LEFT JOIN property_overrides po + ON po.property_id = p.id + AND po.override_component = 'property_type' + AND po.building_part = 0 + WHERE p.portfolio_id = :p + ORDER BY p.id + """ + ), + {"p": portfolio_id}, + ).mappings().all() + + records: list[dict[str, Any]] = [] + for i, r in enumerate(rows, 1): + base: dict[str, Any] = dict(r) + uprn = int(base["uprn"]) if base.get("uprn") is not None else None + for key, value in epc_details_from_service(svc, uprn).items(): + if base.get(key) is None: + base[key] = value + records.append(base) + if i % 50 == 0: + print(f" EPC fetched {i}/{len(rows)}") + df = pd.DataFrame(records) + df["uprn"] = df["uprn"].astype("string") + return df + + +def load_recommendations(engine: Engine, scenario_id: int) -> pd.DataFrame: + """Default, not-already-installed recommendations on each property's default + plan for the scenario, with the material type/battery flag joined.""" + with engine.connect() as conn: + rows = conn.execute( + text( + """ + SELECT pl.property_id, + r.measure_type, + r.description, + r.estimated_cost, + r.sap_points, + r.co2_equivalent_savings, + r.kwh_savings, + r.energy_cost_savings, + m.type AS material_type, + 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.scenario_id = :s + AND pl.is_default = TRUE + AND r.default = TRUE + AND r.already_installed = FALSE + """ + ), + {"s": scenario_id}, + ).mappings().all() + return pd.DataFrame([dict(r) for r in rows]) + + +def load_default_plans(engine: Engine, scenario_id: int) -> pd.DataFrame: + """The chosen (is_default) plan per property — the new SAP calculator's + post-works results.""" + with engine.connect() as conn: + rows = conn.execute( + text( + """ + SELECT property_id, post_sap_points, post_epc_rating, + cost_of_works, contingency_cost, co2_savings, + energy_bill_savings, energy_consumption_savings, + valuation_increase + FROM plan + WHERE scenario_id = :s AND is_default = TRUE + """ + ), + {"s": scenario_id}, + ).mappings().all() + return pd.DataFrame([dict(r) for r in rows]) + + +# --------------------------------------------------------------------------- # +# Sheet building +# --------------------------------------------------------------------------- # +def _apply_battery_suffix(recs: pd.DataFrame) -> pd.DataFrame: + """solar_pv recommendations that carry a battery material become + solar_pv_with_battery (mirrors the old export).""" + if recs.empty: + return recs + is_solar_battery = (recs["material_type"] == "solar_pv") & (recs["includes_battery"]) + recs = recs.copy() + recs["measure_type"] = np.where( + is_solar_battery, + recs["measure_type"].astype(str) + "_with_battery", + recs["measure_type"], + ) + return recs + + +def build_scenario_sheet( + properties_df: pd.DataFrame, recs: pd.DataFrame, plans: pd.DataFrame +) -> pd.DataFrame: + recs = _apply_battery_suffix(recs) + + # Pivot: one column per measure_type holding its estimated_cost. + if not recs.empty: + deduped = recs.drop_duplicates(subset=["property_id", "measure_type"], keep="first") + cost_pivot = deduped.pivot( + index="property_id", columns="measure_type", values="estimated_cost" + ).reset_index() + sap_uplift = ( + recs.groupby("property_id")["sap_points"].sum().reset_index(name="sap_points") + ) + savings = ( + recs.groupby("property_id")[ + ["co2_equivalent_savings", "kwh_savings", "energy_cost_savings"] + ] + .sum() + .reset_index() + ) + else: + cost_pivot = pd.DataFrame({"property_id": []}) + sap_uplift = pd.DataFrame({"property_id": [], "sap_points": []}) + savings = pd.DataFrame( + {"property_id": [], "co2_equivalent_savings": [], "kwh_savings": [], "energy_cost_savings": []} + ) + + id_cols = [ + c + for c in [ + "landlord_property_id", "property_id", "uprn", "address", "postcode", + "property_type", "walls", "roof", "heating", "windows", + "current_epc_rating", "current_sap_points", "original_sap_points", + "total_floor_area", "number_of_rooms", "lodgement_date", "is_expired", "id", + ] + if c in properties_df.columns + ] + + df = ( + properties_df[id_cols] + .merge(cost_pivot, how="left", on="property_id") + .merge(sap_uplift, how="left", on="property_id") + .merge(savings, how="left", on="property_id") + .merge(plans, how="left", on="property_id") + ) + + # total retrofit cost = sum of the per-measure cost columns + measure_cols_present = [c for c in df.columns if c in set(EXPECTED_MEASURE_COLUMNS) + or c.endswith("_with_battery")] + df["total_retrofit_cost"] = df[measure_cols_present].sum(axis=1) if measure_cols_present else 0.0 + + df["sap_points"] = df["sap_points"].fillna(0) + # Post-works SAP/EPC straight from the new SAP calculator's plan row; + # fall back to current + uplift / sap_to_epc only when the plan lacks them. + df["predicted_post_works_sap"] = df["post_sap_points"].where( + df["post_sap_points"].notna(), df.get("current_sap_points", 0) + df["sap_points"] + ) + df["predicted_post_works_epc"] = df["post_epc_rating"].where( + df["post_epc_rating"].notna(), + df["predicted_post_works_sap"].apply(lambda x: sap_to_epc(x) if pd.notna(x) else None), + ) + + # ensure the stable measure column set exists + for col in EXPECTED_MEASURE_COLUMNS: + if col not in df.columns: + df[col] = "" + return df + + +def _safe_sheet_name(name: str, used: set[str]) -> str: + clean = re.sub(r"[:\\/?*\[\]]", "", name or "scenario").strip() or "scenario" + clean = clean[:31] + base, i = clean, 1 + while clean in used: + suffix = f" ({i})" + clean = base[: 31 - len(suffix)] + suffix + i += 1 + used.add(clean) + return clean + + +def export_portfolio(portfolio_id: int, out_path: Path) -> None: + load_env(ENV_PATH) + settings = get_settings() + engine = build_engine() + svc = EpcClientService(auth_token=settings.OPEN_EPC_API_TOKEN) + + with engine.connect() as conn: + pname = conn.execute( + text("SELECT name FROM portfolio WHERE id = :p"), {"p": portfolio_id} + ).scalar() + + scenarios = modelled_scenarios(engine, portfolio_id) + if not scenarios: + raise SystemExit(f"No modelled scenarios (with plans) for portfolio {portfolio_id}.") + print(f"Portfolio {portfolio_id} ({pname}) — {len(scenarios)} modelled scenario(s): " + f"{[s['name'] for s in scenarios]}") + + print("Loading properties + EPC descriptive fields…") + properties_df = load_properties(engine, portfolio_id, svc) + + out_path.parent.mkdir(parents=True, exist_ok=True) + used_names: set[str] = set() + with pd.ExcelWriter(out_path) as writer: + for s in scenarios: + recs = load_recommendations(engine, s["id"]) + plans = load_default_plans(engine, s["id"]) + sheet_df = build_scenario_sheet(properties_df, recs, plans) + sheet = _safe_sheet_name(s["name"] or f"scenario_{s['id']}", used_names) + sheet_df.to_excel(writer, sheet_name=sheet, index=False) + print(f" sheet {sheet!r}: {len(sheet_df)} properties, " + f"{0 if recs.empty else len(recs)} recommendations") + print(f"Wrote {out_path}") + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--portfolio", type=int, required=True) + ap.add_argument("--out", type=Path, default=None, + help="output xlsx path (default: sfr/principal_pitch/.xlsx)") + args = ap.parse_args() + + out = args.out + if out is None: + load_env(ENV_PATH) + with build_engine().connect() as conn: + nm = conn.execute(text("SELECT name FROM portfolio WHERE id=:p"), {"p": args.portfolio}).scalar() + safe = re.sub(r"[\\/:*?\"<>|]", "_", str(nm or f"portfolio_{args.portfolio}")) + out = _REPO_ROOT / "sfr" / "principal_pitch" / f"{safe}.xlsx" + export_portfolio(args.portfolio, out) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())