Model/domain/modelling/scoring/scoring.py

"""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 Callable, Sequence

from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.modelling.scoring.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 cascade_scores(
    scorer: PackageScorer,
    baseline: EpcPropertyData,
    overlays: Sequence[EpcSimulation],
) -> list[Score]:
    """Score the cumulative prefixes of `overlays` in order: index 0 is the
    baseline (empty prefix), index i the state after the first i overlays. The
    list has `len(overlays) + 1` entries — one calculator run each.

    Each Score carries its `SapResult`, so the same cascade powers both the
    role-3 marginal attribution (`marginals_from_scores`) and the telescoping
    per-measure bill cascade — neither needs to re-score (ADR-0014 / ADR-0016)."""
    return [
        scorer.score(baseline, list(overlays[:prefix_length]))
        for prefix_length in range(len(overlays) + 1)
    ]


def marginals_from_scores(scores: Sequence[Score]) -> list[MeasureImpact]:
    """Each measure's marginal impact from a precomputed cumulative-prefix
    cascade (`scores[0]` is the baseline). Signed so positive is an improvement;
    the marginals telescope to `scores[-1]` vs `scores[0]`."""
    impacts: list[MeasureImpact] = []
    for index in range(1, len(scores)):
        previous: Score = scores[index - 1]
        current: Score = scores[index]
        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
                ),
            )
        )
    return impacts


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."""
    return marginals_from_scores(cascade_scores(scorer, baseline, overlays))


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


def independent_option_signals(
    scorer: PackageScorer,
    baseline: EpcPropertyData,
    options: Sequence[MeasureOption],
    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: float | None = next(
            (signal for overlay, signal in scored if overlay == option.overlay),
            None,
        )
        if cached is None:
            cached = objective(scorer.score(baseline, [option.overlay])) - base_value
            scored.append((option.overlay, cached))
        signals.append(cached)
    return signals