mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
26106505be
profile_corpus_error.py buckets signed SAP error by raw-API feature and lists worst over/under-raters with the PE/CO2-vs-cost split (COST-side vs DEMAND-side triage). dive_cert.py dumps one cert's lodged-vs-ours SAP/CO2/PE + full intermediate line refs + mapped inputs. Both run on the committed RdSAP-21.0.1 corpus (no /tmp sample needed). Used to find the stone-wall, per-part-roof, ground-floor-flat and HP-water fixes this session. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
103 lines
3.7 KiB
Python
103 lines
3.7 KiB
Python
"""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_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))
|
|
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={r.co2_kg_per_yr / 1000:.3f} t "
|
|
f"d={r.co2_kg_per_yr / 1000 - lodged_co2:+.3f}"
|
|
)
|
|
if lodged_pe is not None:
|
|
print(
|
|
f" PE lodged={lodged_pe:.1f} ours={r.primary_energy_kwh_per_m2:.1f} "
|
|
f"d={r.primary_energy_kwh_per_m2 - lodged_pe:+.1f} kWh/m2"
|
|
)
|
|
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()
|