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: 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.considered_measures import (  # noqa: E402
    combine_considered_measures,
)
from domain.modelling.measure_type import MeasureType  # noqa: E402
from domain.modelling.plan import Plan, PlanMeasure  # noqa: E402
from domain.modelling.recommendation import Recommendation  # noqa: E402
from domain.modelling.scenario import Scenario  # noqa: E402
from harness.console import candidate_recommendations, 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")
_CANDIDATES_CSV_PATH = Path("modelling_e2e_candidates.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 _resolve_considered(
    allowlist: Optional[frozenset[MeasureType]],
    excluded: Optional[frozenset[MeasureType]],
) -> Optional[frozenset[MeasureType]]:
    """Combine the `--measures` allowlist with the `--exclude-measures` set. With
    no exclusions the allowlist is returned unchanged (None = every measure).
    With exclusions the result is (the allowlist, or every measure) minus the
    excluded types — so `--exclude-measures secondary_heating_removal` considers
    every measure except that one, without enumerating the rest."""
    if not excluded:
        return allowlist
    base = allowlist if allowlist is not None else frozenset(MeasureType)
    return base - excluded


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 _candidate_lines(
    recommendations: list[Recommendation], selected: set[MeasureType]
) -> list[str]:
    """Render every candidate Option (the full menu the Generators produced,
    not just the Plan the Optimiser selected) with its per-Option cost, flagging
    the Options that made it into the Plan — so measures the Optimiser passed
    over (e.g. an ASHP it found too costly for the target band) are visible."""
    lines: list[str] = []
    for recommendation in recommendations:
        for option in recommendation.options:
            cost = option.cost
            cost_note = (
                f"£{cost.total:,.0f} (+{cost.contingency_rate * 100:.0f}% cont.)"
                if cost is not None
                else "no cost"
            )
            flag = " ✓ SELECTED" if option.measure_type in selected else ""
            lines.append(
                f"    [{recommendation.surface}] {option.measure_type} · "
                f"{cost_note}{flag} — {option.description}"
            )
    return lines


def _candidate_csv_rows(
    property_id: int,
    uprn: Optional[int],
    recommendations: list[Recommendation],
    selected: set[MeasureType],
) -> list[str]:
    """One CSV row per candidate Option: the full measure menu with cost,
    contingency, and whether the Optimiser selected it."""
    rows: list[str] = []
    for recommendation in recommendations:
        for option in recommendation.options:
            cost = option.cost
            total = f"{cost.total:.2f}" if cost is not None else ""
            contingency = f"{cost.contingency_rate:.4f}" if cost is not None else ""
            chosen = "yes" if option.measure_type in selected else "no"
            description = option.description.replace(",", ";")
            rows.append(
                f"{property_id},{uprn or ''},{recommendation.surface},"
                f"{option.measure_type},{total},{contingency},{chosen},{description}"
            )
    return rows


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)
        # The live `solar` table is keyed by UPRN and needs the fetch's
        # coordinates; insights are only present when those coordinates were
        # (see `_solar_insights_for`), so `spatial.coordinates` is non-None here.
        if solar_insights is not None:
            assert spatial is not None and spatial.coordinates is not None
            uow.solar.save(
                uprn,
                longitude=spatial.coordinates.longitude,
                latitude=spatial.coordinates.latitude,
                insights=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="optional override: comma-separated measure types to consider. The "
        "Scenario's exclusions already drive this; the flag narrows it further.",
    )
    parser.add_argument(
        "--exclude-measures",
        default=None,
        help="optional override: comma-separated measure types to exclude on top "
        "of the Scenario's own exclusions (e.g. secondary_heating_removal, which "
        "the live catalogue does not yet stock)",
    )
    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)",
        default=False,
    )
    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()
    cli_considered = _resolve_considered(
        _parse_measures(args.measures), _parse_measures(args.exclude_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
    )
    # The Scenario's own exclusions drive which measures the run considers; the
    # --measures/--exclude-measures flags are an optional override layered on top.
    considered = combine_considered_measures(
        scenario.considered_measures() if scenario is not None else None,
        cli_considered,
    )

    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"
    ]
    candidate_csv_rows: list[str] = [
        "property_id,uprn,surface,measure_type,cost_total,contingency_rate,"
        "selected,description"
    ]

    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,
            )
            # The full candidate menu (every Generator Option + its cost), so
            # measures the Optimiser did not select are still visible.
            candidates: list[Recommendation] = candidate_recommendations(
                epc,
                planning_restrictions=restrictions,
                solar_insights=solar_insights,
                considered_measures=considered,
                products=products,
            )
            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
            # A failed catalogue query (e.g. a `material.type` enum mismatch)
            # aborts the shared session's transaction; without a rollback every
            # subsequent property reports `InFailedSqlTransaction` and masks its
            # own real error. Reset so each property surfaces what's wrong.
            catalogue_session.rollback()
            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]
        selected: set[MeasureType] = {m.measure_type for m in plan.measures}
        context = _context_summary(spatial, solar_insights)
        candidate_lines = _candidate_lines(candidates, selected)
        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(f"  candidate measures considered ({len(candidate_lines)} option(s)):")
        for candidate_line in candidate_lines:
            print(candidate_line)
        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.append("**Selected Plan**\n")
        md_lines.extend(_measure_summary(m) for m in plan.measures)
        md_lines.append("")
        md_lines.append("**All candidate measures (cost per measure)**\n")
        md_lines.extend(candidate_lines)
        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}"
        )
        candidate_csv_rows.extend(
            _candidate_csv_rows(property_id, uprn, candidates, selected)
        )

    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")
    _CANDIDATES_CSV_PATH.write_text(
        "\n".join(candidate_csv_rows) + "\n", encoding="utf-8"
    )
    print(f"wrote {_MARKDOWN_PATH.resolve()}")
    print(f"wrote {_CSV_PATH.resolve()}")
    print(f"wrote {_CANDIDATES_CSV_PATH.resolve()}")


if __name__ == "__main__":
    main()