junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-12 13:29:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Pairs harness survives unmappable certs and appends telemetry incrementally 🟩

Browse source 
The 2000-postcode sweep died at ~884 on a strict-raising RdSAP-17.1 cert
(missing multiple_glazed_proportion) inside the pair-check, losing the
whole run's output. get_by_uprn and the cohort fetch now skip-and-log per
item; telemetry rows append as produced and carry raw values (age band,
TFA, fuel text) so band-width questions are answerable post hoc.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-07-06 10:01:57 +00:00
parent ee5bb6c4f2
commit 2903a069ea
1 changed files with 41 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
scripts/expired_prediction_pairs_harness.py
View file

@ -395,16 +395,40 @@ def run( # pragma: no cover - live IO composition
conditioned_scores: list[PairScore] = []
telemetry: list[dict[str, object]] = []
pairs = 0
def emit_telemetry(row: dict[str, object]) -> None:
# Append incrementally so a late crash loses nothing.
telemetry.append(row)
if telemetry_path is None:
return
import dataclasses as _dc
import json as _json
serialisable = {
k: (_dc.asdict(v) if isinstance(v, LadderStep) else v)
for k, v in row.items()
}
with telemetry_path.open("a") as handle:
handle.write(_json.dumps(serialisable) + "\n")
if telemetry_path is not None and telemetry_path.exists():
telemetry_path.unlink()
for index, raw_postcode in enumerate(postcodes):
postcode = str(Postcode(raw_postcode))
historic = latest_pre_2012_by_uprn(
historic_repo.get_for_postcode(Postcode(raw_postcode))
)
# Pair-check first: only a postcode with a SAP-10.2 relodgement of a
# pre-2012 UPRN pays for the cohort fetch.
# pre-2012 UPRN pays for the cohort fetch. A cert the mapper can't yet
# map (strict-raise) can't be a validation target either — skip it and
# keep sweeping; one bad cert must not kill a multi-hour run.
paired: list[tuple[str, HistoricEpc, EpcPropertyData]] = []
for uprn, record in historic.items():
actual = epc_client.get_by_uprn(int(uprn))
try:
actual = epc_client.get_by_uprn(int(uprn))
except Exception as error:
print(f" {uprn}: unmappable cert skipped: {error}", file=sys.stderr)
continue
if actual is not None and actual.sap_version == _VALIDATION_SAP_VERSION:
paired.append((uprn, record, actual))
print(
@ -415,7 +439,11 @@ def run( # pragma: no cover - live IO composition
)
if not paired:
continue
cohort = comparables_repo.candidates_for(postcode)
try:
cohort = comparables_repo.candidates_for(postcode)
except Exception as error:
print(f" {postcode}: cohort fetch failed: {error}", file=sys.stderr)
continue
for uprn, record, actual in paired:
pairs += 1
loo_cohort = [c for c in cohort if c.epc.uprn != int(uprn)]
@ -440,11 +468,20 @@ def run( # pragma: no cover - live IO composition
main_fuel=conditioning.main_fuel,
total_floor_area_m2=conditioning.total_floor_area_m2,
)
telemetry.append(
emit_telemetry(
{
"postcode": postcode,
"uprn": uprn,
"plain_cohort_size": len(base),
# Raw values alongside the booleans, so band-width and
# near-miss questions are answerable post hoc.
"historic_age_band": conditioning.construction_age_band,
"actual_age_band": _actual_age_band(actual),
"historic_tfa": conditioning.total_floor_area_m2,
"actual_tfa": actual.total_floor_area_m2,
"historic_fuel": conditioning.main_fuel,
"actual_fuel": _actual_fuel(actual),
"historic_fuel_text": record.main_fuel,
**{f"ladder_{k}": v for k, v in ladder.items()},
"agrees_property_type": (
None
@ -490,17 +527,6 @@ def run( # pragma: no cover - live IO composition
)
)
if telemetry_path is not None:
import dataclasses as _dc
import json as _json
with telemetry_path.open("w") as handle:
for row in telemetry:
serialisable = {
k: (_dc.asdict(v) if isinstance(v, LadderStep) else v)
for k, v in row.items()
}
handle.write(_json.dumps(serialisable) + "\n")
report = format_report(
aggregate(plain_scores), aggregate(conditioned_scores), pairs
)