Model/scripts/run_modelling_e2e.py

"""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 for inspection: it reads each Property's UPRN from the DB, fetches the
latest EPC **live** from the gov EPC API by UPRN, then runs the Modelling stage
in memory (every Recommendation Generator → the Optimiser → a costed, attributed
Plan). It is read-only on the DB (just the UPRN lookup) and persists nothing —
purely for inspecting recommendations. Prints a per-Property plan table and
writes a Markdown + CSV summary.

Config: loads `backend/.env` for the DB creds (`DB_*`), the EPC API token
(`EPC_AUTH_TOKEN`), 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 so imports
resolve to this checkout:

    python -m scripts.run_modelling_e2e 115 116 117            # goal band C (default)
    python -m scripts.run_modelling_e2e --goal B 115 116 117   # a different target band
    python -m scripts.run_modelling_e2e --no-solar 115 116     # skip the Google Solar leg

Per Property the script resolves the UPRN's spatial reference from the Ordnance
Survey Open-UPRN parquet in S3 (`GeospatialS3Repository`): the planning
protections (conservation/listed/heritage) gate the wall + solar measures, and
the coordinates drive a live Google Solar `buildingInsights` fetch so the Solar
PV Options can fire (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 entirely.
"""

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.plan import Plan, PlanMeasure  # noqa: E402
from harness.console import DEFAULT_CATALOGUE, 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 sqlalchemy import create_engine, text  # 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 _uprns_for(property_ids: list[int]) -> dict[int, Optional[int]]:
    """Read each Property's UPRN from the DB (read-only)."""
    engine = create_engine(
        _db_url(), pool_pre_ping=True, connect_args={"connect_timeout": 10}
    )
    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 _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 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 EPC band (default C)")
    parser.add_argument(
        "--no-solar",
        action="store_true",
        help="skip the live Google Solar fetch (no Solar PV Options)",
    )
    args = parser.parse_args()

    _load_env(_ENV_PATH)
    epc_client = EpcClientService(os.environ["EPC_AUTH_TOKEN"])
    geospatial = GeospatialS3Repository(_s3_parquet_reader(os.environ["DATA_BUCKET"]))
    solar_client = GoogleSolarApiClient(os.environ["GOOGLE_SOLAR_API_KEY"])
    uprns = _uprns_for(args.property_ids)

    print(
        f"modelling {len(args.property_ids)} propertie(s) (goal band {args.goal}); "
        f"EPCs fetched live by UPRN, modelled in memory — no DB writes...\n"
    )

    md_lines: list[str] = [f"# Modelling recommendations (goal band {args.goal})\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,
                catalogue_path=DEFAULT_CATALOGUE,
                planning_restrictions=restrictions,
                solar_insights=solar_insights,
                print_table=False,
            )
        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}"
        )

    _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()