From 31ced27162ad723d1936e0acde4148c4839c638b Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Tue, 16 Jun 2026 15:03:26 +0000 Subject: [PATCH] feat(modelling): surface the full candidate measure menu with per-measure cost MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The run only showed the measures the Optimiser selected, so a candidate it passed over (e.g. an ASHP it found too costly for the target band) and that measure's cost were invisible. Add `harness.console.candidate_recommendations` — every Generator Option with its per-Option cost, before optimisation — and have run_modelling_e2e print the full menu per property (flagging the selected Options), write a "cost per measure" section into the markdown, and emit a per-Option modelling_e2e_candidates.csv. Co-Authored-By: Claude Opus 4.8 --- harness/console.py | 35 ++++++++++++++- scripts/run_modelling_e2e.py | 80 ++++++++++++++++++++++++++++++++++- tests/harness/test_console.py | 32 +++++++++++++- 3 files changed, 144 insertions(+), 3 deletions(-) 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/scripts/run_modelling_e2e.py b/scripts/run_modelling_e2e.py index 1b900ed5..fca5bc11 100644 --- a/scripts/run_modelling_e2e.py +++ b/scripts/run_modelling_e2e.py @@ -61,8 +61,9 @@ from domain.geospatial.planning_restrictions import PlanningRestrictions # noqa 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.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 @@ -86,6 +87,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: @@ -236,6 +238,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, *, @@ -358,6 +407,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) @@ -384,6 +437,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( @@ -412,7 +474,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} " @@ -420,6 +484,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") @@ -428,19 +495,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/harness/test_console.py b/tests/harness/test_console.py index fb7c1c07..4896163f 100644 --- a/tests/harness/test_console.py +++ b/tests/harness/test_console.py @@ -12,7 +12,12 @@ from domain.geospatial.planning_restrictions import PlanningRestrictions from domain.modelling.contingencies import contingency_rate 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 +229,31 @@ 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_sample_catalogue_prices_every_generator_measure_type() -> None: # Arrange — the default offline catalogue. products: ProductJsonRepository = ProductJsonRepository(DEFAULT_CATALOGUE)