diff --git a/scripts/corpus_1000/mapper_audit.py b/scripts/corpus_1000/mapper_audit.py new file mode 100644 index 000000000..6d82c5f1d --- /dev/null +++ b/scripts/corpus_1000/mapper_audit.py @@ -0,0 +1,56 @@ +"""Mapper faithfulness audit: for identity pass-through fields (dims, counts, +thicknesses) compare the RAW cert value to what EpcPropertyDataMapper produced. +Any mismatch is a mapper bug silently feeding the SAP engine wrong inputs.""" +from __future__ import annotations +import json +from collections import defaultdict +from datatypes.epc.domain.mapper import EpcPropertyDataMapper +CORPUS="backend/epc_api/json_samples/RdSAP-Schema-21.0.1/corpus.jsonl" +def gv(x): return x.get('value') if isinstance(x,dict) else x +def num(x): + x=gv(x) + if x is None: return None + s=str(x).lower().replace('mm','').replace('m2','').strip() + try: return float(s) + except: return s # keep strings like 'measured'/'NI' for equality +mism=defaultdict(list) +n=0 +for line in open(CORPUS): + d=json.loads(line) + try: epc=EpcPropertyDataMapper.from_api_response(d) + except Exception: continue + n+=1; u=d.get('uprn') + # top-level counts + checks=[('door_count', d.get('door_count'), epc.door_count), + ('habitable', d.get('habitable_room_count'), getattr(epc,'habitable_room_count',None)), + ('led', d.get('led_fixed_lighting_bulbs_count'), epc.led_fixed_lighting_bulbs_count), + ('cfl', d.get('cfl_fixed_lighting_bulbs_count'), epc.cfl_fixed_lighting_bulbs_count), + ('incand', d.get('incandescent_fixed_lighting_bulbs_count'), epc.incandescent_fixed_lighting_bulbs_count)] + for name,rv,ev in checks: + if rv is not None and ev is not None and num(rv)!=num(ev): + mism[name].append((u,rv,ev)) + # per-building-part fabric + rbps=d.get('sap_building_parts') or []; ebps=epc.sap_building_parts or [] + for i,rbp in enumerate(rbps): + if i>=len(ebps): + mism['DROPPED_building_part'].append((u,len(rbps),len(ebps))); break + ebp=ebps[i] + for name,rv,ev in [('wall_thick',rbp.get('wall_thickness'),getattr(ebp,'wall_thickness_mm',None)), + ('wall_ins_thick',rbp.get('wall_insulation_thickness'),getattr(ebp,'wall_insulation_thickness',None)), + ('roof_ins_thick',rbp.get('roof_insulation_thickness'),getattr(ebp,'roof_insulation_thickness',None)), + ('wall_con',rbp.get('wall_construction'),getattr(ebp,'wall_construction',None)), + ('wall_ins_type',rbp.get('wall_insulation_type'),getattr(ebp,'wall_insulation_type',None))]: + if rv is not None and ev is not None and num(rv)!=num(ev): + mism[name].append((u,rv,ev)) + # windows: count + total area + rw=d.get('sap_windows') or []; ew=epc.sap_windows or [] + if len(rw)!=len(ew): mism['window_count'].append((u,len(rw),len(ew))) + ra=sum((gv(w.get('window_width')) or 0)*(gv(w.get('window_height')) or 0) for w in rw) + ea=sum((float(w.window_width) if w.window_width else 0)*(float(w.window_height) if w.window_height else 0) for w in ew) + if abs(ra-ea)>0.02: mism['window_area'].append((u,round(ra,2),round(ea,2))) +print(f"audited {n} certs") +print("=== field mismatches (raw cert vs EpcPropertyData) ===") +for k,v in sorted(mism.items(), key=lambda x:-len(x[1])): + print(f"\n[{len(v):4d}] {k}") + for u,rv,ev in v[:5]: print(f" {u}: raw={rv} epc={ev}") +if not mism: print(" (no mismatches — mapper is identity-faithful on all checked fields)")