diff --git a/domain/modelling/scoring.py b/domain/modelling/scoring.py new file mode 100644 index 00000000..c0558d8a --- /dev/null +++ b/domain/modelling/scoring.py @@ -0,0 +1,91 @@ +"""Per-measure scoring — the telescoping marginal cascade (ADR-0016). + +`marginal_impacts` applies overlays one at a time in the given order and +reports each measure's marginal contribution. It serves two of the three +scoring roles: + - role 1 (per-Option optimiser signal): call per Option as a 1-element + sequence -> its independent-vs-baseline impact; + - role 3 (final-package display attribution): call once with the selected + overlays in best-practice order -> per-measure impacts that telescope + exactly to the whole-package total. + +Per-Option (role 1) figures are an approximate signal and must not be surfaced +as a measure's true impact — only the final-package cascade (role 3) is +truthful. The whole-package re-score (role 2) is `PackageScorer.score` directly. +""" + +from dataclasses import dataclass +from typing import Sequence + +from datatypes.epc.domain.epc_property_data import EpcPropertyData +from domain.modelling.package_scorer import PackageScorer, Score +from domain.modelling.recommendation import MeasureOption +from domain.modelling.simulation import EpcSimulation + + +@dataclass(frozen=True) +class MeasureImpact: + """One measure's marginal contribution, signed so positive is always an + improvement: `sap_points` is the SAP gain; the savings are reductions + (baseline-at-this-step minus the new value).""" + + sap_points: float + co2_savings_kg_per_yr: float + energy_savings_kwh_per_yr: float + + +def marginal_impacts( + scorer: PackageScorer, + baseline: EpcPropertyData, + overlays: Sequence[EpcSimulation], +) -> list[MeasureImpact]: + """Apply overlays cumulatively in order; return each one's marginal impact + over the running state. The marginals telescope to the whole-package total.""" + impacts: list[MeasureImpact] = [] + previous: Score = scorer.score(baseline, []) + for index in range(len(overlays)): + current: Score = scorer.score(baseline, list(overlays[: index + 1])) + impacts.append( + MeasureImpact( + sap_points=current.sap_continuous - previous.sap_continuous, + co2_savings_kg_per_yr=previous.co2_kg_per_yr - current.co2_kg_per_yr, + energy_savings_kwh_per_yr=( + previous.primary_energy_kwh_per_yr + - current.primary_energy_kwh_per_yr + ), + ) + ) + previous = current + return impacts + + +def independent_option_impacts( + 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] = [] + for option in options: + cached = next( + (impact for overlay, impact 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 + ), + ) + scored.append((option.overlay, cached)) + impacts.append(cached) + return impacts diff --git a/tests/domain/modelling/test_scoring.py b/tests/domain/modelling/test_scoring.py new file mode 100644 index 00000000..af286018 --- /dev/null +++ b/tests/domain/modelling/test_scoring.py @@ -0,0 +1,142 @@ +"""Behaviour of per-measure scoring: the telescoping marginal cascade that +serves both the per-Option optimiser signal (role 1) and the final-package +display attribution (role 3) — ADR-0016. Exercises the real calculator on a +hand-built EPD; no PDF/parser involved. +""" + +from typing import Sequence + +from datatypes.epc.domain.epc_property_data import ( + BuildingPartIdentifier, + EpcPropertyData, +) +from domain.modelling.package_scorer import PackageScorer, Score +from domain.modelling.recommendation import MeasureOption +from domain.modelling.scoring import ( + MeasureImpact, + independent_option_impacts, + marginal_impacts, +) +from domain.modelling.simulation import BuildingPartOverlay, EpcSimulation +from domain.sap10_calculator.calculator import Sap10Calculator +from tests.domain.sap10_calculator.worksheet._elmhurst_worksheet_000490 import ( + build_epc, +) + +_MAIN_CAVITY = EpcSimulation( + building_parts={ + BuildingPartIdentifier.MAIN: BuildingPartOverlay(wall_insulation_type=2) + } +) +_EXT1_CAVITY = EpcSimulation( + building_parts={ + BuildingPartIdentifier.EXTENSION_1: BuildingPartOverlay(wall_insulation_type=2) + } +) + + +class _CountingScorer(PackageScorer): + """A PackageScorer stand-in that counts score() calls; the score is a + deterministic function of the overlays so distinct overlays differ.""" + + def __init__(self) -> None: + self.calls = 0 + + def score( + self, baseline: EpcPropertyData, simulations: Sequence[EpcSimulation] + ) -> Score: + self.calls += 1 + total = 0.0 + for sim in simulations: + for overlay in sim.building_parts.values(): + total += overlay.wall_insulation_type or 0 + return Score( + sap_continuous=total, co2_kg_per_yr=0.0, primary_energy_kwh_per_yr=0.0 + ) + + +def _option(overlay: EpcSimulation) -> MeasureOption: + return MeasureOption( + measure_type="cavity_wall_insulation", description="opt", overlay=overlay + ) + + +def test_independent_option_impacts_score_each_distinct_overlay_once() -> None: + # Arrange + baseline: EpcPropertyData = build_epc() + scorer = _CountingScorer() + overlay_a = EpcSimulation( + building_parts={ + BuildingPartIdentifier.MAIN: BuildingPartOverlay(wall_insulation_type=2) + } + ) + overlay_a_dup = EpcSimulation( + building_parts={ + BuildingPartIdentifier.MAIN: BuildingPartOverlay(wall_insulation_type=2) + } + ) + overlay_b = EpcSimulation( + building_parts={ + BuildingPartIdentifier.MAIN: BuildingPartOverlay(wall_insulation_type=3) + } + ) + options = [_option(overlay_a), _option(overlay_a_dup), _option(overlay_b)] + + # Act + impacts: list[MeasureImpact] = independent_option_impacts( + scorer, baseline, options + ) + + # 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 + + +def test_single_overlay_marginal_is_its_improvement_over_baseline() -> None: + # Arrange + baseline: EpcPropertyData = build_epc() + scorer = PackageScorer(Sap10Calculator()) + base: Score = scorer.score(baseline, []) + filled: Score = scorer.score(baseline, [_MAIN_CAVITY]) + + # Act + impacts: list[MeasureImpact] = marginal_impacts(scorer, baseline, [_MAIN_CAVITY]) + + # Assert + assert len(impacts) == 1 + assert impacts[0].sap_points > 0 # cavity fill improves SAP + assert ( + abs(impacts[0].sap_points - (filled.sap_continuous - base.sap_continuous)) + <= 1e-9 + ) + + +def test_marginals_telescope_to_the_whole_package_total() -> None: + # Arrange + baseline: EpcPropertyData = build_epc() + scorer = PackageScorer(Sap10Calculator()) + overlays = [_MAIN_CAVITY, _EXT1_CAVITY] + base: Score = scorer.score(baseline, []) + full: Score = scorer.score(baseline, overlays) + + # Act + impacts: list[MeasureImpact] = marginal_impacts(scorer, baseline, overlays) + + # Assert + assert len(impacts) == 2 + assert ( + abs( + sum(i.sap_points for i in impacts) + - (full.sap_continuous - base.sap_continuous) + ) + <= 1e-9 + ) + assert ( + abs( + sum(i.energy_savings_kwh_per_yr for i in impacts) + - (base.primary_energy_kwh_per_yr - full.primary_energy_kwh_per_yr) + ) + <= 1e-6 + )