mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
13b75bcb0a
JsonCachingEpcClient caches at the raw layer (cert payloads verbatim, search rows as JSON) so warm re-runs replay through retry+mapper+parsing in minutes AND the cache doubles as integration-test fixture material — the ADR-0030 corpus pattern (anonymised raw JSON through the mapper), which pickled domain objects could never be. Wired into the pairs harness (--cache-dir). roof_construction_code_sweep pins the RdSAP roof code table the basement-code-6 way: single-building-part certs only, code x description-prefix purity table. Also fixes the tfa telemetry key rename that left the agreement row empty. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
114 lines
4 KiB
Python
114 lines
4 KiB
Python
"""Pin the RdSAP ``roof_construction`` code table empirically (ADR-0054 gap).
|
|
|
|
The gov-EPC API lodges ``sap_building_parts[].roof_construction`` as an int
|
|
whose enumeration is published nowhere we hold; the historic dump lodges roofs
|
|
as display text. The wall equivalent was pinned the same way this script works
|
|
(the basement wall code 6: a bulk co-occurrence sweep). Method:
|
|
|
|
- fetch every cert in the given postcodes' cohorts;
|
|
- keep only certs with EXACTLY one building part and one roofs element, so the
|
|
(code, description) pairing is unambiguous;
|
|
- cross-tabulate code x description-prefix (text before the first comma) and
|
|
report each code's dominant prefix with its purity.
|
|
|
|
Usage:
|
|
python scripts/roof_construction_code_sweep.py --postcodes-file pcs.txt \
|
|
--cache-dir .epc_cache --out roof_codes.md
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
from collections import Counter, defaultdict
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from datatypes.epc.domain.epc_property_data import EpcPropertyData # noqa: E402
|
|
|
|
|
|
def prefix_of(description: str) -> str:
|
|
"""The roof description's construction/form half — the text before the
|
|
first comma ("Pitched, 100 mm loft insulation" -> "Pitched")."""
|
|
return description.split(",", 1)[0].strip()
|
|
|
|
|
|
def tabulate(certs: list[EpcPropertyData]) -> dict[int, Counter[str]]:
|
|
"""code -> Counter(description prefix), over unambiguous certs only."""
|
|
table: dict[int, Counter[str]] = defaultdict(Counter)
|
|
for cert in certs:
|
|
if len(cert.sap_building_parts) != 1 or len(cert.roofs) != 1:
|
|
continue
|
|
code = cert.sap_building_parts[0].roof_construction
|
|
if code is None:
|
|
continue
|
|
table[code][prefix_of(cert.roofs[0].description)] += 1
|
|
return dict(table)
|
|
|
|
|
|
def format_table(table: dict[int, Counter[str]]) -> str:
|
|
lines = [
|
|
"# roof_construction code x description-prefix co-occurrence",
|
|
"",
|
|
"| code | n | dominant prefix | purity | runners-up |",
|
|
"|---|---|---|---|---|",
|
|
]
|
|
for code in sorted(table):
|
|
counts = table[code]
|
|
total = sum(counts.values())
|
|
(top, top_n), *rest = counts.most_common()
|
|
runners = ", ".join(f"{p} ({n})" for p, n in rest[:3]) or "—"
|
|
lines.append(
|
|
f"| {code} | {total} | {top} | {top_n / total:.0%} | {runners} |"
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main() -> None: # pragma: no cover - CLI/IO composition
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--postcodes-file", type=Path, required=True)
|
|
parser.add_argument("--cache-dir", type=Path, required=True)
|
|
parser.add_argument("--out", type=Path, default=None)
|
|
args = parser.parse_args()
|
|
|
|
from scripts.e2e_common import load_env
|
|
from scripts.epc_disk_cache import JsonCachingEpcClient
|
|
|
|
load_env()
|
|
client = JsonCachingEpcClient(os.environ["OPEN_EPC_API_TOKEN"], args.cache_dir)
|
|
|
|
certs: list[EpcPropertyData] = []
|
|
postcodes = [
|
|
line.strip()
|
|
for line in args.postcodes_file.read_text().splitlines()
|
|
if line.strip()
|
|
]
|
|
for index, postcode in enumerate(postcodes):
|
|
try:
|
|
results = client.search_by_postcode(postcode)
|
|
except Exception as error:
|
|
print(f"{postcode}: search failed: {error}", file=sys.stderr)
|
|
continue
|
|
for result in results:
|
|
try:
|
|
certs.append(
|
|
client.get_by_certificate_number(result.certificate_number)
|
|
)
|
|
except Exception:
|
|
continue # unmappable cert — not usable evidence
|
|
print(
|
|
f"[{index + 1}/{len(postcodes)}] {postcode}: {len(certs)} certs so far",
|
|
file=sys.stderr,
|
|
flush=True,
|
|
)
|
|
|
|
report = format_table(tabulate(certs))
|
|
if args.out is not None:
|
|
args.out.write_text(report + "\n")
|
|
print(report)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|