From 2c126b2a628d54772c0debfcecaa13e7eb010dbe Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Fri, 5 Jun 2026 19:39:31 +0000 Subject: [PATCH] tooling(debug): add scripts/elmhurst_input_sheet.py worksheet-input dumper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 _elmhurst_input_sheet.md. Pyright strict clean. Co-Authored-By: Claude Opus 4.8 --- scripts/elmhurst_input_sheet.py | 297 ++++++++++++++++++++++++++++++++ 1 file changed, 297 insertions(+) create mode 100644 scripts/elmhurst_input_sheet.py diff --git a/scripts/elmhurst_input_sheet.py b/scripts/elmhurst_input_sheet.py new file mode 100644 index 00000000..940783aa --- /dev/null +++ b/scripts/elmhurst_input_sheet.py @@ -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 [ ...] + + # 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" + +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 `/.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} 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 ----------------------------------- + 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:]))