junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-12 13:29:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1526 from Hestia-Homes/feature/fabric-first-scenario

Browse source 
Fabric-first scenario constraint: two-phase optimisation (ADR-0061)
This commit is contained in:
KhalimCK 2026-07-10 12:48:06 +01:00 • committed by GitHub
commit 660e42b920
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 1445 additions and 301 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

73
docs/adr/0061-fabric-first-is-a-two-phase-optimisation.md Normal file
View file

@ -0,0 +1,73 @@
---
status: accepted (extends ADR-0016)
---
# Fabric First is a two-phase optimisation with strict envelope priority
Landlords with a fabric-first retrofit policy require the building envelope —
insulation and windows — to be treated before heating systems and renewables
are considered. The legacy engine carried this as `enforce_fabric_first` on
the plan-API request body (`funding_optimiser.optimise_with_scenarios`): an
optimiser pass over fabric-only measures with the full budget, then a second
pass over the remainder with the leftover budget and the residual target,
summing approximate per-measure SAP points. The new engine needed the same
capability on the truthful-re-score Optimiser core (ADR-0016).
Decided in a grilling session with Khalim, 2026-07-09.
## Decision
**`fabric_first` is a Scenario attribute, and the Optimiser honours it as two
sequential `optimise_package` phases in which the envelope has strict first
claim on the budget** (`optimise_package_fabric_first`, ADR-0016 core reused
per phase).
- **The flag lives on the `scenario` table** (FE-owned Drizzle schema:
`fabric_first boolean NOT NULL DEFAULT false`), mirrored in `ScenarioModel`
and the domain `Scenario`. A Plan's provenance stays derivable from its
Scenario row alone — unlike the legacy request-body flag, the same
scenario_id cannot produce differently-constrained plans. **Deploy order**:
the Drizzle migration must land before this mirror, or every scenario read
crashes on the missing column.
- **Fabric = the building envelope** (`FABRIC_MEASURE_TYPES`): wall (CWI /
EWI / IWI), roof (loft / sloping-ceiling / flat-roof), floor (suspended /
solid) insulation, plus double / secondary glazing — the legacy list
exactly. Lighting, tune-ups, secondary-heating removal, heating and solar
all wait for phase 2. Mechanical ventilation is unclassified: it is never
selected, only injected as a forced Measure Dependency (ADR-0016) of the
fabric that triggers it, in whichever phase that happens — and only once.
- **Phase 1** runs `optimise_package` over the fabric groups with the full
budget: least-cost-to-target, repair, max-gain fallback. If the truthful
post-fabric score meets the Scenario target, the package is fabric-only —
surplus budget is left unspent, per the ADR-0016 no-overshoot doctrine.
- **Phase 2** (target unmet, or no target) optimises every group phase 1 did
not consume — non-fabric, plus fabric groups it left unpicked, which may
re-enter on their post-fabric worth — under the leftover budget. Candidates
are valued **against the fabric-applied dwelling**: warm-start signals are
re-scored and every package re-score is prefixed with the phase-1 overlays
(`_PrefixedScorer`), so a heating system whose worth changes once the
envelope is treated is re-ranked truthfully, and the returned Score remains
the whole-package figure against the true baseline (bills and the role-3
cascade stay honest).
- **Strict priority, not target-aware compromise**: phase 1 commits the
max-gain fabric package even when a cheaper fabric/heating split would have
reached the target — a £4,000 budget buys floor insulation and leaves the
£3,200 boiler unaffordable, and the target is missed rather than the fabric
skipped. This is the landlord's explicit trade.
- **Every goal honours the flag**, not just Increasing EPC: with no SAP
target the two phases are both max-gain, so fabric still gets first claim
on the budget.
- **No Plan-contract change**: the Scenario row is the provenance; the
best-practice cascade already orders fabric before heating in the persisted
measures.
## Consequences
- The Modelling orchestrator branches once, on `scenario.fabric_first`,
between `optimise_package` and `optimise_package_fabric_first`; everything
downstream (attribution, bills, persistence) is unchanged.
- Phase 2 re-scores each remaining Option once against the post-fabric
dwelling — a handful of extra calculator calls per fabric-first Plan.
- A fabric-first Plan can undershoot a target a plain Plan would have reached
within the same budget. This is by design and should be communicated when
scenario results are compared.

62
docs/adr/0062-goal-aligned-optimiser-objectives.md Normal file
View file

@ -0,0 +1,62 @@
---
status: accepted (extends ADR-0016; composes with ADR-0061)
---
# Goal-aligned Optimiser objectives: each goal maximises its own metric
Every Scenario goal used to optimise SAP. The legacy engine returned no
target for Energy Savings / Reducing CO2 (`optimiser_functions.calculate_gain`
→ `None`) and maximised SAP gain within budget regardless of the goal, and the
new engine inherited that: the goal label changed nothing but the words on the
brief. The scorer already computes each package's carbon and (via SapResult →
EnergyBreakdown → BillDerivation) its annual bill, so aligning the objective
is a selection change, not a calculator change.
Decided in a grilling session with Khalim, 2026-07-09.
## Decision
**The Optimiser maximises the Scenario goal's own metric, as a pluggable
`objective: Callable[[Score], float]` (higher is better), with no target:
goal-aligned briefs are "reduce as much as possible within this budget".**
- **Reducing CO2 emissions** maximises annual kg CO2 saved
(`-score.co2_kg_per_yr`).
- **Energy Savings** maximises the annual bill £ saved, priced at the **live
Fuel Rates snapshot** (ADR-0014), not SAP's internal tariff book — that
difference is the point of the goal. SAP is itself a cost-shaped rating, so
the two frequently agree; they diverge exactly when current tariffs disagree
with SAP's assumptions (e.g. the gas/electricity price ratio).
- **Increasing EPC** keeps its SAP objective and band-target semantics
(least-cost-to-target, repair, max-gain fallback) unchanged.
- **Valuation Improvement / None** stay max-SAP-within-budget — SAP is a
defensible valuation proxy and `None` has no semantics to encode.
- **`goal_value` is ignored for the goal-aligned goals** — no percentage or
absolute target exists yet. If targets arrive later they slot into the
existing target machinery on the objective's scale.
- **A budget is mandatory** for the goal-aligned goals: unconstrained
"as much as possible" would recommend every beneficial measure. A
budget-less Energy/CO2 Scenario raises a `ValueError` naming the scenario
and goal — a loud misconfiguration, not a maximal plan.
- **One currency everywhere**: the role-1 group signals
(`independent_option_signals`), the forced Measure Dependency pricing, the
greedy-repair marginals, and Fabric First's phase-2 re-scoring
(ADR-0061) are all measured by the same objective, so a ventilation that
costs SAP but is carbon-neutral cannot sink a carbon-improving wall, and a
fabric-first phase 2 picks its heating on post-fabric carbon, not
post-fabric SAP.
## Consequences
- Selection changes; truth-telling does not. The Plan's persisted Scores,
Bills, and role-3 SAP attribution are computed exactly as before — only
*which* package is chosen responds to the goal.
- At a £16,000 budget on the uninsulated solid-brick corpus dwelling
(001431), the SAP objective buys wall + floor + gas boiler (SAP 72.9,
2,069 kg CO2/yr, £2,088/yr) while the carbon objective buys wall + floor +
storage heaters (SAP 69.2, 1,098 kg CO2/yr, £2,635/yr) — goals now trade
SAP, carbon and bills against each other visibly.
- The Energy Savings objective inherits the Fuel Rates snapshot's staleness
characteristics (quarterly Ofgem-cap cadence, ADR-0014).
- `independent_option_impacts` (role-1 SAP/CO2/kWh triple) is removed —
superseded by `independent_option_signals` in the objective's currency.

22
domain/modelling/measure_type.py
View file

@ -13,6 +13,7 @@ change. It is also the vocabulary the ``considered_measures`` allowlist speaks
from __future__ import annotations
from enum import StrEnum
from typing import Final
class MeasureType(StrEnum):
@ -37,3 +38,24 @@ class MeasureType(StrEnum):
SYSTEM_TUNE_UP_ZONED = "system_tune_up_zoned"
SOLAR_PV = "solar_pv"
SECONDARY_HEATING_REMOVAL = "secondary_heating_removal"
# The measure types a Fabric First Scenario treats in phase 1 — the building
# envelope: wall / roof / floor insulation and glazing. Everything else
# (heating, solar, lighting, tune-ups, secondary-heating removal) waits for
# phase 2. Mechanical ventilation is deliberately absent: it is never selected,
# only injected as a forced Measure Dependency of the fabric that triggers it.
FABRIC_MEASURE_TYPES: Final[frozenset[MeasureType]] = frozenset(
{
MeasureType.CAVITY_WALL_INSULATION,
MeasureType.EXTERNAL_WALL_INSULATION,
MeasureType.INTERNAL_WALL_INSULATION,
MeasureType.LOFT_INSULATION,
MeasureType.SLOPING_CEILING_INSULATION,
MeasureType.FLAT_ROOF_INSULATION,
MeasureType.SUSPENDED_FLOOR_INSULATION,
MeasureType.SOLID_FLOOR_INSULATION,
MeasureType.DOUBLE_GLAZING,
MeasureType.SECONDARY_GLAZING,
}
)

235
domain/modelling/optimisation/optimiser.py
View file

@ -21,10 +21,10 @@ from __future__ import annotations
import itertools
from dataclasses import dataclass
from typing import Optional, Protocol, Sequence
from typing import Callable, Optional, Protocol, Sequence
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.modelling.measure_type import MeasureType
from domain.modelling.measure_type import FABRIC_MEASURE_TYPES, MeasureType
from domain.modelling.scoring.package_scorer import Score
from domain.modelling.recommendation import MeasureOption
from domain.modelling.simulation import EpcSimulation
@ -33,8 +33,9 @@ from domain.modelling.simulation import EpcSimulation
@dataclass(frozen=True)
class ScoredOption:
"""A candidate Measure Option paired with its role-1 (independent-vs-
baseline) SAP gain — the optimiser's input signal. Cost is read from the
Option; the gain is supplied by scoring."""
baseline) gain in the goal objective's currency — SAP points by default,
kg CO2 / £ saved for the goal-aligned Scenarios (ADR-0062). Cost is read
from the Option; the gain is supplied by scoring."""
option: MeasureOption
sap_gain: float
@ -171,6 +172,12 @@ class OptimisedPackage:
score: Score
def sap_rating(score: Score) -> float:
"""The default Optimiser objective: the un-rounded SAP rating (higher is
better) — what every goal optimised before goal-aligned objectives."""
return score.sap_continuous
def optimise_package(
*,
groups: list[list[ScoredOption]],
@ -179,6 +186,7 @@ def optimise_package(
budget: Optional[float],
target_sap: Optional[float],
dependencies: Sequence[MeasureDependency] = (),
objective: Callable[[Score], float] = sap_rating,
) -> OptimisedPackage:
"""Select the Optimised Package for one Property + Scenario (ADR-0016 +
its amendment).
@ -197,48 +205,218 @@ def optimise_package(
Without a ``target_sap`` (other goals) it is max-gain-within-budget. Either
way forced dependencies are injected on every path and their cost counts
toward the spend; the returned `selected` includes them. ``budget`` of None
means unconstrained."""
baseline_sap: float = _score(scorer, baseline_epc, []).sap_continuous
means unconstrained.
``objective`` is the currency every internally-computed figure is measured
in (ADR-0062): the goal's metric, higher is better — SAP by default, CO2
reduction / bill saving for the goal-aligned Scenarios. The caller must
supply the group signals in the same currency; ``target_sap`` (when given)
is a value on the same scale."""
baseline_value: float = objective(_score(scorer, baseline_epc, []))
# Score each forced dependency's independent (role-1) impact so the selection
# can price the ventilation a wall drags in — negative for ventilation.
deps: list[MeasureDependency] = _with_role1_signals(
dependencies, scorer, baseline_epc, baseline_sap
dependencies, scorer, baseline_epc, baseline_value, objective
)
if target_sap is None:
return _max_gain_package(groups, scorer, baseline_epc, budget, deps)
target_gain: float = target_sap - baseline_sap
target_gain: float = target_sap - baseline_value
chosen: Optional[list[ScoredOption]] = optimise_min_cost(
groups, budget, target_gain, deps
)
if chosen is not None:
package: OptimisedPackage = _repair_to_target(
chosen, groups, deps, scorer, baseline_epc, budget, target_sap
chosen, groups, deps, scorer, baseline_epc, budget, target_sap, objective
)
if package.score.sap_continuous >= target_sap:
if objective(package.score) >= target_sap:
return package
# Target unreachable within budget (warm-start infeasible, or the repaired
# package still falls short) → best effort: the most improvement budget buys.
return _max_gain_package(groups, scorer, baseline_epc, budget, deps)
def optimise_package_fabric_first(
*,
groups: list[list[ScoredOption]],
scorer: Scorer,
baseline_epc: EpcPropertyData,
budget: Optional[float],
target_sap: Optional[float],
dependencies: Sequence[MeasureDependency] = (),
objective: Callable[[Score], float] = sap_rating,
) -> OptimisedPackage:
"""Select the Optimised Package under the Fabric First constraint: optimise
the fabric measures (``FABRIC_MEASURE_TYPES``) first with the full budget;
if the truthful post-fabric score meets ``target_sap``, stop there. Otherwise
optimise the remaining groups on top — the starting point for phase 2 is the
dwelling with the phase-1 fabric applied — within the leftover budget."""
fabric_groups: list[list[ScoredOption]] = [
group for group in groups if _is_fabric_group(group)
]
if not fabric_groups:
# Nothing for phase 1 to claim (the envelope is already treated):
# a plain run is identical and skips the phase-2 re-scoring.
return optimise_package(
groups=groups,
scorer=scorer,
baseline_epc=baseline_epc,
budget=budget,
target_sap=target_sap,
dependencies=dependencies,
)
fabric_package: OptimisedPackage = optimise_package(
groups=fabric_groups,
scorer=scorer,
baseline_epc=baseline_epc,
budget=budget,
target_sap=target_sap,
dependencies=dependencies,
objective=objective,
)
if (
target_sap is not None
and objective(fabric_package.score) >= target_sap
):
return fabric_package
if not fabric_package.selected:
# Phase 1 committed nothing (no fabric affordable or worth having), so
# the phase-2 prefix would be empty and its re-scoring would reproduce
# the signals the groups already carry: a plain run is identical.
return optimise_package(
groups=groups,
scorer=scorer,
baseline_epc=baseline_epc,
budget=budget,
target_sap=target_sap,
dependencies=dependencies,
)
# Phase 2 — the upgrade requirement is not met by fabric alone: optimise
# the remaining groups (non-fabric, plus any fabric group phase 1 left
# unpicked) on top of the committed fabric. Every score call is prefixed
# with the phase-1 overlays, so candidates are valued against the
# fabric-applied dwelling and the resulting package score stays the
# truthful whole-package figure against the original baseline.
consumed: set[int] = _used_group_indices(groups, fabric_package.selected)
remaining_groups: list[list[ScoredOption]] = [
group for index, group in enumerate(groups) if index not in consumed
]
if not remaining_groups:
return fabric_package
post_fabric_scorer = _PrefixedScorer(
scorer, [scored.option.overlay for scored in fabric_package.selected]
)
leftover_budget: Optional[float] = (
None if budget is None else budget - _package_cost(fabric_package.selected)
)
# Everything phase 1 committed — its picks plus the dependencies it
# injected. A dependency already injected (e.g. the wall's ventilation) is
# satisfied for the whole package: phase 2 must not force it in again.
phase_one_types: set[MeasureType] = {
scored.option.measure_type for scored in fabric_package.selected
}
outstanding_dependencies: list[MeasureDependency] = [
dependency
for dependency in dependencies
if dependency.required.option.measure_type not in phase_one_types
]
top_up: OptimisedPackage = optimise_package(
groups=_rescored_groups(
remaining_groups,
post_fabric_scorer,
baseline_epc,
objective=objective,
start_value=objective(fabric_package.score),
),
scorer=post_fabric_scorer,
baseline_epc=baseline_epc,
budget=leftover_budget,
target_sap=target_sap,
dependencies=outstanding_dependencies,
objective=objective,
)
return OptimisedPackage(
selected=[*fabric_package.selected, *top_up.selected],
score=top_up.score,
)
def _is_fabric_group(group: list[ScoredOption]) -> bool:
"""A group belongs to phase 1 when every Option in it is a fabric measure
(groups are one Recommendation each, so they are homogeneous in kind)."""
return all(
scored.option.measure_type in FABRIC_MEASURE_TYPES for scored in group
)
def _rescored_groups(
groups: list[list[ScoredOption]],
scorer: Scorer,
baseline_epc: EpcPropertyData,
*,
objective: Callable[[Score], float],
start_value: float,
) -> list[list[ScoredOption]]:
"""The groups with every Option's role-1 warm-start signal re-scored
through ``scorer`` in the ``objective``'s currency — for phase 2, its
independent gain on the post-fabric dwelling rather than the raw baseline,
so options whose worth changes once the envelope is treated (a boiler on
an insulated home) are re-ranked. ``start_value`` is the objective value of
``baseline_epc`` through ``scorer`` with no candidate applied — the caller
already has it (the phase-1 package score in the objective's currency), so
it is threaded in rather than re-computed."""
return [
[
ScoredOption(
option=scored.option,
sap_gain=objective(
scorer.score(baseline_epc, [scored.option.overlay])
)
- start_value,
)
for scored in group
]
for group in groups
]
class _PrefixedScorer:
"""A Scorer view of the dwelling with a committed package already applied:
every score call sees the ``prefix`` overlays before the candidate's own.
Phase 2 of Fabric First scores through this, so its candidates are valued
against the post-fabric dwelling while the returned Score remains the
truthful whole-package figure against the true baseline."""
def __init__(self, inner: Scorer, prefix: Sequence[EpcSimulation]) -> None:
self._inner = inner
self._prefix = list(prefix)
def score(
self, baseline: EpcPropertyData, simulations: Sequence[EpcSimulation]
) -> Score:
return self._inner.score(baseline, [*self._prefix, *simulations])
def _with_role1_signals(
dependencies: Sequence[MeasureDependency],
scorer: Scorer,
baseline_epc: EpcPropertyData,
baseline_sap: float,
baseline_value: float,
objective: Callable[[Score], float],
) -> list[MeasureDependency]:
"""Replace each dependency's placeholder role-1 signal with its true
independent-vs-baseline SAP impact, so the selectors price what the
dependency really does to the package (ADR-0016 amendment)."""
independent-vs-baseline impact **in the objective's currency**, so the
selectors price what the dependency really does to the package (ADR-0016
amendment; ADR-0062 for the currency)."""
scored: list[MeasureDependency] = []
for dependency in dependencies:
signal: float = (
scorer.score(
baseline_epc, [dependency.required.option.overlay]
).sap_continuous
- baseline_sap
objective(
scorer.score(baseline_epc, [dependency.required.option.overlay])
)
- baseline_value
)
scored.append(
MeasureDependency(
@ -276,16 +454,17 @@ def _repair_to_target(
baseline_epc: EpcPropertyData,
budget: Optional[float],
target_sap: float,
objective: Callable[[Score], float],
) -> OptimisedPackage:
"""Inject dependencies onto the warm-start, re-score for the truth, then
greedy-add the untreated-group Option with the best marginal SAP-per-£ (its
own dependency folded in) until the true SAP clears ``target_sap`` or no
affordable improving Option remains."""
greedy-add the untreated-group Option with the best marginal objective-per-£
(its own dependency folded in) until the true objective value clears
``target_sap`` or no affordable improving Option remains."""
selected: list[ScoredOption] = _inject(chosen, dependencies)
score: Score = _score(scorer, baseline_epc, selected)
while score.sap_continuous < target_sap:
while objective(score) < target_sap:
candidate = _best_repair_candidate(
groups, chosen, dependencies, scorer, baseline_epc, score, budget
groups, chosen, dependencies, scorer, baseline_epc, score, budget, objective
)
if candidate is None:
break
@ -343,14 +522,16 @@ def _best_repair_candidate(
baseline_epc: EpcPropertyData,
current: Score,
budget: Optional[float],
objective: Callable[[Score], float],
) -> Optional[ScoredOption]:
"""The untreated-group Option giving the best **marginal** SAP-per-£ when
added to the current package — re-scored (not the role-1 signal) with any
ventilation dependency it newly triggers folded in, so both its SAP and its
incremental cost are truthful. Affordable when the resulting whole-package
"""The untreated-group Option giving the best **marginal** objective-per-£
when added to the current package — re-scored (not the role-1 signal) with
any ventilation dependency it newly triggers folded in, so both its gain and
its incremental cost are truthful. Affordable when the resulting whole-package
cost is within ``budget`` and strictly improving. None if there is none."""
used: set[int] = _used_group_indices(groups, chosen)
base_cost: float = _package_cost(_inject(chosen, dependencies))
current_value: float = objective(current)
best: Optional[ScoredOption] = None
best_ratio: float = 0.0
for index, group in enumerate(groups):
@ -364,7 +545,7 @@ def _best_repair_candidate(
if budget is not None and package_cost > budget:
continue
trial: Score = _score(scorer, baseline_epc, trial_selected)
marginal: float = trial.sap_continuous - current.sap_continuous
marginal: float = objective(trial) - current_value
if marginal <= 0.0:
continue
incremental: float = package_cost - base_cost

8
domain/modelling/scenario.py
View file

@ -26,7 +26,12 @@ class Scenario:
`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."""
inclusions column). Empty means nothing is barred.
`fabric_first` constrains the Optimiser to treat the building envelope
first: fabric measures are optimised with the full budget, and heating /
renewables are only considered on top of the committed fabric when the
fabric alone does not meet the brief's target."""
id: int
goal: str
@ -34,6 +39,7 @@ class Scenario:
budget: Optional[float]
is_default: bool
exclusions: frozenset[MeasureType] = _NO_EXCLUSIONS
fabric_first: bool = False
def considered_measures(self) -> Optional[frozenset[MeasureType]]:
"""The measure-type allowlist the Scenario's exclusions imply: every

43
domain/modelling/scoring/scoring.py
View file

@ -15,7 +15,7 @@ truthful. The whole-package re-score (role 2) is `PackageScorer.score` directly.
"""
from dataclasses import dataclass
from typing import Sequence
from typing import Callable, Optional, Sequence
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.modelling.scoring.package_scorer import PackageScorer, Score
@ -83,33 +83,28 @@ def marginal_impacts(
return marginals_from_scores(cascade_scores(scorer, baseline, overlays))
def independent_option_impacts(
def independent_option_signals(
scorer: PackageScorer,
baseline: EpcPropertyData,
options: Sequence[MeasureOption],
) -> list[MeasureImpact]:
"""Score each Option's overlay independently against the baseline (role 1 —
the optimiser's approximate input signal). Each *distinct* Simulation Overlay
is scored once (Options sharing an overlay reuse the result), so the baseline
is scored once plus one score per distinct overlay. Results follow the input
order. These figures are an approximate signal — never surface them as a
measure's true impact."""
base: Score = scorer.score(baseline, [])
scored: list[tuple[EpcSimulation, MeasureImpact]] = []
impacts: list[MeasureImpact] = []
objective: Callable[[Score], float],
) -> list[float]:
"""Each Option's independent-vs-baseline gain **in the objective's
currency** (role 1 — the optimiser's approximate input signal, ADR-0062):
SAP points for an Increasing-EPC goal, kg CO2 saved for Reducing CO2, £
saved for Energy Savings. Each distinct Simulation Overlay is scored once
(Options sharing an overlay reuse the result); results follow the input
order."""
base_value: float = objective(scorer.score(baseline, []))
scored: list[tuple[EpcSimulation, float]] = []
signals: list[float] = []
for option in options:
cached = next(
(impact for overlay, impact in scored if overlay == option.overlay), None
cached: Optional[float] = next(
(signal for overlay, signal in scored if overlay == option.overlay),
None,
)
if cached is None:
current: Score = scorer.score(baseline, [option.overlay])
cached = MeasureImpact(
sap_points=current.sap_continuous - base.sap_continuous,
co2_savings_kg_per_yr=base.co2_kg_per_yr - current.co2_kg_per_yr,
energy_savings_kwh_per_yr=(
base.primary_energy_kwh_per_yr - current.primary_energy_kwh_per_yr
),
)
cached = objective(scorer.score(baseline, [option.overlay])) - base_value
scored.append((option.overlay, cached))
impacts.append(cached)
return impacts
signals.append(cached)
return signals

4
harness/console.py
View file

@ -39,6 +39,7 @@ from orchestration.modelling_orchestrator import (
_candidate_recommendations, # pyright: ignore[reportPrivateUsage]
)
from orchestration.property_baseline_orchestrator import PropertyBaselineOrchestrator
from repositories.fuel_rates.fuel_rates_repository import FuelRatesRepository
from repositories.fuel_rates.fuel_rates_static_file_repository import (
FuelRatesStaticFileRepository,
)
@ -182,6 +183,7 @@ def run_modelling(
considered_measures: Optional[frozenset[MeasureType]] = None,
products: Optional[ProductRepository] = None,
scenario: Optional[Scenario] = None,
fuel_rates: Optional[FuelRatesRepository] = None,
print_table: bool = True,
) -> Plan:
"""Run ONLY the Modelling stage over ``epc`` with no database — skipping
@ -240,7 +242,7 @@ def run_modelling(
ModellingOrchestrator(
unit_of_work=lambda: unit,
calculator=Sap10Calculator(),
fuel_rates=FuelRatesStaticFileRepository(),
fuel_rates=fuel_rates or FuelRatesStaticFileRepository(),
).run(
property_ids=[_PROPERTY_ID],
scenario_ids=[scenario_id],

4
infrastructure/postgres/modelling/scenario_table.py
View file

@ -78,6 +78,9 @@ class ScenarioModel(SQLModel, table=True):
exclusions: Optional[str] = Field(default=None)
multi_plan: bool = False
is_default: bool = False
# Fabric First constraint (owned by the FE Drizzle schema: boolean NOT
# NULL DEFAULT false — do not deploy this mirror before that migration).
fabric_first: bool = False
# Portfolio-level aggregates stored against the Scenario.
cost: Optional[float] = Field(default=None)
@ -115,4 +118,5 @@ class ScenarioModel(SQLModel, table=True):
budget=self.budget,
is_default=self.is_default,
exclusions=_parse_exclusions(self.exclusions),
fabric_first=self.fabric_first,
)

98
orchestration/modelling_orchestrator.py
View file

@ -19,16 +19,19 @@ from domain.modelling.optimisation.optimiser import (
OptimisedPackage,
ScoredOption,
optimise_package,
optimise_package_fabric_first,
sap_rating,
)
from domain.modelling.scoring.package_scorer import PackageScorer, Score
from domain.modelling.plan import Plan, PlanMeasure
from domain.modelling.recommendation import MeasureOption, Recommendation
from domain.modelling.generators.roof_recommendation import recommend_roof_insulation
from domain.modelling.portfolio_goal import PortfolioGoal
from domain.modelling.scenario import Scenario
from domain.modelling.scoring.scoring import (
MeasureImpact,
cascade_scores,
independent_option_impacts,
independent_option_signals,
marginals_from_scores,
)
from domain.modelling.generators.wall_recommendation import recommend_cavity_wall
@ -48,12 +51,6 @@ from repositories.product.product_repository import ProductRepository
from repositories.solar.solar_repository import SolarRepository
from repositories.unit_of_work import UnitOfWork
# The PortfolioGoal value that targets a SAP band (cf.
# backend.app.db.models.portfolio.PortfolioGoal.INCREASING_EPC). Other goals
# (Energy Savings, Reducing CO2 emissions) don't yet set a SAP repair target —
# the optimiser just maximises SAP gain within budget for them (later slice).
_INCREASING_EPC_GOAL: Final[str] = "Increasing EPC"
# Best-practice install sequence for the role-3 attribution cascade (ADR-0016):
# walls → roof → ventilation → floor, per the legacy `Recommendations` class.
# Ventilation sits after the fabric that triggers it so its (negative) marginal
@ -175,6 +172,11 @@ class ModellingOrchestrator:
considered: Optional[frozenset[MeasureType]] = combine_considered_measures(
scenario.considered_measures(), considered_measures
)
# The Optimiser speaks the goal's currency (ADR-0062): group signals,
# dependency pricing and repair marginals are all measured by this
# objective — SAP by default, carbon reduction for a Reducing-CO2 goal.
_require_budget_for_goal_aligned(scenario)
objective: Callable[[Score], float] = _objective_for(scenario, bill_derivation)
groups: list[list[ScoredOption]] = _scored_candidate_groups(
scorer,
effective_epc,
@ -182,19 +184,27 @@ class ModellingOrchestrator:
planning_restrictions,
solar_potential,
considered,
objective,
)
# 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
)
package: OptimisedPackage = optimise_package(
# A Fabric First brief optimises the envelope with the full budget
# before heating / renewables are considered on top (mirroring the
# legacy engine's enforce_fabric_first).
optimise = (
optimise_package_fabric_first if scenario.fabric_first else optimise_package
)
package: OptimisedPackage = optimise(
groups=groups,
scorer=scorer,
baseline_epc=effective_epc,
budget=scenario.budget,
target_sap=_target_sap(scenario),
dependencies=dependencies,
objective=objective,
)
# Role-3 attribution: re-apply the *selected* set in best-practice order
@ -388,9 +398,11 @@ def _scored_candidate_groups(
planning_restrictions: PlanningRestrictions,
solar_potential: Optional[SolarPotential],
considered_measures: Optional[frozenset[MeasureType]],
objective: Callable[[Score], float],
) -> list[list[ScoredOption]]:
"""One group per Recommendation: each Option scored independently against
the baseline (role-1 warm-start signal, ADR-0016)."""
the baseline (role-1 warm-start signal, ADR-0016), in the goal objective's
currency (ADR-0062)."""
# The SAP design heat loss sizes the ASHP to the dwelling (ADR-0049); read it
# off a baseline score, which the group scoring computes anyway.
baseline_result = scorer.score(effective_epc, []).sap_result
@ -407,22 +419,80 @@ def _scored_candidate_groups(
design_heat_loss_kw,
):
options = list(recommendation.options)
impacts: list[MeasureImpact] = independent_option_impacts(
scorer, effective_epc, options
signals: list[float] = independent_option_signals(
scorer, effective_epc, options, objective
)
groups.append(
[
ScoredOption(option=option, sap_gain=impact.sap_points)
for option, impact in zip(options, impacts, strict=True)
ScoredOption(option=option, sap_gain=signal)
for option, signal in zip(options, signals, strict=True)
]
)
return groups
def _carbon_reduction(score: Score) -> float:
"""The Reducing-CO2 objective: annual kg CO2 below zero-point, negated so
higher is better (a saved kg scores +1)."""
return -score.co2_kg_per_yr
def _bill_saving(bill_derivation: BillDerivation) -> Callable[[Score], float]:
"""The Energy-Savings objective: the annual Bill at the current Fuel Rates
snapshot, negated so higher is better (a saved £ scores +1). Priced at the
live snapshot, not SAP's internal tariff book — that difference is the
point of the goal (ADR-0062)."""
def objective(score: Score) -> float:
return -_bill_for(bill_derivation, score).total_gbp
return objective
# The goal-aligned goals (ADR-0062): each maximises its own metric within the
# Scenario budget and sets no SAP target. One table is the single source of
# "which goals are goal-aligned" — both the objective dispatch and the
# budget-required guard read it, so a new goal-aligned goal cannot be added to
# one without the other. Each entry builds its objective from the plan's
# BillDerivation (the carbon objective ignores it; the bill objective needs it).
# A goal absent from the table optimises SAP, as every goal did before.
_GOAL_ALIGNED_OBJECTIVES: Final[
dict[str, Callable[[BillDerivation], Callable[[Score], float]]]
] = {
PortfolioGoal.REDUCING_CO2_EMISSIONS.value: lambda _bill_derivation: (
_carbon_reduction
),
PortfolioGoal.ENERGY_SAVINGS.value: _bill_saving,
}
def _require_budget_for_goal_aligned(scenario: Scenario) -> None:
"""A goal-aligned Scenario is 'reduce as much as possible within this
budget' — undefined without one (unconstrained, it would recommend every
beneficial measure). Fail the misconfiguration loudly (ADR-0062)."""
if scenario.budget is None and scenario.goal in _GOAL_ALIGNED_OBJECTIVES:
raise ValueError(
f"scenario {scenario.id} has goal {scenario.goal!r} but no budget; "
"goal-aligned scenarios require a budget"
)
def _objective_for(
scenario: Scenario, bill_derivation: BillDerivation
) -> Callable[[Score], float]:
"""The metric the Scenario's goal maximises (ADR-0062), as an Optimiser
objective (higher is better). Goals without an aligned metric optimise
SAP, as every goal did before."""
build_objective = _GOAL_ALIGNED_OBJECTIVES.get(scenario.goal)
if build_objective is None:
return sap_rating
return build_objective(bill_derivation)
def _target_sap(scenario: Scenario) -> Optional[float]:
"""The SAP rating the Optimiser repairs toward — the floor of the goal
band for an INCREASING_EPC goal, else None (no SAP target)."""
if scenario.goal != _INCREASING_EPC_GOAL:
if scenario.goal != PortfolioGoal.INCREASING_EPC.value:
return None
return float(Epc(scenario.goal_value).sap_lower_bound())

144
tests/domain/modelling/_optimiser_fixtures.py Normal file
View file

@ -0,0 +1,144 @@
"""Shared fixtures for the Optimiser test files: distinguishable Simulation
Overlays (so a stub scorer can attribute a true gain per measure kind), the
ScoredOption builder, the additive per-kind stub scorer, and the forced
ventilation Measure Dependency edge.
Fixture values stay domain-plausible: overlay heating codes are real SAP
Table 4a codes (104 = mains-gas combi boiler, heat pumps carry a PCDF index),
and tests price measures at realistic magnitudes (a CWI around £1,000, an
ASHP around £8,000)."""
from __future__ import annotations
from typing import Optional, Sequence
from datatypes.epc.domain.epc_property_data import (
BuildingPartIdentifier,
EpcPropertyData,
)
from domain.modelling.measure_type import MeasureType
from domain.modelling.optimisation.optimiser import MeasureDependency, ScoredOption
from domain.modelling.recommendation import Cost, MeasureOption
from domain.modelling.scoring.package_scorer import Score
from domain.modelling.simulation import (
BuildingPartOverlay,
EpcSimulation,
GlazingOverlay,
HeatingOverlay,
VentilationOverlay,
)
WALL_OVERLAY = EpcSimulation(
building_parts={
BuildingPartIdentifier.MAIN: BuildingPartOverlay(wall_insulation_type=2)
}
)
ROOF_OVERLAY = EpcSimulation(
building_parts={
BuildingPartIdentifier.MAIN: BuildingPartOverlay(roof_insulation_thickness=300)
}
)
FLOOR_OVERLAY = EpcSimulation(
building_parts={
BuildingPartIdentifier.MAIN: BuildingPartOverlay(floor_insulation_thickness=100)
}
)
GLAZING_OVERLAY = EpcSimulation(glazing=GlazingOverlay(glazing_type=2))
VENT_OVERLAY = EpcSimulation(
ventilation=VentilationOverlay(mechanical_ventilation_kind="EXTRACT_OR_PIV_OUTSIDE")
)
# SAP Table 4a code 104: a mains-gas combi boiler.
BOILER_OVERLAY = EpcSimulation(heating=HeatingOverlay(sap_main_heating_code=104))
# Heat pumps are expressed as a PCDF product index, as the generator emits them.
ASHP_OVERLAY = EpcSimulation(heating=HeatingOverlay(main_heating_index_number=13000))
_WALL_TRIGGERS: frozenset[MeasureType] = frozenset(
{MeasureType.CAVITY_WALL_INSULATION, MeasureType.EXTERNAL_WALL_INSULATION}
)
def scored_option(
measure_type: str,
*,
gain: float,
cost: float,
overlay: Optional[EpcSimulation] = None,
) -> ScoredOption:
"""A one-Option fixture: ``gain`` is the role-1 warm-start signal, ``cost``
the total install cost. Omit ``overlay`` where the test never re-scores."""
return ScoredOption(
option=MeasureOption(
measure_type=MeasureType(measure_type),
description=measure_type,
overlay=overlay if overlay is not None else EpcSimulation(),
cost=Cost(total=cost, contingency_rate=0.0),
),
sap_gain=gain,
)
class StubScorer:
"""A deterministic stand-in for PackageScorer: the package SAP is a base
plus a fixed *true* gain per measure kind present (detected by overlay
field), decoupled from the role-1 signal — so selection, repair and the
two-phase split are exercised without the calculator. Kinds a test does
not use default to 0."""
def __init__(
self,
*,
base: float,
wall: float = 0.0,
roof: float = 0.0,
floor: float = 0.0,
heating: float = 0.0,
vent: float = 0.0,
) -> None:
self._base = base
self._wall = wall
self._roof = roof
self._floor = floor
self._heating = heating
self._vent = vent
def score(
self, baseline: EpcPropertyData, simulations: Sequence[EpcSimulation]
) -> Score:
sap = self._base
for sim in simulations:
if sim.heating is not None:
sap += self._heating
if sim.ventilation is not None:
sap += self._vent
for part in sim.building_parts.values():
if part.wall_insulation_type is not None:
sap += self._wall
if part.roof_insulation_thickness is not None:
sap += self._roof
if part.floor_insulation_thickness is not None:
sap += self._floor
return Score(
sap_continuous=sap, co2_kg_per_yr=0.0, primary_energy_kwh_per_yr=0.0
)
def selected_types(selection: Sequence[ScoredOption]) -> set[str]:
return {scored.option.measure_type for scored in selection}
def ventilation_dependency(
*, cost: float, triggers: frozenset[MeasureType] = _WALL_TRIGGERS
) -> MeasureDependency:
"""A forced 'airtightness requires ventilation' edge for the tests."""
return MeasureDependency(
triggers=triggers,
required=ScoredOption(
option=MeasureOption(
measure_type=MeasureType.MECHANICAL_VENTILATION,
description="mechanical_ventilation",
overlay=VENT_OVERLAY,
cost=Cost(total=cost, contingency_rate=0.0),
),
sap_gain=0.0, # placeholder; optimise_package scores the real signal
),
)

315
tests/domain/modelling/test_optimiser.py
View file

@ -8,14 +8,7 @@ selection with synthetic scores and no calculator.
from __future__ import annotations
from typing import Sequence
from datatypes.epc.domain.epc_property_data import (
BuildingPartIdentifier,
EpcPropertyData,
)
from domain.modelling.optimisation.optimiser import (
MeasureDependency,
OptimisedPackage,
ScoredOption,
optimise,
@ -23,104 +16,30 @@ from domain.modelling.optimisation.optimiser import (
optimise_package,
)
from domain.modelling.measure_type import MeasureType
from domain.modelling.scoring.package_scorer import Score
from domain.modelling.recommendation import Cost, MeasureOption
from domain.modelling.simulation import (
BuildingPartOverlay,
EpcSimulation,
VentilationOverlay,
from tests.domain.modelling._optimiser_fixtures import (
FLOOR_OVERLAY,
ROOF_OVERLAY,
WALL_OVERLAY,
StubScorer,
scored_option,
selected_types,
ventilation_dependency,
)
from tests.domain.sap10_calculator.worksheet._elmhurst_worksheet_000490 import (
build_epc,
)
def _scored(measure_type: str, *, gain: float, cost: float) -> ScoredOption:
return ScoredOption(
option=MeasureOption(
measure_type=MeasureType(measure_type),
description=measure_type,
overlay=EpcSimulation(),
cost=Cost(total=cost, contingency_rate=0.0),
),
sap_gain=gain,
)
# Distinguishable overlays so the stub scorer can attribute a true gain per
# measure (wall / roof / floor) regardless of the role-1 signal.
_WALL_OVERLAY = EpcSimulation(
building_parts={
BuildingPartIdentifier.MAIN: BuildingPartOverlay(wall_insulation_type=2)
}
)
_ROOF_OVERLAY = EpcSimulation(
building_parts={
BuildingPartIdentifier.MAIN: BuildingPartOverlay(roof_insulation_thickness=300)
}
)
_FLOOR_OVERLAY = EpcSimulation(
building_parts={
BuildingPartIdentifier.MAIN: BuildingPartOverlay(floor_insulation_thickness=100)
}
)
def _scored_overlay(
measure_type: str, *, gain: float, cost: float, overlay: EpcSimulation
) -> ScoredOption:
return ScoredOption(
option=MeasureOption(
measure_type=MeasureType(measure_type),
description=measure_type,
overlay=overlay,
cost=Cost(total=cost, contingency_rate=0.0),
),
sap_gain=gain,
)
class _StubScorer:
"""A deterministic stand-in for PackageScorer: the package SAP is a base
plus a fixed *true* gain per measure present (by overlay field), decoupled
from the role-1 signal — so the repair loop is exercised without the
calculator (ADR-0016)."""
def __init__(self, *, base: float, wall: float, roof: float, floor: float) -> None:
self._base = base
self._wall = wall
self._roof = roof
self._floor = floor
def score(
self, baseline: EpcPropertyData, simulations: Sequence[EpcSimulation]
) -> Score:
sap = self._base
for sim in simulations:
part = sim.building_parts[BuildingPartIdentifier.MAIN]
if part.wall_insulation_type is not None:
sap += self._wall
if part.roof_insulation_thickness is not None:
sap += self._roof
if part.floor_insulation_thickness is not None:
sap += self._floor
return Score(sap_continuous=sap, co2_kg_per_yr=0.0, primary_energy_kwh_per_yr=0.0)
def _selected_types(selection: list[ScoredOption]) -> set[str]:
return {scored.option.measure_type for scored in selection}
def test_grouped_knapsack_maximises_gain_within_budget() -> None:
# Arrange — wall group has two mutually-exclusive options; roof + floor one
# each. EWI has the best gain but is unaffordable alongside the rest.
groups: list[list[ScoredOption]] = [
[
_scored("external_wall_insulation", gain=10.0, cost=8000.0),
_scored("cavity_wall_insulation", gain=6.0, cost=1000.0),
scored_option("external_wall_insulation", gain=10.0, cost=8000.0),
scored_option("cavity_wall_insulation", gain=6.0, cost=1000.0),
],
[_scored("loft_insulation", gain=4.0, cost=1500.0)],
[_scored("suspended_floor_insulation", gain=3.0, cost=2000.0)],
[scored_option("loft_insulation", gain=4.0, cost=1500.0)],
[scored_option("suspended_floor_insulation", gain=3.0, cost=2000.0)],
]
# Act
@ -128,7 +47,7 @@ def test_grouped_knapsack_maximises_gain_within_budget() -> None:
# Assert — cavity + loft + floor (cost 4500, gain 13) beats any package
# containing the 8000 EWI option within the 5000 budget.
assert _selected_types(selection) == {
assert selected_types(selection) == {
"cavity_wall_insulation",
"loft_insulation",
"suspended_floor_insulation",
@ -139,8 +58,8 @@ def test_picks_at_most_one_option_per_group() -> None:
# Arrange — both wall options are individually affordable.
groups: list[list[ScoredOption]] = [
[
_scored("external_wall_insulation", gain=10.0, cost=2000.0),
_scored("cavity_wall_insulation", gain=6.0, cost=1000.0),
scored_option("external_wall_insulation", gain=10.0, cost=2000.0),
scored_option("cavity_wall_insulation", gain=6.0, cost=1000.0),
],
]
@ -149,24 +68,24 @@ def test_picks_at_most_one_option_per_group() -> None:
# Assert — never both treatments of the same wall; the higher-gain one wins.
assert len(selection) == 1
assert _selected_types(selection) == {"external_wall_insulation"}
assert selected_types(selection) == {"external_wall_insulation"}
def test_no_budget_picks_the_best_option_in_every_group() -> None:
# Arrange
groups: list[list[ScoredOption]] = [
[
_scored("external_wall_insulation", gain=10.0, cost=8000.0),
_scored("cavity_wall_insulation", gain=6.0, cost=1000.0),
scored_option("external_wall_insulation", gain=10.0, cost=8000.0),
scored_option("cavity_wall_insulation", gain=6.0, cost=1000.0),
],
[_scored("loft_insulation", gain=4.0, cost=1500.0)],
[scored_option("loft_insulation", gain=4.0, cost=1500.0)],
]
# Act — None budget = unconstrained.
selection: list[ScoredOption] = optimise(groups, budget=None)
# Assert
assert _selected_types(selection) == {
assert selected_types(selection) == {
"external_wall_insulation",
"loft_insulation",
}
@ -175,8 +94,8 @@ def test_no_budget_picks_the_best_option_in_every_group() -> None:
def test_budget_too_small_for_any_option_selects_nothing() -> None:
# Arrange
groups: list[list[ScoredOption]] = [
[_scored("cavity_wall_insulation", gain=6.0, cost=1000.0)],
[_scored("loft_insulation", gain=4.0, cost=1500.0)],
[scored_option("cavity_wall_insulation", gain=6.0, cost=1000.0)],
[scored_option("loft_insulation", gain=4.0, cost=1500.0)],
]
# Act
@ -194,15 +113,15 @@ def test_no_groups_selects_nothing() -> None:
def test_within_budget_partial_selection_prefers_the_higher_gain_option() -> None:
# Arrange — only one of the two fits the budget; pick the affordable best.
groups: list[list[ScoredOption]] = [
[_scored("external_wall_insulation", gain=10.0, cost=8000.0)],
[_scored("loft_insulation", gain=4.0, cost=1500.0)],
[scored_option("external_wall_insulation", gain=10.0, cost=8000.0)],
[scored_option("loft_insulation", gain=4.0, cost=1500.0)],
]
# Act
selection: list[ScoredOption] = optimise(groups, budget=2000.0)
# Assert — EWI is unaffordable; loft alone is the best within £2000.
assert _selected_types(selection) == {"loft_insulation"}
assert selected_types(selection) == {"loft_insulation"}
# --- optimise_min_cost: least-cost-to-target selection (ADR-0016 amendment) ---
@ -212,8 +131,8 @@ def test_min_cost_picks_the_cheapest_package_that_reaches_the_target() -> None:
# Arrange — two packages both clear the target gain; one is cheaper.
groups: list[list[ScoredOption]] = [
[
_scored("loft_insulation", gain=10.0, cost=2000.0),
_scored("external_wall_insulation", gain=15.0, cost=3000.0),
scored_option("loft_insulation", gain=10.0, cost=2000.0),
scored_option("external_wall_insulation", gain=15.0, cost=3000.0),
],
]
@ -223,7 +142,7 @@ def test_min_cost_picks_the_cheapest_package_that_reaches_the_target() -> None:
# Assert — least-cost-to-target takes the +10 @ £2000, NOT the higher-gain
# +15 @ £3000 (no overshoot, surplus budget unspent).
assert selection is not None
assert _selected_types(selection) == {"loft_insulation"}
assert selected_types(selection) == {"loft_insulation"}
def test_min_cost_combines_groups_to_reach_the_target_at_least_cost() -> None:
@ -232,10 +151,10 @@ def test_min_cost_combines_groups_to_reach_the_target_at_least_cost() -> None:
# £8000).
groups: list[list[ScoredOption]] = [
[
_scored("cavity_wall_insulation", gain=6.0, cost=1000.0),
_scored("external_wall_insulation", gain=10.0, cost=8000.0),
scored_option("cavity_wall_insulation", gain=6.0, cost=1000.0),
scored_option("external_wall_insulation", gain=10.0, cost=8000.0),
],
[_scored("loft_insulation", gain=4.0, cost=1500.0)],
[scored_option("loft_insulation", gain=4.0, cost=1500.0)],
]
# Act
@ -243,7 +162,7 @@ def test_min_cost_combines_groups_to_reach_the_target_at_least_cost() -> None:
# Assert
assert selection is not None
assert _selected_types(selection) == {
assert selected_types(selection) == {
"cavity_wall_insulation",
"loft_insulation",
}
@ -254,8 +173,8 @@ def test_min_cost_breaks_cost_ties_toward_the_higher_gain() -> None:
# one with more headroom ("recommend more" on a tie).
groups: list[list[ScoredOption]] = [
[
_scored("cavity_wall_insulation", gain=10.0, cost=2000.0),
_scored("external_wall_insulation", gain=14.0, cost=2000.0),
scored_option("cavity_wall_insulation", gain=10.0, cost=2000.0),
scored_option("external_wall_insulation", gain=14.0, cost=2000.0),
],
]
@ -264,13 +183,13 @@ def test_min_cost_breaks_cost_ties_toward_the_higher_gain() -> None:
# Assert
assert selection is not None
assert _selected_types(selection) == {"external_wall_insulation"}
assert selected_types(selection) == {"external_wall_insulation"}
def test_min_cost_returns_none_when_target_unreachable_within_budget() -> None:
# Arrange — the only target-reaching package costs more than the budget.
groups: list[list[ScoredOption]] = [
[_scored("external_wall_insulation", gain=10.0, cost=8000.0)],
[scored_option("external_wall_insulation", gain=10.0, cost=8000.0)],
]
# Act
@ -283,8 +202,8 @@ def test_min_cost_returns_none_when_target_unreachable_within_budget() -> None:
def test_min_cost_returns_none_when_no_package_reaches_the_target() -> None:
# Arrange — even everything together falls short of the target gain.
groups: list[list[ScoredOption]] = [
[_scored("cavity_wall_insulation", gain=6.0, cost=1000.0)],
[_scored("loft_insulation", gain=3.0, cost=1500.0)],
[scored_option("cavity_wall_insulation", gain=6.0, cost=1000.0)],
[scored_option("loft_insulation", gain=3.0, cost=1500.0)],
]
# Act
@ -298,8 +217,8 @@ def test_min_cost_unbudgeted_picks_cheapest_reaching_target_not_everything() ->
# Arrange — no budget cap, but min-cost still means cheapest-to-target, not
# "install everything".
groups: list[list[ScoredOption]] = [
[_scored("cavity_wall_insulation", gain=10.0, cost=1000.0)],
[_scored("loft_insulation", gain=4.0, cost=1500.0)],
[scored_option("cavity_wall_insulation", gain=10.0, cost=1000.0)],
[scored_option("loft_insulation", gain=4.0, cost=1500.0)],
]
# Act — cavity alone (+10 @ £1000) already reaches the target.
@ -307,14 +226,14 @@ def test_min_cost_unbudgeted_picks_cheapest_reaching_target_not_everything() ->
# Assert — loft is left off; it would only add cost past the target.
assert selection is not None
assert _selected_types(selection) == {"cavity_wall_insulation"}
assert selected_types(selection) == {"cavity_wall_insulation"}
def test_min_cost_non_positive_target_selects_nothing() -> None:
# Arrange — a target already met (gain 0 needed) is reached by the empty
# package at zero cost.
groups: list[list[ScoredOption]] = [
[_scored("cavity_wall_insulation", gain=6.0, cost=1000.0)],
[scored_option("cavity_wall_insulation", gain=6.0, cost=1000.0)],
]
# Act
@ -328,11 +247,11 @@ def test_repair_adds_an_untreated_group_option_to_close_the_undershoot() -> None
# Arrange — role-1 under-counts roof (signal 0 → warm-start skips it), but
# its true re-scored gain (+4) is what closes the target.
groups: list[list[ScoredOption]] = [
[_scored_overlay("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=_WALL_OVERLAY)],
[_scored_overlay("loft_insulation", gain=0.0, cost=1000.0, overlay=_ROOF_OVERLAY)],
[_scored_overlay("suspended_floor_insulation", gain=8.0, cost=1000.0, overlay=_FLOOR_OVERLAY)],
[scored_option("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=WALL_OVERLAY)],
[scored_option("loft_insulation", gain=0.0, cost=1000.0, overlay=ROOF_OVERLAY)],
[scored_option("suspended_floor_insulation", gain=8.0, cost=1000.0, overlay=FLOOR_OVERLAY)],
]
scorer = _StubScorer(base=40.0, wall=5.0, roof=4.0, floor=3.0)
scorer = StubScorer(base=40.0, wall=5.0, roof=4.0, floor=3.0)
# Act
package: OptimisedPackage = optimise_package(
@ -358,9 +277,9 @@ def test_repair_adds_an_untreated_group_option_to_close_the_undershoot() -> None
def test_no_target_returns_the_warm_start_package_without_repair() -> None:
# Arrange
groups: list[list[ScoredOption]] = [
[_scored_overlay("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=_WALL_OVERLAY)],
[scored_option("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=WALL_OVERLAY)],
]
scorer = _StubScorer(base=40.0, wall=5.0, roof=4.0, floor=3.0)
scorer = StubScorer(base=40.0, wall=5.0, roof=4.0, floor=3.0)
# Act
package: OptimisedPackage = optimise_package(
@ -381,10 +300,10 @@ def test_no_target_returns_the_warm_start_package_without_repair() -> None:
def test_repair_stops_when_no_affordable_improving_option_remains() -> None:
# Arrange — the only untreated-group option costs more than the budget left.
groups: list[list[ScoredOption]] = [
[_scored_overlay("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=_WALL_OVERLAY)],
[_scored_overlay("loft_insulation", gain=0.0, cost=5000.0, overlay=_ROOF_OVERLAY)],
[scored_option("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=WALL_OVERLAY)],
[scored_option("loft_insulation", gain=0.0, cost=5000.0, overlay=ROOF_OVERLAY)],
]
scorer = _StubScorer(base=40.0, wall=5.0, roof=4.0, floor=3.0)
scorer = StubScorer(base=40.0, wall=5.0, roof=4.0, floor=3.0)
# Act
package: OptimisedPackage = optimise_package(
@ -410,11 +329,11 @@ def test_package_stops_at_the_target_and_does_not_overshoot() -> None:
# Arrange — wall alone already clears the target; max-gain would add roof +
# floor too. Least-cost-to-target must stop at the wall.
groups: list[list[ScoredOption]] = [
[_scored_overlay("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=_WALL_OVERLAY)],
[_scored_overlay("loft_insulation", gain=5.0, cost=1000.0, overlay=_ROOF_OVERLAY)],
[_scored_overlay("suspended_floor_insulation", gain=5.0, cost=1000.0, overlay=_FLOOR_OVERLAY)],
[scored_option("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=WALL_OVERLAY)],
[scored_option("loft_insulation", gain=5.0, cost=1000.0, overlay=ROOF_OVERLAY)],
[scored_option("suspended_floor_insulation", gain=5.0, cost=1000.0, overlay=FLOOR_OVERLAY)],
]
scorer = _StubScorer(base=60.0, wall=10.0, roof=5.0, floor=5.0)
scorer = StubScorer(base=60.0, wall=10.0, roof=5.0, floor=5.0)
# Act — target 69 (gain 9); wall (+10 → 70) reaches it for £1000.
package: OptimisedPackage = optimise_package(
@ -427,18 +346,18 @@ def test_package_stops_at_the_target_and_does_not_overshoot() -> None:
# Assert — just the wall; roof + floor (which would reach 80) are left off,
# surplus budget unspent.
assert _selected_types(package.selected) == {"cavity_wall_insulation"}
assert selected_types(package.selected) == {"cavity_wall_insulation"}
assert abs(package.score.sap_continuous - 70.0) <= 1e-9
def test_package_falls_back_to_max_gain_when_target_unreachable() -> None:
# Arrange — even all three measures (+20 → 80) cannot reach the target.
groups: list[list[ScoredOption]] = [
[_scored_overlay("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=_WALL_OVERLAY)],
[_scored_overlay("loft_insulation", gain=5.0, cost=1000.0, overlay=_ROOF_OVERLAY)],
[_scored_overlay("suspended_floor_insulation", gain=5.0, cost=1000.0, overlay=_FLOOR_OVERLAY)],
[scored_option("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=WALL_OVERLAY)],
[scored_option("loft_insulation", gain=5.0, cost=1000.0, overlay=ROOF_OVERLAY)],
[scored_option("suspended_floor_insulation", gain=5.0, cost=1000.0, overlay=FLOOR_OVERLAY)],
]
scorer = _StubScorer(base=60.0, wall=10.0, roof=5.0, floor=5.0)
scorer = StubScorer(base=60.0, wall=10.0, roof=5.0, floor=5.0)
# Act — target 90 is out of reach; best effort is the most SAP budget buys.
package: OptimisedPackage = optimise_package(
@ -450,7 +369,7 @@ def test_package_falls_back_to_max_gain_when_target_unreachable() -> None:
)
# Assert — max-gain: all three, SAP 80 (below target, best effort).
assert _selected_types(package.selected) == {
assert selected_types(package.selected) == {
"cavity_wall_insulation",
"loft_insulation",
"suspended_floor_insulation",
@ -463,10 +382,10 @@ def test_package_repairs_when_the_signal_overshoots_the_true_score() -> None:
# min-cost warm-start picks it alone; but its true gain is only +5, so the
# package undershoots and repair must top it up.
groups: list[list[ScoredOption]] = [
[_scored_overlay("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=_WALL_OVERLAY)],
[_scored_overlay("loft_insulation", gain=0.0, cost=1000.0, overlay=_ROOF_OVERLAY)],
[scored_option("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=WALL_OVERLAY)],
[scored_option("loft_insulation", gain=0.0, cost=1000.0, overlay=ROOF_OVERLAY)],
]
scorer = _StubScorer(base=60.0, wall=5.0, roof=4.0, floor=0.0)
scorer = StubScorer(base=60.0, wall=5.0, roof=4.0, floor=0.0)
# Act — target 69 (gain 9). Warm-start {wall} (signal 10) → true 65 < 69 →
# repair adds the roof (+4) → 69.
@ -479,7 +398,7 @@ def test_package_repairs_when_the_signal_overshoots_the_true_score() -> None:
)
# Assert
assert _selected_types(package.selected) == {
assert selected_types(package.selected) == {
"cavity_wall_insulation",
"loft_insulation",
}
@ -488,57 +407,6 @@ def test_package_repairs_when_the_signal_overshoots_the_true_score() -> None:
# --- Measure Dependency injection (ADR-0016) -------------------------------
_VENT_OVERLAY = EpcSimulation(
ventilation=VentilationOverlay(
mechanical_ventilation_kind="EXTRACT_OR_PIV_OUTSIDE"
)
)
class _VentStubScorer:
"""A stub that adds a fixed gain per wall overlay present and a fixed
(negative) `vent` contribution when a ventilation overlay is present —
so the Measure Dependency's effect on the truthful package total and the
repair decision is exercised without the calculator."""
def __init__(self, *, base: float, wall: float, roof: float, vent: float) -> None:
self._base = base
self._wall = wall
self._roof = roof
self._vent = vent
def score(
self, baseline: EpcPropertyData, simulations: Sequence[EpcSimulation]
) -> Score:
sap = self._base
for sim in simulations:
if sim.ventilation is not None:
sap += self._vent
for part in sim.building_parts.values():
if part.wall_insulation_type is not None:
sap += self._wall
if part.roof_insulation_thickness is not None:
sap += self._roof
return Score(sap_continuous=sap, co2_kg_per_yr=0.0, primary_energy_kwh_per_yr=0.0)
def _ventilation_dependency(*, cost: float) -> MeasureDependency:
"""A forced 'fabric requires ventilation' edge for the tests."""
return MeasureDependency(
triggers=frozenset(
{MeasureType.CAVITY_WALL_INSULATION, MeasureType.EXTERNAL_WALL_INSULATION}
),
required=ScoredOption(
option=MeasureOption(
measure_type=MeasureType.MECHANICAL_VENTILATION,
description="mechanical_ventilation",
overlay=_VENT_OVERLAY,
cost=Cost(total=cost, contingency_rate=0.0),
),
sap_gain=0.0,
),
)
def test_min_cost_warm_start_avoids_a_wall_whose_forced_ventilation_dooms_it() -> None:
# Arrange — cavity is dirt cheap (£100) and its role-1 signal (+6) alone
@ -547,21 +415,12 @@ def test_min_cost_warm_start_avoids_a_wall_whose_forced_ventilation_dooms_it() -
# package below target. A ventilation-AWARE warm-start prices that −5 into
# the candidate and instead takes the wall-free loft path.
groups: list[list[ScoredOption]] = [
[_scored_overlay("cavity_wall_insulation", gain=6.0, cost=100.0, overlay=_WALL_OVERLAY)],
[_scored_overlay("loft_insulation", gain=8.0, cost=1500.0, overlay=_ROOF_OVERLAY)],
[scored_option("cavity_wall_insulation", gain=6.0, cost=100.0, overlay=WALL_OVERLAY)],
[scored_option("loft_insulation", gain=8.0, cost=1500.0, overlay=ROOF_OVERLAY)],
]
scorer = _VentStubScorer(base=60.0, wall=6.0, roof=8.0, vent=-5.0)
dependency = MeasureDependency(
triggers=frozenset({MeasureType.CAVITY_WALL_INSULATION}),
required=ScoredOption(
option=MeasureOption(
measure_type=MeasureType.MECHANICAL_VENTILATION,
description="mechanical_ventilation",
overlay=_VENT_OVERLAY,
cost=Cost(total=300.0, contingency_rate=0.0),
),
sap_gain=0.0, # placeholder; optimise_package scores the real signal
),
scorer = StubScorer(base=60.0, wall=6.0, roof=8.0, vent=-5.0)
dependency = ventilation_dependency(
cost=300.0, triggers=frozenset({MeasureType.CAVITY_WALL_INSULATION})
)
# Act — target 66 (gain 6 over the 60 baseline).
@ -576,7 +435,7 @@ def test_min_cost_warm_start_avoids_a_wall_whose_forced_ventilation_dooms_it() -
# Assert — the loft path (true 68, £1500), NOT cavity + forced ventilation:
# cavity's signal (+6) is cancelled by ventilation (−5) to +1 < target.
assert _selected_types(package.selected) == {"loft_insulation"}
assert selected_types(package.selected) == {"loft_insulation"}
assert abs(package.score.sap_continuous - 68.0) <= 1e-9
@ -584,9 +443,9 @@ def test_dependency_injected_when_a_trigger_measure_is_selected() -> None:
# Arrange — the wall is selected, so its ventilation dependency must be
# injected before the re-score; ventilation never competes in the pool.
groups: list[list[ScoredOption]] = [
[_scored_overlay("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=_WALL_OVERLAY)],
[scored_option("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=WALL_OVERLAY)],
]
scorer = _VentStubScorer(base=40.0, wall=5.0, roof=4.0, vent=-2.0)
scorer = StubScorer(base=40.0, wall=5.0, roof=4.0, vent=-2.0)
# Act
package: OptimisedPackage = optimise_package(
@ -595,12 +454,12 @@ def test_dependency_injected_when_a_trigger_measure_is_selected() -> None:
baseline_epc=build_epc(),
budget=None,
target_sap=None,
dependencies=[_ventilation_dependency(cost=900.0)],
dependencies=[ventilation_dependency(cost=900.0)],
)
# Assert — ventilation is in the package and its negative contribution lands
# in the truthful total: 40 base + 5 wall − 2 ventilation = 43.
assert _selected_types(package.selected) == {
assert selected_types(package.selected) == {
"cavity_wall_insulation",
"mechanical_ventilation",
}
@ -611,9 +470,9 @@ def test_dependency_not_injected_without_a_trigger_measure() -> None:
# Arrange — only loft is selected; the wall-triggered ventilation dependency
# must not fire.
groups: list[list[ScoredOption]] = [
[_scored_overlay("loft_insulation", gain=4.0, cost=1000.0, overlay=_ROOF_OVERLAY)],
[scored_option("loft_insulation", gain=4.0, cost=1000.0, overlay=ROOF_OVERLAY)],
]
scorer = _VentStubScorer(base=40.0, wall=5.0, roof=4.0, vent=-2.0)
scorer = StubScorer(base=40.0, wall=5.0, roof=4.0, vent=-2.0)
# Act
package: OptimisedPackage = optimise_package(
@ -622,11 +481,11 @@ def test_dependency_not_injected_without_a_trigger_measure() -> None:
baseline_epc=build_epc(),
budget=None,
target_sap=None,
dependencies=[_ventilation_dependency(cost=900.0)],
dependencies=[ventilation_dependency(cost=900.0)],
)
# Assert — no trigger, no ventilation; 40 base + 4 roof = 44.
assert _selected_types(package.selected) == {"loft_insulation"}
assert selected_types(package.selected) == {"loft_insulation"}
assert abs(package.score.sap_continuous - 44.0) <= 1e-9
@ -636,9 +495,9 @@ def test_wall_dropped_when_it_cannot_be_ventilated_within_budget() -> None:
# wall we can't afford to ventilate is a wall we can't afford, so it is
# dropped (the budget is a hard envelope, ventilation is not forced over it).
groups: list[list[ScoredOption]] = [
[_scored_overlay("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=_WALL_OVERLAY)],
[scored_option("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=WALL_OVERLAY)],
]
scorer = _VentStubScorer(base=40.0, wall=5.0, roof=4.0, vent=-2.0)
scorer = StubScorer(base=40.0, wall=5.0, roof=4.0, vent=-2.0)
# Act — tight budget; ventilation-aware selection prices the £900 in.
package: OptimisedPackage = optimise_package(
@ -647,7 +506,7 @@ def test_wall_dropped_when_it_cannot_be_ventilated_within_budget() -> None:
baseline_epc=build_epc(),
budget=1000.0,
target_sap=None,
dependencies=[_ventilation_dependency(cost=900.0)],
dependencies=[ventilation_dependency(cost=900.0)],
)
# Assert — nothing recommended; the budget is respected and the wall is
@ -660,10 +519,10 @@ def test_injected_ventilation_penalty_drives_extra_repair() -> None:
# Repair adds the roof (true +4) to reach 47, paying for the ventilation
# penalty out of the budget the dependency's cost has already eaten into.
groups: list[list[ScoredOption]] = [
[_scored_overlay("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=_WALL_OVERLAY)],
[_scored_overlay("loft_insulation", gain=0.0, cost=1000.0, overlay=_ROOF_OVERLAY)],
[scored_option("cavity_wall_insulation", gain=10.0, cost=1000.0, overlay=WALL_OVERLAY)],
[scored_option("loft_insulation", gain=0.0, cost=1000.0, overlay=ROOF_OVERLAY)],
]
scorer = _VentStubScorer(base=40.0, wall=5.0, roof=4.0, vent=-2.0)
scorer = StubScorer(base=40.0, wall=5.0, roof=4.0, vent=-2.0)
# Act
package: OptimisedPackage = optimise_package(
@ -672,12 +531,12 @@ def test_injected_ventilation_penalty_drives_extra_repair() -> None:
baseline_epc=build_epc(),
budget=5000.0,
target_sap=46.0,
dependencies=[_ventilation_dependency(cost=900.0)],
dependencies=[ventilation_dependency(cost=900.0)],
)
# Assert — repair pulled the roof in to clear the target net of ventilation:
# 40 + 5 wall − 2 vent + 4 roof = 47.
assert _selected_types(package.selected) == {
assert selected_types(package.selected) == {
"cavity_wall_insulation",
"loft_insulation",
"mechanical_ventilation",

349
tests/domain/modelling/test_optimiser_fabric_first.py Normal file
View file

@ -0,0 +1,349 @@
"""Behaviour of the Fabric First two-phase Optimiser: phase 1 optimises the
fabric measures (wall / roof / floor insulation + glazing) with the full
budget; if the truthful post-fabric score meets the Scenario target the
package is fabric-only. Otherwise phase 2 optimises the remaining measures on
top, where the starting point is the dwelling with the phase-1 fabric applied
and only the leftover budget is spendable. Mirrors the legacy engine's
``enforce_fabric_first`` (funding_optimiser.optimise_with_scenarios) on the
new truthful-re-score core (ADR-0016).
"""
from __future__ import annotations
from typing import Sequence
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.modelling.measure_type import MeasureType
from domain.modelling.optimisation.optimiser import (
OptimisedPackage,
ScoredOption,
optimise_package_fabric_first,
)
from domain.modelling.scoring.package_scorer import Score
from domain.modelling.simulation import EpcSimulation
from tests.domain.modelling._optimiser_fixtures import (
ASHP_OVERLAY,
BOILER_OVERLAY,
GLAZING_OVERLAY,
WALL_OVERLAY,
StubScorer,
scored_option,
selected_types,
ventilation_dependency,
)
from tests.domain.sap10_calculator.worksheet._elmhurst_worksheet_000490 import (
build_epc,
)
_AIRTIGHTNESS_TRIGGERS: frozenset[MeasureType] = frozenset(
{MeasureType.CAVITY_WALL_INSULATION, MeasureType.DOUBLE_GLAZING}
)
def test_fabric_reaching_the_target_excludes_non_fabric_measures() -> None:
# Arrange — the £3,200 boiler is the cheapest route to the target (a plain
# least-cost-to-target run would take it alone), but the wall by itself
# reaches the target: fabric first means the package stops at the fabric.
groups: list[list[ScoredOption]] = [
[scored_option("external_wall_insulation", gain=12.0, cost=12000.0, overlay=WALL_OVERLAY)],
[scored_option("gas_boiler_upgrade", gain=15.0, cost=3200.0, overlay=BOILER_OVERLAY)],
]
scorer = StubScorer(base=60.0, wall=12.0, heating=15.0)
# Act — target 69 (gain 9 over the 60 baseline).
package: OptimisedPackage = optimise_package_fabric_first(
groups=groups,
scorer=scorer,
baseline_epc=build_epc(),
budget=15000.0,
target_sap=69.0,
)
# Assert — fabric only: the wall (true 72 ≥ 69); the boiler is never
# considered because the upgrade requirement is already met.
assert selected_types(package.selected) == {"external_wall_insulation"}
assert abs(package.score.sap_continuous - 72.0) <= 1e-9
def test_fabric_short_of_target_is_topped_up_with_non_fabric_measures() -> None:
# Arrange — all the fabric there is (the wall, +5) cannot reach the target;
# phase 2 must add the heat pump on top of the retained fabric.
groups: list[list[ScoredOption]] = [
[scored_option("cavity_wall_insulation", gain=5.0, cost=1000.0, overlay=WALL_OVERLAY)],
[scored_option("air_source_heat_pump", gain=20.0, cost=8000.0, overlay=ASHP_OVERLAY)],
]
scorer = StubScorer(base=60.0, wall=5.0, heating=20.0)
# Act — target 75 (gain 15); fabric alone tops out at 65.
package: OptimisedPackage = optimise_package_fabric_first(
groups=groups,
scorer=scorer,
baseline_epc=build_epc(),
budget=20000.0,
target_sap=75.0,
)
# Assert — the fabric is kept and the heat pump lands on top of it; the
# score is the truthful whole-package figure (60 + 5 + 20).
assert selected_types(package.selected) == {
"cavity_wall_insulation",
"air_source_heat_pump",
}
assert abs(package.score.sap_continuous - 85.0) <= 1e-9
def test_fabric_spend_comes_out_of_the_shared_budget_before_phase_two() -> None:
# Arrange — the £8000 heat pump alone would fit the £8500 budget and reach
# the target, but fabric first commits the £1000 wall first, leaving £7500:
# the heat pump no longer fits. Fabric priority wins over the target.
groups: list[list[ScoredOption]] = [
[scored_option("cavity_wall_insulation", gain=5.0, cost=1000.0, overlay=WALL_OVERLAY)],
[scored_option("air_source_heat_pump", gain=20.0, cost=8000.0, overlay=ASHP_OVERLAY)],
]
scorer = StubScorer(base=60.0, wall=5.0, heating=20.0)
# Act — target 78 (gain 18).
package: OptimisedPackage = optimise_package_fabric_first(
groups=groups,
scorer=scorer,
baseline_epc=build_epc(),
budget=8500.0,
target_sap=78.0,
)
# Assert — wall only; the target is missed rather than the fabric skipped.
assert selected_types(package.selected) == {"cavity_wall_insulation"}
assert abs(package.score.sap_continuous - 65.0) <= 1e-9
class _AirtightnessScorer:
"""A stub where tightening the envelope demands ventilation: the cavity
wall is +5 SAP, the new double glazing is worthless on the raw dwelling
but +4 once the wall is insulated, and every ventilation overlay present
costs −1 — so a double injection is visible in the package score."""
def score(
self, baseline: EpcPropertyData, simulations: Sequence[EpcSimulation]
) -> Score:
wall = any(
part.wall_insulation_type is not None
for sim in simulations
for part in sim.building_parts.values()
)
glazing = any(sim.glazing is not None for sim in simulations)
vents = sum(1 for sim in simulations if sim.ventilation is not None)
sap = 60.0
if wall:
sap += 5.0
if wall and glazing:
sap += 4.0
sap -= float(vents)
return Score(
sap_continuous=sap, co2_kg_per_yr=0.0, primary_energy_kwh_per_yr=0.0
)
def test_ventilation_dependency_is_injected_once_across_both_phases() -> None:
# Arrange — the cavity wall (phase 1) and the double glazing (skipped in
# phase 1 on merit, picked in phase 2 on its post-fabric worth) both
# trigger the same forced ventilation. It must land in the package exactly
# once — phase 2 sees the phase-1 dwelling as already ventilated.
groups: list[list[ScoredOption]] = [
[scored_option("cavity_wall_insulation", gain=5.0, cost=1000.0, overlay=WALL_OVERLAY)],
[scored_option("double_glazing", gain=0.0, cost=3500.0, overlay=GLAZING_OVERLAY)],
]
scorer = _AirtightnessScorer()
# Act — target 68: phase 1 gives 60 + 5 − 1 = 64; the glazing's
# post-fabric +4 closes it, but only if ventilation is not double-counted.
package: OptimisedPackage = optimise_package_fabric_first(
groups=groups,
scorer=scorer,
baseline_epc=build_epc(),
budget=10000.0,
target_sap=68.0,
dependencies=[
ventilation_dependency(cost=300.0, triggers=_AIRTIGHTNESS_TRIGGERS)
],
)
# Assert — one ventilation, and the truthful total counts its penalty once:
# 60 + 5 wall + 4 glazing − 1 ventilation = 68.
ventilation_count = sum(
1
for scored in package.selected
if scored.option.measure_type == MeasureType.MECHANICAL_VENTILATION
)
assert ventilation_count == 1
assert selected_types(package.selected) == {
"cavity_wall_insulation",
"double_glazing",
"mechanical_ventilation",
}
assert abs(package.score.sap_continuous - 68.0) <= 1e-9
def test_no_fabric_candidates_proceeds_straight_to_the_full_pool() -> None:
# Arrange — the envelope work is already done (no fabric Recommendation
# survives generation); fabric first must not veto the run, it just means
# phase 1 has nothing to do.
groups: list[list[ScoredOption]] = [
[scored_option("air_source_heat_pump", gain=20.0, cost=8000.0, overlay=ASHP_OVERLAY)],
]
scorer = StubScorer(base=60.0, heating=20.0)
# Act — target 75.
package: OptimisedPackage = optimise_package_fabric_first(
groups=groups,
scorer=scorer,
baseline_epc=build_epc(),
budget=20000.0,
target_sap=75.0,
)
# Assert — the heat pump package, exactly as a plain run would produce.
assert selected_types(package.selected) == {"air_source_heat_pump"}
assert abs(package.score.sap_continuous - 80.0) <= 1e-9
def test_without_a_target_fabric_still_gets_first_claim_on_the_budget() -> None:
# Arrange — a max-gain goal (no SAP target). Plain max-gain would spend the
# whole £8000 on the heat pump (+20); fabric first commits the wall (+5)
# before the remainder is considered, pricing the heat pump out.
groups: list[list[ScoredOption]] = [
[scored_option("cavity_wall_insulation", gain=5.0, cost=1000.0, overlay=WALL_OVERLAY)],
[scored_option("air_source_heat_pump", gain=20.0, cost=8000.0, overlay=ASHP_OVERLAY)],
]
scorer = StubScorer(base=60.0, wall=5.0, heating=20.0)
# Act — no target: the flag applies to every goal, not just Increasing EPC.
package: OptimisedPackage = optimise_package_fabric_first(
groups=groups,
scorer=scorer,
baseline_epc=build_epc(),
budget=8000.0,
target_sap=None,
)
# Assert — wall first; the heat pump no longer fits the leftover £7000.
assert selected_types(package.selected) == {"cavity_wall_insulation"}
assert abs(package.score.sap_continuous - 65.0) <= 1e-9
class _InteractionScorer:
"""A stub whose boiler gain collapses once the wall is insulated (+10 raw,
+3 post-fabric) while the heat pump's holds (+8 either way) — so a phase 2
that keeps valuing candidates against the raw baseline picks the wrong
heating system."""
def score(
self, baseline: EpcPropertyData, simulations: Sequence[EpcSimulation]
) -> Score:
wall_present = any(
part.wall_insulation_type is not None
for sim in simulations
for part in sim.building_parts.values()
)
sap = 60.0 + (5.0 if wall_present else 0.0)
for sim in simulations:
if sim.heating is None:
continue
if sim.heating.sap_main_heating_code is not None:
sap += 3.0 if wall_present else 10.0
if sim.heating.main_heating_index_number is not None:
sap += 8.0
return Score(
sap_continuous=sap, co2_kg_per_yr=0.0, primary_energy_kwh_per_yr=0.0
)
class _GlazingInteractionScorer:
"""A stub where glazing is worthless on the raw dwelling (+0) but worth +4
once the wall is insulated — so phase 1's max-gain fabric pass leaves it
out, and only a phase 2 that re-admits unpicked fabric can close the
target with it."""
def score(
self, baseline: EpcPropertyData, simulations: Sequence[EpcSimulation]
) -> Score:
wall_present = any(
part.wall_insulation_type is not None
for sim in simulations
for part in sim.building_parts.values()
)
glazing_present = any(sim.glazing is not None for sim in simulations)
heating_present = any(sim.heating is not None for sim in simulations)
sap = 60.0
if wall_present:
sap += 5.0
if wall_present and glazing_present:
sap += 4.0
if heating_present:
sap += 10.0
return Score(
sap_continuous=sap, co2_kg_per_yr=0.0, primary_energy_kwh_per_yr=0.0
)
def test_fabric_unpicked_in_phase_one_can_reenter_phase_two() -> None:
# Arrange — glazing loses phase 1 on merit (it scores nothing on the raw
# dwelling), but post-wall it is the only affordable way to the target:
# the heat pump that could also close it does not fit the leftover budget.
groups: list[list[ScoredOption]] = [
[scored_option("cavity_wall_insulation", gain=5.0, cost=1000.0, overlay=WALL_OVERLAY)],
[scored_option("double_glazing", gain=0.0, cost=3500.0, overlay=GLAZING_OVERLAY)],
[scored_option("air_source_heat_pump", gain=10.0, cost=8000.0, overlay=ASHP_OVERLAY)],
]
scorer = _GlazingInteractionScorer()
# Act — target 69 (gain 9); budget £5000 keeps the heat pump out of reach
# after the wall's £1000.
package: OptimisedPackage = optimise_package_fabric_first(
groups=groups,
scorer=scorer,
baseline_epc=build_epc(),
budget=5000.0,
target_sap=69.0,
)
# Assert — the skipped glazing re-enters on its post-fabric worth: 60 + 5
# wall + 4 glazing = 69, target met.
assert selected_types(package.selected) == {
"cavity_wall_insulation",
"double_glazing",
}
assert abs(package.score.sap_continuous - 69.0) <= 1e-9
def test_phase_two_values_candidates_against_the_post_fabric_dwelling() -> None:
# Arrange — one heating Recommendation, two Options. The boiler's role-1
# signal (vs the raw baseline, +10) beats the heat pump's (+8) and it is
# cheaper — but on the insulated dwelling the boiler is only worth +3.
# Only a heat pump gets the fabric-applied dwelling to the target.
groups: list[list[ScoredOption]] = [
[scored_option("cavity_wall_insulation", gain=5.0, cost=1000.0, overlay=WALL_OVERLAY)],
[
scored_option("gas_boiler_upgrade", gain=10.0, cost=3200.0, overlay=BOILER_OVERLAY),
scored_option("air_source_heat_pump", gain=8.0, cost=8000.0, overlay=ASHP_OVERLAY),
],
]
scorer = _InteractionScorer()
# Act — target 73: wall (65) + boiler-post-fabric (+3) = 68 misses; wall +
# heat pump (+8) = 73 reaches. The heating group is consumed by whichever
# option phase 2 warm-starts with, so the choice must be made on
# post-fabric values, not raw-baseline signals.
package: OptimisedPackage = optimise_package_fabric_first(
groups=groups,
scorer=scorer,
baseline_epc=build_epc(),
budget=20000.0,
target_sap=73.0,
)
# Assert
assert selected_types(package.selected) == {
"cavity_wall_insulation",
"air_source_heat_pump",
}
assert abs(package.score.sap_continuous - 73.0) <= 1e-9

162
tests/domain/modelling/test_optimiser_goal_objective.py Normal file
View file

@ -0,0 +1,162 @@
"""Behaviour of the Optimiser under a goal-aligned objective (ADR-0062): a
Scenario whose goal is Reducing CO2 emissions / Energy Savings optimises its
own metric, not SAP. The caller supplies group signals already measured in the
objective's currency; the optimiser must price everything *it* computes — the
forced Measure Dependency signals — in the same currency, so a ventilation
that costs SAP but is carbon-neutral cannot sink a carbon-improving wall.
"""
from __future__ import annotations
from typing import Sequence
from datatypes.epc.domain.epc_property_data import (
BuildingPartIdentifier,
EpcPropertyData,
)
from domain.modelling.measure_type import MeasureType
from domain.modelling.optimisation.optimiser import (
OptimisedPackage,
ScoredOption,
optimise_package,
optimise_package_fabric_first,
)
from domain.modelling.scoring.package_scorer import Score
from domain.modelling.simulation import BuildingPartOverlay, EpcSimulation
from tests.domain.modelling._optimiser_fixtures import (
ASHP_OVERLAY,
BOILER_OVERLAY,
WALL_OVERLAY,
scored_option,
selected_types,
ventilation_dependency,
)
from tests.domain.sap10_calculator.worksheet._elmhurst_worksheet_000490 import (
build_epc,
)
class _CarbonScorer:
"""A stub where the wall is a small carbon win (−20 kg/yr) and a large SAP
win (+6), while its forced ventilation is carbon-neutral but SAP-ruinous
(−30): SAP-priced dependency signals sink the wall; carbon-priced ones
keep it."""
def score(
self, baseline: EpcPropertyData, simulations: Sequence[EpcSimulation]
) -> Score:
sap, co2 = 60.0, 500.0
for sim in simulations:
if sim.ventilation is not None:
sap -= 30.0
for part in sim.building_parts.values():
if part.wall_insulation_type is not None:
sap += 6.0
co2 -= 20.0
return Score(
sap_continuous=sap, co2_kg_per_yr=co2, primary_energy_kwh_per_yr=0.0
)
def _carbon_reduction(score: Score) -> float:
return -score.co2_kg_per_yr
def test_dependency_signals_are_priced_in_the_objective_currency() -> None:
# Arrange — the wall's signal (supplied by the caller, +20 kg CO2 saved)
# and the ventilation it forces in (carbon-neutral). Under legacy SAP
# pricing the ventilation's −30 SAP would outweigh the wall's +20 signal
# and the package would collapse to nothing.
groups: list[list[ScoredOption]] = [
[scored_option("cavity_wall_insulation", gain=20.0, cost=1000.0, overlay=WALL_OVERLAY)],
]
dependency = ventilation_dependency(
cost=300.0, triggers=frozenset({MeasureType.CAVITY_WALL_INSULATION})
)
# Act — a Reducing-CO2 brief: maximise carbon reduction within budget.
package: OptimisedPackage = optimise_package(
groups=groups,
scorer=_CarbonScorer(),
baseline_epc=build_epc(),
budget=5000.0,
target_sap=None,
dependencies=[dependency],
objective=_carbon_reduction,
)
# Assert — the wall survives with its ventilation: the dependency is worth
# 0 kg CO2, not −30 SAP, so the package is a net +20 kg saving.
assert selected_types(package.selected) == {
"cavity_wall_insulation",
"mechanical_ventilation",
}
assert abs(package.score.co2_kg_per_yr - 480.0) <= 1e-9
# Internal wall insulation — a distinct fabric overlay so the fabric-first
# phase-1 pick is unambiguous. No shared fixture (the shared WALL_OVERLAY is a
# cavity fill, type 2); this is a solid-wall internal treatment, type 3.
_IWI_OVERLAY = EpcSimulation(
building_parts={
BuildingPartIdentifier.MAIN: BuildingPartOverlay(wall_insulation_type=3)
}
)
class _CarbonHeatingScorer:
"""A stub where the boiler wins on SAP (+10 vs +2) but the heat pump wins
on carbon (−50 vs −5 kg/yr): a fabric-first phase 2 that re-scores its
candidates in SAP picks the wrong heating for a Reducing-CO2 brief."""
def score(
self, baseline: EpcPropertyData, simulations: Sequence[EpcSimulation]
) -> Score:
sap, co2 = 60.0, 500.0
for sim in simulations:
for part in sim.building_parts.values():
if part.wall_insulation_type is not None:
sap += 5.0
co2 -= 10.0
if sim.heating is None:
continue
if sim.heating.sap_main_heating_code is not None:
sap += 10.0
co2 -= 5.0
if sim.heating.main_heating_index_number is not None:
sap += 2.0
co2 -= 50.0
return Score(
sap_continuous=sap, co2_kg_per_yr=co2, primary_energy_kwh_per_yr=0.0
)
def test_fabric_first_phase_two_rescores_in_the_objective_currency() -> None:
# Arrange — a fabric-first Reducing-CO2 brief. Phase 1 commits the wall;
# phase 2 must choose the heating on its post-fabric *carbon* worth, not
# its SAP worth. Signals are supplied in kg CO2 saved (the caller's job).
groups: list[list[ScoredOption]] = [
[scored_option("internal_wall_insulation", gain=10.0, cost=1000.0, overlay=_IWI_OVERLAY)],
[
scored_option("gas_boiler_upgrade", gain=5.0, cost=2000.0, overlay=BOILER_OVERLAY),
scored_option("air_source_heat_pump", gain=50.0, cost=6000.0, overlay=ASHP_OVERLAY),
],
]
# Act — no target (goal-aligned briefs have none), generous budget.
package: OptimisedPackage = optimise_package_fabric_first(
groups=groups,
scorer=_CarbonHeatingScorer(),
baseline_epc=build_epc(),
budget=10000.0,
target_sap=None,
objective=_carbon_reduction,
)
# Assert — the wall plus the heat pump (−50 kg), not the SAP-favoured
# boiler; the truthful package carbon is 500 − 10 − 50 = 440.
assert selected_types(package.selected) == {
"internal_wall_insulation",
"air_source_heat_pump",
}
assert abs(package.score.co2_kg_per_yr - 440.0) <= 1e-9

12
tests/domain/modelling/test_scoring.py
View file

@ -16,7 +16,7 @@ from domain.modelling.recommendation import MeasureOption
from domain.modelling.scoring.scoring import (
MeasureImpact,
cascade_scores,
independent_option_impacts,
independent_option_signals,
marginal_impacts,
marginals_from_scores,
)
@ -64,7 +64,7 @@ def _option(overlay: EpcSimulation) -> MeasureOption:
)
def test_independent_option_impacts_score_each_distinct_overlay_once() -> None:
def test_independent_option_signals_score_each_distinct_overlay_once() -> None:
# Arrange
baseline: EpcPropertyData = build_epc()
scorer = _CountingScorer()
@ -86,15 +86,15 @@ def test_independent_option_impacts_score_each_distinct_overlay_once() -> None:
options = [_option(overlay_a), _option(overlay_a_dup), _option(overlay_b)]
# Act
impacts: list[MeasureImpact] = independent_option_impacts(
scorer, baseline, options
signals: list[float] = independent_option_signals(
scorer, baseline, options, lambda score: score.sap_continuous
)
# Assert
# baseline scored once + one score per DISTINCT overlay (a, b) = 3, not 4
assert scorer.calls == 3
assert impacts[0].sap_points == impacts[1].sap_points == 2.0
assert impacts[2].sap_points == 3.0
assert signals[0] == signals[1] == 2.0
assert signals[2] == 3.0
def test_single_overlay_marginal_is_its_improvement_over_baseline() -> None:

60
tests/orchestration/test_modelling_fabric_first.py Normal file
View file

@ -0,0 +1,60 @@
"""The ModellingOrchestrator honours a Fabric First Scenario: when
``scenario.fabric_first`` is set, the Optimiser treats the building envelope
with the full budget before heating / renewables are considered, so a budget
the plain optimiser would spend on a heating system is spent on fabric
instead. End-to-end through ``run_modelling`` (no database) with the real
calculator, against the uninsulated solid-brick 001431 dwelling whose plain
plan is heating-led.
"""
from __future__ import annotations
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.modelling.measure_type import FABRIC_MEASURE_TYPES, MeasureType
from domain.modelling.plan import Plan
from domain.modelling.scenario import Scenario
from harness.console import run_modelling
from tests.domain.modelling._elmhurst_recommendation import (
parse_recommendation_summary,
)
_HEATING_MEASURES: frozenset[MeasureType] = frozenset(
{
MeasureType.GAS_BOILER_UPGRADE,
MeasureType.AIR_SOURCE_HEAT_PUMP,
MeasureType.HIGH_HEAT_RETENTION_STORAGE_HEATERS,
MeasureType.SOLAR_PV,
}
)
def _fabric_first_scenario(*, budget: float) -> Scenario:
return Scenario(
id=999,
goal="Increasing EPC",
goal_value="C",
budget=budget,
is_default=True,
fabric_first=True,
)
def test_fabric_first_scenario_spends_the_budget_on_fabric_before_heating() -> None:
# Arrange — uninsulated solid brick: at a £4000 budget the plain optimiser
# buys the £3200 gas boiler (the cheapest route toward band C). Fabric
# first must commit the envelope work first, after which the boiler no
# longer fits the leftover budget.
epc: EpcPropertyData = parse_recommendation_summary(
"solid_brick_ewi_001431_before.pdf"
)
# Act
plan: Plan = run_modelling(
epc, scenario=_fabric_first_scenario(budget=4000.0), print_table=False
)
# Assert — the plan is fabric-led: at least one envelope measure, and none
# of the budget leaked to a heating system or renewables.
selected = {measure.measure_type for measure in plan.measures}
assert selected & FABRIC_MEASURE_TYPES
assert not selected & _HEATING_MEASURES

133
tests/orchestration/test_modelling_goal_objectives.py Normal file
View file

@ -0,0 +1,133 @@
"""The ModellingOrchestrator aligns the Optimiser's objective with the
Scenario's goal (ADR-0062): Reducing CO2 emissions maximises the carbon
reduction the budget buys, Energy Savings maximises the annual bill saving,
and Increasing EPC keeps its SAP target semantics. End-to-end through
``run_modelling`` (no database) with the real calculator, against the
uninsulated solid-brick 001431 dwelling where the SAP-optimal and
carbon-optimal packages diverge at a £16,000 budget.
"""
from __future__ import annotations
import dataclasses
import pytest
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.fuel_rates.fuel import Fuel
from domain.fuel_rates.fuel_rates import FuelRate, FuelRates
from domain.modelling.measure_type import MeasureType
from domain.modelling.plan import Plan
from domain.modelling.scenario import Scenario
from harness.console import run_modelling
from repositories.fuel_rates.fuel_rates_repository import FuelRatesRepository
from repositories.fuel_rates.fuel_rates_static_file_repository import (
FuelRatesStaticFileRepository,
)
from tests.domain.modelling._elmhurst_recommendation import (
parse_recommendation_summary,
)
def _solid_brick_dwelling() -> EpcPropertyData:
return parse_recommendation_summary("solid_brick_ewi_001431_before.pdf")
def _scenario(goal: str, *, budget: float) -> Scenario:
return Scenario(
id=999, goal=goal, goal_value="", budget=budget, is_default=True
)
def test_reducing_co2_scenario_buys_carbon_not_sap() -> None:
# Arrange — at £16,000 the SAP objective buys the wall + floor + £3,200
# gas boiler package (~2,069 kg CO2/yr, SAP 72.9). The carbon objective
# swaps the boiler for electric storage heaters (~1,098 kg/yr) — a lower
# SAP, but ~970 kg/yr less carbon on the low-carbon grid.
epc = _solid_brick_dwelling()
# Act — the same dwelling and budget under each goal.
sap_led: Plan = run_modelling(
epc,
scenario=_scenario("Valuation Improvement", budget=16000.0),
print_table=False,
)
carbon_led: Plan = run_modelling(
epc,
scenario=_scenario("Reducing CO2 emissions", budget=16000.0),
print_table=False,
)
# Assert — the goal changes the outcome in the goal's favour: the carbon
# plan cuts materially more CO2 than the SAP plan buys with the same
# money, and the gas boiler that wins on SAP-per-£ is rejected.
assert (
carbon_led.post_retrofit.co2_kg_per_yr
< sap_led.post_retrofit.co2_kg_per_yr - 500.0
)
selected = {measure.measure_type for measure in carbon_led.measures}
assert MeasureType.GAS_BOILER_UPGRADE not in selected
def test_a_goal_aligned_scenario_without_a_budget_fails_loudly() -> None:
# Arrange — 'reduce as much as possible within this budget' is undefined
# without a budget: unconstrained it would recommend every beneficial
# measure. A budget-less goal-aligned Scenario is a misconfiguration and
# must fail visibly, not produce a maximal plan.
epc = _solid_brick_dwelling()
budgetless = Scenario(
id=999,
goal="Reducing CO2 emissions",
goal_value="",
budget=None,
is_default=True,
)
# Act / Assert
with pytest.raises(ValueError, match="budget"):
run_modelling(epc, scenario=budgetless, print_table=False)
class _FixedFuelRates(FuelRatesRepository):
def __init__(self, rates: FuelRates) -> None:
self._rates = rates
def get_current(self) -> FuelRates:
return self._rates
def _cheap_electricity_snapshot() -> FuelRates:
"""The committed snapshot with electricity at 1p/kWh — a world where any
electric heating out-bills gas, while SAP's internal price book (which the
calculator rates against) is unmoved."""
base = FuelRatesStaticFileRepository().get_current()
rates = dict(base.rates)
rates[Fuel.ELECTRICITY] = FuelRate(
unit_rate_p_per_kwh=1.0,
standing_charge_p_per_day=rates[Fuel.ELECTRICITY].standing_charge_p_per_day,
)
return dataclasses.replace(base, rates=rates)
def test_energy_savings_scenario_prices_packages_at_the_live_fuel_rates() -> None:
# Arrange — SAP is itself a cost metric, but it prices energy from its
# internal tariff book. The Energy Savings goal must price at the *live*
# Fuel Rates snapshot: with 1p/kWh electricity, electric heating slashes
# the bill even though SAP still scores the gas boiler package higher.
epc = _solid_brick_dwelling()
# Act
plan: Plan = run_modelling(
epc,
scenario=_scenario("Energy Savings", budget=16000.0),
fuel_rates=_FixedFuelRates(_cheap_electricity_snapshot()),
print_table=False,
)
# Assert — the bill objective abandons the boiler for electric heating.
selected = {measure.measure_type for measure in plan.measures}
assert MeasureType.GAS_BOILER_UPGRADE not in selected
assert selected & {
MeasureType.AIR_SOURCE_HEAT_PUMP,
MeasureType.HIGH_HEAT_RETENTION_STORAGE_HEATERS,
}

22
tests/repositories/scenario/test_scenario_postgres_repository.py
View file

@ -142,6 +142,28 @@ def test_get_many_raises_on_an_exclusion_that_is_not_a_measure_type(
ScenarioPostgresRepository(session).get_many([7])
def test_get_many_maps_the_fabric_first_flag(db_engine: Engine) -> None:
# Arrange — a Fabric First brief created in the scenario-builder.
with Session(db_engine) as session:
session.add(
ScenarioModel(
id=7,
goal=PortfolioGoal.INCREASING_EPC,
goal_value="C",
is_default=True,
fabric_first=True,
)
)
session.commit()
# Act
with Session(db_engine) as session:
scenario: Scenario = ScenarioPostgresRepository(session).get_many([7])[0]
# Assert
assert scenario.fabric_first is True
def test_get_many_raises_when_a_scenario_id_is_missing(db_engine: Engine) -> None:
# Arrange
with Session(db_engine) as session: