junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-08 11:17:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(baseline): CalculatorRebaseliner — calculator goes load-bearing (ADR-0013 amend)

Browse source 
Slice 5a: the promotion. Replaces StubRebaseliner in production and collapses the
shadow runner into the rebaseliner (ADR-0013 amendment).

- CalculatorRebaseliner runs Sap10Calculator on every Property:
  * sap_version < 10.2 -> Effective Performance IS the calculator output
    (band via Epc.from_sap_score, CO2 kg->t, PEUI rounded), reason "pre_sap10".
  * sap_version >= 10.2 -> Effective = lodged (API figures on-target), and the
    calculator only logs divergence (SAP>0.5, PEUI/CO2 1%) as a validation signal.
  * a calculator raise propagates -> batch aborts (ADR-0012); fix the cert at once.
- Rebaseliner.rebaseline gains property_id (for the divergence log).
- LoggingCalculatorShadow / the calculator_shadow seam removed from the
  orchestrator; its divergence-comparison logic now lives in the rebaseliner.
- StubRebaseliner kept (signature updated) for orchestrator/repo unit tests.

The SapResult->EnergyBreakdown adapter + BillDerivation wiring (to populate the
bill block) follow once the appliances/cooking SapResult fields land.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-06-02 10:04:24 +00:00
parent 5f65b9be62
commit 15da2d3970
11 changed files with 262 additions and 384 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
applications/ara_first_run/handler.py
View file

@ -10,8 +10,7 @@ from sqlmodel import Session
from applications.ara_first_run.ara_first_run_trigger_body import (
AraFirstRunTriggerBody,
)
from domain.property_baseline.calculator_shadow import LoggingCalculatorShadow
from domain.property_baseline.rebaseliner import StubRebaseliner
from domain.property_baseline.calculator_rebaseliner import CalculatorRebaseliner
from domain.sap10_calculator.calculator import Sap10Calculator
from infrastructure.postgres.config import PostgresConfig
from infrastructure.postgres.engine import make_engine
@ -82,10 +81,10 @@ def build_first_run_pipeline(
),
baseline=PropertyBaselineOrchestrator(
unit_of_work=unit_of_work,
rebaseliner=StubRebaseliner(),
# Shadow only: validates the calculator over the wild cohort without
# gating the load-bearing baseline write (ADR-0013).
calculator_shadow=LoggingCalculatorShadow(Sap10Calculator()),
# The calculator is load-bearing: effective=calculated for pre-10.2
# certs, lodged + divergence-logged at/above 10.2; a raise aborts the
# batch (ADR-0013 amendment).
rebaseliner=CalculatorRebaseliner(Sap10Calculator()),
),
modelling=ModellingOrchestrator(
scenario_repo=ScenarioRepository(),

113
domain/property_baseline/calculator_rebaseliner.py Normal file
View file

@ -0,0 +1,113 @@
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Optional, Protocol
from datatypes.epc.domain.epc import Epc
from domain.property_baseline.performance import Performance
from domain.property_baseline.rebaseliner import Rebaseliner, RebaselineReason
if TYPE_CHECKING:
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.sap10_calculator.calculator import SapResult
logger = logging.getLogger(__name__)
# The calculator targets SAP 10.2 (14-03-2025). A cert lodged below this carries
# a superseded methodology and is rebaselined to the calculator's output; at or
# above it, the API's lodged figures are kept and the calculator only validates.
_SAP10_2_FLOOR = 10.2
_SAP_ABS_TOL = 0.5
_REL_TOL = 0.01
_KG_PER_TONNE = 1000.0
class Calculator(Protocol):
"""The slice of `Sap10Calculator` the rebaseliner needs — `Sap10Calculator`
satisfies it structurally, so this module does not import the calculator."""
def calculate(self, epc: "EpcPropertyData") -> "SapResult": ...
def performance_from_sap_result(result: "SapResult") -> Performance:
"""The four rated quantities, read off a `SapResult`: band derived from the
score, CO2 converted kg→tonnes, PEUI rounded to the lodged integer scale."""
return Performance(
sap_score=result.sap_score,
epc_band=Epc.from_sap_score(result.sap_score),
co2_emissions=result.co2_kg_per_yr / _KG_PER_TONNE,
primary_energy_intensity=round(result.primary_energy_kwh_per_m2),
)
def _relative_diff(calculated: float, lodged: float) -> float:
if lodged == 0:
return 0.0 if calculated == 0 else float("inf")
return abs(calculated - lodged) / abs(lodged)
class CalculatorRebaseliner(Rebaseliner):
"""Produces Effective Performance from the deterministic `Sap10Calculator`
(ADR-0013 amendment — the calculator is load-bearing).
Runs the calculator on every Property. For a cert lodged under a superseded
methodology (``sap_version < 10.2``) the calculator's output **is** Effective
Performance. At or above 10.2 the API's lodged figures are kept and the
calculator only **logs divergence** (a validation signal). A calculator
strict-raise propagates — the batch aborts (ADR-0012) and the un-mapped cert
is fixed immediately.
"""
def __init__(self, calculator: Calculator) -> None:
self._calculator = calculator
def rebaseline(
self, property_id: int, effective_epc: "EpcPropertyData", lodged: Performance
) -> tuple[Performance, RebaselineReason]:
# A raise (UnmappedSapCode, etc.) propagates: the calculator is
# load-bearing, so the batch aborts and the cert is fixed at once.
result = self._calculator.calculate(effective_epc)
sap_version = effective_epc.sap_version
if sap_version is not None and sap_version < _SAP10_2_FLOOR:
return performance_from_sap_result(result), "pre_sap10"
self._log_divergence(
property_id=property_id, sap_version=sap_version, result=result, lodged=lodged
)
return lodged, "none"
def _log_divergence(
self,
*,
property_id: int,
sap_version: Optional[float],
result: "SapResult",
lodged: Performance,
) -> None:
if abs(result.sap_score_continuous - lodged.sap_score) > _SAP_ABS_TOL:
self._warn(property_id, sap_version, "sap_score", lodged.sap_score, result.sap_score_continuous)
if _relative_diff(result.primary_energy_kwh_per_m2, lodged.primary_energy_intensity) > _REL_TOL:
self._warn(
property_id, sap_version, "primary_energy_intensity",
lodged.primary_energy_intensity, result.primary_energy_kwh_per_m2,
)
calculated_co2_t = result.co2_kg_per_yr / _KG_PER_TONNE
if _relative_diff(calculated_co2_t, lodged.co2_emissions) > _REL_TOL:
self._warn(property_id, sap_version, "co2_emissions", lodged.co2_emissions, calculated_co2_t)
def _warn(
self,
property_id: int,
sap_version: Optional[float],
quantity: str,
lodged: float,
calculated: float,
) -> None:
logger.warning(
"SAP10 calculator divergence on %s for property_id=%s sap_version=%s: "
"lodged=%s calculated=%s",
quantity,
property_id,
sap_version,
lodged,
calculated,
)

141
domain/property_baseline/calculator_shadow.py
View file

@ -1,141 +0,0 @@
from __future__ import annotations
import logging
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Optional, Protocol
from domain.property_baseline.performance import Performance
if TYPE_CHECKING:
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.sap10_calculator.calculator import SapResult
logger = logging.getLogger(__name__)
# A continuous SAP this far from the lodged integer would round to a different
# band-driving score; PEUI / CO2 scale with dwelling size so they use a relative
# tolerance (ADR-0013). Starting dials — tune against the wild-cohort logs.
_SAP_ABS_TOL = 0.5
_REL_TOL = 0.01
_KG_PER_TONNE = 1000.0
class CalculatorShadow(ABC):
"""Runs SAP10 Calculation in shadow beside the load-bearing baseline write
and reports divergence from Lodged Performance (ADR-0013).
The calculator is not yet load-bearing — it is still being hardened, and a
large test cohort is about to flow through baselining. So an implementation
**must never raise**: a shadow failure may not abort the batch (ADR-0012's
all-or-nothing governs only the load-bearing Lodged/Effective write). It
observes, compares against Lodged, and logs; it does not feed Effective
Performance. The seam is retired when the calculator is promoted to the
Rebaseliner and its output *becomes* Effective Performance.
"""
@abstractmethod
def observe(
self,
*,
property_id: int,
effective_epc: "EpcPropertyData",
lodged: Performance,
) -> None: ...
def _relative_diff(calculated: float, lodged: float) -> float:
"""|calculated − lodged| / |lodged|; a zero lodged value diverges iff
calculated is non-zero (avoids a divide-by-zero on degenerate certs)."""
if lodged == 0:
return 0.0 if calculated == 0 else float("inf")
return abs(calculated - lodged) / abs(lodged)
class Calculator(Protocol):
"""The slice of `Sap10Calculator` the shadow needs: cert in, result out.
`Sap10Calculator` satisfies it structurally — no coupling to its module."""
def calculate(self, epc: "EpcPropertyData") -> "SapResult": ...
class LoggingCalculatorShadow(CalculatorShadow):
"""Runs the calculator and logs, never persists, never raises (ADR-0013).
A strict-raise (an un-mapped cert) is caught and logged at ``error`` so the
wild-cohort gap is greppable; a successful result whose SAP / PEUI / CO2
diverges from Lodged beyond tolerance is logged at ``warning``. Every line
is tagged with ``property_id`` and the cert's ``sap_version`` so SAP-10.2
divergence (a real calculator signal) is separable from older-spec drift.
"""
def __init__(self, calculator: Calculator) -> None:
self._calculator = calculator
def observe(
self,
*,
property_id: int,
effective_epc: "EpcPropertyData",
lodged: Performance,
) -> None:
sap_version = effective_epc.sap_version
try:
# Broad by design: the point is to discover *what* breaks in the
# wild, and a shadow failure must never abort the batch (ADR-0013).
result = self._calculator.calculate(effective_epc)
except Exception as exc:
logger.error(
"SAP10 shadow calculation failed for property_id=%s "
"sap_version=%s: %r",
property_id,
sap_version,
exc,
)
return
if abs(result.sap_score_continuous - lodged.sap_score) > _SAP_ABS_TOL:
self._warn_divergence(
quantity="sap_score",
property_id=property_id,
sap_version=sap_version,
lodged=lodged.sap_score,
calculated=result.sap_score_continuous,
)
if _relative_diff(
result.primary_energy_kwh_per_m2, lodged.primary_energy_intensity
) > _REL_TOL:
self._warn_divergence(
quantity="primary_energy_intensity",
property_id=property_id,
sap_version=sap_version,
lodged=lodged.primary_energy_intensity,
calculated=result.primary_energy_kwh_per_m2,
)
# Lodged CO2 is tonnes/yr; the calculator emits kg/yr (ADR-0013).
calculated_co2_t = result.co2_kg_per_yr / _KG_PER_TONNE
if _relative_diff(calculated_co2_t, lodged.co2_emissions) > _REL_TOL:
self._warn_divergence(
quantity="co2_emissions",
property_id=property_id,
sap_version=sap_version,
lodged=lodged.co2_emissions,
calculated=calculated_co2_t,
)
def _warn_divergence(
self,
*,
quantity: str,
property_id: int,
sap_version: Optional[float],
lodged: float,
calculated: float,
) -> None:
logger.warning(
"SAP10 shadow divergence on %s for property_id=%s sap_version=%s: "
"lodged=%s calculated=%s",
quantity,
property_id,
sap_version,
lodged,
calculated,
)

12
domain/property_baseline/rebaseliner.py
View file

@ -36,20 +36,22 @@ class Rebaseliner(ABC):
@abstractmethod
def rebaseline(
self, effective_epc: EpcPropertyData, lodged: Performance
self, property_id: int, effective_epc: EpcPropertyData, lodged: Performance
) -> tuple[Performance, RebaselineReason]: ...
class StubRebaseliner(Rebaseliner):
"""The no-ML stub for the validation phase.
"""A no-calculator stub for tests that don't want the real calculator.
SAP10 certs pass through untouched — Effective Performance equals Lodged,
reason ``"none"``. A pre-SAP10 cert genuinely needs ML rebaselining, which is
not implemented yet (#1135), so it raises rather than fabricating a "none".
reason ``"none"``. A pre-SAP10 cert genuinely needs rebaselining, which this
stub does not do, so it raises rather than fabricating a "none". Production
uses ``CalculatorRebaseliner`` (the calculator is load-bearing — ADR-0013
amendment); this stub stays for orchestrator/repo unit tests.
"""
def rebaseline(
self, effective_epc: EpcPropertyData, lodged: Performance
self, property_id: int, effective_epc: EpcPropertyData, lodged: Performance
) -> tuple[Performance, RebaselineReason]:
sap_version = effective_epc.sap_version
if sap_version is not None and sap_version < _SAP10_FLOOR:

13
orchestration/property_baseline_orchestrator.py
View file

@ -6,7 +6,6 @@ from datatypes.epc.domain.epc_property_data import (
EpcPropertyData,
RenewableHeatIncentive,
)
from domain.property_baseline.calculator_shadow import CalculatorShadow
from domain.property_baseline.property_baseline_performance import PropertyBaselinePerformance
from domain.property_baseline.performance import lodged_performance
from domain.property_baseline.rebaseliner import Rebaseliner
@ -33,11 +32,9 @@ class PropertyBaselineOrchestrator:
*,
unit_of_work: Callable[[], UnitOfWork],
rebaseliner: Rebaseliner,
calculator_shadow: CalculatorShadow,
) -> None:
self._unit_of_work = unit_of_work
self._rebaseliner = rebaseliner
self._calculator_shadow = calculator_shadow
def run(self, property_ids: list[int]) -> None:
with self._unit_of_work() as uow:
@ -46,7 +43,7 @@ class PropertyBaselineOrchestrator:
effective_epc = prop.effective_epc
lodged = lodged_performance(effective_epc)
effective, reason = self._rebaseliner.rebaseline(
effective_epc, lodged
property_id, effective_epc, lodged
)
rhi = _require_rhi(effective_epc)
baseline = PropertyBaselinePerformance(
@ -57,14 +54,6 @@ class PropertyBaselineOrchestrator:
water_heating_kwh=rhi.water_heating_kwh,
)
uow.property_baseline.save(baseline, property_id)
# Shadow only: validate the calculator in the wild without
# gating the load-bearing write above (ADR-0013). `observe`
# never raises, so it cannot abort the batch.
self._calculator_shadow.observe(
property_id=property_id,
effective_epc=effective_epc,
lodged=lodged,
)
uow.commit()

134
tests/domain/property_baseline/test_calculator_rebaseliner.py Normal file
View file

@ -0,0 +1,134 @@
from __future__ import annotations
import logging
from typing import Optional
import pytest
from datatypes.epc.domain.epc import Epc
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.property_baseline.calculator_rebaseliner import CalculatorRebaseliner
from domain.property_baseline.performance import Performance
from domain.sap10_calculator.calculator import SapResult
from domain.sap10_calculator.exceptions import UnmappedSapCode
def _epc(*, sap_version: Optional[float]) -> EpcPropertyData:
epc = object.__new__(EpcPropertyData)
epc.sap_version = sap_version
return epc
def _lodged() -> Performance:
return Performance(
sap_score=72, epc_band=Epc.C, co2_emissions=1.8, primary_energy_intensity=180
)
def _sap_result(
*,
sap_score: int = 72,
co2_kg_per_yr: float = 1800.0,
primary_energy_kwh_per_m2: float = 180.0,
) -> SapResult:
return SapResult(
sap_score=sap_score,
sap_score_continuous=float(sap_score),
ecf=0.0,
total_fuel_cost_gbp=0.0,
co2_kg_per_yr=co2_kg_per_yr,
space_heating_kwh_per_yr=0.0,
space_cooling_kwh_per_yr=0.0,
fabric_energy_efficiency_kwh_per_m2_yr=0.0,
main_heating_fuel_kwh_per_yr=0.0,
main_2_heating_fuel_kwh_per_yr=0.0,
secondary_heating_fuel_kwh_per_yr=0.0,
space_cooling_fuel_kwh_per_yr=0.0,
hot_water_kwh_per_yr=0.0,
pumps_fans_kwh_per_yr=0.0,
lighting_kwh_per_yr=0.0,
primary_energy_kwh_per_yr=0.0,
primary_energy_kwh_per_m2=primary_energy_kwh_per_m2,
monthly=(),
intermediate={},
)
class _StubCalculator:
def __init__(self, result: SapResult) -> None:
self._result = result
def calculate(self, epc: EpcPropertyData) -> SapResult:
return self._result
def test_pre_10_2_cert_is_rebaselined_to_the_calculator_output() -> None:
# Arrange — a SAP 10.0 cert: lodged figures are a superseded methodology, so
# the calculator's output becomes Effective Performance (ADR-0013 amendment).
calculator = _StubCalculator(
_sap_result(sap_score=70, co2_kg_per_yr=1900.0, primary_energy_kwh_per_m2=185.4)
)
rebaseliner = CalculatorRebaseliner(calculator)
epc = _epc(sap_version=10.0)
# Act
effective, reason = rebaseliner.rebaseline(
property_id=10, effective_epc=epc, lodged=_lodged()
)
# Assert — calculated Performance: band from the score, CO2 kg->t, PEUI rounded.
assert effective == Performance(
sap_score=70, epc_band=Epc.C, co2_emissions=1.9, primary_energy_intensity=185
)
assert reason == "pre_sap10"
def test_a_10_2_cert_keeps_the_lodged_figures() -> None:
# Arrange — a SAP 10.2 cert: the API's lodged figures are on-target, so they
# stand; the calculator runs only to validate.
calculator = _StubCalculator(_sap_result(sap_score=72))
rebaseliner = CalculatorRebaseliner(calculator)
epc = _epc(sap_version=10.2)
# Act
effective, reason = rebaseliner.rebaseline(
property_id=10, effective_epc=epc, lodged=_lodged()
)
# Assert
assert effective == _lodged()
assert reason == "none"
def test_a_10_2_cert_logs_divergence_when_the_calculator_disagrees(
caplog: pytest.LogCaptureFixture,
) -> None:
# Arrange — calculated SAP 76 vs lodged 72 (> 0.5 out) on a 10.2 cert.
calculator = _StubCalculator(_sap_result(sap_score=76))
rebaseliner = CalculatorRebaseliner(calculator)
epc = _epc(sap_version=10.2)
# Act
with caplog.at_level(logging.WARNING):
rebaseliner.rebaseline(property_id=42, effective_epc=epc, lodged=_lodged())
# Assert — a divergence warning, tagged with property_id + sap_version.
assert len(caplog.records) == 1
message = caplog.records[0].getMessage()
assert "sap_score" in message
assert "property_id=42" in message
assert "sap_version=10.2" in message
def test_a_calculator_raise_propagates_and_aborts() -> None:
# Arrange — the calculator is load-bearing, so a raise is not swallowed.
class _Raising:
def calculate(self, epc: EpcPropertyData) -> SapResult:
raise UnmappedSapCode("heat_emitter_type", 99)
rebaseliner = CalculatorRebaseliner(_Raising())
epc = _epc(sap_version=10.0)
# Act / Assert
with pytest.raises(UnmappedSapCode):
rebaseliner.rebaseline(property_id=10, effective_epc=epc, lodged=_lodged())

166
tests/domain/property_baseline/test_calculator_shadow.py
View file

@ -1,166 +0,0 @@
from __future__ import annotations
import logging
from typing import Optional
import pytest
from datatypes.epc.domain.epc import Epc
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.property_baseline.calculator_shadow import LoggingCalculatorShadow
from domain.property_baseline.performance import Performance
from domain.sap10_calculator.calculator import SapResult
from domain.sap10_calculator.exceptions import UnmappedSapCode
def _epc(*, sap_version: Optional[float]) -> EpcPropertyData:
epc = object.__new__(EpcPropertyData)
epc.sap_version = sap_version
return epc
def _lodged() -> Performance:
return Performance(
sap_score=72, epc_band=Epc.C, co2_emissions=1.8, primary_energy_intensity=180
)
def _sap_result(
*,
sap_score_continuous: float = 72.0,
primary_energy_kwh_per_m2: float = 180.0,
co2_kg_per_yr: float = 1800.0,
) -> SapResult:
"""A `SapResult` whose three compared quantities default to *matching*
`_lodged()`; each test perturbs one axis."""
return SapResult(
sap_score=round(sap_score_continuous),
sap_score_continuous=sap_score_continuous,
ecf=0.0,
total_fuel_cost_gbp=0.0,
co2_kg_per_yr=co2_kg_per_yr,
space_heating_kwh_per_yr=0.0,
space_cooling_kwh_per_yr=0.0,
fabric_energy_efficiency_kwh_per_m2_yr=0.0,
main_heating_fuel_kwh_per_yr=0.0,
main_2_heating_fuel_kwh_per_yr=0.0,
secondary_heating_fuel_kwh_per_yr=0.0,
space_cooling_fuel_kwh_per_yr=0.0,
hot_water_kwh_per_yr=0.0,
pumps_fans_kwh_per_yr=0.0,
lighting_kwh_per_yr=0.0,
primary_energy_kwh_per_yr=0.0,
primary_energy_kwh_per_m2=primary_energy_kwh_per_m2,
monthly=(),
intermediate={},
)
class _RaisingCalculator:
def calculate(self, epc: EpcPropertyData) -> SapResult:
raise UnmappedSapCode("heat_emitter_type", 99)
class _StubCalculator:
def __init__(self, result: SapResult) -> None:
self._result = result
def calculate(self, epc: EpcPropertyData) -> SapResult:
return self._result
def test_observe_swallows_a_calculator_raise_and_logs_error(
caplog: pytest.LogCaptureFixture,
) -> None:
# Arrange — the calculator strict-raises on a cert it cannot yet map.
shadow = LoggingCalculatorShadow(_RaisingCalculator())
epc = _epc(sap_version=10.2)
# Act — observe must not propagate the raise (ADR-0013: shadow is not
# load-bearing, so it cannot abort the batch).
with caplog.at_level(logging.ERROR):
shadow.observe(property_id=42, effective_epc=epc, lodged=_lodged())
# Assert — exactly one error record, tagged with property_id + sap_version
# and carrying the exception so the wild-cohort gap is greppable.
assert len(caplog.records) == 1
message = caplog.records[0].getMessage()
assert caplog.records[0].levelno == logging.ERROR
assert "property_id=42" in message
assert "sap_version=10.2" in message
assert "heat_emitter_type" in message
def test_observe_warns_when_sap_diverges_beyond_half_a_point(
caplog: pytest.LogCaptureFixture,
) -> None:
# Arrange — calculated SAP 75.0 vs lodged 72 is 3.0 out (> 0.5).
shadow = LoggingCalculatorShadow(
_StubCalculator(_sap_result(sap_score_continuous=75.0))
)
epc = _epc(sap_version=10.2)
# Act
with caplog.at_level(logging.WARNING):
shadow.observe(property_id=42, effective_epc=epc, lodged=_lodged())
# Assert — one warning, naming the diverging quantity + the tags.
assert len(caplog.records) == 1
message = caplog.records[0].getMessage()
assert caplog.records[0].levelno == logging.WARNING
assert "sap_score" in message
assert "property_id=42" in message
assert "sap_version=10.2" in message
def test_observe_warns_when_peui_diverges_beyond_one_percent(
caplog: pytest.LogCaptureFixture,
) -> None:
# Arrange — calculated PEUI 200 vs lodged 180 is ~11% out (> 1%).
shadow = LoggingCalculatorShadow(
_StubCalculator(_sap_result(primary_energy_kwh_per_m2=200.0))
)
epc = _epc(sap_version=10.2)
# Act
with caplog.at_level(logging.WARNING):
shadow.observe(property_id=42, effective_epc=epc, lodged=_lodged())
# Assert
assert len(caplog.records) == 1
assert "primary_energy_intensity" in caplog.records[0].getMessage()
def test_observe_warns_when_co2_diverges_beyond_one_percent_after_kg_to_tonnes(
caplog: pytest.LogCaptureFixture,
) -> None:
# Arrange — calculator emits kg/yr; 2000 kg = 2.0 t vs lodged 1.8 t (~11%).
shadow = LoggingCalculatorShadow(
_StubCalculator(_sap_result(co2_kg_per_yr=2000.0))
)
epc = _epc(sap_version=10.2)
# Act
with caplog.at_level(logging.WARNING):
shadow.observe(property_id=42, effective_epc=epc, lodged=_lodged())
# Assert — the kg→tonnes conversion is applied before comparison, so a
# matching 1800 kg would *not* fire (guarded by the silent-when-aligned test).
assert len(caplog.records) == 1
assert "co2_emissions" in caplog.records[0].getMessage()
def test_observe_is_silent_when_the_calculator_agrees_with_lodged(
caplog: pytest.LogCaptureFixture,
) -> None:
# Arrange — all three quantities at the matching defaults (SAP 72, PEUI 180,
# 1800 kg ≡ 1.8 t): nothing should be logged.
shadow = LoggingCalculatorShadow(_StubCalculator(_sap_result()))
epc = _epc(sap_version=10.2)
# Act
with caplog.at_level(logging.WARNING):
shadow.observe(property_id=42, effective_epc=epc, lodged=_lodged())
# Assert
assert caplog.records == []

4
tests/domain/property_baseline/test_rebaseliner.py
View file

@ -29,7 +29,7 @@ def test_sap10_epc_is_not_rebaselined_so_effective_equals_lodged() -> None:
rebaseliner = StubRebaseliner()
# Act
effective, reason = rebaseliner.rebaseline(epc, lodged)
effective, reason = rebaseliner.rebaseline(10, epc, lodged)
# Assert — Effective Performance equals Lodged, reason "none".
assert effective == lodged
@ -45,4 +45,4 @@ def test_pre_sap10_epc_raises_because_rebaselining_is_not_implemented() -> None:
# Act / Assert
with pytest.raises(RebaselineNotImplemented):
rebaseliner.rebaseline(epc, _lodged())
rebaseliner.rebaseline(10, epc, _lodged())

19
tests/orchestration/fakes.py
View file

@ -10,8 +10,6 @@ from types import TracebackType
from typing import Any, Optional
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.property_baseline.calculator_shadow import CalculatorShadow
from domain.property_baseline.performance import Performance
from domain.property_baseline.property_baseline_performance import PropertyBaselinePerformance
from domain.property.properties import Properties
from domain.property.property import Property
@ -90,23 +88,6 @@ class FakePropertyBaselineRepo(PropertyBaselineRepository):
raise NotImplementedError
class FakeCalculatorShadow(CalculatorShadow):
"""Records each `observe` call so a test can assert the orchestrator runs
the shadow per property without dragging in the real calculator."""
def __init__(self) -> None:
self.observed: list[tuple[int, EpcPropertyData, Performance]] = []
def observe(
self,
*,
property_id: int,
effective_epc: EpcPropertyData,
lodged: Performance,
) -> None:
self.observed.append((property_id, effective_epc, lodged))
class FakeUnitOfWork(UnitOfWork):
"""A unit that holds in-memory repos and counts commits."""

2
tests/orchestration/test_ara_first_run_pipeline_integration.py
View file

@ -36,7 +36,6 @@ from repositories.geospatial.geospatial_repository import GeospatialRepository
from repositories.materials.materials_repository import MaterialsRepository
from repositories.postgres_unit_of_work import PostgresUnitOfWork
from repositories.scenario.scenario_repository import ScenarioRepository
from tests.orchestration.fakes import FakeCalculatorShadow
_JSON_SAMPLES = Path(__file__).resolve().parents[2] / "backend/epc_api/json_samples"
@ -114,7 +113,6 @@ def test_first_run_baselines_through_repos_and_is_idempotent_on_rerun(
baseline=PropertyBaselineOrchestrator(
unit_of_work=unit_of_work,
rebaseliner=StubRebaseliner(),
calculator_shadow=FakeCalculatorShadow(),
),
modelling=ModellingOrchestrator(
scenario_repo=ScenarioRepository(),

31
tests/orchestration/test_property_baseline_orchestrator.py
View file

@ -13,7 +13,6 @@ from domain.property_baseline.rebaseliner import RebaselineNotImplemented, StubR
from domain.property.property import Property, PropertyIdentity
from orchestration.property_baseline_orchestrator import PropertyBaselineOrchestrator
from tests.orchestration.fakes import (
FakeCalculatorShadow,
FakePropertyBaselineRepo,
FakePropertyRepo,
FakeUnitOfWork,
@ -38,34 +37,6 @@ def _property(*, sap_version: float) -> Property:
)
def test_run_invokes_the_calculator_shadow_per_property_and_still_persists() -> None:
# Arrange
property_baseline_repo = FakePropertyBaselineRepo()
shadow = FakeCalculatorShadow()
prop = _property(sap_version=10.2)
uow = FakeUnitOfWork(
property=FakePropertyRepo({10: prop}),
property_baseline=property_baseline_repo,
)
orchestrator = PropertyBaselineOrchestrator(
unit_of_work=lambda: uow,
rebaseliner=StubRebaseliner(),
calculator_shadow=shadow,
)
# Act
orchestrator.run([10])
# Assert — the load-bearing write + single commit are unchanged, and the
# shadow observed the Effective EPC + Lodged Performance once (ADR-0013).
lodged = Performance(
sap_score=72, epc_band=Epc.C, co2_emissions=1.8, primary_energy_intensity=180
)
assert len(property_baseline_repo.saved) == 1
assert uow.commits == 1
assert shadow.observed == [(10, prop.effective_epc, lodged)]
def test_run_establishes_persists_and_commits_the_batch_once() -> None:
# Arrange
property_baseline_repo = FakePropertyBaselineRepo()
@ -76,7 +47,6 @@ def test_run_establishes_persists_and_commits_the_batch_once() -> None:
orchestrator = PropertyBaselineOrchestrator(
unit_of_work=lambda: uow,
rebaseliner=StubRebaseliner(),
calculator_shadow=FakeCalculatorShadow(),
)
# Act
@ -112,7 +82,6 @@ def test_run_raises_on_a_pre_sap10_property_and_does_not_commit() -> None:
orchestrator = PropertyBaselineOrchestrator(
unit_of_work=lambda: uow,
rebaseliner=StubRebaseliner(),
calculator_shadow=FakeCalculatorShadow(),
)
# Act / Assert — the raise propagates; the batch is neither persisted nor