mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
4006753620
The new gov EPC API (api.get-energy-performance-data..., Bearer auth) returns 403 "Bad authentication header" with EPC_AUTH_TOKEN but 200 with OPEN_EPC_API_TOKEN — the token name is misleading (it is the Bearer token for the new API, not the open-data API). Verified live against /api/domestic/search. Unblocks the live EPC fetch in run_modelling_e2e. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
398 lines
17 KiB
Python
398 lines
17 KiB
Python
"""Run Modelling end-to-end for specific Properties (by ``property_id``) and
|
|
print the recommendations for inspection.
|
|
|
|
The local DB's Properties have no linked, ingested EPC yet (Ingestion's source
|
|
clients are still stubbed — #1136), so this script does the ingestion step
|
|
inline: it reads each Property's UPRN from the DB, fetches the latest EPC
|
|
**live** from the gov EPC API by UPRN, resolves the UPRN's spatial reference
|
|
from S3, and fetches Google Solar — then runs the Modelling stage (every
|
|
Recommendation Generator → the Optimiser → a costed, attributed Plan). The same
|
|
local computation runs whether or not you store the result: by default it
|
|
persists **nothing** (the run is for inspecting recommendations); pass
|
|
`--persist` to write the inputs + the Plan to the DB.
|
|
|
|
To keep the inspected recommendations identical to what gets stored, **both
|
|
modes price against the live ``material`` catalogue (read-only)** and model
|
|
against a real **Scenario** read from the DB — not the JSON sample catalogue.
|
|
Pass `--scenario-id` to target a real Scenario (its ``goal_value`` drives the
|
|
band); without it the run synthesises an Increasing-EPC-to-``--goal`` Scenario.
|
|
``--measures`` restricts the run to a comma-separated set of measure types
|
|
(mirroring the legacy `inclusions`) — e.g. only HHRSH + Solar PV.
|
|
|
|
Config: loads `backend/.env` for the DB creds (`DB_*`), the EPC API token
|
|
(`OPEN_EPC_API_TOKEN` — the Bearer token for the new gov API), the Google Solar
|
|
key (`GOOGLE_SOLAR_API_KEY`) and the S3
|
|
reference bucket (`DATA_BUCKET`) — the agent never sees the secrets. AWS creds
|
|
come from the ambient `~/.aws` profile. Run from the worktree root:
|
|
|
|
# inspect only (no DB writes), HHRSH + Solar PV, against Scenario 1263:
|
|
python -m scripts.run_modelling_e2e --scenario-id 1263 \
|
|
--measures high_heat_retention_storage_heaters,solar_pv 115 116 117
|
|
# same run, but persist the Plans (needs --portfolio-id):
|
|
python -m scripts.run_modelling_e2e --scenario-id 1263 --portfolio-id 4 \
|
|
--measures high_heat_retention_storage_heaters,solar_pv --persist 115 116 117
|
|
python -m scripts.run_modelling_e2e --no-solar 115 116 # skip the Google leg
|
|
|
|
Per Property the spatial reference (S3 Open-UPRN parquet) gives the planning
|
|
protections (conservation/listed/heritage — gate the wall + solar measures) and
|
|
the coordinates that drive the Google Solar fetch (ADR-0026). Buildings S3
|
|
doesn't cover, or that Google has no solar coverage for, fall back to
|
|
unrestricted / no-solar and are still modelled. Pass `--no-solar` to skip the
|
|
Google leg.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import io
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Optional, cast
|
|
|
|
import boto3
|
|
import pandas as pd
|
|
|
|
_REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(_REPO_ROOT)) # worktree root first — avoid the import trap
|
|
|
|
from datatypes.epc.domain.epc_property_data import EpcPropertyData # noqa: E402
|
|
from domain.geospatial.planning_restrictions import PlanningRestrictions # noqa: E402
|
|
from domain.geospatial.spatial_reference import SpatialReference # noqa: E402
|
|
from domain.modelling.measure_type import MeasureType # noqa: E402
|
|
from domain.modelling.plan import Plan, PlanMeasure # noqa: E402
|
|
from domain.modelling.scenario import Scenario # noqa: E402
|
|
from harness.console import run_modelling # noqa: E402
|
|
from harness.plan_table import format_plan_table # noqa: E402
|
|
from infrastructure.epc_client.epc_client_service import EpcClientService # noqa: E402
|
|
from infrastructure.solar.google_solar_api_client import ( # noqa: E402
|
|
BuildingInsightsNotFoundError,
|
|
GoogleSolarApiClient,
|
|
)
|
|
from repositories.geospatial.geospatial_s3_repository import ( # noqa: E402
|
|
GeospatialS3Repository,
|
|
ParquetReader,
|
|
)
|
|
from repositories.product.product_postgres_repository import ( # noqa: E402
|
|
ProductPostgresRepository,
|
|
)
|
|
from repositories.postgres_unit_of_work import PostgresUnitOfWork # noqa: E402
|
|
from repositories.scenario.scenario_postgres_repository import ( # noqa: E402
|
|
ScenarioPostgresRepository,
|
|
)
|
|
from sqlalchemy import Engine, create_engine, text # noqa: E402
|
|
from sqlmodel import Session # noqa: E402
|
|
|
|
_ENV_PATH = _REPO_ROOT / "backend" / ".env"
|
|
_MARKDOWN_PATH = Path("modelling_e2e.md")
|
|
_CSV_PATH = Path("modelling_e2e.csv")
|
|
|
|
|
|
def _load_env(path: Path) -> None:
|
|
"""Load `KEY=value` lines from `backend/.env` into the environment (without
|
|
overriding anything already set), so the DB creds + EPC token are present."""
|
|
if not path.exists():
|
|
return
|
|
for raw in path.read_text(encoding="utf-8").splitlines():
|
|
line = raw.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
|
|
|
|
|
|
def _db_url() -> str:
|
|
"""The connection string from the FastAPI-layer `DB_*` env vars."""
|
|
env = os.environ
|
|
return (
|
|
f"postgresql+psycopg2://{env['DB_USERNAME']}:{env['DB_PASSWORD']}"
|
|
f"@{env['DB_HOST']}:{env['DB_PORT']}/{env['DB_NAME']}"
|
|
)
|
|
|
|
|
|
def _s3_parquet_reader(bucket: str) -> ParquetReader:
|
|
"""A `ParquetReader` (key -> DataFrame) backed by `bucket` in S3, for the
|
|
`GeospatialS3Repository`. AWS creds come from the ambient `~/.aws` profile;
|
|
pyarrow reads the parquet bytes (s3fs is not installed here)."""
|
|
# boto3 ships only partial type stubs, so the client is an untyped boundary.
|
|
client = cast(Any, boto3.client("s3")) # pyright: ignore[reportUnknownMemberType]
|
|
|
|
def read(key: str) -> pd.DataFrame:
|
|
body = cast(bytes, client.get_object(Bucket=bucket, Key=key)["Body"].read())
|
|
return pd.read_parquet(io.BytesIO(body))
|
|
|
|
return read
|
|
|
|
|
|
def _spatial_for(
|
|
repo: GeospatialS3Repository, uprn: int
|
|
) -> Optional[SpatialReference]:
|
|
"""The UPRN's spatial reference (coordinates + planning protections), or
|
|
None when S3 doesn't cover it — a missing reference must not abort the run,
|
|
so a lookup error degrades to None (unrestricted, no solar)."""
|
|
try:
|
|
return repo.spatial_for(uprn)
|
|
except Exception as error: # noqa: BLE001 — S3/parquet hiccup is non-fatal
|
|
print(f" spatial lookup failed for uprn {uprn}: {type(error).__name__}: {error}")
|
|
return None
|
|
|
|
|
|
def _solar_insights_for(
|
|
solar_client: GoogleSolarApiClient, spatial: Optional[SpatialReference]
|
|
) -> Optional[dict[str, Any]]:
|
|
"""The raw Google Solar `buildingInsights` for the reference's coordinates,
|
|
or None when there are no coordinates / Google has no coverage there."""
|
|
if spatial is None or spatial.coordinates is None:
|
|
return None
|
|
try:
|
|
return solar_client.get_building_insights(
|
|
spatial.coordinates.longitude, spatial.coordinates.latitude
|
|
)
|
|
except BuildingInsightsNotFoundError:
|
|
return None # no Google solar coverage at this point — model without it
|
|
|
|
|
|
def _engine() -> Engine:
|
|
"""A connection-pooled engine to DevAssessmentModelDB (DB_* creds)."""
|
|
return create_engine(
|
|
_db_url(), pool_pre_ping=True, connect_args={"connect_timeout": 10}
|
|
)
|
|
|
|
|
|
def _uprns_for(engine: Engine, property_ids: list[int]) -> dict[int, Optional[int]]:
|
|
"""Read each Property's UPRN from the DB (read-only)."""
|
|
with engine.connect() as conn:
|
|
rows = conn.execute(
|
|
text("SELECT id, uprn FROM property WHERE id = ANY(:ids)"),
|
|
{"ids": property_ids},
|
|
).fetchall()
|
|
return {int(pid): (int(uprn) if uprn is not None else None) for pid, uprn in rows}
|
|
|
|
|
|
def _scenario_for(session: Session, scenario_id: int) -> Scenario:
|
|
"""Read the Scenario the run targets (read-only). An Increasing-EPC Scenario
|
|
must carry a ``goal_value`` (band) — the old null-band rows were a fixed bug
|
|
and crash the Optimiser's target — so reject one that does not."""
|
|
scenario: Scenario = ScenarioPostgresRepository(session).get_many([scenario_id])[0]
|
|
if scenario.goal == "Increasing EPC" and not scenario.goal_value:
|
|
raise ValueError(
|
|
f"scenario {scenario_id} has no goal_value (band); pick a recent one"
|
|
)
|
|
return scenario
|
|
|
|
|
|
def _parse_measures(raw: Optional[str]) -> Optional[frozenset[MeasureType]]:
|
|
"""Parse `--measures a,b,c` into a `considered_measures` allowlist, or None
|
|
(consider every modelled measure) when unset. Raises on an unknown type."""
|
|
if raw is None:
|
|
return None
|
|
return frozenset(MeasureType(token.strip()) for token in raw.split(",") if token.strip())
|
|
|
|
|
|
def _context_summary(
|
|
spatial: Optional[SpatialReference], solar_insights: Optional[dict[str, Any]]
|
|
) -> str:
|
|
"""A one-line note on what the geospatial leg contributed: which planning
|
|
protections gated the measures, and whether Google Solar potential fired."""
|
|
if spatial is None:
|
|
restrictions_note = "no spatial reference"
|
|
else:
|
|
flags = [
|
|
name
|
|
for name, on in (
|
|
("conservation", spatial.restrictions.in_conservation_area),
|
|
("listed", spatial.restrictions.is_listed),
|
|
("heritage", spatial.restrictions.is_heritage),
|
|
)
|
|
if on
|
|
]
|
|
restrictions_note = ", ".join(flags) if flags else "unrestricted"
|
|
solar_note = "solar ✓" if solar_insights is not None else "no solar"
|
|
return f"{restrictions_note}; {solar_note}"
|
|
|
|
|
|
def _measure_summary(measure: PlanMeasure) -> str:
|
|
return (
|
|
f" - {measure.measure_type}: "
|
|
f"+{measure.impact.sap_points:.2f} SAP · £{measure.cost.total:,.0f} "
|
|
f"— {measure.description}"
|
|
)
|
|
|
|
|
|
def _persist(
|
|
engine: Engine,
|
|
*,
|
|
property_id: int,
|
|
uprn: int,
|
|
portfolio_id: int,
|
|
scenario: Scenario,
|
|
epc: EpcPropertyData,
|
|
spatial: Optional[SpatialReference],
|
|
solar_insights: Optional[dict[str, Any]],
|
|
plan: Plan,
|
|
) -> None:
|
|
"""Write the run's inputs (EPC + spatial + solar) and the computed Plan to
|
|
the DB in one Unit of Work, then commit. ``PlanPostgresRepository`` replaces
|
|
any existing Plan for ``(property_id, scenario.id)`` (idempotent re-run)."""
|
|
with PostgresUnitOfWork(lambda: Session(engine)) as uow:
|
|
uow.epc.save(epc, property_id=property_id, portfolio_id=portfolio_id)
|
|
if spatial is not None:
|
|
uow.spatial.save(uprn, spatial)
|
|
if solar_insights is not None:
|
|
uow.solar.save(property_id, solar_insights)
|
|
uow.plan.save(
|
|
plan,
|
|
property_id=property_id,
|
|
scenario_id=scenario.id,
|
|
portfolio_id=portfolio_id,
|
|
is_default=scenario.is_default,
|
|
)
|
|
uow.commit()
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("property_ids", type=int, nargs="+", help="Property ids to model")
|
|
parser.add_argument("--goal", default="C", help="target band when no --scenario-id (default C)")
|
|
parser.add_argument(
|
|
"--scenario-id", type=int, default=None, help="model against this DB Scenario"
|
|
)
|
|
parser.add_argument(
|
|
"--measures",
|
|
default=None,
|
|
help="comma-separated measure types to consider (default: all)",
|
|
)
|
|
parser.add_argument(
|
|
"--portfolio-id", type=int, default=None, help="portfolio id (required for --persist)"
|
|
)
|
|
parser.add_argument(
|
|
"--persist",
|
|
action="store_true",
|
|
help="WRITE the inputs + Plan to the DB (default: inspect only, no writes)",
|
|
)
|
|
parser.add_argument(
|
|
"--no-solar",
|
|
action="store_true",
|
|
help="skip the live Google Solar fetch (no Solar PV Options)",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
if args.persist and (args.scenario_id is None or args.portfolio_id is None):
|
|
parser.error("--persist requires --scenario-id and --portfolio-id")
|
|
|
|
_load_env(_ENV_PATH)
|
|
# The new gov EPC API (Bearer) authenticates with OPEN_EPC_API_TOKEN — the
|
|
# name is misleading; EPC_AUTH_TOKEN is dead (403). Verified against the
|
|
# /api/domestic/search endpoint.
|
|
epc_client = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"])
|
|
geospatial = GeospatialS3Repository(_s3_parquet_reader(os.environ["DATA_BUCKET"]))
|
|
solar_client = GoogleSolarApiClient(os.environ["GOOGLE_SOLAR_API_KEY"])
|
|
engine = _engine()
|
|
considered = _parse_measures(args.measures)
|
|
uprns = _uprns_for(engine, args.property_ids)
|
|
# One read-only session for the live `material` catalogue, reused across the
|
|
# batch so both store and no-store runs price against the same DB rows.
|
|
catalogue_session = Session(engine)
|
|
products = ProductPostgresRepository(catalogue_session)
|
|
scenario: Optional[Scenario] = (
|
|
_scenario_for(catalogue_session, args.scenario_id)
|
|
if args.scenario_id is not None
|
|
else None
|
|
)
|
|
|
|
target = (
|
|
f"scenario {scenario.id} (band {scenario.goal_value})"
|
|
if scenario is not None
|
|
else f"synthesised Increasing-EPC band {args.goal}"
|
|
)
|
|
measures_note = ",".join(sorted(considered)) if considered else "all measures"
|
|
mode = "PERSISTING to DB" if args.persist else "no DB writes"
|
|
print(
|
|
f"modelling {len(args.property_ids)} propertie(s) · {target} · {measures_note} · "
|
|
f"{mode} (DB material catalogue, live EPC/solar)...\n"
|
|
)
|
|
|
|
md_lines: list[str] = [f"# Modelling recommendations ({target}, {measures_note})\n"]
|
|
csv_rows: list[str] = [
|
|
"property_id,uprn,baseline_sap,post_sap,measures,measure_types,cost_of_works"
|
|
]
|
|
|
|
for property_id in args.property_ids:
|
|
uprn = uprns.get(property_id)
|
|
try:
|
|
if uprn is None:
|
|
raise ValueError("no UPRN on the property row")
|
|
epc: Optional[EpcPropertyData] = epc_client.get_by_uprn(uprn)
|
|
if epc is None:
|
|
raise ValueError(f"no EPC found for UPRN {uprn}")
|
|
spatial: Optional[SpatialReference] = _spatial_for(geospatial, uprn)
|
|
restrictions: PlanningRestrictions = (
|
|
spatial.restrictions if spatial is not None else PlanningRestrictions()
|
|
)
|
|
solar_insights: Optional[dict[str, Any]] = (
|
|
None if args.no_solar else _solar_insights_for(solar_client, spatial)
|
|
)
|
|
plan: Plan = run_modelling(
|
|
epc,
|
|
goal_band=args.goal,
|
|
planning_restrictions=restrictions,
|
|
solar_insights=solar_insights,
|
|
considered_measures=considered,
|
|
products=products,
|
|
scenario=scenario,
|
|
print_table=False,
|
|
)
|
|
if args.persist:
|
|
assert scenario is not None # guaranteed by the --persist guard
|
|
_persist(
|
|
engine,
|
|
property_id=property_id,
|
|
uprn=uprn,
|
|
portfolio_id=args.portfolio_id,
|
|
scenario=scenario,
|
|
epc=epc,
|
|
spatial=spatial,
|
|
solar_insights=solar_insights,
|
|
plan=plan,
|
|
)
|
|
except Exception as error: # noqa: BLE001 — one bad property must not stop the run
|
|
line = f"property {property_id} (uprn {uprn}): ERROR — {type(error).__name__}: {error}"
|
|
print(line + "\n")
|
|
md_lines.append(f"## Property {property_id}\n\n`{line}`\n")
|
|
csv_rows.append(f"{property_id},{uprn or ''},,,,ERROR,")
|
|
continue
|
|
|
|
measure_types = [m.measure_type for m in plan.measures]
|
|
context = _context_summary(spatial, solar_insights)
|
|
header = (
|
|
f"=== Property {property_id} (uprn {uprn}) === "
|
|
f"SAP {plan.baseline.sap_continuous:.1f} -> {plan.post_sap_continuous:.1f} "
|
|
f"· {len(plan.measures)} measure(s) · £{plan.cost_of_works:,.0f} · {context}"
|
|
)
|
|
print(header)
|
|
print(format_plan_table(plan))
|
|
print()
|
|
|
|
md_lines.append(f"## Property {property_id} (uprn {uprn})\n")
|
|
md_lines.append(
|
|
f"SAP {plan.baseline.sap_continuous:.1f} → {plan.post_sap_continuous:.1f} "
|
|
f"· {len(plan.measures)} measure(s) · cost £{plan.cost_of_works:,.0f} "
|
|
f"· {context}\n"
|
|
)
|
|
md_lines.extend(_measure_summary(m) for m in plan.measures)
|
|
md_lines.append("")
|
|
csv_rows.append(
|
|
f"{property_id},{uprn},{plan.baseline.sap_continuous:.2f},"
|
|
f"{plan.post_sap_continuous:.2f},{len(plan.measures)},"
|
|
f"{'|'.join(measure_types)},{plan.cost_of_works:.0f}"
|
|
)
|
|
|
|
catalogue_session.close()
|
|
_MARKDOWN_PATH.write_text("\n".join(md_lines) + "\n", encoding="utf-8")
|
|
_CSV_PATH.write_text("\n".join(csv_rows) + "\n", encoding="utf-8")
|
|
print(f"wrote {_MARKDOWN_PATH.resolve()}")
|
|
print(f"wrote {_CSV_PATH.resolve()}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|