Model/harness/console.py

"""Run one property through the full First Run pipeline with no database.

The interactive inspection entrypoint: hand it an `EpcPropertyData` (e.g.
`EpcPropertyDataMapper.from_api_response(json)`), and it wires the whole
`AraFirstRunPipeline` (Ingestion -> Baseline -> Modelling) against in-memory
fakes — no Postgres, no network — runs it, prints the sense-check table, and
returns the `Plan` for further poking.

Dev tooling, not deployed: it reuses the in-memory test fakes, so run it from a
REPL at the worktree root::

    from datatypes.epc.domain.mapper import EpcPropertyDataMapper
    from harness.console import run_one
    plan = run_one(EpcPropertyDataMapper.from_api_response(my_api_json), goal_band="C")
"""

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Any, Optional

from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.geospatial.coordinates import Coordinates
from domain.geospatial.planning_restrictions import PlanningRestrictions
from domain.modelling.measure_type import MeasureType
from domain.modelling.plan import Plan
from domain.modelling.scenario import Scenario
from domain.property.property import Property, PropertyIdentity
from domain.property_baseline.rebaseliner import StubRebaseliner
from domain.sap10_calculator.calculator import Sap10Calculator
from harness.plan_table import format_plan_table
from orchestration.ara_first_run_pipeline import AraFirstRunPipeline
from orchestration.ingestion_orchestrator import IngestionOrchestrator
from orchestration.modelling_orchestrator import ModellingOrchestrator
from orchestration.property_baseline_orchestrator import PropertyBaselineOrchestrator
from repositories.fuel_rates.fuel_rates_static_file_repository import (
    FuelRatesStaticFileRepository,
)
from repositories.geospatial.geospatial_repository import GeospatialRepository
from repositories.product.product_json_repository import ProductJsonRepository
from tests.orchestration.fakes import (
    FakeEpcRepo,
    FakePlanRepository,
    FakePropertyRepo,
    FakeScenarioRepository,
    FakeSolarRepo,
    FakeUnitOfWork,
)

DEFAULT_CATALOGUE = Path(__file__).resolve().parent / "sample_catalogue.json"

_PROPERTY_ID = 1
_SCENARIO_ID = 7
_PORTFOLIO_ID = 1


@dataclass
class _Command:
    portfolio_id: int
    property_ids: list[int]
    scenario_ids: list[int]


class _FetcherReturning:
    def __init__(self, epc: EpcPropertyData) -> None:
        self._epc = epc

    def get_by_uprn(self, uprn: int) -> Optional[EpcPropertyData]:
        return self._epc


class _NoCoordinates(GeospatialRepository):
    def coordinates_for(self, uprn: int) -> Optional[Coordinates]:
        return None  # skip the solar leg


class _UnusedSolarFetcher:
    def get_building_insights(
        self, longitude: float, latitude: float
    ) -> dict[str, Any]:  # pragma: no cover
        return {}


def run_one(
    epc: EpcPropertyData,
    *,
    goal_band: str = "C",
    catalogue_path: Path = DEFAULT_CATALOGUE,
    current_market_value: Optional[float] = None,
    print_table: bool = True,
) -> Plan:
    """Run ``epc`` through the full First Run pipeline with no database and
    return its Plan for the default Increasing-EPC Scenario targeting
    ``goal_band``. Prints the sense-check table unless ``print_table`` is False.

    Pass ``current_market_value`` (a Property Valuation) to value the Plan's
    Valuation Uplift in £ — otherwise the uplift is percentage-only (ADR-0018).
    ``epc`` must carry lodged recorded-performance + the RHI block (a real lodged
    EPC does) so the Baseline stage can run."""
    epc_repo = FakeEpcRepo()
    plan_repo = FakePlanRepository()
    property_repo = FakePropertyRepo(
        {
            _PROPERTY_ID: Property(
                identity=PropertyIdentity(
                    portfolio_id=_PORTFOLIO_ID,
                    postcode="A0 0AA",
                    address="1 Some Street",
                    uprn=12345,
                ),
                current_market_value=current_market_value,
            )
        },
        epc_repo=epc_repo,
    )
    unit = FakeUnitOfWork(
        property=property_repo,
        epc=epc_repo,
        scenario=FakeScenarioRepository(
            {
                _SCENARIO_ID: Scenario(
                    id=_SCENARIO_ID,
                    goal="Increasing EPC",
                    goal_value=goal_band,
                    budget=None,
                    is_default=True,
                )
            }
        ),
        product=ProductJsonRepository(catalogue_path),
        plan=plan_repo,
    )

    pipeline = AraFirstRunPipeline(
        ingestion=IngestionOrchestrator(
            unit_of_work=lambda: unit,
            epc_fetcher=_FetcherReturning(epc),
            geospatial_repo=_NoCoordinates(),
            solar_fetcher=_UnusedSolarFetcher(),
        ),
        baseline=PropertyBaselineOrchestrator(
            unit_of_work=lambda: unit,
            rebaseliner=StubRebaseliner(),
            fuel_rates=FuelRatesStaticFileRepository(),
        ),
        modelling=ModellingOrchestrator(
            unit_of_work=lambda: unit,
            calculator=Sap10Calculator(),
            fuel_rates=FuelRatesStaticFileRepository(),
        ),
    )
    pipeline.run(
        _Command(
            portfolio_id=_PORTFOLIO_ID,
            property_ids=[_PROPERTY_ID],
            scenario_ids=[_SCENARIO_ID],
        )
    )

    plan = plan_repo.saved[(_PROPERTY_ID, _SCENARIO_ID)]
    if print_table:
        print("\n" + format_plan_table(plan))
    return plan


def run_modelling(
    epc: EpcPropertyData,
    *,
    goal_band: str = "C",
    catalogue_path: Path = DEFAULT_CATALOGUE,
    current_market_value: Optional[float] = None,
    planning_restrictions: PlanningRestrictions = PlanningRestrictions(),
    solar_insights: Optional[dict[str, Any]] = None,
    considered_measures: Optional[frozenset[MeasureType]] = None,
    print_table: bool = True,
) -> Plan:
    """Run ONLY the Modelling stage over ``epc`` with no database — skipping
    Ingestion and Baseline. Modelling re-scores the EPC itself, so unlike
    `run_one` this needs no lodged recorded-performance / RHI: it runs on any
    EPC the calculator can score, which is what you want for inspecting
    recommendations across an arbitrary EPC dump offline.

    ``solar_insights`` is the Property's raw Google Solar ``buildingInsights``
    JSON (as persisted by ``SolarRepository``); when given, the solar
    Recommendation Generator sees the dwelling's potential and can offer Solar
    PV Options (ADR-0026)."""
    plan_repo = FakePlanRepository()
    property_repo = FakePropertyRepo(
        {
            _PROPERTY_ID: Property(
                identity=PropertyIdentity(
                    portfolio_id=_PORTFOLIO_ID,
                    postcode="A0 0AA",
                    address="1 Some Street",
                    uprn=12345,
                ),
                epc=epc,
                current_market_value=current_market_value,
                planning_restrictions=planning_restrictions,
            )
        },
    )
    unit = FakeUnitOfWork(
        property=property_repo,
        solar=FakeSolarRepo(
            by_property={_PROPERTY_ID: solar_insights}
            if solar_insights is not None
            else None
        ),
        scenario=FakeScenarioRepository(
            {
                _SCENARIO_ID: Scenario(
                    id=_SCENARIO_ID,
                    goal="Increasing EPC",
                    goal_value=goal_band,
                    budget=None,
                    is_default=True,
                )
            }
        ),
        product=ProductJsonRepository(catalogue_path),
        plan=plan_repo,
    )

    ModellingOrchestrator(
        unit_of_work=lambda: unit,
        calculator=Sap10Calculator(),
        fuel_rates=FuelRatesStaticFileRepository(),
    ).run(
        property_ids=[_PROPERTY_ID],
        scenario_ids=[_SCENARIO_ID],
        portfolio_id=_PORTFOLIO_ID,
        considered_measures=considered_measures,
    )

    plan = plan_repo.saved[(_PROPERTY_ID, _SCENARIO_ID)]
    if print_table:
        print("\n" + format_plan_table(plan))
    return plan