"""Two checks of the matching rules, published with the sample release so the figures can be reproduced. 1. Match precision by identifier hold-out. OpenStreetMap features that carry the Wikidata ID (wikidata tag) of an item within 500 m give known pairs. For each one the tag is hidden and the name + distance + category rule of build.py (score_pair) scores every Wikidata item within 500 m; the best-scoring item is the rule's answer: auto_merge score >= 0.85: correct when it is the tagged item review 0.60 <= score < 0.85: sent to a reviewer no_match no item scores 0.60 or more Precision = correct auto-merges / auto-merges (Wilson 95% interval). Tagged features lean towards well-known places, so this tests the rule; it does not replace a human-labelled sample. 2. Same name nearby. Pairs of records (after matching) whose normalised names are identical, within 150 m, under two rules, both reported: text rule within 100 m, any name (the check reported in the tender response): same category = candidate missed duplicate, different category = different products at one site release rule within 150 m, not already a review-band pair, names used 5 or more times in one LVEP skipped (postprocess_review.py; the check same_name_nearby and the same_name_candidate lineage events) Reads the raw downloads (data/raw, not published) and data/processed. Writes to sample-release/: match_precision_pairs.csv, same_name_pairs.csv, match_precision_summary.json, and a copy of this script. Usage: python3 pipeline/match_precision_test.py """ from __future__ import annotations import csv import io import json import math import shutil import sys from collections import Counter from pathlib import Path HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) import build # noqa: E402 from common import RAW, load_config, read_json # noqa: E402 from postprocess_review import GENERIC_NAME_MIN_COUNT, norm # noqa: E402 REPO = HERE.parent OUT = REPO / "sample-release" PROCESSED = REPO / "data" / "processed" HOLDOUT_RADIUS_M = 500 def wilson(k, n, z=1.96): if not n: return None, None p = k / n d = 1 + z * z / n c = p + z * z / (2 * n) h = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) return round(100 * (c - h) / d, 2), round(100 * (c + h) / d, 2) def records_by_lvep(): """Standardised source records assigned to sample LVEPs, as in build.py main() before matching.""" cfg = load_config() lad_to_lvep = {la["code"]: l["id"] for l in cfg["lveps"] for la in l["local_authorities"]} bgc = read_json(RAW / "ons_lad_bgc.geojson") lads = {f["properties"]["LAD26CD"]: build.PolyIndex(f["geometry"]) for f in bgc["features"]} loaded = {} for l in cfg["lveps"]: for fname, fn in ((f"osm_{l['id']}.json", build.osm_records), (f"wikidata_{l['id']}.json", build.wd_records), (f"nhle_{l['id']}.json", build.nhle_records)): raw = read_json(RAW / fname) for r in fn(raw, raw["_meta"]["fetched_at"]): loaded.setdefault(r["uid"], r) grouped, _, _ = build.group_numbered_units(list(loaded.values())) out = {l["id"]: [] for l in cfg["lveps"]} for r in grouped: code, _ = build.assign_geography(r["lat"], r["lon"], lads) if code: out[lad_to_lvep[code]].append(r) return out def holdout(recs_by_lvep): rows = [] for lvep, recs in recs_by_lvep.items(): wd = {r["sid"]: r for r in recs if r["src"] == "wd"} grid = build.Grid(list(wd.values()), cell=0.005) for o in sorted((r for r in recs if r["src"] == "osm" and r.get("wikidata")), key=lambda r: r["uid"]): true = wd.get(o["wikidata"]) if not true: continue d_true = build.haversine_m(o["lat"], o["lon"], true["lat"], true["lon"]) if d_true > HOLDOUT_RADIUS_M: continue hidden = dict(o, wikidata=None) best = None for w in grid.near(o["lat"], o["lon"], rings=2): d = build.haversine_m(o["lat"], o["lon"], w["lat"], w["lon"]) if d > HOLDOUT_RADIUS_M: continue s, ns = build.score_pair(hidden, w, d) if best is None or (s, -d, w["sid"]) > (best[0], -best[2], best[3]["sid"]): best = (s, ns, d, w) s = best[0] if best else None band = "auto_merge" if s is not None and s >= build.BANDS["auto_merge"] else "review" if s is not None and s >= build.BANDS["review"] else "no_match" chosen = best[3] if best and band != "no_match" else None rows.append({ "lvep_id": lvep, "osm_id": o["sid"], "osm_name": o["name"], "osm_category": o["cat"], "tagged_wikidata_id": true["sid"], "tagged_wikidata_name": true["name"], "tagged_wikidata_category": true["cat"], "tagged_distance_m": round(d_true), "rule_best_wikidata_id": best[3]["sid"] if best else "", "rule_best_wikidata_name": best[3]["name"] if best else "", "rule_best_score": s if s is not None else "", "rule_best_name_similarity": best[1] if best else "", "rule_best_distance_m": round(best[2]) if best else "", "band": band, "chosen_is_tagged_item": "" if chosen is None else str(chosen["sid"] == true["sid"]).lower(), }) return rows def same_name(meta): """Pairs within 150 m whose names match under either normalisation: text rule: build.py slug_name (accents and punctuation removed, stop words such as 'the', 'st' and 'ltd' dropped), within 100 m, every pair; release rule: postprocess_review.py norm (no stop words dropped), within 150 m, not a review-band pair, names used 5 or more times in one LVEP skipped.""" rows = [] for l in meta["lveps"]: recs = json.loads((PROCESSED / f"records-{l['id']}.json").read_text(encoding="utf-8"))["records"] names = Counter(norm(r["name"]) for r in recs if not r.get("units")) grid = build.Grid(recs, cell=0.002) for a in sorted(recs, key=lambda r: r["id"]): na, sa = norm(a["name"]), build.slug_name(a["name"]) for b in grid.near(a["lat"], a["lon"]): if b["id"] <= a["id"]: continue slug_same = bool(sa) and build.slug_name(b["name"]) == sa norm_same = bool(na) and not a.get("units") and not b.get("units") and norm(b["name"]) == na if not (slug_same or norm_same): continue d = build.haversine_m(a["lat"], a["lon"], b["lat"], b["lon"]) if d > 150: continue review = any(x["id"] == b["id"] for x in a.get("dups", [])) text_rule = slug_same and d <= 100 release_rule = norm_same and not review and names[na] < GENERIC_NAME_MIN_COUNT if not (text_rule or release_rule): continue rows.append({ "lvep_id": l["id"], "prototype_id_a": a["id"], "ve_product_id_a": a.get("vid", ""), "prototype_id_b": b["id"], "ve_product_id_b": b.get("vid", ""), "name_a": a["name"], "name_b": b["name"], "category_a": a["cat"], "category_b": b["cat"], "distance_m": round(d), "same_category": str(a["cat"] == b["cat"]).lower(), "review_band_pair": str(review).lower(), "name_used_in_lvep": names[na] if norm_same else "", "text_rule_within_100m": str(text_rule).lower(), "release_rule_same_name_candidate": str(release_rule).lower(), }) return rows def write_csv(path, rows): buf = io.StringIO(newline="") w = csv.DictWriter(buf, fieldnames=list(rows[0]), lineterminator="\r\n") w.writeheader() w.writerows(rows) path.write_bytes(buf.getvalue().encode("utf-8")) def main(): meta = json.loads((PROCESSED / "meta.json").read_text(encoding="utf-8")) pairs = holdout(records_by_lvep()) bands = Counter(r["band"] for r in pairs) correct = sum(1 for r in pairs if r["band"] == "auto_merge" and r["chosen_is_tagged_item"] == "true") review_right = sum(1 for r in pairs if r["band"] == "review" and r["chosen_is_tagged_item"] == "true") lo, hi = wilson(correct, bands["auto_merge"]) errors = [f"{r['osm_name']} ({r['osm_id']}) -> {r['rule_best_wikidata_name']} ({r['rule_best_wikidata_id']}), tagged {r['tagged_wikidata_name']}" for r in pairs if r["band"] == "auto_merge" and r["chosen_is_tagged_item"] == "false"] sn = same_name(meta) t100 = [r for r in sn if r["text_rule_within_100m"] == "true"] rel = [r for r in sn if r["release_rule_same_name_candidate"] == "true"] both = sum(1 for r in sn if r["text_rule_within_100m"] == "true" and r["release_rule_same_name_candidate"] == "true") summary = { "snapshot": meta["generated"][:10], "match_precision": { "pairs": len(pairs), "auto_merge": bands["auto_merge"], "auto_merge_correct": correct, "precision_pct": round(100 * correct / bands["auto_merge"], 1) if bands["auto_merge"] else None, "precision_wilson_95_pct": [lo, hi], "review": bands["review"], "review_with_tagged_item_first": review_right, "no_match": bands["no_match"], "auto_merge_recall_pct": round(100 * correct / len(pairs), 1) if pairs else None, "merged_correctly_or_sent_to_review_pct": round(100 * (correct + bands["review"]) / len(pairs), 1) if pairs else None, "wrong_auto_merges": errors, "method": "OSM features with a Wikidata tag whose item lies within 500 m; tag hidden; build.py score_pair against every Wikidata item within 500 m; best score decides (>= 0.85 auto-merge, 0.60 to 0.85 review).", }, "same_name_nearby": { "rows_in_file": len(sn), "pairs_under_both_rules": both, "text_rule_within_100m": {"pairs": len(t100), "same_category_candidates": sum(1 for r in t100 if r["same_category"] == "true"), "different_category": sum(1 for r in t100 if r["same_category"] == "false"), "of_which_review_band_pairs": sum(1 for r in t100 if r["review_band_pair"] == "true"), "rule": "names equal after build.py slug_name (stop words such as the, st, ltd dropped), within 100 m, every pair of records"}, "release_rule_within_150m": {"pairs": len(rel), "within_100m": sum(1 for r in rel if int(r["distance_m"]) <= 100), "rule": f"names equal after postprocess_review.py norm, within 150 m, not a review-band pair, names used {GENERIC_NAME_MIN_COUNT}+ times in one LVEP skipped (check same_name_nearby)"}, }, "files": ["match_precision_pairs.csv", "same_name_pairs.csv", "match_precision_test.py"], } OUT.mkdir(exist_ok=True) write_csv(OUT / "match_precision_pairs.csv", pairs) write_csv(OUT / "same_name_pairs.csv", sn) (OUT / "match_precision_summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=1) + "\n", encoding="utf-8") shutil.copyfile(Path(__file__), OUT / "match_precision_test.py") print(json.dumps(summary, ensure_ascii=False, indent=1)) if __name__ == "__main__": main()