tooling(debug): add scripts/elmhurst_input_sheet.py worksheet-input dumper

Reconstructs the per-cert "Elmhurst SAP input sheet" generator that the
API-accuracy debugging loop relied on (the worked example survives at
'sap worksheets/golden fixture debugging/6035_elmhurst_input_sheet.md'); the
original was a throwaway and never committed. Companion to
eval_api_sap_accuracy.py: once that names a worst-offender cert, this dumps
the codes the mapper hands the calculator (from_api_response → EpcPropertyData)
in the 6035 layout — header, lodged element descriptions, building parts +
dimensions, windows, doors/heating/water/vent — plus the lodged reference
outputs and OUR continuous SAP next to the lodged value, to read side-by-side
with the Elmhurst Summary / P960 worksheet PDF.

Reads the fetch_2026_epc_sample.py cache (EPC_SAMPLE_CACHE, default
/tmp/epc_2026_sample). `--out-dir` writes <cert>_elmhurst_input_sheet.md.
Pyright strict clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-06-05 19:39:31 +00:00
parent e09bed31bc
commit 2c126b2a62

View file

@ -0,0 +1,297 @@
"""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 _g(obj: Any, *names: str, default: Any = None) -> Any:
"""First present attribute from `names`, else `default`."""
for n in names:
if hasattr(obj, n):
return getattr(obj, n)
return default
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}"
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 -----------------------------------
w("\n## Building parts / dimensions")
for bp in epc.sap_building_parts or []:
name = _g(bp, "identifier", default=f"part {_g(bp, 'building_part_number')}")
age = _g(bp, "construction_age_band")
w(f"### {name} (part {_g(bp, 'building_part_number')}, age {age})")
w(
f" wall_construction={_g(bp, 'wall_construction')} "
f"insulation_type={_g(bp, 'wall_insulation_type')} "
f"ins_thick={_g(bp, 'wall_insulation_thickness')} "
f"wall_thickness={_g(bp, 'wall_thickness')}mm "
f"measured={_g(bp, 'wall_thickness_measured')} "
f"dry_lined={_g(bp, 'wall_dry_lined')}"
)
w(f" party_wall_construction={_g(bp, 'party_wall_construction')}")
w(
f" roof_construction={_g(bp, 'roof_construction')} "
f"ins_location={_g(bp, 'roof_insulation_location')} "
f"ins_thick={_g(bp, 'roof_insulation_thickness')}"
)
w(
f" floor_heat_loss={_g(bp, 'floor_heat_loss')} "
f"floor_ins_thick={_g(bp, 'floor_insulation_thickness')}"
)
rir = _g(bp, "sap_room_in_roof")
if rir is not None:
w(
f" ROOM-IN-ROOF: floor_area={_num(_g(rir, 'floor_area'))} "
f"age={_g(rir, 'construction_age_band')} "
f"insulation={_g(rir, 'insulation')} "
f"connected={_g(rir, 'roof_room_connected')}"
)
floor_dims: list[Any] = _g(bp, "sap_floor_dimensions", default=[]) or []
for fd in floor_dims:
w(
f" floor {_g(fd, 'floor')}: area={_num(_g(fd, 'total_floor_area'))} "
f"height={_num(_g(fd, 'room_height'))} "
f"HLP={_num(_g(fd, 'heat_loss_perimeter'))} "
f"party_wall_len={_num(_g(fd, 'party_wall_length'))} "
f"floor_constr={_g(fd, 'floor_construction')} "
f"floor_ins={_g(fd, 'floor_insulation')}"
)
# --- windows -------------------------------------------------------
windows = epc.sap_windows or []
w(f"\n## Windows ({len(windows)})")
for i, win in enumerate(windows):
w(
f" W{i}: {_g(win, 'window_width')}x{_g(win, 'window_height')}m "
f"orient={_g(win, 'orientation')} "
f"glazing_type={_g(win, 'glazing_type')} "
f"gap={_g(win, 'glazing_gap')} "
f"pvc={_g(win, 'pvc_frame')} "
f"draught={_g(win, 'draught_proofed')} "
f"loc(bp)={_g(win, 'window_location')} "
f"wall_type={_g(win, 'window_wall_type')} "
f"frame_factor={_g(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
mh_details: list[Any] = _g(sh, "main_heating_details", default=[]) or []
for mh in mh_details:
w(
f" MAIN: sap_code={_g(mh, 'sap_main_heating_code')} "
f"fuel={_g(mh, 'main_fuel_type')} "
f"category={_g(mh, 'main_heating_category')} "
f"emitter={_g(mh, 'heat_emitter_type')} "
f"emit_temp={_g(mh, 'emitter_temperature')} "
f"control={_g(mh, 'main_heating_control')} "
f"fghrs={_g(mh, 'has_fghrs')} "
f"fan_flue={_g(mh, 'fan_flue_present')} "
f"flue_type={_g(mh, 'boiler_flue_type')} "
f"pump_age={_g(mh, 'central_heating_pump_age')} "
f"data_source={_g(mh, 'main_heating_data_source')} "
f"idx={_g(mh, 'main_heating_index_number')} "
f"fraction={_g(mh, 'main_heating_fraction')}"
)
w(
f" WATER: code={_g(sh, 'water_heating_code')} "
f"fuel={_g(sh, 'water_heating_fuel')} "
f"cylinder_size={_g(sh, 'cylinder_size')} "
f"has_cyl={str(epc.has_hot_water_cylinder).lower()} "
f"cyl_ins_type={_g(sh, 'cylinder_insulation_type')} "
f"cyl_ins_thick={_g(sh, 'cylinder_insulation_thickness')} "
f"immersion={_g(sh, 'immersion_heating_type')} "
f"solar_wh={str(epc.solar_water_heating).lower()} "
f"secondary_fuel={_g(sh, 'secondary_fuel_type')} "
f"secondary_type={_g(sh, 'secondary_heating_type')}"
)
es = epc.sap_energy_source
w(
f" ENERGY SOURCE: mains_gas={_g(es, 'mains_gas')} "
f"meter_type={_g(es, 'meter_type')} "
f"wind_turbines={_g(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:]))