junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-30 13:10:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'feature/bill-derivation' into feature/junte+khalim

Browse source
This commit is contained in:
Jun-te Kim 2026-06-09 10:06:40 +00:00
commit 06cb4f7b6e
29 changed files with 579 additions and 104 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

42
domain/modelling/considered_measures.py Normal file
View file

@ -0,0 +1,42 @@
"""Restricting a modelling run to a chosen set of measure types.
The allowlist a run "considers" — mirroring the legacy engine's `inclusions`
(`backend/app/plan/schemas.py`). It filters the candidate Recommendations at the
Option level so a multi-option Recommendation (e.g. Heating & Hot Water competing
HHRSH against an ASHP bundle) is kept with only its allowed Options; a
Recommendation left with no allowed Option is dropped. The Optimiser still
freely chooses among what survives — including choosing nothing.
A `None` allowlist means "consider every modelled measure" (the unrestricted
default).
"""
from __future__ import annotations
from collections.abc import Iterable
from typing import Optional
from domain.modelling.measure_type import MeasureType
from domain.modelling.recommendation import Recommendation
def restrict_to_considered_measures(
recommendations: Iterable[Recommendation],
considered_measures: Optional[frozenset[MeasureType]],
) -> list[Recommendation]:
"""Keep only the Options whose measure type is in ``considered_measures``,
dropping any Recommendation left with none. ``None`` keeps everything."""
if considered_measures is None:
return list(recommendations)
restricted: list[Recommendation] = []
for recommendation in recommendations:
kept = tuple(
option
for option in recommendation.options
if option.measure_type in considered_measures
)
if kept:
restricted.append(
Recommendation(surface=recommendation.surface, options=kept)
)
return restricted

7
domain/modelling/generators/floor_recommendation.py
View file

@ -15,6 +15,7 @@ from datatypes.epc.domain.epc_property_data import (
SapBuildingPart,
)
from domain.building_geometry import ground_floor_area
from domain.modelling.measure_type import MeasureType
from domain.modelling.recommendation import Cost, MeasureOption, Recommendation
from domain.modelling.simulation import BuildingPartOverlay, EpcSimulation
from repositories.product.product_repository import ProductRepository
@ -39,14 +40,14 @@ def _is_uninsulated(thickness: Optional[Union[str, int]]) -> bool:
return thickness.strip() in ("", "0")
def _floor_measure_type(construction_type: Optional[str]) -> Optional[str]:
def _floor_measure_type(construction_type: Optional[str]) -> Optional[MeasureType]:
"""Map the lodged floor construction to the insulation Measure Type, or
None when the construction is not a treatable suspended/solid floor."""
text = (construction_type or "").lower()
if "suspended" in text:
return "suspended_floor_insulation"
return MeasureType.SUSPENDED_FLOOR_INSULATION
if "solid" in text:
return "solid_floor_insulation"
return MeasureType.SOLID_FLOOR_INSULATION
return None

7
domain/modelling/generators/glazing_recommendation.py
View file

@ -23,6 +23,7 @@ from typing import Final, Optional
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.geospatial.planning_restrictions import PlanningRestrictions
from domain.modelling.measure_type import MeasureType
from domain.modelling.recommendation import Cost, MeasureOption, Recommendation
from domain.modelling.simulation import EpcSimulation, WindowOverlay
from repositories.product.product_repository import ProductRepository
@ -42,7 +43,7 @@ class _GlazingTarget:
`solar_transmittance` — SAP10.2 Table U2 code, then heat-loss U and solar g).
"""
measure_type: str
measure_type: MeasureType
description: str
glazing_type: int
u_value: float
@ -52,7 +53,7 @@ class _GlazingTarget:
# Unrestricted: replace the units with double glazing (gt=5 "Double post 2022";
# U 4.80→1.40, g 0.85→0.72).
_DOUBLE: Final[_GlazingTarget] = _GlazingTarget(
measure_type="double_glazing",
measure_type=MeasureType.DOUBLE_GLAZING,
description="Replace the single-glazed windows with double glazing",
glazing_type=5,
u_value=1.40,
@ -64,7 +65,7 @@ _DOUBLE: Final[_GlazingTarget] = _GlazingTarget(
# solar gain). The external units can't be replaced on a protected/over-looked
# building, so this is the planning-picked Measure.
_SECONDARY: Final[_GlazingTarget] = _GlazingTarget(
measure_type="secondary_glazing",
measure_type=MeasureType.SECONDARY_GLAZING,
description="Fit secondary glazing to the single-glazed windows",
glazing_type=11,
u_value=2.90,

5
domain/modelling/generators/heating_recommendation.py
View file

@ -18,14 +18,15 @@ from datatypes.epc.domain.epc_property_data import EpcPropertyData, MainHeatingD
from datatypes.epc.domain.field_mappings import PROPERTY_TYPE_LOOKUP
from domain.geospatial.planning_restrictions import PlanningRestrictions
from domain.modelling.products import AshpCostInputs, AshpExistingSystem, Products
from domain.modelling.measure_type import MeasureType
from domain.modelling.recommendation import Cost, MeasureOption, Recommendation
from domain.modelling.simulation import EpcSimulation, HeatingOverlay
from repositories.product.product_repository import ProductRepository
_HEATING_SURFACE = "Heating & Hot Water"
_HHR_STORAGE_MEASURE_TYPE = "high_heat_retention_storage_heaters"
_ASHP_MEASURE_TYPE = "air_source_heat_pump"
_HHR_STORAGE_MEASURE_TYPE = MeasureType.HIGH_HEAT_RETENTION_STORAGE_HEATERS
_ASHP_MEASURE_TYPE = MeasureType.AIR_SOURCE_HEAT_PUMP
# Electricity main-fuel code (Elmhurst → SAP10 Table 12).
_ELECTRICITY_FUEL = 30

3
domain/modelling/generators/lighting_recommendation.py
View file

@ -18,11 +18,12 @@ scoring (ADR-0016).
from typing import Final, Optional
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.modelling.measure_type import MeasureType
from domain.modelling.recommendation import Cost, MeasureOption, Recommendation
from domain.modelling.simulation import EpcSimulation, LightingOverlay
from repositories.product.product_repository import ProductRepository
_LIGHTING_MEASURE_TYPE: Final[str] = "low_energy_lighting"
_LIGHTING_MEASURE_TYPE: Final[MeasureType] = MeasureType.LOW_ENERGY_LIGHTING
def recommend_lighting(

9
domain/modelling/generators/roof_recommendation.py
View file

@ -17,12 +17,13 @@ from datatypes.epc.domain.epc_property_data import (
EpcPropertyData,
)
from domain.building_geometry import roof_area
from domain.modelling.measure_type import MeasureType
from domain.modelling.recommendation import Cost, MeasureOption, Recommendation
from domain.modelling.simulation import BuildingPartOverlay, EpcSimulation
from repositories.product.product_repository import ProductRepository
_LOFT_MEASURE_TYPE = "loft_insulation"
_SLOPING_CEILING_MEASURE_TYPE = "sloping_ceiling_insulation"
_LOFT_MEASURE_TYPE = MeasureType.LOFT_INSULATION
_SLOPING_CEILING_MEASURE_TYPE = MeasureType.SLOPING_CEILING_INSULATION
# RdSAP 10 Table 16: 0 mm lodged roof insulation is an uninsulated loft. The
# Elmhurst mapper resolves "As Built" to 0 for pitched/sloping/loft roofs.
_ROOF_UNINSULATED_MM = 0
@ -32,7 +33,7 @@ _ROOF_UNINSULATED_MM = 0
_RECOMMENDED_LOFT_THICKNESS_MM = 300
# Recommended sloping-ceiling depth (mm); Elmhurst re-lodges 100 mm (ADR-0021).
_RECOMMENDED_SLOPING_CEILING_THICKNESS_MM = 100
_FLAT_ROOF_MEASURE_TYPE = "flat_roof_insulation"
_FLAT_ROOF_MEASURE_TYPE = MeasureType.FLAT_ROOF_INSULATION
# Recommended flat-roof depth (mm); Elmhurst re-lodges 200 mm (ADR-0021).
_RECOMMENDED_FLAT_ROOF_THICKNESS_MM = 200
@ -109,7 +110,7 @@ def _roof_recommendation(
epc: EpcPropertyData,
products: ProductRepository,
*,
measure_type: str,
measure_type: MeasureType,
description: str,
thickness_mm: int,
) -> Recommendation:

3
domain/modelling/generators/solar_recommendation.py
View file

@ -24,6 +24,7 @@ from datatypes.epc.domain.epc_property_data import (
from datatypes.epc.domain.field_mappings import PROPERTY_TYPE_LOOKUP
from domain.geospatial.planning_restrictions import PlanningRestrictions
from domain.modelling.products import Products, SolarCostInputs
from domain.modelling.measure_type import MeasureType
from domain.modelling.recommendation import Cost, MeasureOption, Recommendation
from domain.modelling.simulation import EpcSimulation, SolarOverlay
from domain.modelling.solar_potential import (
@ -37,7 +38,7 @@ from domain.sap10_calculator.rdsap.cert_to_inputs import (
from repositories.product.product_repository import ProductRepository
_SOLAR_SURFACE = "Solar PV"
_SOLAR_MEASURE_TYPE = "solar_pv"
_SOLAR_MEASURE_TYPE = MeasureType.SOLAR_PV
# The fixed, representative battery capacity for the with-battery variant
# (ADR-0026) — a flagged estimate (see the rate sheet), 5 kWh.

9
domain/modelling/generators/solid_wall_recommendation.py
View file

@ -24,12 +24,13 @@ from datatypes.epc.domain.epc_property_data import (
from datatypes.epc.domain.field_mappings import PROPERTY_TYPE_LOOKUP
from domain.building_geometry import gross_heat_loss_wall_area
from domain.geospatial.planning_restrictions import PlanningRestrictions
from domain.modelling.measure_type import MeasureType
from domain.modelling.recommendation import Cost, MeasureOption, Recommendation
from domain.modelling.simulation import BuildingPartOverlay, EpcSimulation
from repositories.product.product_repository import ProductRepository
_EXTERNAL_MEASURE_TYPE: Final[str] = "external_wall_insulation"
_INTERNAL_MEASURE_TYPE: Final[str] = "internal_wall_insulation"
_EXTERNAL_MEASURE_TYPE: Final[MeasureType] = MeasureType.EXTERNAL_WALL_INSULATION
_INTERNAL_MEASURE_TYPE: Final[MeasureType] = MeasureType.INTERNAL_WALL_INSULATION
# RdSAP `wall_construction` codes (consistent across paths for 1-5).
_WALL_SOLID_BRICK: Final[int] = 3
@ -56,7 +57,7 @@ _SOLID_WALL_INSULATION_MM: Final[int] = 100
# Which solid-wall Options each construction can take (ADR-0019). Solid brick
# and system-built take both; timber-frame takes IWI only (EWI not
# constructable). The breathable cob/stone exclusions take neither (never keyed).
_CONSTRUCTABLE_OPTIONS: Final[dict[int, tuple[str, ...]]] = {
_CONSTRUCTABLE_OPTIONS: Final[dict[int, tuple[MeasureType, ...]]] = {
_WALL_SOLID_BRICK: (_EXTERNAL_MEASURE_TYPE, _INTERNAL_MEASURE_TYPE),
_WALL_SYSTEM_BUILT: (_EXTERNAL_MEASURE_TYPE, _INTERNAL_MEASURE_TYPE),
_WALL_TIMBER_FRAME: (_INTERNAL_MEASURE_TYPE,),
@ -74,7 +75,7 @@ _DESCRIPTION: Final[dict[str, str]] = {
def _solid_wall_option(
epc: EpcPropertyData, products: ProductRepository, measure_type: str
epc: EpcPropertyData, products: ProductRepository, measure_type: MeasureType
) -> MeasureOption:
"""Build one solid-wall Measure Option: its insulation overlay (100 mm at the
External/Internal `wall_insulation_type`) priced at the heat-loss wall area."""

3
domain/modelling/generators/ventilation_recommendation.py
View file

@ -16,11 +16,12 @@ mirrors legacy ``Property.has_ventilation``.
from typing import Optional
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.modelling.measure_type import MeasureType
from domain.modelling.recommendation import Cost, MeasureOption, Recommendation
from domain.modelling.simulation import EpcSimulation, VentilationOverlay
from repositories.product.product_repository import ProductRepository
_VENTILATION_MEASURE_TYPE = "mechanical_ventilation"
_VENTILATION_MEASURE_TYPE = MeasureType.MECHANICAL_VENTILATION
# The SAP10.2 §2 mechanical-ventilation kind installed: decentralised MEV
# ("mechanical extract, decentralised (MEV dc)" → MechanicalVentilationKind

3
domain/modelling/generators/wall_recommendation.py
View file

@ -12,11 +12,12 @@ from datatypes.epc.domain.epc_property_data import (
EpcPropertyData,
)
from domain.building_geometry import gross_heat_loss_wall_area
from domain.modelling.measure_type import MeasureType
from domain.modelling.recommendation import Cost, MeasureOption, Recommendation
from domain.modelling.simulation import BuildingPartOverlay, EpcSimulation
from repositories.product.product_repository import ProductRepository
_CAVITY_MEASURE_TYPE = "cavity_wall_insulation"
_CAVITY_MEASURE_TYPE = MeasureType.CAVITY_WALL_INSULATION
# RdSAP 10 Table 5 wall_construction: 4 = "Cavity". Table 6
# wall_insulation_type: 4 = "as-built / assumed" (uninsulated), 2 = "Filled

35
domain/modelling/measure_type.py Normal file
View file

@ -0,0 +1,35 @@
"""MeasureType — the canonical vocabulary of the measures the Modelling stage
models.
One member per Recommendation Generator option. A ``StrEnum`` so each member
*is* its string value: it persists straight into the ``recommendation`` varchar
column, is the optimiser's group-by key, and compares equal to the raw strings
the catalogue and EPC carry — so it can replace the per-generator string
constants as the single source of truth without a persistence or optimiser
change. It is also the vocabulary the ``considered_measures`` allowlist speaks
(mirroring the legacy engine's ``inclusions``).
"""
from __future__ import annotations
from enum import StrEnum
class MeasureType(StrEnum):
"""A measure the Modelling stage can recommend (CONTEXT.md)."""
CAVITY_WALL_INSULATION = "cavity_wall_insulation"
EXTERNAL_WALL_INSULATION = "external_wall_insulation"
INTERNAL_WALL_INSULATION = "internal_wall_insulation"
LOFT_INSULATION = "loft_insulation"
SLOPING_CEILING_INSULATION = "sloping_ceiling_insulation"
FLAT_ROOF_INSULATION = "flat_roof_insulation"
SUSPENDED_FLOOR_INSULATION = "suspended_floor_insulation"
SOLID_FLOOR_INSULATION = "solid_floor_insulation"
DOUBLE_GLAZING = "double_glazing"
SECONDARY_GLAZING = "secondary_glazing"
LOW_ENERGY_LIGHTING = "low_energy_lighting"
MECHANICAL_VENTILATION = "mechanical_ventilation"
HIGH_HEAT_RETENTION_STORAGE_HEATERS = "high_heat_retention_storage_heaters"
AIR_SOURCE_HEAT_PUMP = "air_source_heat_pump"
SOLAR_PV = "solar_pv"

9
domain/modelling/optimisation/measure_dependency.py
View file

@ -24,17 +24,18 @@ from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.modelling.generators.ventilation_recommendation import (
recommend_ventilation,
)
from domain.modelling.measure_type import MeasureType
from domain.modelling.optimisation.optimiser import MeasureDependency, ScoredOption
from domain.modelling.recommendation import MeasureOption, Recommendation
from repositories.product.product_repository import ProductRepository
# The measure types that force a ventilation dependency (cf. legacy
# `assumptions.measures_needing_ventilation`).
MEASURES_NEEDING_VENTILATION: frozenset[str] = frozenset(
MEASURES_NEEDING_VENTILATION: frozenset[MeasureType] = frozenset(
{
"cavity_wall_insulation",
"internal_wall_insulation",
"external_wall_insulation",
MeasureType.CAVITY_WALL_INSULATION,
MeasureType.INTERNAL_WALL_INSULATION,
MeasureType.EXTERNAL_WALL_INSULATION,
}
)

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

@ -24,6 +24,7 @@ from dataclasses import dataclass
from typing import Optional, Protocol, Sequence
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.modelling.measure_type import MeasureType
from domain.modelling.scoring.package_scorer import Score
from domain.modelling.recommendation import MeasureOption
from domain.modelling.simulation import EpcSimulation
@ -48,7 +49,7 @@ class MeasureDependency:
figure, the repair decision, and the persisted package. Held as data so
extending the triggers is a data edit, not control flow."""
triggers: frozenset[str]
triggers: frozenset[MeasureType]
required: ScoredOption
@ -300,11 +301,11 @@ def _inject(
"""``chosen`` plus every forced dependency whose triggers intersect the
chosen measure-types, de-duplicated by required measure-type (a dependency
several measures trigger is injected once)."""
chosen_types: set[str] = {s.option.measure_type for s in chosen}
chosen_types: set[MeasureType] = {s.option.measure_type for s in chosen}
injected: list[ScoredOption] = list(chosen)
present: set[str] = set(chosen_types)
present: set[MeasureType] = set(chosen_types)
for dependency in dependencies:
required_type: str = dependency.required.option.measure_type
required_type: MeasureType = dependency.required.option.measure_type
if dependency.triggers & chosen_types and required_type not in present:
injected.append(dependency.required)
present.add(required_type)

3
domain/modelling/plan.py
View file

@ -15,6 +15,7 @@ from typing import Optional
from datatypes.epc.domain.epc import Epc
from domain.billing.bill import Bill
from domain.modelling.measure_type import MeasureType
from domain.modelling.scoring.package_scorer import Score
from domain.modelling.recommendation import Cost
from domain.modelling.scoring.scoring import MeasureImpact
@ -33,7 +34,7 @@ class PlanMeasure:
billing has run (persisted as NULL — ADR-0014 amendment). They are distinct
from `impact.energy_savings_kwh_per_yr`, which is *primary* energy."""
measure_type: str
measure_type: MeasureType
description: str
cost: Cost
impact: MeasureImpact

3
domain/modelling/recommendation.py
View file

@ -11,6 +11,7 @@ CONTEXT.md.
from dataclasses import dataclass
from typing import Optional
from domain.modelling.measure_type import MeasureType
from domain.modelling.simulation import EpcSimulation
@ -28,7 +29,7 @@ class Cost:
class MeasureOption:
"""One mutually-exclusive way to treat a Recommendation's surface."""
measure_type: str
measure_type: MeasureType
description: str
overlay: EpcSimulation
cost: Optional[Cost] = None

41
harness/console.py
View file

@ -23,6 +23,7 @@ from typing import Any, Optional
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.geospatial.coordinates import Coordinates
from domain.geospatial.planning_restrictions import PlanningRestrictions
from domain.modelling.measure_type import MeasureType
from domain.modelling.plan import Plan
from domain.modelling.scenario import Scenario
from domain.property.property import Property, PropertyIdentity
@ -38,6 +39,7 @@ from repositories.fuel_rates.fuel_rates_static_file_repository import (
)
from repositories.geospatial.geospatial_repository import GeospatialRepository
from repositories.product.product_json_repository import ProductJsonRepository
from repositories.product.product_repository import ProductRepository
from tests.orchestration.fakes import (
FakeEpcRepo,
FakePlanRepository,
@ -171,6 +173,9 @@ def run_modelling(
current_market_value: Optional[float] = None,
planning_restrictions: PlanningRestrictions = PlanningRestrictions(),
solar_insights: Optional[dict[str, Any]] = None,
considered_measures: Optional[frozenset[MeasureType]] = None,
products: Optional[ProductRepository] = None,
scenario: Optional[Scenario] = None,
print_table: bool = True,
) -> Plan:
"""Run ONLY the Modelling stage over ``epc`` with no database — skipping
@ -182,7 +187,22 @@ def run_modelling(
``solar_insights`` is the Property's raw Google Solar ``buildingInsights``
JSON (as persisted by ``SolarRepository``); when given, the solar
Recommendation Generator sees the dwelling's potential and can offer Solar
PV Options (ADR-0026)."""
PV Options (ADR-0026).
``products`` overrides the Product catalogue source (default: the JSON
sample catalogue) — pass a read-only ``ProductPostgresRepository`` to price
against the live ``material`` table. ``scenario`` overrides the default
Increasing-EPC-to-``goal_band`` Scenario — pass a Scenario read from the DB
so the run targets a real ``scenario_id`` (its ``goal_value``/budget drive
the Optimiser); the computed Plan is then keyed by that Scenario's id."""
scenario_obj = scenario or Scenario(
id=_SCENARIO_ID,
goal="Increasing EPC",
goal_value=goal_band,
budget=None,
is_default=True,
)
scenario_id = scenario_obj.id
plan_repo = FakePlanRepository()
property_repo = FakePropertyRepo(
{
@ -206,18 +226,8 @@ def run_modelling(
if solar_insights is not None
else None
),
scenario=FakeScenarioRepository(
{
_SCENARIO_ID: Scenario(
id=_SCENARIO_ID,
goal="Increasing EPC",
goal_value=goal_band,
budget=None,
is_default=True,
)
}
),
product=ProductJsonRepository(catalogue_path),
scenario=FakeScenarioRepository({scenario_id: scenario_obj}),
product=products or ProductJsonRepository(catalogue_path),
plan=plan_repo,
)
@ -227,11 +237,12 @@ def run_modelling(
fuel_rates=FuelRatesStaticFileRepository(),
).run(
property_ids=[_PROPERTY_ID],
scenario_ids=[_SCENARIO_ID],
scenario_ids=[scenario_id],
portfolio_id=_PORTFOLIO_ID,
considered_measures=considered_measures,
)
plan = plan_repo.saved[(_PROPERTY_ID, _SCENARIO_ID)]
plan = plan_repo.saved[(_PROPERTY_ID, scenario_id)]
if print_table:
print("\n" + format_plan_table(plan))
return plan

52
orchestration/modelling_orchestrator.py
View file

@ -7,7 +7,9 @@ from datatypes.epc.domain.epc import Epc
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.billing.bill import Bill, EnergyBreakdown
from domain.billing.bill_derivation import BillDerivation
from domain.modelling.considered_measures import restrict_to_considered_measures
from domain.modelling.generators.floor_recommendation import recommend_floor_insulation
from domain.modelling.measure_type import MeasureType
from domain.modelling.optimisation.measure_dependency import ventilation_dependency
from domain.modelling.optimisation.optimiser import (
MeasureDependency,
@ -92,8 +94,16 @@ class ModellingOrchestrator:
self._fuel_rates = fuel_rates
def run(
self, property_ids: list[int], scenario_ids: list[int], portfolio_id: int
self,
property_ids: list[int],
scenario_ids: list[int],
portfolio_id: int,
*,
considered_measures: Optional[frozenset[MeasureType]] = None,
) -> None:
"""Model the batch. ``considered_measures`` restricts the run to those
measure types (mirroring the legacy `inclusions`); None considers every
modelled measure."""
scorer = PackageScorer(self._calculator)
# Resolve Fuel Rates once and reuse the BillDerivation across the batch,
# so every baseline/post bill is priced at the same snapshot (ADR-0014).
@ -120,6 +130,7 @@ class ModellingOrchestrator:
current_market_value=prop.current_market_value,
planning_restrictions=prop.planning_restrictions,
solar_potential=solar_potential,
considered_measures=considered_measures,
)
uow.plan.save(
plan,
@ -141,16 +152,22 @@ class ModellingOrchestrator:
current_market_value: Optional[float],
planning_restrictions: PlanningRestrictions,
solar_potential: Optional[SolarPotential],
considered_measures: Optional[frozenset[MeasureType]],
) -> Plan:
"""Generate → score → optimise → re-score/repair → attribute → bill →
assemble the Plan for one Property + Scenario."""
groups: list[list[ScoredOption]] = _scored_candidate_groups(
scorer, effective_epc, products, planning_restrictions, solar_potential
scorer,
effective_epc,
products,
planning_restrictions,
solar_potential,
considered_measures,
)
# 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
effective_epc, products, considered_measures
)
package: OptimisedPackage = optimise_package(
groups=groups,
@ -224,11 +241,13 @@ def _candidate_recommendations(
products: ProductRepository,
planning_restrictions: PlanningRestrictions,
solar_potential: Optional[SolarPotential],
considered_measures: Optional[frozenset[MeasureType]],
) -> list[Recommendation]:
"""Run every Recommendation Generator; keep the ones that apply. Solid-wall
insulation, glazing, heating and solar are additionally gated by the
Property's planning protections (ADR-0019 / ADR-0022 / ADR-0024 / ADR-0026);
solar also needs the Property's Google solar potential."""
solar also needs the Property's Google solar potential. ``considered_measures``
then restricts the survivors to the run's allowlist (None = all)."""
found = (
recommend_cavity_wall(effective_epc, products),
recommend_solid_wall(effective_epc, products, planning_restrictions),
@ -241,19 +260,33 @@ def _candidate_recommendations(
effective_epc, products, solar_potential, planning_restrictions
),
)
return [recommendation for recommendation in found if recommendation is not None]
applicable = [
recommendation for recommendation in found if recommendation is not None
]
return restrict_to_considered_measures(applicable, considered_measures)
def _measure_dependencies(
effective_epc: EpcPropertyData, products: ProductRepository
effective_epc: EpcPropertyData,
products: ProductRepository,
considered_measures: Optional[frozenset[MeasureType]],
) -> list[MeasureDependency]:
"""The forced Measure Dependencies for this Property — currently just
ventilation, suppressed when the dwelling is already mechanically
ventilated (ADR-0016)."""
ventilated (ADR-0016). A dependency whose required measure is outside the
run's allowlist is also suppressed, so a restricted run forces nothing it is
not considering."""
dependency: Optional[MeasureDependency] = ventilation_dependency(
effective_epc, products
)
return [dependency] if dependency is not None else []
if dependency is None:
return []
if (
considered_measures is not None
and dependency.required.option.measure_type not in considered_measures
):
return []
return [dependency]
def _scored_candidate_groups(
@ -262,12 +295,13 @@ def _scored_candidate_groups(
products: ProductRepository,
planning_restrictions: PlanningRestrictions,
solar_potential: Optional[SolarPotential],
considered_measures: Optional[frozenset[MeasureType]],
) -> list[list[ScoredOption]]:
"""One group per Recommendation: each Option scored independently against
the baseline (role-1 warm-start signal, ADR-0016)."""
groups: list[list[ScoredOption]] = []
for recommendation in _candidate_recommendations(
effective_epc, products, planning_restrictions, solar_potential
effective_epc, products, planning_restrictions, solar_potential, considered_measures
):
options = list(recommendation.options)
impacts: list[MeasureImpact] = independent_option_impacts(

7
repositories/product/product_postgres_repository.py
View file

@ -17,11 +17,16 @@ class ProductPostgresRepository(ProductRepository):
self._session = session
def get(self, measure_type: str) -> Product:
# The live catalogue holds many active rows per type; order by id so the
# pick is deterministic (a re-seed prices the same) rather than relying
# on the database's physical row order.
row: MaterialRow | None = self._session.exec(
select(MaterialRow).where(
select(MaterialRow)
.where(
col(MaterialRow.type) == measure_type,
col(MaterialRow.is_active).is_(True),
)
.order_by(col(MaterialRow.id))
).first()
if row is None:
raise ValueError(f"no active product for measure type {measure_type!r}")

199
scripts/run_modelling_e2e.py
View file

@ -3,30 +3,42 @@ print the recommendations for inspection.
The local DB's Properties have no linked, ingested EPC yet (Ingestion's source
clients are still stubbed — #1136), so this script does the ingestion step
inline for inspection: it reads each Property's UPRN from the DB, fetches the
latest EPC **live** from the gov EPC API by UPRN, then runs the Modelling stage
in memory (every Recommendation Generator → the Optimiser → a costed, attributed
Plan). It is read-only on the DB (just the UPRN lookup) and persists nothing —
purely for inspecting recommendations. Prints a per-Property plan table and
writes a Markdown + CSV summary.
inline: it reads each Property's UPRN from the DB, fetches the latest EPC
**live** from the gov EPC API by UPRN, resolves the UPRN's spatial reference
from S3, and fetches Google Solar — then runs the Modelling stage (every
Recommendation Generator → the Optimiser → a costed, attributed Plan). The same
local computation runs whether or not you store the result: by default it
persists **nothing** (the run is for inspecting recommendations); pass
`--persist` to write the inputs + the Plan to the DB.
To keep the inspected recommendations identical to what gets stored, **both
modes price against the live ``material`` catalogue (read-only)** and model
against a real **Scenario** read from the DB — not the JSON sample catalogue.
Pass `--scenario-id` to target a real Scenario (its ``goal_value`` drives the
band); without it the run synthesises an Increasing-EPC-to-``--goal`` Scenario.
``--measures`` restricts the run to a comma-separated set of measure types
(mirroring the legacy `inclusions`) — e.g. only HHRSH + Solar PV.
Config: loads `backend/.env` for the DB creds (`DB_*`), the EPC API token
(`EPC_AUTH_TOKEN`), the Google Solar key (`GOOGLE_SOLAR_API_KEY`) and the S3
(`OPEN_EPC_API_TOKEN` — the Bearer token for the new gov API), the Google Solar
key (`GOOGLE_SOLAR_API_KEY`) and the S3
reference bucket (`DATA_BUCKET`) — the agent never sees the secrets. AWS creds
come from the ambient `~/.aws` profile. Run from the worktree root so imports
resolve to this checkout:
come from the ambient `~/.aws` profile. Run from the worktree root:
python -m scripts.run_modelling_e2e 115 116 117 # goal band C (default)
python -m scripts.run_modelling_e2e --goal B 115 116 117 # a different target band
python -m scripts.run_modelling_e2e --no-solar 115 116 # skip the Google Solar leg
# inspect only (no DB writes), HHRSH + Solar PV, against Scenario 1263:
python -m scripts.run_modelling_e2e --scenario-id 1263 \
--measures high_heat_retention_storage_heaters,solar_pv 115 116 117
# same run, but persist the Plans (needs --portfolio-id):
python -m scripts.run_modelling_e2e --scenario-id 1263 --portfolio-id 4 \
--measures high_heat_retention_storage_heaters,solar_pv --persist 115 116 117
python -m scripts.run_modelling_e2e --no-solar 115 116 # skip the Google leg
Per Property the script resolves the UPRN's spatial reference from the Ordnance
Survey Open-UPRN parquet in S3 (`GeospatialS3Repository`): the planning
protections (conservation/listed/heritage) gate the wall + solar measures, and
the coordinates drive a live Google Solar `buildingInsights` fetch so the Solar
PV Options can fire (ADR-0026). Buildings S3 doesn't cover, or that Google has
no solar coverage for, fall back to unrestricted / no-solar and are still
modelled. Pass `--no-solar` to skip the Google leg entirely.
Per Property the spatial reference (S3 Open-UPRN parquet) gives the planning
protections (conservation/listed/heritage — gate the wall + solar measures) and
the coordinates that drive the Google Solar fetch (ADR-0026). Buildings S3
doesn't cover, or that Google has no solar coverage for, fall back to
unrestricted / no-solar and are still modelled. Pass `--no-solar` to skip the
Google leg.
"""
from __future__ import annotations
@ -47,8 +59,10 @@ sys.path.insert(0, str(_REPO_ROOT)) # worktree root first — avoid the import
from datatypes.epc.domain.epc_property_data import EpcPropertyData # noqa: E402
from domain.geospatial.planning_restrictions import PlanningRestrictions # noqa: E402
from domain.geospatial.spatial_reference import SpatialReference # noqa: E402
from domain.modelling.measure_type import MeasureType # noqa: E402
from domain.modelling.plan import Plan, PlanMeasure # noqa: E402
from harness.console import DEFAULT_CATALOGUE, run_modelling # noqa: E402
from domain.modelling.scenario import Scenario # noqa: E402
from harness.console import run_modelling # noqa: E402
from harness.plan_table import format_plan_table # noqa: E402
from infrastructure.epc_client.epc_client_service import EpcClientService # noqa: E402
from infrastructure.solar.google_solar_api_client import ( # noqa: E402
@ -59,7 +73,15 @@ from repositories.geospatial.geospatial_s3_repository import ( # noqa: E402
GeospatialS3Repository,
ParquetReader,
)
from sqlalchemy import create_engine, text # noqa: E402
from repositories.product.product_postgres_repository import ( # noqa: E402
ProductPostgresRepository,
)
from repositories.postgres_unit_of_work import PostgresUnitOfWork # noqa: E402
from repositories.scenario.scenario_postgres_repository import ( # noqa: E402
ScenarioPostgresRepository,
)
from sqlalchemy import Engine, create_engine, text # noqa: E402
from sqlmodel import Session # noqa: E402
_ENV_PATH = _REPO_ROOT / "backend" / ".env"
_MARKDOWN_PATH = Path("modelling_e2e.md")
@ -130,11 +152,15 @@ def _solar_insights_for(
return None # no Google solar coverage at this point — model without it
def _uprns_for(property_ids: list[int]) -> dict[int, Optional[int]]:
"""Read each Property's UPRN from the DB (read-only)."""
engine = create_engine(
def _engine() -> Engine:
"""A connection-pooled engine to DevAssessmentModelDB (DB_* creds)."""
return create_engine(
_db_url(), pool_pre_ping=True, connect_args={"connect_timeout": 10}
)
def _uprns_for(engine: Engine, property_ids: list[int]) -> dict[int, Optional[int]]:
"""Read each Property's UPRN from the DB (read-only)."""
with engine.connect() as conn:
rows = conn.execute(
text("SELECT id, uprn FROM property WHERE id = ANY(:ids)"),
@ -143,6 +169,26 @@ def _uprns_for(property_ids: list[int]) -> dict[int, Optional[int]]:
return {int(pid): (int(uprn) if uprn is not None else None) for pid, uprn in rows}
def _scenario_for(session: Session, scenario_id: int) -> Scenario:
"""Read the Scenario the run targets (read-only). An Increasing-EPC Scenario
must carry a ``goal_value`` (band) — the old null-band rows were a fixed bug
and crash the Optimiser's target — so reject one that does not."""
scenario: Scenario = ScenarioPostgresRepository(session).get_many([scenario_id])[0]
if scenario.goal == "Increasing EPC" and not scenario.goal_value:
raise ValueError(
f"scenario {scenario_id} has no goal_value (band); pick a recent one"
)
return scenario
def _parse_measures(raw: Optional[str]) -> Optional[frozenset[MeasureType]]:
"""Parse `--measures a,b,c` into a `considered_measures` allowlist, or None
(consider every modelled measure) when unset. Raises on an unknown type."""
if raw is None:
return None
return frozenset(MeasureType(token.strip()) for token in raw.split(",") if token.strip())
def _context_summary(
spatial: Optional[SpatialReference], solar_insights: Optional[dict[str, Any]]
) -> str:
@ -173,10 +219,57 @@ def _measure_summary(measure: PlanMeasure) -> str:
)
def _persist(
engine: Engine,
*,
property_id: int,
uprn: int,
portfolio_id: int,
scenario: Scenario,
epc: EpcPropertyData,
spatial: Optional[SpatialReference],
solar_insights: Optional[dict[str, Any]],
plan: Plan,
) -> None:
"""Write the run's inputs (EPC + spatial + solar) and the computed Plan to
the DB in one Unit of Work, then commit. ``PlanPostgresRepository`` replaces
any existing Plan for ``(property_id, scenario.id)`` (idempotent re-run)."""
with PostgresUnitOfWork(lambda: Session(engine)) as uow:
uow.epc.save(epc, property_id=property_id, portfolio_id=portfolio_id)
if spatial is not None:
uow.spatial.save(uprn, spatial)
if solar_insights is not None:
uow.solar.save(property_id, solar_insights)
uow.plan.save(
plan,
property_id=property_id,
scenario_id=scenario.id,
portfolio_id=portfolio_id,
is_default=scenario.is_default,
)
uow.commit()
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("property_ids", type=int, nargs="+", help="Property ids to model")
parser.add_argument("--goal", default="C", help="target EPC band (default C)")
parser.add_argument("--goal", default="C", help="target band when no --scenario-id (default C)")
parser.add_argument(
"--scenario-id", type=int, default=None, help="model against this DB Scenario"
)
parser.add_argument(
"--measures",
default=None,
help="comma-separated measure types to consider (default: all)",
)
parser.add_argument(
"--portfolio-id", type=int, default=None, help="portfolio id (required for --persist)"
)
parser.add_argument(
"--persist",
action="store_true",
help="WRITE the inputs + Plan to the DB (default: inspect only, no writes)",
)
parser.add_argument(
"--no-solar",
action="store_true",
@ -184,18 +277,42 @@ def main() -> None:
)
args = parser.parse_args()
if args.persist and (args.scenario_id is None or args.portfolio_id is None):
parser.error("--persist requires --scenario-id and --portfolio-id")
_load_env(_ENV_PATH)
epc_client = EpcClientService(os.environ["EPC_AUTH_TOKEN"])
# The new gov EPC API (Bearer) authenticates with OPEN_EPC_API_TOKEN — the
# name is misleading; EPC_AUTH_TOKEN is dead (403). Verified against the
# /api/domestic/search endpoint.
epc_client = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"])
geospatial = GeospatialS3Repository(_s3_parquet_reader(os.environ["DATA_BUCKET"]))
solar_client = GoogleSolarApiClient(os.environ["GOOGLE_SOLAR_API_KEY"])
uprns = _uprns_for(args.property_ids)
print(
f"modelling {len(args.property_ids)} propertie(s) (goal band {args.goal}); "
f"EPCs fetched live by UPRN, modelled in memory — no DB writes...\n"
engine = _engine()
considered = _parse_measures(args.measures)
uprns = _uprns_for(engine, args.property_ids)
# One read-only session for the live `material` catalogue, reused across the
# batch so both store and no-store runs price against the same DB rows.
catalogue_session = Session(engine)
products = ProductPostgresRepository(catalogue_session)
scenario: Optional[Scenario] = (
_scenario_for(catalogue_session, args.scenario_id)
if args.scenario_id is not None
else None
)
md_lines: list[str] = [f"# Modelling recommendations (goal band {args.goal})\n"]
target = (
f"scenario {scenario.id} (band {scenario.goal_value})"
if scenario is not None
else f"synthesised Increasing-EPC band {args.goal}"
)
measures_note = ",".join(sorted(considered)) if considered else "all measures"
mode = "PERSISTING to DB" if args.persist else "no DB writes"
print(
f"modelling {len(args.property_ids)} propertie(s) · {target} · {measures_note} · "
f"{mode} (DB material catalogue, live EPC/solar)...\n"
)
md_lines: list[str] = [f"# Modelling recommendations ({target}, {measures_note})\n"]
csv_rows: list[str] = [
"property_id,uprn,baseline_sap,post_sap,measures,measure_types,cost_of_works"
]
@ -218,11 +335,26 @@ def main() -> None:
plan: Plan = run_modelling(
epc,
goal_band=args.goal,
catalogue_path=DEFAULT_CATALOGUE,
planning_restrictions=restrictions,
solar_insights=solar_insights,
considered_measures=considered,
products=products,
scenario=scenario,
print_table=False,
)
if args.persist:
assert scenario is not None # guaranteed by the --persist guard
_persist(
engine,
property_id=property_id,
uprn=uprn,
portfolio_id=args.portfolio_id,
scenario=scenario,
epc=epc,
spatial=spatial,
solar_insights=solar_insights,
plan=plan,
)
except Exception as error: # noqa: BLE001 — one bad property must not stop the run
line = f"property {property_id} (uprn {uprn}): ERROR — {type(error).__name__}: {error}"
print(line + "\n")
@ -255,6 +387,7 @@ def main() -> None:
f"{'|'.join(measure_types)},{plan.cost_of_works:.0f}"
)
catalogue_session.close()
_MARKDOWN_PATH.write_text("\n".join(md_lines) + "\n", encoding="utf-8")
_CSV_PATH.write_text("\n".join(csv_rows) + "\n", encoding="utf-8")
print(f"wrote {_MARKDOWN_PATH.resolve()}")

83
tests/domain/modelling/test_considered_measures.py Normal file
View file

@ -0,0 +1,83 @@
"""Slice 3 — `restrict_to_considered_measures`, the pure allowlist that limits a
run to a chosen set of measure types (mirroring the legacy engine's
`inclusions`).
It filters at the Option level, so a multi-option Recommendation (e.g. Heating &
Hot Water offering both HHRSH and an ASHP bundle) is kept with only its allowed
Options; a Recommendation left with no allowed Option is dropped entirely. A
None allowlist means "consider everything" (today's unrestricted behaviour).
"""
from domain.modelling.considered_measures import restrict_to_considered_measures
from domain.modelling.measure_type import MeasureType
from domain.modelling.recommendation import MeasureOption, Recommendation
from domain.modelling.simulation import EpcSimulation
def _option(measure_type: MeasureType) -> MeasureOption:
return MeasureOption(
measure_type=measure_type, description=str(measure_type), overlay=EpcSimulation()
)
def _heating_rec() -> Recommendation:
# Heating & Hot Water competes HHRSH against an ASHP bundle in one rec.
return Recommendation(
surface="Heating & Hot Water",
options=(
_option(MeasureType.HIGH_HEAT_RETENTION_STORAGE_HEATERS),
_option(MeasureType.AIR_SOURCE_HEAT_PUMP),
),
)
def _solar_rec() -> Recommendation:
return Recommendation(surface="Solar PV", options=(_option(MeasureType.SOLAR_PV),))
def _wall_rec() -> Recommendation:
return Recommendation(
surface="Wall", options=(_option(MeasureType.CAVITY_WALL_INSULATION),)
)
def test_none_allowlist_keeps_everything() -> None:
# Arrange
recommendations = [_heating_rec(), _solar_rec(), _wall_rec()]
# Act
kept = restrict_to_considered_measures(recommendations, None)
# Assert
assert kept == recommendations
def test_drops_recommendations_with_no_allowed_option() -> None:
# Arrange
considered = frozenset(
{MeasureType.HIGH_HEAT_RETENTION_STORAGE_HEATERS, MeasureType.SOLAR_PV}
)
# Act
kept = restrict_to_considered_measures(
[_heating_rec(), _solar_rec(), _wall_rec()], considered
)
# Assert — the wall rec is gone; heating + solar survive.
surfaces = {rec.surface for rec in kept}
assert surfaces == {"Heating & Hot Water", "Solar PV"}
def test_filters_options_within_a_kept_recommendation() -> None:
# Arrange — HHRSH is allowed but the competing ASHP bundle is not.
considered = frozenset(
{MeasureType.HIGH_HEAT_RETENTION_STORAGE_HEATERS, MeasureType.SOLAR_PV}
)
# Act
kept = restrict_to_considered_measures([_heating_rec()], considered)
# Assert — the heating rec keeps ONLY its HHRSH option, ASHP is dropped.
assert len(kept) == 1
kept_types = [option.measure_type for option in kept[0].options]
assert kept_types == [MeasureType.HIGH_HEAT_RETENTION_STORAGE_HEATERS]

4
tests/domain/modelling/test_heating_recommendation.py
View file

@ -55,7 +55,7 @@ def test_electric_storage_dwelling_yields_an_hhr_storage_bundle() -> None:
# bundle, whose overlay is the absolute HHR end-state.
assert recommendation is not None
assert recommendation.surface == "Heating & Hot Water"
options = {o.measure_type: o for o in recommendation.options}
options = {o.measure_type.value: o for o in recommendation.options}
assert "high_heat_retention_storage_heaters" in options
assert options["high_heat_retention_storage_heaters"].overlay.heating == HeatingOverlay(
main_fuel_type=30,
@ -150,7 +150,7 @@ def test_gas_boiler_house_yields_an_ashp_bundle() -> None:
# Assert — the ASHP bundle carries the absolute heat-pump end-state.
assert recommendation is not None
options = {o.measure_type: o for o in recommendation.options}
options = {o.measure_type.value: o for o in recommendation.options}
assert "air_source_heat_pump" in options
assert options["air_source_heat_pump"].overlay.heating == HeatingOverlay(
main_fuel_type=30,

52
tests/domain/modelling/test_measure_type.py Normal file
View file

@ -0,0 +1,52 @@
"""Slice 1 — `MeasureType`, the canonical enum of the measures the Modelling
stage actually models.
A `StrEnum` so each member *is* its persisted/optimiser string value (e.g.
`MeasureType.SOLAR_PV == "solar_pv"`), letting it flow through the `recommendation`
varchar column and the optimiser's group-by-type key unchanged. Replaces the
per-generator string constants as the single source of truth, and is the
vocabulary the `considered_measures` allowlist speaks.
"""
from domain.modelling.measure_type import MeasureType
# The full set of measures the generators emit today (one per Generator option).
_EXPECTED_VALUES = {
"cavity_wall_insulation",
"external_wall_insulation",
"internal_wall_insulation",
"loft_insulation",
"sloping_ceiling_insulation",
"flat_roof_insulation",
"suspended_floor_insulation",
"solid_floor_insulation",
"double_glazing",
"secondary_glazing",
"low_energy_lighting",
"mechanical_ventilation",
"high_heat_retention_storage_heaters",
"air_source_heat_pump",
"solar_pv",
}
def test_members_cover_exactly_the_modelled_measures() -> None:
# Arrange / Act
values = {member.value for member in MeasureType}
# Assert
assert values == _EXPECTED_VALUES
def test_member_is_its_string_value() -> None:
# Arrange / Act / Assert — a StrEnum member compares and persists as its str.
assert MeasureType.SOLAR_PV == "solar_pv"
assert MeasureType.HIGH_HEAT_RETENTION_STORAGE_HEATERS == (
"high_heat_retention_storage_heaters"
)
assert isinstance(MeasureType.SOLAR_PV, str)
def test_round_trips_through_its_value() -> None:
# Arrange / Act / Assert — a raw catalogue/DB string maps back to the member.
assert MeasureType("cavity_wall_insulation") is MeasureType.CAVITY_WALL_INSULATION

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

@ -22,6 +22,7 @@ from domain.modelling.optimisation.optimiser import (
optimise_min_cost,
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 (
@ -37,7 +38,7 @@ from tests.domain.sap10_calculator.worksheet._elmhurst_worksheet_000490 import (
def _scored(measure_type: str, *, gain: float, cost: float) -> ScoredOption:
return ScoredOption(
option=MeasureOption(
measure_type=measure_type,
measure_type=MeasureType(measure_type),
description=measure_type,
overlay=EpcSimulation(),
cost=Cost(total=cost, contingency_rate=0.0),
@ -70,7 +71,7 @@ def _scored_overlay(
) -> ScoredOption:
return ScoredOption(
option=MeasureOption(
measure_type=measure_type,
measure_type=MeasureType(measure_type),
description=measure_type,
overlay=overlay,
cost=Cost(total=cost, contingency_rate=0.0),
@ -524,10 +525,12 @@ class _VentStubScorer:
def _ventilation_dependency(*, cost: float) -> MeasureDependency:
"""A forced 'fabric requires ventilation' edge for the tests."""
return MeasureDependency(
triggers=frozenset({"cavity_wall_insulation", "external_wall_insulation"}),
triggers=frozenset(
{MeasureType.CAVITY_WALL_INSULATION, MeasureType.EXTERNAL_WALL_INSULATION}
),
required=ScoredOption(
option=MeasureOption(
measure_type="mechanical_ventilation",
measure_type=MeasureType.MECHANICAL_VENTILATION,
description="mechanical_ventilation",
overlay=_VENT_OVERLAY,
cost=Cost(total=cost, contingency_rate=0.0),
@ -549,10 +552,10 @@ def test_min_cost_warm_start_avoids_a_wall_whose_forced_ventilation_dooms_it() -
]
scorer = _VentStubScorer(base=60.0, wall=6.0, roof=8.0, vent=-5.0)
dependency = MeasureDependency(
triggers=frozenset({"cavity_wall_insulation"}),
triggers=frozenset({MeasureType.CAVITY_WALL_INSULATION}),
required=ScoredOption(
option=MeasureOption(
measure_type="mechanical_ventilation",
measure_type=MeasureType.MECHANICAL_VENTILATION,
description="mechanical_ventilation",
overlay=_VENT_OVERLAY,
cost=Cost(total=300.0, contingency_rate=0.0),

3
tests/domain/modelling/test_plan.py
View file

@ -9,6 +9,7 @@ from __future__ import annotations
from datatypes.epc.domain.epc import Epc
from domain.billing.bill import Bill, BillSection, BillSectionCost
from domain.modelling.measure_type import MeasureType
from domain.modelling.scoring.package_scorer import Score
from domain.modelling.plan import Plan, PlanMeasure
from domain.modelling.recommendation import Cost
@ -28,7 +29,7 @@ def _bill(*, heating_kwh: float, total_gbp: float) -> Bill:
def _measure(measure_type: str, total: float, rate: float) -> PlanMeasure:
return PlanMeasure(
measure_type=measure_type,
measure_type=MeasureType(measure_type),
description=measure_type.replace("_", " "),
cost=Cost(total=total, contingency_rate=rate),
impact=MeasureImpact(

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

@ -10,6 +10,7 @@ from datatypes.epc.domain.epc_property_data import (
BuildingPartIdentifier,
EpcPropertyData,
)
from domain.modelling.measure_type import MeasureType
from domain.modelling.scoring.package_scorer import PackageScorer, Score
from domain.modelling.recommendation import MeasureOption
from domain.modelling.scoring.scoring import (
@ -59,7 +60,7 @@ class _CountingScorer(PackageScorer):
def _option(overlay: EpcSimulation) -> MeasureOption:
return MeasureOption(
measure_type="cavity_wall_insulation", description="opt", overlay=overlay
measure_type=MeasureType.CAVITY_WALL_INSULATION, description="opt", overlay=overlay
)

5
tests/harness/test_plan_table.py
View file

@ -2,6 +2,7 @@
from __future__ import annotations
from domain.modelling.measure_type import MeasureType
from domain.modelling.plan import Plan, PlanMeasure
from domain.modelling.recommendation import Cost
from domain.modelling.scoring.package_scorer import Score
@ -18,7 +19,7 @@ def _plan() -> Plan:
)
measures = (
PlanMeasure(
measure_type="cavity_wall_insulation",
measure_type=MeasureType.CAVITY_WALL_INSULATION,
description="Cavity wall insulation",
cost=Cost(total=500.0, contingency_rate=0.1),
impact=MeasureImpact(
@ -30,7 +31,7 @@ def _plan() -> Plan:
energy_cost_savings=120.0,
),
PlanMeasure(
measure_type="mechanical_ventilation",
measure_type=MeasureType.MECHANICAL_VENTILATION,
description="Mechanical extract ventilation",
cost=Cost(total=900.0, contingency_rate=0.26),
impact=MeasureImpact(

27
tests/orchestration/test_modelling_solar_threading.py
View file

@ -15,6 +15,7 @@ from typing import Any
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.geospatial.planning_restrictions import PlanningRestrictions
from domain.modelling.measure_type import MeasureType
from domain.modelling.product import Product
from domain.modelling.recommendation import Recommendation
from orchestration.modelling_orchestrator import (
@ -95,7 +96,7 @@ def test_candidate_recommendations_includes_solar_when_potential_present() -> No
# Act
recommendations: list[Recommendation] = _candidate_recommendations(
epc, _StubProducts(), PlanningRestrictions(), potential
epc, _StubProducts(), PlanningRestrictions(), potential, None
)
# Assert — a "Solar PV" Recommendation is among the candidates.
@ -108,8 +109,30 @@ def test_candidate_recommendations_excludes_solar_without_potential() -> None:
# Act
recommendations = _candidate_recommendations(
epc, _StubProducts(), PlanningRestrictions(), None
epc, _StubProducts(), PlanningRestrictions(), None, None
)
# Assert
assert "Solar PV" not in {r.surface for r in recommendations}
def test_considered_measures_restricts_candidates_to_the_allowlist() -> None:
# Arrange — a solar-eligible house, with its solar potential present, so the
# unrestricted run offers Solar PV alongside any fabric/heating candidates.
epc = _eligible_house()
potential = _solar_potential_for(
FakeSolarRepo(by_property={1: json.loads(_INSIGHTS_FIXTURE.read_text())}), 1
)
# Act — restrict the run to Solar PV only.
recommendations = _candidate_recommendations(
epc, _StubProducts(), PlanningRestrictions(), potential, frozenset({MeasureType.SOLAR_PV})
)
# Assert — every surviving Option is solar_pv; nothing else leaks through.
option_types = {
option.measure_type
for recommendation in recommendations
for option in recommendation.options
}
assert option_types == {MeasureType.SOLAR_PV}

5
tests/repositories/plan/test_plan_postgres_repository.py
View file

@ -14,6 +14,7 @@ from sqlalchemy import Engine
from sqlmodel import Session, col, select
from datatypes.epc.domain.epc import Epc
from domain.modelling.measure_type import MeasureType
from domain.modelling.scoring.package_scorer import Score
from domain.modelling.plan import Plan, PlanMeasure
from domain.modelling.recommendation import Cost
@ -25,7 +26,7 @@ from repositories.plan.plan_postgres_repository import PlanPostgresRepository
def _plan() -> Plan:
measures: tuple[PlanMeasure, ...] = (
PlanMeasure(
measure_type="cavity_wall_insulation",
measure_type=MeasureType.CAVITY_WALL_INSULATION,
description="Cavity wall insulation",
cost=Cost(total=1000.0, contingency_rate=0.10),
impact=MeasureImpact(
@ -112,7 +113,7 @@ def test_save_persists_null_per_measure_savings_when_unbilled(
) -> None:
# Arrange — a Plan Measure whose per-measure bills were never derived.
measure = PlanMeasure(
measure_type="loft_insulation",
measure_type=MeasureType.LOFT_INSULATION,
description="Loft insulation",
cost=Cost(total=500.0, contingency_rate=0.20),
impact=MeasureImpact(

37
tests/repositories/product/test_product_postgres_repository.py
View file

@ -42,6 +42,43 @@ def test_get_maps_active_material_to_product_with_contingency(
assert abs(product.contingency_rate - 0.10) <= 1e-9
def test_get_picks_the_lowest_id_when_several_active_rows_share_a_type(
db_engine: Engine,
) -> None:
# Arrange — the live catalogue holds many active rows per type (e.g. 74
# solar_pv); the choice must be deterministic so a re-run prices the same.
with Session(db_engine) as session:
session.add_all(
[
MaterialRow(
id=7,
type="solar_pv",
total_cost=99.0,
cost_unit="gbp_per_unit",
is_active=True,
description="Solar PV (higher id)",
),
MaterialRow(
id=3,
type="solar_pv",
total_cost=42.0,
cost_unit="gbp_per_unit",
is_active=True,
description="Solar PV (lowest id)",
),
]
)
session.commit()
# Act
with Session(db_engine) as session:
product: Product = ProductPostgresRepository(session).get("solar_pv")
# Assert — the lowest-id active row wins, deterministically.
assert product.id == 3
assert abs(product.unit_cost_per_m2 - 42.0) <= 1e-9
def test_get_raises_when_only_an_inactive_product_exists(db_engine: Engine) -> None:
# Arrange
with Session(db_engine) as session: