Model/scripts/dive_cert.py

"""Deep-dive a single corpus cert: lodged vs computed SAP/CO2/PE + the full
intermediate line-ref dump + the mapped fabric/heat-loss inputs, so the
diverging line is visible WITHOUT an Elmhurst worksheet.

USAGE
    PYTHONPATH=/workspaces/model python scripts/dive_cert.py <cert_number_or_substr>
    PYTHONPATH=/workspaces/model python scripts/dive_cert.py --filter wall_insulation_type=3  [--n 8]
"""
from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import Any

from datatypes.epc.domain.mapper import EpcPropertyDataMapper
from domain.sap10_calculator.calculator import calculate_sap_from_inputs
from domain.sap10_calculator.rdsap.cert_to_inputs import (
    SAP_10_2_SPEC_PRICES,
    cert_to_demand_inputs,
    cert_to_inputs,
)
from scripts.profile_api_error import features

_CORPUS = Path("backend/epc_api/json_samples/RdSAP-Schema-21.0.1/corpus.jsonl")


def _cert_id(doc: dict[str, Any]) -> str:
    return str(
        doc.get("certificate_number")
        or doc.get("lmk_key")
        or doc.get("uprn")
        or "?"
    )


def _dump(doc: dict[str, Any]) -> None:
    cert = _cert_id(doc)
    lodged_sap = doc.get("energy_rating_current")
    lodged_co2 = doc.get("co2_emissions_current")
    lodged_pe = doc.get("energy_consumption_current")
    epc = EpcPropertyDataMapper.from_api_response(doc)
    r = calculate_sap_from_inputs(cert_to_inputs(epc, prices=SAP_10_2_SPEC_PRICES))
    # SAP/EI rating is the UK-average rating cascade (`r`); EPC CO2/PE use the
    # postcode demand cascade (SAP 10.2 Appendix U p.124). Display CO2/PE from
    # the demand cascade so they compare like-for-like with the lodged EPC.
    d = calculate_sap_from_inputs(cert_to_demand_inputs(epc, prices=SAP_10_2_SPEC_PRICES))
    print("=" * 90)
    print(f"CERT {cert}")
    print(
        f"  SAP  lodged={lodged_sap}  ours={r.sap_score_continuous:.2f}  "
        f"d={r.sap_score_continuous - (lodged_sap or 0):+.2f}"
    )
    if lodged_co2 is not None:
        print(
            f"  CO2  lodged={lodged_co2:.3f}  ours={d.co2_kg_per_yr / 1000:.3f} t  "
            f"d={d.co2_kg_per_yr / 1000 - lodged_co2:+.3f}  (demand cascade)"
        )
    if lodged_pe is not None:
        print(
            f"  PE   lodged={lodged_pe:.1f}  ours={d.primary_energy_kwh_per_m2:.1f}  "
            f"d={d.primary_energy_kwh_per_m2 - lodged_pe:+.1f} kWh/m2  (demand cascade)"
        )
    print(
        f"  energy kWh/yr: spaceheat={r.space_heating_kwh_per_yr:.0f} "
        f"main={r.main_heating_fuel_kwh_per_yr:.0f} "
        f"sec={r.secondary_heating_fuel_kwh_per_yr:.0f} "
        f"hw={r.hot_water_kwh_per_yr:.0f} light={r.lighting_kwh_per_yr:.0f} "
        f"pumpfan={r.pumps_fans_kwh_per_yr:.0f}"
    )
    d = epc.__dict__
    print("  --- key mapped inputs ---")
    f = features(doc)
    for k in (
        "property_type", "built_form", "age_band", "main_sap_code",
        "main_heat_cat", "main_fuel", "has_pcdb_main", "main_data_source",
        "wall_construction", "wall_insulation_type", "roof_codes",
        "roof_insulation_thickness", "whc", "water_fuel", "immersion_type",
        "has_cylinder", "has_secondary", "has_pv", "mains_gas", "n_building_parts",
    ):
        print(f"    {k:26s}= {f.get(k)}")
    print("  --- intermediate line refs ---")
    inter = r.intermediate or {}
    for k in sorted(inter):
        print(f"    {k:34s}= {inter[k]:.4f}")


def main() -> None:
    docs = [json.loads(l) for l in _CORPUS.read_text().splitlines() if l.strip()]
    if "--filter" in sys.argv:
        spec = sys.argv[sys.argv.index("--filter") + 1]
        key, _, val = spec.partition("=")
        n = int(sys.argv[sys.argv.index("--n") + 1]) if "--n" in sys.argv else 6
        hits = [d for d in docs if str(features(d).get(key)) == val]
        print(f"{len(hits)} certs match {spec}; dumping first {n}")
        for d in hits[:n]:
            _dump(d)
        return
    target = sys.argv[1]
    for d in docs:
        if target in _cert_id(d):
            _dump(d)
            return
    print(f"no cert matching {target}")


if __name__ == "__main__":
    main()