diff --git a/domain/modelling/considered_measures.py b/domain/modelling/considered_measures.py index 530ca053..81fe20cb 100644 --- a/domain/modelling/considered_measures.py +++ b/domain/modelling/considered_measures.py @@ -20,6 +20,20 @@ from domain.modelling.measure_type import MeasureType from domain.modelling.recommendation import Recommendation +def combine_considered_measures( + a: Optional[frozenset[MeasureType]], + b: Optional[frozenset[MeasureType]], +) -> Optional[frozenset[MeasureType]]: + """Intersect two allowlists, treating ``None`` as "all measures". Used to + layer an explicit override over the allowlist a Scenario's exclusions imply: + None ∧ x = x, and both present narrows to their intersection.""" + if a is None: + return b + if b is None: + return a + return a & b + + def restrict_to_considered_measures( recommendations: Iterable[Recommendation], considered_measures: Optional[frozenset[MeasureType]], diff --git a/domain/modelling/generators/heating_recommendation.py b/domain/modelling/generators/heating_recommendation.py index b10e1b76..475039df 100644 --- a/domain/modelling/generators/heating_recommendation.py +++ b/domain/modelling/generators/heating_recommendation.py @@ -25,6 +25,7 @@ from domain.modelling.products import ( TuneUpCostInputs, ) from domain.modelling.measure_type import MeasureType +from domain.modelling.product import Product from domain.modelling.recommendation import Cost, MeasureOption, Recommendation from domain.modelling.simulation import EpcSimulation, HeatingOverlay from domain.sap10_calculator.tables.table_4b import ( @@ -641,8 +642,9 @@ def _ashp_option( if not _ashp_eligible(epc, restrictions): return None # Cost is composed per-dwelling from the rate sheet (ADR-0025), not the - # single catalogue scalar; the catalogue row is still read for its id. - product = products.get(_ASHP_MEASURE_TYPE) + # single catalogue scalar; the catalogue row is read only for its id, so an + # absent ASHP row must not suppress the bundle — it just carries no id. + product: Optional[Product] = products.get_optional(_ASHP_MEASURE_TYPE) cost: Cost = Products().ashp_bundle_cost(ashp_cost_inputs(epc)) return MeasureOption( measure_type=_ASHP_MEASURE_TYPE, @@ -652,7 +654,7 @@ def _ashp_option( ), overlay=EpcSimulation(heating=_ASHP_OVERLAY), cost=cost, - material_id=product.id, + material_id=product.id if product is not None else None, ) diff --git a/domain/modelling/scenario.py b/domain/modelling/scenario.py index 07f95ecb..6792e268 100644 --- a/domain/modelling/scenario.py +++ b/domain/modelling/scenario.py @@ -12,16 +12,33 @@ columns are not modelled. Carries no phases — multi-phase is deferred from dataclasses import dataclass from typing import Optional +from domain.modelling.measure_type import MeasureType + +_NO_EXCLUSIONS: frozenset[MeasureType] = frozenset() + @dataclass(frozen=True) class Scenario: """A retrofit brief: its goal, optional budget, and whether it is the Property's default Scenario. `goal` / `goal_value` are the lodged target (e.g. "INCREASING_EPC" → band "C"); carried for the Optimiser, not yet - enforced.""" + enforced. + + `exclusions` are the measure types the brief bars from the run (the only + measure-scoping the live ``scenario`` table persists — there is no + inclusions column). Empty means nothing is barred.""" id: int goal: str goal_value: str budget: Optional[float] is_default: bool + exclusions: frozenset[MeasureType] = _NO_EXCLUSIONS + + def considered_measures(self) -> Optional[frozenset[MeasureType]]: + """The measure-type allowlist the Scenario's exclusions imply: every + modelled measure minus the excluded ones, or None (consider every + measure) when nothing is excluded.""" + if not self.exclusions: + return None + return frozenset(MeasureType) - self.exclusions diff --git a/harness/console.py b/harness/console.py index 5a408c75..b6052883 100644 --- a/harness/console.py +++ b/harness/console.py @@ -25,14 +25,19 @@ 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.recommendation import Recommendation from domain.modelling.scenario import Scenario +from domain.modelling.solar_potential import SolarPotential 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.modelling_orchestrator import ( + ModellingOrchestrator, + _candidate_recommendations, # pyright: ignore[reportPrivateUsage] +) from orchestration.property_baseline_orchestrator import PropertyBaselineOrchestrator from repositories.fuel_rates.fuel_rates_static_file_repository import ( FuelRatesStaticFileRepository, @@ -247,3 +252,31 @@ def run_modelling( if print_table: print("\n" + format_plan_table(plan)) return plan + + +def candidate_recommendations( + epc: EpcPropertyData, + *, + catalogue_path: Path = DEFAULT_CATALOGUE, + planning_restrictions: PlanningRestrictions = PlanningRestrictions(), + solar_insights: Optional[dict[str, Any]] = None, + considered_measures: Optional[frozenset[MeasureType]] = None, + products: Optional[ProductRepository] = None, +) -> list[Recommendation]: + """Every candidate Recommendation the Generators produce for ``epc`` — the + full menu of Measure Options with their per-Option cost, *before* the + Optimiser selects a Plan. Use this to inspect measures (and their cost) that + a Plan does not end up selecting, e.g. an ASHP the Optimiser passed over for + a cheaper route to the target band. Inputs mirror `run_modelling`.""" + solar_potential: Optional[SolarPotential] = ( + SolarPotential.from_building_insights(solar_insights) + if solar_insights is not None and "solarPotential" in solar_insights + else None + ) + return _candidate_recommendations( + epc, + products or ProductJsonRepository(catalogue_path), + planning_restrictions, + solar_potential, + considered_measures, + ) diff --git a/infrastructure/postgres/modelling/scenario_table.py b/infrastructure/postgres/modelling/scenario_table.py index 47b40b73..5f7197ba 100644 --- a/infrastructure/postgres/modelling/scenario_table.py +++ b/infrastructure/postgres/modelling/scenario_table.py @@ -8,10 +8,35 @@ from sqlalchemy import Enum as SAEnum from sqlalchemy.sql import func from sqlmodel import Field, SQLModel +from domain.modelling.measure_type import MeasureType from domain.modelling.portfolio_goal import PortfolioGoal from domain.modelling.scenario import Scenario +def _parse_exclusions(raw: Optional[str]) -> frozenset[MeasureType]: + """Parse the live ``scenario.exclusions`` column — a Postgres text-array + literal like ``{solar_pv,internal_wall_insulation}`` — into the excluded + MeasureTypes. Each token must be an exact MeasureType value (no high-level + category expansion); an unknown token is a data error and raises, matching + the repo's strict-enum convention.""" + if not raw: + return frozenset() + inner = raw.strip() + if inner.startswith("{") and inner.endswith("}"): + inner = inner[1:-1] + tokens = [token.strip().strip('"') for token in inner.split(",") if token.strip()] + excluded: set[MeasureType] = set() + for token in tokens: + try: + excluded.add(MeasureType(token)) + except ValueError as error: + raise ValueError( + f"scenario excludes unknown measure type {token!r}; the " + f"exclusions column must hold exact MeasureType values" + ) from error + return frozenset(excluded) + + class ScenarioModel(SQLModel, table=True): """The single SQLModel definition of the live ``scenario`` table (ADR-0017 amendment). Full legacy column parity; ``goal`` is the ``PortfolioGoal`` @@ -89,4 +114,5 @@ class ScenarioModel(SQLModel, table=True): goal_value=self.goal_value, budget=self.budget, is_default=self.is_default, + exclusions=_parse_exclusions(self.exclusions), ) diff --git a/orchestration/modelling_orchestrator.py b/orchestration/modelling_orchestrator.py index c1a9ea44..867cb8b2 100644 --- a/orchestration/modelling_orchestrator.py +++ b/orchestration/modelling_orchestrator.py @@ -7,7 +7,10 @@ from datatypes.epc.domain.epc import Epc from datatypes.epc.domain.epc_property_data import EpcPropertyData from domain.billing.bill import Bill, EnergyBreakdown from domain.billing.bill_derivation import BillDerivation -from domain.modelling.considered_measures import restrict_to_considered_measures +from domain.modelling.considered_measures import ( + combine_considered_measures, + restrict_to_considered_measures, +) from domain.modelling.generators.floor_recommendation import recommend_floor_insulation from domain.modelling.measure_type import MeasureType from domain.modelling.optimisation.measure_dependency import ventilation_dependency @@ -159,18 +162,23 @@ class ModellingOrchestrator: ) -> Plan: """Generate → score → optimise → re-score/repair → attribute → bill → assemble the Plan for one Property + Scenario.""" + # The Scenario's own exclusions scope the run; an explicit + # ``considered_measures`` (e.g. from a harness) narrows it further. + considered: Optional[frozenset[MeasureType]] = combine_considered_measures( + scenario.considered_measures(), considered_measures + ) groups: list[list[ScoredOption]] = _scored_candidate_groups( scorer, effective_epc, products, planning_restrictions, solar_potential, - considered_measures, + considered, ) # Forced Measure Dependencies (ventilation) are excluded from the pool # but injected into the package before the re-score (ADR-0016). dependencies: list[MeasureDependency] = _measure_dependencies( - effective_epc, products, considered_measures + effective_epc, products, considered ) package: OptimisedPackage = optimise_package( groups=groups, @@ -249,24 +257,86 @@ def _candidate_recommendations( solar_potential: Optional[SolarPotential], considered_measures: Optional[frozenset[MeasureType]], ) -> list[Recommendation]: - """Run every Recommendation Generator; keep the ones that apply. Solid-wall - insulation, glazing, heating and solar are additionally gated by the - Property's planning protections (ADR-0019 / ADR-0022 / ADR-0024 / ADR-0026); - solar also needs the Property's Google solar potential. ``considered_measures`` - then restricts the survivors to the run's allowlist (None = all).""" - found = ( - recommend_cavity_wall(effective_epc, products), - recommend_solid_wall(effective_epc, products, planning_restrictions), - recommend_roof_insulation(effective_epc, products), - recommend_floor_insulation(effective_epc, products), - recommend_glazing(effective_epc, products, planning_restrictions), - recommend_lighting(effective_epc, products), - recommend_heating(effective_epc, products, planning_restrictions), - recommend_secondary_heating_removal(effective_epc, products), - recommend_solar( - effective_epc, products, solar_potential, planning_restrictions + """Run the applicable Recommendation Generators; keep the ones that apply. + Solid-wall insulation, glazing, heating and solar are additionally gated by + the Property's planning protections (ADR-0019 / ADR-0022 / ADR-0024 / + ADR-0026); solar also needs the Property's Google solar potential. + + ``considered_measures`` gates generation *up front*: a generator runs only + when the allowlist admits at least one of the measure types it can emit + (None = every measure), so an excluded measure never reaches the catalogue — + which matters when the live ``material.type`` enum cannot even represent it + (e.g. ``secondary_heating_removal``). ``restrict_to_considered_measures`` + then trims any disallowed Options off the multi-Option survivors.""" + + def admitted(*emits: MeasureType) -> bool: + return considered_measures is None or any( + measure in considered_measures for measure in emits + ) + + # Each generator paired with the measure types it can emit, so the allowlist + # can skip a generator whose every type is excluded before it is invoked. + generators: tuple[ + tuple[bool, Callable[[], Optional[Recommendation]]], ... + ] = ( + ( + admitted(MeasureType.CAVITY_WALL_INSULATION), + lambda: recommend_cavity_wall(effective_epc, products), + ), + ( + admitted( + MeasureType.INTERNAL_WALL_INSULATION, + MeasureType.EXTERNAL_WALL_INSULATION, + ), + lambda: recommend_solid_wall( + effective_epc, products, planning_restrictions + ), + ), + ( + admitted( + MeasureType.LOFT_INSULATION, + MeasureType.SLOPING_CEILING_INSULATION, + MeasureType.FLAT_ROOF_INSULATION, + ), + lambda: recommend_roof_insulation(effective_epc, products), + ), + ( + admitted( + MeasureType.SUSPENDED_FLOOR_INSULATION, + MeasureType.SOLID_FLOOR_INSULATION, + ), + lambda: recommend_floor_insulation(effective_epc, products), + ), + ( + admitted(MeasureType.DOUBLE_GLAZING, MeasureType.SECONDARY_GLAZING), + lambda: recommend_glazing(effective_epc, products, planning_restrictions), + ), + ( + admitted(MeasureType.LOW_ENERGY_LIGHTING), + lambda: recommend_lighting(effective_epc, products), + ), + ( + admitted( + MeasureType.HIGH_HEAT_RETENTION_STORAGE_HEATERS, + MeasureType.AIR_SOURCE_HEAT_PUMP, + MeasureType.GAS_BOILER_UPGRADE, + MeasureType.SYSTEM_TUNE_UP, + MeasureType.SYSTEM_TUNE_UP_ZONED, + ), + lambda: recommend_heating(effective_epc, products, planning_restrictions), + ), + ( + admitted(MeasureType.SECONDARY_HEATING_REMOVAL), + lambda: recommend_secondary_heating_removal(effective_epc, products), + ), + ( + admitted(MeasureType.SOLAR_PV), + lambda: recommend_solar( + effective_epc, products, solar_potential, planning_restrictions + ), ), ) + found = [thunk() for is_admitted, thunk in generators if is_admitted] applicable = [ recommendation for recommendation in found if recommendation is not None ] diff --git a/repositories/product/product_json_repository.py b/repositories/product/product_json_repository.py index 902f931f..5c566a92 100644 --- a/repositories/product/product_json_repository.py +++ b/repositories/product/product_json_repository.py @@ -6,7 +6,7 @@ from typing import Any, cast from domain.modelling.contingencies import contingency_rate from domain.modelling.product import Product -from repositories.product.product_repository import ProductRepository +from repositories.product.product_repository import ProductNotFound, ProductRepository class ProductJsonRepository(ProductRepository): @@ -33,7 +33,7 @@ class ProductJsonRepository(ProductRepository): def get(self, measure_type: str) -> Product: entry: Any = self._entries.get(measure_type) if entry is None: - raise ValueError(f"no product for measure type {measure_type!r}") + raise ProductNotFound(f"no product for measure type {measure_type!r}") if not isinstance(entry, dict): raise ValueError(f"product {measure_type!r} entry is not an object") typed_entry: dict[str, Any] = cast("dict[str, Any]", entry) diff --git a/repositories/product/product_postgres_repository.py b/repositories/product/product_postgres_repository.py index c43a0ae1..5f27a68e 100644 --- a/repositories/product/product_postgres_repository.py +++ b/repositories/product/product_postgres_repository.py @@ -5,7 +5,7 @@ from sqlmodel import Session, col, select from domain.modelling.contingencies import contingency_rate from domain.modelling.product import Product from infrastructure.postgres.product_table import MaterialRow -from repositories.product.product_repository import ProductRepository +from repositories.product.product_repository import ProductNotFound, ProductRepository # The domain ``MeasureType`` vocabulary and the catalogue's ``material.type`` @@ -47,7 +47,9 @@ class ProductPostgresRepository(ProductRepository): .order_by(col(MaterialRow.id)) ).first() if row is None: - raise ValueError(f"no active product for measure type {measure_type!r}") + raise ProductNotFound( + f"no active product for measure type {measure_type!r}" + ) if row.total_cost is None: raise ValueError(f"product {measure_type!r} has no total_cost") return Product( diff --git a/repositories/product/product_repository.py b/repositories/product/product_repository.py index eab7b202..8aabd07d 100644 --- a/repositories/product/product_repository.py +++ b/repositories/product/product_repository.py @@ -1,10 +1,18 @@ from __future__ import annotations from abc import ABC, abstractmethod +from typing import Optional from domain.modelling.product import Product +class ProductNotFound(ValueError): + """Raised when the catalogue has no active entry for a Measure Type. A + subclass of ``ValueError`` so existing callers that catch ``ValueError`` + keep working, while callers that only want to know *whether* a row exists + (see ``get_optional``) can catch this case alone.""" + + class ProductRepository(ABC): """Loads Products from the catalogue, abstracting the data source (a Postgres-backed materials table today; a JSON file for costs the ETL does @@ -13,6 +21,17 @@ class ProductRepository(ABC): @abstractmethod def get(self, measure_type: str) -> Product: - """Return the Product for a Measure Type, raising if there is no active - catalogue entry.""" + """Return the Product for a Measure Type, raising ``ProductNotFound`` + if there is no active catalogue entry.""" ... + + def get_optional(self, measure_type: str) -> Optional[Product]: + """Return the Product for a Measure Type, or None when the catalogue has + no active entry. For measures whose cost is composed off-catalogue (e.g. + ASHP, priced from the rate sheet per ADR-0025) the catalogue row is read + only for its id, so a missing row is not an error — the measure is still + offered, just without a ``material_id``.""" + try: + return self.get(measure_type) + except ProductNotFound: + return None diff --git a/scripts/run_modelling_e2e.py b/scripts/run_modelling_e2e.py index 097a3109..c18a2d1b 100644 --- a/scripts/run_modelling_e2e.py +++ b/scripts/run_modelling_e2e.py @@ -14,10 +14,19 @@ persists **nothing** (the run is for inspecting recommendations); pass 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. +Pass `--scenario-id` to target a real Scenario; its ``goal_value`` drives the +band and **its ``exclusions`` drive which measures the run considers** (the live +scenario table persists exclusions only, no inclusions). Without `--scenario-id` +the run synthesises an Increasing-EPC-to-``--goal`` Scenario with no exclusions. +`--measures` / `--exclude-measures` are optional overlays layered on top of the +Scenario's own exclusions. + +KNOWN GOTCHA: the live ``material.type`` enum does not yet carry +``secondary_heating_removal``, so a Property with a lodged secondary heater +crashes the catalogue read for that measure. Until the catalogue stocks it, pass +``--exclude-measures secondary_heating_removal`` (an ASHP row is also absent, but +ASHP is priced off the rate sheet so it degrades to ``material_id=None`` rather +than crashing — no flag needed). 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 @@ -25,13 +34,13 @@ 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 + # inspect only (no DB writes), Scenario 1266, measures from the Scenario: + python -m scripts.run_modelling_e2e --scenario-id 1266 \ + --exclude-measures secondary_heating_removal 709634 709635 709636 + # same run, but persist EPC + spatial + solar + Plan (needs --portfolio-id): + python -m scripts.run_modelling_e2e --scenario-id 1266 --portfolio-id 785 \ + --persist --exclude-measures secondary_heating_removal 709634 709635 + python -m scripts.run_modelling_e2e --no-solar 709634 709635 # skip Google leg Per Property the spatial reference (S3 Open-UPRN parquet) gives the planning protections (conservation/listed/heritage — gate the wall + solar measures) and @@ -66,10 +75,14 @@ from repositories.property.property_overrides_postgres_reader import ( # noqa: ) 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 run_modelling # 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 @@ -93,6 +106,7 @@ 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: @@ -219,6 +233,21 @@ def _parse_measures(raw: Optional[str]) -> Optional[frozenset[MeasureType]]: ) +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: @@ -249,6 +278,53 @@ def _measure_summary(measure: PlanMeasure) -> str: ) +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, *, @@ -303,7 +379,15 @@ def main() -> None: parser.add_argument( "--measures", default=None, - help="comma-separated measure types to consider (default: all)", + 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", @@ -335,7 +419,9 @@ def main() -> None: 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) + cli_considered = _resolve_considered( + _parse_measures(args.measures), _parse_measures(args.exclude_measures) + ) uprns = _uprns_for(engine, args.property_ids) # Landlord Overrides are read from property_overrides and folded onto the lodged # EPC to form the Effective EPC the calculator scores (ADR-0032). @@ -350,6 +436,12 @@ def main() -> None: 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})" @@ -367,6 +459,10 @@ def main() -> None: 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) @@ -408,6 +504,15 @@ def main() -> None: 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( @@ -436,7 +541,9 @@ def main() -> None: 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} " @@ -444,6 +551,9 @@ def main() -> None: ) 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") @@ -452,19 +562,30 @@ def main() -> None: 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__": diff --git a/tests/domain/modelling/test_considered_measures.py b/tests/domain/modelling/test_considered_measures.py index b5b5b4e7..e2ef9ad9 100644 --- a/tests/domain/modelling/test_considered_measures.py +++ b/tests/domain/modelling/test_considered_measures.py @@ -8,7 +8,10 @@ Options; a Recommendation left with no allowed Option is dropped entirely. A None allowlist means "consider everything" (today's unrestricted behaviour). """ -from domain.modelling.considered_measures import restrict_to_considered_measures +from domain.modelling.considered_measures import ( + combine_considered_measures, + restrict_to_considered_measures, +) from domain.modelling.measure_type import MeasureType from domain.modelling.recommendation import MeasureOption, Recommendation from domain.modelling.simulation import EpcSimulation @@ -68,6 +71,18 @@ def test_drops_recommendations_with_no_allowed_option() -> None: assert surfaces == {"Heating & Hot Water", "Solar PV"} +def test_combine_treats_none_as_all_and_intersects_two_allowlists() -> None: + # Arrange + a = frozenset({MeasureType.SOLAR_PV, MeasureType.LOFT_INSULATION}) + b = frozenset({MeasureType.SOLAR_PV, MeasureType.CAVITY_WALL_INSULATION}) + + # Act / Assert — None means "all", so it never narrows; two sets intersect. + assert combine_considered_measures(None, b) == b + assert combine_considered_measures(a, None) == a + assert combine_considered_measures(None, None) is None + assert combine_considered_measures(a, b) == frozenset({MeasureType.SOLAR_PV}) + + def test_filters_options_within_a_kept_recommendation() -> None: # Arrange — HHRSH is allowed but the competing ASHP bundle is not. considered = frozenset( diff --git a/tests/domain/modelling/test_heating_recommendation.py b/tests/domain/modelling/test_heating_recommendation.py index 81b9f692..4f1a321a 100644 --- a/tests/domain/modelling/test_heating_recommendation.py +++ b/tests/domain/modelling/test_heating_recommendation.py @@ -8,10 +8,14 @@ later slices. Detection + pricing only; impact is produced by scoring (ADR-0016) from datatypes.epc.domain.epc_property_data import EpcPropertyData from domain.geospatial.planning_restrictions import PlanningRestrictions from domain.modelling.generators.heating_recommendation import recommend_heating +from domain.modelling.measure_type import MeasureType from domain.modelling.product import Product from domain.modelling.recommendation import Recommendation from domain.modelling.simulation import HeatingOverlay -from repositories.product.product_repository import ProductRepository +from repositories.product.product_repository import ( + ProductNotFound, + ProductRepository, +) from tests.domain.modelling._elmhurst_recommendation import ( parse_recommendation_summary, ) @@ -170,6 +174,41 @@ def test_gas_boiler_house_yields_an_ashp_bundle() -> None: ) +class _StubProductsWithoutAshp(ProductRepository): + """A catalogue with no ASHP row. ASHP's cost is composed from the rate sheet + (ADR-0025) and the catalogue row is read only for its id, so a missing row + must not suppress the bundle — it just carries no material_id.""" + + def get(self, measure_type: str) -> Product: + if measure_type == MeasureType.AIR_SOURCE_HEAT_PUMP: + raise ProductNotFound(f"no active product for {measure_type!r}") + return Product( + measure_type=measure_type, unit_cost_per_m2=3500.0, contingency_rate=0.26 + ) + + +def test_ashp_bundle_offered_when_catalogue_lacks_an_ashp_product() -> None: + # Arrange — a mains-gas house (ASHP-eligible) priced against a catalogue with + # no ASHP row; ASHP is costed from the rate sheet, so the bundle must still + # be offered, just without a material id. + baseline: EpcPropertyData = _gas_boiler_house() + + # Act + recommendation: Recommendation | None = recommend_heating( + baseline, _StubProductsWithoutAshp() + ) + + # Assert — the ASHP bundle is still offered, carrying its composite cost and + # no material id. + assert recommendation is not None + option = next( + o for o in recommendation.options if o.measure_type == "air_source_heat_pump" + ) + assert option.material_id is None + assert option.cost is not None + assert option.cost.total > 0.0 + + def test_ashp_bundle_carries_the_composite_per_dwelling_cost() -> None: # Arrange — a mains-gas regular boiler with a cylinder (90 m2, 7 habitable # rooms): the ASHP reuses the existing wet system (ADR-0025). diff --git a/tests/domain/modelling/test_scenario.py b/tests/domain/modelling/test_scenario.py new file mode 100644 index 00000000..f159aff2 --- /dev/null +++ b/tests/domain/modelling/test_scenario.py @@ -0,0 +1,43 @@ +"""The Scenario's measure scoping: its exclusions imply the allowlist the run +considers (the live `scenario` table persists exclusions only — no inclusions).""" + +from domain.modelling.measure_type import MeasureType +from domain.modelling.scenario import Scenario + + +def _scenario(exclusions: frozenset[MeasureType]) -> Scenario: + return Scenario( + id=1, + goal="Increasing EPC", + goal_value="C", + budget=None, + is_default=True, + exclusions=exclusions, + ) + + +def test_no_exclusions_considers_every_measure() -> None: + # Arrange + scenario = _scenario(frozenset()) + + # Act + considered = scenario.considered_measures() + + # Assert — None means "consider all" (the unrestricted default). + assert considered is None + + +def test_exclusions_imply_the_complement_allowlist() -> None: + # Arrange — exclude solar PV and ASHP. + scenario = _scenario( + frozenset({MeasureType.SOLAR_PV, MeasureType.AIR_SOURCE_HEAT_PUMP}) + ) + + # Act + considered = scenario.considered_measures() + + # Assert — every modelled measure survives except the two excluded ones. + assert considered is not None + assert MeasureType.SOLAR_PV not in considered + assert MeasureType.AIR_SOURCE_HEAT_PUMP not in considered + assert considered == frozenset(MeasureType) - scenario.exclusions diff --git a/tests/harness/test_console.py b/tests/harness/test_console.py index fb7c1c07..3a8a80a6 100644 --- a/tests/harness/test_console.py +++ b/tests/harness/test_console.py @@ -10,9 +10,16 @@ from datatypes.epc.domain.epc import Epc from datatypes.epc.domain.epc_property_data import EpcPropertyData from domain.geospatial.planning_restrictions import PlanningRestrictions from domain.modelling.contingencies import contingency_rate +from domain.modelling.measure_type import MeasureType +from domain.modelling.scenario import Scenario from domain.modelling.generators.heating_recommendation import recommend_heating from domain.modelling.generators.solid_wall_recommendation import recommend_solid_wall -from harness.console import DEFAULT_CATALOGUE, run_modelling, run_one +from harness.console import ( + DEFAULT_CATALOGUE, + candidate_recommendations, + run_modelling, + run_one, +) from repositories.product.product_json_repository import ProductJsonRepository from tests.domain.modelling._elmhurst_recommendation import ( parse_recommendation_summary, @@ -224,6 +231,54 @@ def test_run_modelling_recommends_hhr_storage_for_an_electric_dwelling() -> None assert "air_source_heat_pump" in {m.measure_type for m in plan.measures} +def test_candidate_recommendations_surface_unselected_options_with_cost() -> None: + # Arrange — an electric dwelling whose heating Recommendation offers both an + # ASHP and an HHR-storage bundle; the Optimiser selects only one of them. + epc: EpcPropertyData = _electric_storage_lit_epc() + + # Act — the full candidate menu (every Generator Option, pre-optimisation) + # alongside the optimised Plan. + candidates = candidate_recommendations(epc) + plan = run_modelling(epc, goal_band="C", print_table=False) + + # Assert — the menu carries every offered Option (so a measure the Plan did + # not select, like the passed-over HHR bundle, is still inspectable), and + # every Option carries a cost so "cost per measure" is always available. + offered = { + option.measure_type for rec in candidates for option in rec.options + } + selected = {measure.measure_type for measure in plan.measures} + assert "high_heat_retention_storage_heaters" in offered + assert "air_source_heat_pump" in offered + assert offered - selected # at least one offered measure was not selected + assert all( + option.cost is not None for rec in candidates for option in rec.options + ) + + +def test_run_modelling_honours_a_scenarios_exclusions() -> None: + # Arrange — an electric dwelling whose optimised Plan normally leads with the + # ASHP bundle; a Scenario that excludes ASHP must keep it out of the Plan. + epc: EpcPropertyData = _electric_storage_lit_epc() + scenario = Scenario( + id=1, + goal="Increasing EPC", + goal_value="C", + budget=None, + is_default=True, + exclusions=frozenset({MeasureType.AIR_SOURCE_HEAT_PUMP}), + ) + + # Act — the orchestrator derives the allowlist from the Scenario's exclusions. + plan = run_modelling(epc, scenario=scenario, print_table=False) + + # Assert — ASHP is excluded; the Plan still improves the dwelling via other + # measures (e.g. the HHR storage bundle). + measure_types = {measure.measure_type for measure in plan.measures} + assert MeasureType.AIR_SOURCE_HEAT_PUMP not in measure_types + assert measure_types # a non-empty Plan still came back + + def test_sample_catalogue_prices_every_generator_measure_type() -> None: # Arrange — the default offline catalogue. products: ProductJsonRepository = ProductJsonRepository(DEFAULT_CATALOGUE) diff --git a/tests/orchestration/test_modelling_solar_threading.py b/tests/orchestration/test_modelling_solar_threading.py index 68884d27..988f4229 100644 --- a/tests/orchestration/test_modelling_solar_threading.py +++ b/tests/orchestration/test_modelling_solar_threading.py @@ -116,6 +116,53 @@ def test_candidate_recommendations_excludes_solar_without_potential() -> None: assert "Solar PV" not in {r.surface for r in recommendations} +class _ProductsRaisingFor(ProductRepository): + """A catalogue that raises for one measure type — mirroring the live DB, + whose ``material.type`` enum does not carry ``secondary_heating_removal``. + Invoking that measure's generator would raise, so this proves an excluded + generator is never run.""" + + def __init__(self, forbidden: MeasureType) -> None: + self._forbidden = forbidden + + def get(self, measure_type: str) -> Product: + if measure_type == self._forbidden: + raise ValueError(f"catalogue cannot represent {measure_type!r}") + return Product( + measure_type=measure_type, + unit_cost_per_m2=0.0, + contingency_rate=0.15, + id=909, + ) + + +def test_an_excluded_measures_generator_is_not_invoked() -> None: + # Arrange — a dwelling with a lodged secondary heating system (so the + # secondary-heating generator is eligible to fire) priced against a catalogue + # that raises for that type, exactly as the live `material.type` enum does. + epc = _eligible_house() + epc.sap_heating.secondary_heating_type = 631 + allowlist = frozenset(MeasureType) - {MeasureType.SECONDARY_HEATING_REMOVAL} + + # Act — excluding the measure must stop its generator running at all (it would + # otherwise query the catalogue and raise). + recommendations = _candidate_recommendations( + epc, + _ProductsRaisingFor(MeasureType.SECONDARY_HEATING_REMOVAL), + PlanningRestrictions(), + None, + allowlist, + ) + + # Assert — the run completes and no secondary-heating option leaks through. + option_types = { + option.measure_type + for recommendation in recommendations + for option in recommendation.options + } + assert MeasureType.SECONDARY_HEATING_REMOVAL not in option_types + + def test_considered_measures_restricts_candidates_to_the_allowlist() -> None: # Arrange — a solar-eligible house, with its solar potential present, so the # unrestricted run offers Solar PV alongside any fabric/heating candidates. diff --git a/tests/repositories/scenario/test_scenario_postgres_repository.py b/tests/repositories/scenario/test_scenario_postgres_repository.py index 8e0df21c..8b5804c7 100644 --- a/tests/repositories/scenario/test_scenario_postgres_repository.py +++ b/tests/repositories/scenario/test_scenario_postgres_repository.py @@ -14,6 +14,7 @@ import pytest from sqlalchemy import Engine from sqlmodel import Session +from domain.modelling.measure_type import MeasureType from domain.modelling.portfolio_goal import PortfolioGoal from domain.modelling.scenario import Scenario from infrastructure.postgres.modelling import ScenarioModel @@ -67,6 +68,80 @@ def test_get_many_maps_live_scenario_rows_to_domain_in_input_order( ) +def test_get_many_parses_the_exclusions_array_into_measure_types( + db_engine: Engine, +) -> None: + # Arrange — the live `exclusions` column is a Postgres text-array literal of + # exact MeasureType values. + with Session(db_engine) as session: + session.add( + ScenarioModel( + id=7, + goal=PortfolioGoal.INCREASING_EPC, + goal_value="C", + is_default=True, + exclusions="{solar_pv,internal_wall_insulation}", + ) + ) + session.commit() + + # Act + with Session(db_engine) as session: + scenario: Scenario = ScenarioPostgresRepository(session).get_many([7])[0] + + # Assert + assert scenario.exclusions == frozenset( + {MeasureType.SOLAR_PV, MeasureType.INTERNAL_WALL_INSULATION} + ) + + +def test_get_many_treats_a_null_exclusions_column_as_no_exclusions( + db_engine: Engine, +) -> None: + # Arrange + with Session(db_engine) as session: + session.add( + ScenarioModel( + id=7, + goal=PortfolioGoal.INCREASING_EPC, + goal_value="C", + is_default=True, + exclusions=None, + ) + ) + session.commit() + + # Act + with Session(db_engine) as session: + scenario: Scenario = ScenarioPostgresRepository(session).get_many([7])[0] + + # Assert + assert scenario.exclusions == frozenset() + + +def test_get_many_raises_on_an_exclusion_that_is_not_a_measure_type( + db_engine: Engine, +) -> None: + # Arrange — a legacy high-level category (`heating`) is not an exact + # MeasureType value; exact-only resolution must reject it loudly. + with Session(db_engine) as session: + session.add( + ScenarioModel( + id=7, + goal=PortfolioGoal.INCREASING_EPC, + goal_value="C", + is_default=True, + exclusions="{heating}", + ) + ) + session.commit() + + # Act / Assert + with Session(db_engine) as session: + with pytest.raises(ValueError): + ScenarioPostgresRepository(session).get_many([7]) + + def test_get_many_raises_when_a_scenario_id_is_missing(db_engine: Engine) -> None: # Arrange with Session(db_engine) as session: