mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
114 lines
4.6 KiB
Python
114 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
import pandas as pd
|
|
from dotenv import load_dotenv
|
|
|
|
from datatypes.epc.domain.epc import Epc
|
|
from datatypes.epc.domain.mapper import EpcPropertyDataMapper
|
|
from domain.sap10_calculator.calculator import calculate_sap_from_inputs
|
|
from domain.sap10_calculator.rdsap.cert_to_inputs import cert_to_inputs
|
|
from infrastructure.epc_client.epc_client_service import EpcClientService
|
|
|
|
# UPRNs to compare. Most are RdSAP 20.0.0 (pre-SAP10) certs — the ones the
|
|
# Reduced-Field Synthesis mapper (ADR-0027) re-maps so the SAP10 calculator can
|
|
# re-score them. The commented rows are non-20.0.0 neighbours kept for context.
|
|
UPRNS: list[int] = [
|
|
10003318624, # 20.0.0 Flat 1, 6 Alexandra Gardens, PO38 1EE
|
|
# 10003318625, # 20.0.0 Flat 2, 6 Alexandra Gardens, PO38 1EE
|
|
# 10003318626, # 20.0.0 Flat 3, 6 Alexandra Gardens, PO38 1EE
|
|
# 10003318698, # 17.1 Flat 4, 6 Alexandra Gardens, PO38 1EE
|
|
# 100062430247, # 20.0.0 Flat 5, Adelaide Court, Adelaide Place, PO33 3DG
|
|
# 100062430248, # 20.0.0 Flat 6, Adelaide Court, Adelaide Place, PO33 3DG
|
|
# 100062430250, # 20.0.0 Flat 8, Adelaide Court, Adelaide Place, PO33 3DG
|
|
# 100062429797, # 20.0.0 Flat 1, 10-11 Cross Street, PO33 2AD
|
|
# 10003320577, # 20.0.0 Flat 3, 10-11 Cross Street, PO33 2AD
|
|
# 10003320573, # 18.0 Flat 7, 10-11 Cross Street, PO33 2AD
|
|
# 10024248769, # 20.0.0 Flat 8, 10-11 Cross Street, PO33 2AD
|
|
# 10024248772, # 18.0 Flat 9, 10-11 Cross Street, PO33 2AD
|
|
]
|
|
|
|
|
|
def fetch_raw_cert(service: EpcClientService, uprn: int) -> Optional[dict[str, Any]]:
|
|
"""Pull the latest raw certificate dict for a UPRN straight off the EPC
|
|
client. We want the RAW cert (not the mapped EpcPropertyData) because the
|
|
lodged SAP score lives there as `energy_rating_current` — the mapper does
|
|
not carry it onto the domain object.
|
|
"""
|
|
results = service._search(uprn=uprn) # pyright: ignore[reportPrivateUsage]
|
|
if not results:
|
|
return None
|
|
latest = max(results, key=lambda r: r.registration_date)
|
|
return service._fetch_certificate( # pyright: ignore[reportPrivateUsage]
|
|
latest.certificate_number
|
|
)
|
|
|
|
|
|
def compare_sap(raw: dict[str, Any]) -> dict[str, object]:
|
|
"""Re-score a raw cert through our SAP10 calculator and line it up against
|
|
the figure the surveyor lodged. For a 20.0.0 cert the calculated value is
|
|
the counterfactual "what EPC would this get under today's spec" (ADR-0027).
|
|
"""
|
|
epc = EpcPropertyDataMapper.from_api_response(raw)
|
|
result = calculate_sap_from_inputs(cert_to_inputs(epc))
|
|
|
|
# Lodged Performance: the surveyor's original SAP score, read directly from
|
|
# the raw cert. Bands are derived from the score the same way for both sides.
|
|
lodged_score = raw.get("energy_rating_current")
|
|
lodged_band = (
|
|
Epc.from_sap_score(lodged_score).value if lodged_score is not None else "?"
|
|
)
|
|
our_band = Epc.from_sap_score(result.sap_score).value
|
|
|
|
return {
|
|
"address": epc.address_line_1,
|
|
"postcode": epc.postcode,
|
|
# The SAP methodology version (RdSAP 2012 lodges 9.9x); the *schema*
|
|
# version (20.0.0) is annotated in the UPRNS list above.
|
|
"sap_ver": raw.get("sap_version"),
|
|
"lodged_sap": lodged_score,
|
|
"lodged_band": lodged_band,
|
|
"our_sap": result.sap_score,
|
|
"our_band": our_band,
|
|
"delta": (
|
|
result.sap_score - lodged_score if lodged_score is not None else None
|
|
),
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
# Mirror conftest.py: pull OPEN_EPC_API_TOKEN out of backend/.env so the
|
|
# script runs standalone (`python scripts/eon/find_epc_data.py`).
|
|
repo_root = Path(__file__).resolve().parents[2]
|
|
load_dotenv(repo_root / "backend" / ".env")
|
|
|
|
token = os.getenv("OPEN_EPC_API_TOKEN")
|
|
if token is None:
|
|
raise RuntimeError("OPEN_EPC_API_TOKEN not defined in env")
|
|
service = EpcClientService(auth_token=token)
|
|
|
|
rows: list[dict[str, object]] = []
|
|
for uprn in UPRNS:
|
|
raw = fetch_raw_cert(service, uprn)
|
|
if raw is None:
|
|
print(f"UPRN {uprn}: no EPC found")
|
|
continue
|
|
try:
|
|
rows.append({"uprn": uprn, **compare_sap(raw)})
|
|
except Exception as exc: # surface, don't abort the whole sweep
|
|
print(f"UPRN {uprn}: failed to score — {type(exc).__name__}: {exc}")
|
|
|
|
if not rows:
|
|
print("No certs scored.")
|
|
return
|
|
|
|
table = pd.DataFrame(rows)
|
|
with pd.option_context("display.max_columns", None, "display.width", None):
|
|
print(table.to_string(index=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|