Merge pull request #1371 from Hestia-Homes/feat/data-export-script

Add data-export script on the new DDD model (replaces principal_pitch/2_export_data.py)

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/<portfolio name>.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())