Model/scripts/elmhurst_input_sheet.py

"""Render a human-readable "Elmhurst SAP input sheet" for one or more certs.

WHAT THIS IS FOR
----------------
The debugging companion to `eval_api_sap_accuracy.py`: once that script names a
worst-offender cert, this dumps everything the mapper hands the calculator —
the *codes the calculator actually sees* (`from_api_response` →
`EpcPropertyData`) — in the same readable layout as the worked
`sap worksheets/golden fixture debugging/6035_elmhurst_input_sheet.md`, plus
the lodged reference outputs the worksheet must reproduce and our own
continuous SAP next to the lodged value. You read it side-by-side with the
real Elmhurst Summary / P960 worksheet PDF to localise where we diverge.

USAGE
-----
    PYTHONPATH=/workspaces/model python scripts/elmhurst_input_sheet.py <cert> [<cert> ...]

    # write each sheet to a file instead of stdout:
    PYTHONPATH=/workspaces/model python scripts/elmhurst_input_sheet.py --out-dir "sap worksheets/golden fixture debugging" <cert>

Certs are read from the cache built by `fetch_2026_epc_sample.py` (default
`/tmp/epc_2026_sample`, overridable via `EPC_SAMPLE_CACHE`). A bare cert
number resolves to `<cache>/<cert>.json`; an explicit path is also accepted.
"""
from __future__ import annotations

import json
import math
import os
import sys
from pathlib import Path
from typing import Any, Optional

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_inputs

CACHE = Path(os.environ.get("EPC_SAMPLE_CACHE", "/tmp/epc_2026_sample"))


def _num(v: Any) -> Any:
    """Unwrap a Measurement (`.value`) or pass an int/float/str through."""
    return getattr(v, "value", v)


def _resolve(cert_arg: str) -> Path:
    p = Path(cert_arg)
    if p.suffix == ".json" and p.exists():
        return p
    cached = CACHE / f"{cert_arg}.json"
    if cached.exists():
        return cached
    raise FileNotFoundError(
        f"No cached JSON for {cert_arg!r} (looked at {cached}). "
        f"Run scripts/fetch_2026_epc_sample.py or set EPC_SAMPLE_CACHE."
    )


def _our_sap(epc: Any) -> str:
    """Our continuous SAP, or the exception that blocks it."""
    try:
        result = calculate_sap_from_inputs(cert_to_inputs(epc, prices=SAP_10_2_SPEC_PRICES))
        cont: float = result.sap_score_continuous
        return f"{cont:.4f}" if math.isfinite(cont) else f"non-finite ({cont})"
    except Exception as e:  # debugging tool — surface, don't swallow
        return f"RAISED {type(e).__name__}: {e}"


def render(cert: str, doc: dict[str, Any]) -> str:
    epc = EpcPropertyDataMapper.from_api_response(doc)
    out: list[str] = []
    w = out.append

    # --- header --------------------------------------------------------
    w(f"# Cert {cert} — Elmhurst SAP input sheet\n")
    addr = ", ".join(
        str(x) for x in (epc.address_line_1, epc.post_town, epc.postcode) if x
    )
    w(f"Address: {addr}")
    w(
        f"Dwelling: {epc.dwelling_type}  built_form={epc.built_form}  "
        f"property_type={epc.property_type}"
    )
    w(
        f"TFA: {epc.total_floor_area_m2} m²   "
        f"habitable_rooms={epc.habitable_rooms_count}  "
        f"heated_rooms={epc.heated_rooms_count}"
    )
    w(
        f"Extensions: {epc.extensions_count}   region_code={epc.region_code}  "
        f"measurement_type={epc.measurement_type}"
    )
    w(
        f"Pressure test: {epc.pressure_test if epc.pressure_test is not None else '(not tested)'}  "
        f"door_count={epc.door_count}"
    )
    w(
        f"Conservatory: type={epc.conservatory_type}  "
        f"heated_sep_consv={str(epc.has_heated_separate_conservatory).lower()}"
    )

    # --- our vs lodged (debug aid) -------------------------------------
    lodged = doc.get("energy_rating_current")
    our = _our_sap(epc)
    delta = ""
    try:
        if lodged is not None and not our.startswith(("RAISED", "non-finite")):
            delta = f"   Δ={float(our) - float(lodged):+.4f} (we − lodged)"
    except ValueError:
        pass
    w(f"\n## SAP: OURS={our}  LODGED={lodged}{delta}")

    # --- element descriptions (lodged) ---------------------------------
    w("\n## Element descriptions (lodged)")
    for label, elems in (
        ("WALL", epc.walls), ("ROOF", epc.roofs), ("FLOOR", epc.floors),
    ):
        for el in elems or []:
            w(f"  {label}: {el.description}")
    if epc.window:
        w(f"  WINDOW: {epc.window.description}")
    for el in epc.main_heating or []:
        w(f"  MAIN HEATING: {el.description}")
    if epc.hot_water:
        w(f"  HOT WATER: {epc.hot_water.description}")
    if epc.lighting:
        w(f"  LIGHTING: {epc.lighting.description}")
    if epc.secondary_heating:
        w(f"  SECONDARY: {epc.secondary_heating.description}")

    # --- building parts / dimensions -----------------------------------
    # NB direct attribute access (not getattr-with-default) so a future
    # domain rename fails loudly here rather than silently printing None
    # over real data. Field names are the `epc_property_data` domain
    # types the mapper emits (NOT the `schema` dataclasses).
    w("\n## Building parts / dimensions")
    for bp in epc.sap_building_parts or []:
        w(f"### {bp.identifier} (part {bp.building_part_number}, age {bp.construction_age_band})")
        w(
            f"  wall_construction={bp.wall_construction} "
            f"insulation_type={bp.wall_insulation_type} "
            f"ins_thick={bp.wall_insulation_thickness} "
            f"wall_thickness={bp.wall_thickness_mm}mm "
            f"measured={bp.wall_thickness_measured} "
            f"dry_lined={bp.wall_dry_lined}"
        )
        w(f"  party_wall_construction={bp.party_wall_construction}")
        w(
            f"  roof_construction={bp.roof_construction} ({bp.roof_construction_type}) "
            f"ins_location={bp.roof_insulation_location} "
            f"ins_thick={bp.roof_insulation_thickness}"
        )
        w(
            f"  floor_heat_loss={bp.floor_heat_loss} ({bp.floor_type}) "
            f"floor_ins_thick={bp.floor_insulation_thickness}"
        )
        rir = bp.sap_room_in_roof
        if rir is not None:
            w(
                f"  ROOM-IN-ROOF: floor_area={_num(rir.floor_area)} "
                f"age={rir.construction_age_band} "
                f"gable1={rir.gable_1_length_m}x{rir.gable_1_height_m}m "
                f"gable2={rir.gable_2_length_m}x{rir.gable_2_height_m}m "
                f"common_wall={rir.common_wall_length_m}m"
            )
        for fd in bp.sap_floor_dimensions or []:
            w(
                f"  floor {fd.floor}: area={fd.total_floor_area_m2} "
                f"height={fd.room_height_m} "
                f"HLP={fd.heat_loss_perimeter_m} "
                f"party_wall_len={fd.party_wall_length_m} "
                f"floor_constr={fd.floor_construction} floor_ins={fd.floor_insulation} "
                f"exposed={fd.is_exposed_floor}"
            )

    # --- windows -------------------------------------------------------
    windows = epc.sap_windows or []
    w(f"\n## Windows ({len(windows)})")
    for i, win in enumerate(windows):
        w(
            f"  W{i}: {win.window_width}x{win.window_height}m "
            f"orient={win.orientation} "
            f"glazing_type={win.glazing_type} "
            f"gap={win.glazing_gap} "
            f"frame={win.frame_material} "
            f"draught={win.draught_proofed} "
            f"loc(bp)={win.window_location} "
            f"wall_type={win.window_wall_type} "
            f"frame_factor={win.frame_factor}"
        )

    # --- doors / heating / water / vent --------------------------------
    w("\n## Doors / heating / water / vent")
    w(
        f"  door_count={epc.door_count} "
        f"insulated_door_count={epc.insulated_door_count}"
    )
    sh = epc.sap_heating
    for mh in sh.main_heating_details or []:
        w(
            f"  MAIN: sap_code={mh.sap_main_heating_code} "
            f"fuel={mh.main_fuel_type} "
            f"category={mh.main_heating_category} "
            f"emitter={mh.heat_emitter_type} "
            f"emit_temp={mh.emitter_temperature} "
            f"control={mh.main_heating_control} "
            f"fghrs={mh.has_fghrs} "
            f"fan_flue={mh.fan_flue_present} "
            f"flue_type={mh.boiler_flue_type} "
            f"pump_age={mh.central_heating_pump_age} "
            f"data_source={mh.main_heating_data_source} "
            f"idx={mh.main_heating_index_number} "
            f"fraction={mh.main_heating_fraction}"
        )
    w(
        f"  WATER: code={sh.water_heating_code} "
        f"fuel={sh.water_heating_fuel} "
        f"cylinder_size={sh.cylinder_size} "
        f"has_cyl={str(epc.has_hot_water_cylinder).lower()} "
        f"cyl_ins_type={sh.cylinder_insulation_type} "
        f"cyl_ins_thick={sh.cylinder_insulation_thickness_mm} "
        f"immersion={sh.immersion_heating_type} "
        f"solar_wh={str(epc.solar_water_heating).lower()} "
        f"secondary_fuel={sh.secondary_fuel_type} "
        f"secondary_type={sh.secondary_heating_type}"
    )
    es = epc.sap_energy_source
    w(
        f"  ENERGY SOURCE: mains_gas={es.mains_gas} "
        f"meter_type={es.meter_type} "
        f"wind_turbines={es.wind_turbines_count}  "
        f"pv_raw={json.dumps(doc.get('sap_energy_source', {}).get('photovoltaic_supply'))}"
    )
    w(
        f"  VENT: fixed_AC={str(epc.has_fixed_air_conditioning).lower()}  "
        f"LIGHTING: led={epc.led_fixed_lighting_bulbs_count} "
        f"cfl={epc.cfl_fixed_lighting_bulbs_count} "
        f"incandescent={epc.incandescent_fixed_lighting_bulbs_count}"
    )

    # --- lodged reference outputs (the target) -------------------------
    w("\n## Lodged reference outputs (the target a worksheet must reproduce)")

    def _d(k: str) -> Any:
        return doc.get(k)

    w(
        f"  energy_rating_current={_d('energy_rating_current')} "
        f"env_impact_current={_d('environmental_impact_current')}"
    )
    w(
        f"  energy_consumption_current={_d('energy_consumption_current')} "
        f"co2_emissions_current={_d('co2_emissions_current')} "
        f"(per_floor_area={_d('co2_emissions_current_per_floor_area')})"
    )
    w(
        f"  heating_cost_current={_d('heating_cost_current')} "
        f"hot_water_cost_current={_d('hot_water_cost_current')} "
        f"lighting_cost_current={_d('lighting_cost_current')}"
    )
    return "\n".join(out) + "\n"


def main(argv: list[str]) -> int:
    args = [a for a in argv if not a.startswith("--")]
    out_dir: Optional[Path] = None
    if "--out-dir" in argv:
        i = argv.index("--out-dir")
        out_dir = Path(argv[i + 1])
        args = [a for a in args if a != str(out_dir)]
    if not args:
        print(__doc__)
        return 2
    for cert_arg in args:
        path = _resolve(cert_arg)
        cert = path.stem
        doc = json.loads(path.read_text())
        sheet = render(cert, doc)
        if out_dir is not None:
            out_dir.mkdir(parents=True, exist_ok=True)
            dest = out_dir / f"{cert}_elmhurst_input_sheet.md"
            dest.write_text(sheet)
            print(f"wrote {dest}")
        else:
            print(sheet)
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))