diff --git a/scripts/epc_disk_cache.py b/scripts/epc_disk_cache.py new file mode 100644 index 000000000..c3224da67 --- /dev/null +++ b/scripts/epc_disk_cache.py @@ -0,0 +1,81 @@ +"""A raw-JSON disk cache around the EPC-API client, for harness scripts. + +Caches at the RAW layer — certificate payloads exactly as the API returned +them, search results as plain JSON rows — so that: + +1. **Iteration is fast**: the pairs harness re-reads the same certs on every + run (pairs and cohorts don't change), so each re-run was re-paying 10-20k + sequential HTTP calls; a warm cache replays in minutes. +2. **The cache is fixture material**: a committed integration-test corpus + should hold anonymised raw API JSON loaded through + ``EpcPropertyDataMapper.from_api_response`` (the ADR-0030 corpus pattern), + so replays keep exercising the mapper and survive domain-dataclass changes. + Pickles of internal dataclasses would do neither. + +Successful responses only — errors propagate uncached. Script-support code, +not a repository: production ingestion must keep reading live (a re-lodgement +must be seen), so this stays out of the DDD layers. +""" + +from __future__ import annotations + +import dataclasses +import json +import re +from pathlib import Path +from typing import Any, Optional, cast + +from datatypes.epc.search.epc_search_result import EpcSearchResult +from infrastructure.epc_client.epc_client_service import EpcClientService + + +def _safe(key: str) -> str: + return re.sub(r"[^A-Za-z0-9_-]", "_", key) + + +class JsonCachingEpcClient(EpcClientService): + """An ``EpcClientService`` whose raw HTTP layer is disk-cached as JSON. + + Overrides the two private fetchers, so retry, mapping and search parsing + all still run on every call — a warm replay exercises the exact production + code path minus the network. + """ + + def __init__(self, auth_token: str, cache_dir: Path) -> None: + super().__init__(auth_token) + self._cache_dir = cache_dir + cache_dir.mkdir(parents=True, exist_ok=True) + + def _path(self, key: str) -> Path: + return self._cache_dir / f"{_safe(key)}.json" + + def _read(self, key: str) -> Optional[Any]: + path = self._path(key) + if not path.exists(): + return None + return json.loads(path.read_text()) + + def _write(self, key: str, value: Any) -> None: + self._path(key).write_text(json.dumps(value)) + + def _fetch_certificate(self, cert_num: str) -> dict[str, Any]: + cached = self._read(f"cert_{cert_num}") + if cached is not None: + return cast(dict[str, Any], cached) + raw = super()._fetch_certificate(cert_num) + self._write(f"cert_{cert_num}", raw) + return raw + + def _search( + self, + postcode: Optional[str] = None, + uprn: Optional[int] = None, + ) -> list[EpcSearchResult]: + key = f"search_pc_{postcode}" if postcode else f"search_uprn_{uprn}" + cached = self._read(key) + if cached is not None: + rows = cast(list[dict[str, Any]], cached) + return [EpcSearchResult(**row) for row in rows] + results = super()._search(postcode=postcode, uprn=uprn) + self._write(key, [dataclasses.asdict(r) for r in results]) + return results diff --git a/scripts/expired_prediction_pairs_harness.py b/scripts/expired_prediction_pairs_harness.py index 267e4092a..3b113a449 100644 --- a/scripts/expired_prediction_pairs_harness.py +++ b/scripts/expired_prediction_pairs_harness.py @@ -358,7 +358,9 @@ def format_diagnosis(rows: list[dict[str, object]]) -> str: def run( # pragma: no cover - live IO composition - postcodes: list[str], telemetry_path: Optional[Path] = None + postcodes: list[str], + telemetry_path: Optional[Path] = None, + cache_dir: Optional[Path] = None, ) -> str: from domain.sap10_calculator.calculator import Sap10Calculator from infrastructure.epc_client.epc_client_service import EpcClientService @@ -374,7 +376,14 @@ def run( # pragma: no cover - live IO composition from scripts.e2e_common import load_env, s3_parquet_reader load_env() - epc_client = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"]) + if cache_dir is not None: + from scripts.epc_disk_cache import JsonCachingEpcClient + + epc_client: EpcClientService = JsonCachingEpcClient( + os.environ["OPEN_EPC_API_TOKEN"], cache_dir + ) + else: + epc_client = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"]) geospatial = GeospatialS3Repository(s3_parquet_reader(os.environ["DATA_BUCKET"])) comparables_repo = EpcComparablePropertiesRepository(epc_client, geospatial) historic_repo = HistoricEpcS3Repository.with_default_s3_client() @@ -511,7 +520,7 @@ def run( # pragma: no cover - live IO composition if conditioning.main_fuel is None else conditioning.main_fuel == _actual_fuel(actual) ), - "agrees_tfa_within_5pct": _tfa_within_band( + "agrees_tfa_within_band": _tfa_within_band( actual.total_floor_area_m2, conditioning.total_floor_area_m2, ), @@ -543,6 +552,12 @@ def main() -> None: # pragma: no cover - CLI entry parser.add_argument( "--telemetry", type=Path, default=None, help="write per-pair JSONL here" ) + parser.add_argument( + "--cache-dir", + type=Path, + default=None, + help="raw-JSON disk cache for API responses (fast re-runs; fixture material)", + ) args = parser.parse_args() postcodes: list[str] = list(args.postcodes) if args.postcodes_file is not None: @@ -553,7 +568,7 @@ def main() -> None: # pragma: no cover - CLI entry ] if not postcodes: parser.error("no postcodes given") - report = run(postcodes, telemetry_path=args.telemetry) + report = run(postcodes, telemetry_path=args.telemetry, cache_dir=args.cache_dir) if args.out is not None: args.out.write_text(report + "\n") print(report) diff --git a/scripts/roof_construction_code_sweep.py b/scripts/roof_construction_code_sweep.py new file mode 100644 index 000000000..b3c406790 --- /dev/null +++ b/scripts/roof_construction_code_sweep.py @@ -0,0 +1,114 @@ +"""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()