mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-30 13:10:47 +00:00
Merge remote-tracking branch 'origin/main' into feature/landlord-overrides
This commit is contained in:
Jun-te Kim
commit
364867b05e
16 changed files with 625 additions and 47 deletions
|
|
@ -20,6 +20,20 @@ from domain.modelling.measure_type import MeasureType
|
|||
from domain.modelling.recommendation import Recommendation
|
||||
|
||||
|
||||
def combine_considered_measures(
|
||||
a: Optional[frozenset[MeasureType]],
|
||||
b: Optional[frozenset[MeasureType]],
|
||||
) -> Optional[frozenset[MeasureType]]:
|
||||
"""Intersect two allowlists, treating ``None`` as "all measures". Used to
|
||||
layer an explicit override over the allowlist a Scenario's exclusions imply:
|
||||
None ∧ x = x, and both present narrows to their intersection."""
|
||||
if a is None:
|
||||
return b
|
||||
if b is None:
|
||||
return a
|
||||
return a & b
|
||||
|
||||
|
||||
def restrict_to_considered_measures(
|
||||
recommendations: Iterable[Recommendation],
|
||||
considered_measures: Optional[frozenset[MeasureType]],
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ from domain.modelling.products import (
|
|||
TuneUpCostInputs,
|
||||
)
|
||||
from domain.modelling.measure_type import MeasureType
|
||||
from domain.modelling.product import Product
|
||||
from domain.modelling.recommendation import Cost, MeasureOption, Recommendation
|
||||
from domain.modelling.simulation import EpcSimulation, HeatingOverlay
|
||||
from domain.sap10_calculator.tables.table_4b import (
|
||||
|
|
@ -641,8 +642,9 @@ def _ashp_option(
|
|||
if not _ashp_eligible(epc, restrictions):
|
||||
return None
|
||||
# Cost is composed per-dwelling from the rate sheet (ADR-0025), not the
|
||||
# single catalogue scalar; the catalogue row is still read for its id.
|
||||
product = products.get(_ASHP_MEASURE_TYPE)
|
||||
# single catalogue scalar; the catalogue row is read only for its id, so an
|
||||
# absent ASHP row must not suppress the bundle — it just carries no id.
|
||||
product: Optional[Product] = products.get_optional(_ASHP_MEASURE_TYPE)
|
||||
cost: Cost = Products().ashp_bundle_cost(ashp_cost_inputs(epc))
|
||||
return MeasureOption(
|
||||
measure_type=_ASHP_MEASURE_TYPE,
|
||||
|
|
@ -652,7 +654,7 @@ def _ashp_option(
|
|||
),
|
||||
overlay=EpcSimulation(heating=_ASHP_OVERLAY),
|
||||
cost=cost,
|
||||
material_id=product.id,
|
||||
material_id=product.id if product is not None else None,
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -12,16 +12,33 @@ columns are not modelled. Carries no phases — multi-phase is deferred
|
|||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
from domain.modelling.measure_type import MeasureType
|
||||
|
||||
_NO_EXCLUSIONS: frozenset[MeasureType] = frozenset()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Scenario:
|
||||
"""A retrofit brief: its goal, optional budget, and whether it is the
|
||||
Property's default Scenario. `goal` / `goal_value` are the lodged target
|
||||
(e.g. "INCREASING_EPC" → band "C"); carried for the Optimiser, not yet
|
||||
enforced."""
|
||||
enforced.
|
||||
|
||||
`exclusions` are the measure types the brief bars from the run (the only
|
||||
measure-scoping the live ``scenario`` table persists — there is no
|
||||
inclusions column). Empty means nothing is barred."""
|
||||
|
||||
id: int
|
||||
goal: str
|
||||
goal_value: str
|
||||
budget: Optional[float]
|
||||
is_default: bool
|
||||
exclusions: frozenset[MeasureType] = _NO_EXCLUSIONS
|
||||
|
||||
def considered_measures(self) -> Optional[frozenset[MeasureType]]:
|
||||
"""The measure-type allowlist the Scenario's exclusions imply: every
|
||||
modelled measure minus the excluded ones, or None (consider every
|
||||
measure) when nothing is excluded."""
|
||||
if not self.exclusions:
|
||||
return None
|
||||
return frozenset(MeasureType) - self.exclusions
|
||||
|
|
|
|||
|
|
@ -25,14 +25,19 @@ 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.recommendation import Recommendation
|
||||
from domain.modelling.scenario import Scenario
|
||||
from domain.modelling.solar_potential import SolarPotential
|
||||
from domain.property.property import Property, PropertyIdentity
|
||||
from domain.property_baseline.rebaseliner import StubRebaseliner
|
||||
from domain.sap10_calculator.calculator import Sap10Calculator
|
||||
from harness.plan_table import format_plan_table
|
||||
from orchestration.ara_first_run_pipeline import AraFirstRunPipeline
|
||||
from orchestration.ingestion_orchestrator import IngestionOrchestrator
|
||||
from orchestration.modelling_orchestrator import ModellingOrchestrator
|
||||
from orchestration.modelling_orchestrator import (
|
||||
ModellingOrchestrator,
|
||||
_candidate_recommendations, # pyright: ignore[reportPrivateUsage]
|
||||
)
|
||||
from orchestration.property_baseline_orchestrator import PropertyBaselineOrchestrator
|
||||
from repositories.fuel_rates.fuel_rates_static_file_repository import (
|
||||
FuelRatesStaticFileRepository,
|
||||
|
|
@ -247,3 +252,31 @@ def run_modelling(
|
|||
if print_table:
|
||||
print("\n" + format_plan_table(plan))
|
||||
return plan
|
||||
|
||||
|
||||
def candidate_recommendations(
|
||||
epc: EpcPropertyData,
|
||||
*,
|
||||
catalogue_path: Path = DEFAULT_CATALOGUE,
|
||||
planning_restrictions: PlanningRestrictions = PlanningRestrictions(),
|
||||
solar_insights: Optional[dict[str, Any]] = None,
|
||||
considered_measures: Optional[frozenset[MeasureType]] = None,
|
||||
products: Optional[ProductRepository] = None,
|
||||
) -> list[Recommendation]:
|
||||
"""Every candidate Recommendation the Generators produce for ``epc`` — the
|
||||
full menu of Measure Options with their per-Option cost, *before* the
|
||||
Optimiser selects a Plan. Use this to inspect measures (and their cost) that
|
||||
a Plan does not end up selecting, e.g. an ASHP the Optimiser passed over for
|
||||
a cheaper route to the target band. Inputs mirror `run_modelling`."""
|
||||
solar_potential: Optional[SolarPotential] = (
|
||||
SolarPotential.from_building_insights(solar_insights)
|
||||
if solar_insights is not None and "solarPotential" in solar_insights
|
||||
else None
|
||||
)
|
||||
return _candidate_recommendations(
|
||||
epc,
|
||||
products or ProductJsonRepository(catalogue_path),
|
||||
planning_restrictions,
|
||||
solar_potential,
|
||||
considered_measures,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -8,10 +8,35 @@ from sqlalchemy import Enum as SAEnum
|
|||
from sqlalchemy.sql import func
|
||||
from sqlmodel import Field, SQLModel
|
||||
|
||||
from domain.modelling.measure_type import MeasureType
|
||||
from domain.modelling.portfolio_goal import PortfolioGoal
|
||||
from domain.modelling.scenario import Scenario
|
||||
|
||||
|
||||
def _parse_exclusions(raw: Optional[str]) -> frozenset[MeasureType]:
|
||||
"""Parse the live ``scenario.exclusions`` column — a Postgres text-array
|
||||
literal like ``{solar_pv,internal_wall_insulation}`` — into the excluded
|
||||
MeasureTypes. Each token must be an exact MeasureType value (no high-level
|
||||
category expansion); an unknown token is a data error and raises, matching
|
||||
the repo's strict-enum convention."""
|
||||
if not raw:
|
||||
return frozenset()
|
||||
inner = raw.strip()
|
||||
if inner.startswith("{") and inner.endswith("}"):
|
||||
inner = inner[1:-1]
|
||||
tokens = [token.strip().strip('"') for token in inner.split(",") if token.strip()]
|
||||
excluded: set[MeasureType] = set()
|
||||
for token in tokens:
|
||||
try:
|
||||
excluded.add(MeasureType(token))
|
||||
except ValueError as error:
|
||||
raise ValueError(
|
||||
f"scenario excludes unknown measure type {token!r}; the "
|
||||
f"exclusions column must hold exact MeasureType values"
|
||||
) from error
|
||||
return frozenset(excluded)
|
||||
|
||||
|
||||
class ScenarioModel(SQLModel, table=True):
|
||||
"""The single SQLModel definition of the live ``scenario`` table (ADR-0017
|
||||
amendment). Full legacy column parity; ``goal`` is the ``PortfolioGoal``
|
||||
|
|
@ -89,4 +114,5 @@ class ScenarioModel(SQLModel, table=True):
|
|||
goal_value=self.goal_value,
|
||||
budget=self.budget,
|
||||
is_default=self.is_default,
|
||||
exclusions=_parse_exclusions(self.exclusions),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -7,7 +7,10 @@ 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.considered_measures import (
|
||||
combine_considered_measures,
|
||||
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
|
||||
|
|
@ -159,18 +162,23 @@ class ModellingOrchestrator:
|
|||
) -> Plan:
|
||||
"""Generate → score → optimise → re-score/repair → attribute → bill →
|
||||
assemble the Plan for one Property + Scenario."""
|
||||
# The Scenario's own exclusions scope the run; an explicit
|
||||
# ``considered_measures`` (e.g. from a harness) narrows it further.
|
||||
considered: Optional[frozenset[MeasureType]] = combine_considered_measures(
|
||||
scenario.considered_measures(), considered_measures
|
||||
)
|
||||
groups: list[list[ScoredOption]] = _scored_candidate_groups(
|
||||
scorer,
|
||||
effective_epc,
|
||||
products,
|
||||
planning_restrictions,
|
||||
solar_potential,
|
||||
considered_measures,
|
||||
considered,
|
||||
)
|
||||
# Forced Measure Dependencies (ventilation) are excluded from the pool
|
||||
# but injected into the package before the re-score (ADR-0016).
|
||||
dependencies: list[MeasureDependency] = _measure_dependencies(
|
||||
effective_epc, products, considered_measures
|
||||
effective_epc, products, considered
|
||||
)
|
||||
package: OptimisedPackage = optimise_package(
|
||||
groups=groups,
|
||||
|
|
@ -249,24 +257,86 @@ def _candidate_recommendations(
|
|||
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. ``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),
|
||||
recommend_roof_insulation(effective_epc, products),
|
||||
recommend_floor_insulation(effective_epc, products),
|
||||
recommend_glazing(effective_epc, products, planning_restrictions),
|
||||
recommend_lighting(effective_epc, products),
|
||||
recommend_heating(effective_epc, products, planning_restrictions),
|
||||
recommend_secondary_heating_removal(effective_epc, products),
|
||||
recommend_solar(
|
||||
effective_epc, products, solar_potential, planning_restrictions
|
||||
"""Run the applicable Recommendation Generators; 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.
|
||||
|
||||
``considered_measures`` gates generation *up front*: a generator runs only
|
||||
when the allowlist admits at least one of the measure types it can emit
|
||||
(None = every measure), so an excluded measure never reaches the catalogue —
|
||||
which matters when the live ``material.type`` enum cannot even represent it
|
||||
(e.g. ``secondary_heating_removal``). ``restrict_to_considered_measures``
|
||||
then trims any disallowed Options off the multi-Option survivors."""
|
||||
|
||||
def admitted(*emits: MeasureType) -> bool:
|
||||
return considered_measures is None or any(
|
||||
measure in considered_measures for measure in emits
|
||||
)
|
||||
|
||||
# Each generator paired with the measure types it can emit, so the allowlist
|
||||
# can skip a generator whose every type is excluded before it is invoked.
|
||||
generators: tuple[
|
||||
tuple[bool, Callable[[], Optional[Recommendation]]], ...
|
||||
] = (
|
||||
(
|
||||
admitted(MeasureType.CAVITY_WALL_INSULATION),
|
||||
lambda: recommend_cavity_wall(effective_epc, products),
|
||||
),
|
||||
(
|
||||
admitted(
|
||||
MeasureType.INTERNAL_WALL_INSULATION,
|
||||
MeasureType.EXTERNAL_WALL_INSULATION,
|
||||
),
|
||||
lambda: recommend_solid_wall(
|
||||
effective_epc, products, planning_restrictions
|
||||
),
|
||||
),
|
||||
(
|
||||
admitted(
|
||||
MeasureType.LOFT_INSULATION,
|
||||
MeasureType.SLOPING_CEILING_INSULATION,
|
||||
MeasureType.FLAT_ROOF_INSULATION,
|
||||
),
|
||||
lambda: recommend_roof_insulation(effective_epc, products),
|
||||
),
|
||||
(
|
||||
admitted(
|
||||
MeasureType.SUSPENDED_FLOOR_INSULATION,
|
||||
MeasureType.SOLID_FLOOR_INSULATION,
|
||||
),
|
||||
lambda: recommend_floor_insulation(effective_epc, products),
|
||||
),
|
||||
(
|
||||
admitted(MeasureType.DOUBLE_GLAZING, MeasureType.SECONDARY_GLAZING),
|
||||
lambda: recommend_glazing(effective_epc, products, planning_restrictions),
|
||||
),
|
||||
(
|
||||
admitted(MeasureType.LOW_ENERGY_LIGHTING),
|
||||
lambda: recommend_lighting(effective_epc, products),
|
||||
),
|
||||
(
|
||||
admitted(
|
||||
MeasureType.HIGH_HEAT_RETENTION_STORAGE_HEATERS,
|
||||
MeasureType.AIR_SOURCE_HEAT_PUMP,
|
||||
MeasureType.GAS_BOILER_UPGRADE,
|
||||
MeasureType.SYSTEM_TUNE_UP,
|
||||
MeasureType.SYSTEM_TUNE_UP_ZONED,
|
||||
),
|
||||
lambda: recommend_heating(effective_epc, products, planning_restrictions),
|
||||
),
|
||||
(
|
||||
admitted(MeasureType.SECONDARY_HEATING_REMOVAL),
|
||||
lambda: recommend_secondary_heating_removal(effective_epc, products),
|
||||
),
|
||||
(
|
||||
admitted(MeasureType.SOLAR_PV),
|
||||
lambda: recommend_solar(
|
||||
effective_epc, products, solar_potential, planning_restrictions
|
||||
),
|
||||
),
|
||||
)
|
||||
found = [thunk() for is_admitted, thunk in generators if is_admitted]
|
||||
applicable = [
|
||||
recommendation for recommendation in found if recommendation is not None
|
||||
]
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ from typing import Any, cast
|
|||
|
||||
from domain.modelling.contingencies import contingency_rate
|
||||
from domain.modelling.product import Product
|
||||
from repositories.product.product_repository import ProductRepository
|
||||
from repositories.product.product_repository import ProductNotFound, ProductRepository
|
||||
|
||||
|
||||
class ProductJsonRepository(ProductRepository):
|
||||
|
|
@ -33,7 +33,7 @@ class ProductJsonRepository(ProductRepository):
|
|||
def get(self, measure_type: str) -> Product:
|
||||
entry: Any = self._entries.get(measure_type)
|
||||
if entry is None:
|
||||
raise ValueError(f"no product for measure type {measure_type!r}")
|
||||
raise ProductNotFound(f"no product for measure type {measure_type!r}")
|
||||
if not isinstance(entry, dict):
|
||||
raise ValueError(f"product {measure_type!r} entry is not an object")
|
||||
typed_entry: dict[str, Any] = cast("dict[str, Any]", entry)
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ from sqlmodel import Session, col, select
|
|||
from domain.modelling.contingencies import contingency_rate
|
||||
from domain.modelling.product import Product
|
||||
from infrastructure.postgres.product_table import MaterialRow
|
||||
from repositories.product.product_repository import ProductRepository
|
||||
from repositories.product.product_repository import ProductNotFound, ProductRepository
|
||||
|
||||
|
||||
# The domain ``MeasureType`` vocabulary and the catalogue's ``material.type``
|
||||
|
|
@ -47,7 +47,9 @@ class ProductPostgresRepository(ProductRepository):
|
|||
.order_by(col(MaterialRow.id))
|
||||
).first()
|
||||
if row is None:
|
||||
raise ValueError(f"no active product for measure type {measure_type!r}")
|
||||
raise ProductNotFound(
|
||||
f"no active product for measure type {measure_type!r}"
|
||||
)
|
||||
if row.total_cost is None:
|
||||
raise ValueError(f"product {measure_type!r} has no total_cost")
|
||||
return Product(
|
||||
|
|
|
|||
|
|
@ -1,10 +1,18 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Optional
|
||||
|
||||
from domain.modelling.product import Product
|
||||
|
||||
|
||||
class ProductNotFound(ValueError):
|
||||
"""Raised when the catalogue has no active entry for a Measure Type. A
|
||||
subclass of ``ValueError`` so existing callers that catch ``ValueError``
|
||||
keep working, while callers that only want to know *whether* a row exists
|
||||
(see ``get_optional``) can catch this case alone."""
|
||||
|
||||
|
||||
class ProductRepository(ABC):
|
||||
"""Loads Products from the catalogue, abstracting the data source (a
|
||||
Postgres-backed materials table today; a JSON file for costs the ETL does
|
||||
|
|
@ -13,6 +21,17 @@ class ProductRepository(ABC):
|
|||
|
||||
@abstractmethod
|
||||
def get(self, measure_type: str) -> Product:
|
||||
"""Return the Product for a Measure Type, raising if there is no active
|
||||
catalogue entry."""
|
||||
"""Return the Product for a Measure Type, raising ``ProductNotFound``
|
||||
if there is no active catalogue entry."""
|
||||
...
|
||||
|
||||
def get_optional(self, measure_type: str) -> Optional[Product]:
|
||||
"""Return the Product for a Measure Type, or None when the catalogue has
|
||||
no active entry. For measures whose cost is composed off-catalogue (e.g.
|
||||
ASHP, priced from the rate sheet per ADR-0025) the catalogue row is read
|
||||
only for its id, so a missing row is not an error — the measure is still
|
||||
offered, just without a ``material_id``."""
|
||||
try:
|
||||
return self.get(measure_type)
|
||||
except ProductNotFound:
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -14,10 +14,19 @@ persists **nothing** (the run is for inspecting recommendations); pass
|
|||
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.
|
||||
Pass `--scenario-id` to target a real Scenario; its ``goal_value`` drives the
|
||||
band and **its ``exclusions`` drive which measures the run considers** (the live
|
||||
scenario table persists exclusions only, no inclusions). Without `--scenario-id`
|
||||
the run synthesises an Increasing-EPC-to-``--goal`` Scenario with no exclusions.
|
||||
`--measures` / `--exclude-measures` are optional overlays layered on top of the
|
||||
Scenario's own exclusions.
|
||||
|
||||
KNOWN GOTCHA: the live ``material.type`` enum does not yet carry
|
||||
``secondary_heating_removal``, so a Property with a lodged secondary heater
|
||||
crashes the catalogue read for that measure. Until the catalogue stocks it, pass
|
||||
``--exclude-measures secondary_heating_removal`` (an ASHP row is also absent, but
|
||||
ASHP is priced off the rate sheet so it degrades to ``material_id=None`` rather
|
||||
than crashing — no flag needed).
|
||||
|
||||
Config: loads `backend/.env` for the DB creds (`DB_*`), the EPC API token
|
||||
(`OPEN_EPC_API_TOKEN` — the Bearer token for the new gov API), the Google Solar
|
||||
|
|
@ -25,13 +34,13 @@ 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:
|
||||
|
||||
# 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
|
||||
# inspect only (no DB writes), Scenario 1266, measures from the Scenario:
|
||||
python -m scripts.run_modelling_e2e --scenario-id 1266 \
|
||||
--exclude-measures secondary_heating_removal 709634 709635 709636
|
||||
# same run, but persist EPC + spatial + solar + Plan (needs --portfolio-id):
|
||||
python -m scripts.run_modelling_e2e --scenario-id 1266 --portfolio-id 785 \
|
||||
--persist --exclude-measures secondary_heating_removal 709634 709635
|
||||
python -m scripts.run_modelling_e2e --no-solar 709634 709635 # skip Google leg
|
||||
|
||||
Per Property the spatial reference (S3 Open-UPRN parquet) gives the planning
|
||||
protections (conservation/listed/heritage — gate the wall + solar measures) and
|
||||
|
|
@ -66,10 +75,14 @@ from repositories.property.property_overrides_postgres_reader import ( # noqa:
|
|||
)
|
||||
from domain.geospatial.planning_restrictions import PlanningRestrictions # noqa: E402
|
||||
from domain.geospatial.spatial_reference import SpatialReference # noqa: E402
|
||||
from domain.modelling.considered_measures import ( # noqa: E402
|
||||
combine_considered_measures,
|
||||
)
|
||||
from domain.modelling.measure_type import MeasureType # noqa: E402
|
||||
from domain.modelling.plan import Plan, PlanMeasure # noqa: E402
|
||||
from domain.modelling.recommendation import Recommendation # noqa: E402
|
||||
from domain.modelling.scenario import Scenario # noqa: E402
|
||||
from harness.console import run_modelling # noqa: E402
|
||||
from harness.console import candidate_recommendations, 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
|
||||
|
|
@ -93,6 +106,7 @@ from sqlmodel import Session # noqa: E402
|
|||
_ENV_PATH = _REPO_ROOT / "backend" / ".env"
|
||||
_MARKDOWN_PATH = Path("modelling_e2e.md")
|
||||
_CSV_PATH = Path("modelling_e2e.csv")
|
||||
_CANDIDATES_CSV_PATH = Path("modelling_e2e_candidates.csv")
|
||||
|
||||
|
||||
def _load_env(path: Path) -> None:
|
||||
|
|
@ -219,6 +233,21 @@ def _parse_measures(raw: Optional[str]) -> Optional[frozenset[MeasureType]]:
|
|||
)
|
||||
|
||||
|
||||
def _resolve_considered(
|
||||
allowlist: Optional[frozenset[MeasureType]],
|
||||
excluded: Optional[frozenset[MeasureType]],
|
||||
) -> Optional[frozenset[MeasureType]]:
|
||||
"""Combine the `--measures` allowlist with the `--exclude-measures` set. With
|
||||
no exclusions the allowlist is returned unchanged (None = every measure).
|
||||
With exclusions the result is (the allowlist, or every measure) minus the
|
||||
excluded types — so `--exclude-measures secondary_heating_removal` considers
|
||||
every measure except that one, without enumerating the rest."""
|
||||
if not excluded:
|
||||
return allowlist
|
||||
base = allowlist if allowlist is not None else frozenset(MeasureType)
|
||||
return base - excluded
|
||||
|
||||
|
||||
def _context_summary(
|
||||
spatial: Optional[SpatialReference], solar_insights: Optional[dict[str, Any]]
|
||||
) -> str:
|
||||
|
|
@ -249,6 +278,53 @@ def _measure_summary(measure: PlanMeasure) -> str:
|
|||
)
|
||||
|
||||
|
||||
def _candidate_lines(
|
||||
recommendations: list[Recommendation], selected: set[MeasureType]
|
||||
) -> list[str]:
|
||||
"""Render every candidate Option (the full menu the Generators produced,
|
||||
not just the Plan the Optimiser selected) with its per-Option cost, flagging
|
||||
the Options that made it into the Plan — so measures the Optimiser passed
|
||||
over (e.g. an ASHP it found too costly for the target band) are visible."""
|
||||
lines: list[str] = []
|
||||
for recommendation in recommendations:
|
||||
for option in recommendation.options:
|
||||
cost = option.cost
|
||||
cost_note = (
|
||||
f"£{cost.total:,.0f} (+{cost.contingency_rate * 100:.0f}% cont.)"
|
||||
if cost is not None
|
||||
else "no cost"
|
||||
)
|
||||
flag = " ✓ SELECTED" if option.measure_type in selected else ""
|
||||
lines.append(
|
||||
f" [{recommendation.surface}] {option.measure_type} · "
|
||||
f"{cost_note}{flag} — {option.description}"
|
||||
)
|
||||
return lines
|
||||
|
||||
|
||||
def _candidate_csv_rows(
|
||||
property_id: int,
|
||||
uprn: Optional[int],
|
||||
recommendations: list[Recommendation],
|
||||
selected: set[MeasureType],
|
||||
) -> list[str]:
|
||||
"""One CSV row per candidate Option: the full measure menu with cost,
|
||||
contingency, and whether the Optimiser selected it."""
|
||||
rows: list[str] = []
|
||||
for recommendation in recommendations:
|
||||
for option in recommendation.options:
|
||||
cost = option.cost
|
||||
total = f"{cost.total:.2f}" if cost is not None else ""
|
||||
contingency = f"{cost.contingency_rate:.4f}" if cost is not None else ""
|
||||
chosen = "yes" if option.measure_type in selected else "no"
|
||||
description = option.description.replace(",", ";")
|
||||
rows.append(
|
||||
f"{property_id},{uprn or ''},{recommendation.surface},"
|
||||
f"{option.measure_type},{total},{contingency},{chosen},{description}"
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
def _persist(
|
||||
engine: Engine,
|
||||
*,
|
||||
|
|
@ -303,7 +379,15 @@ def main() -> None:
|
|||
parser.add_argument(
|
||||
"--measures",
|
||||
default=None,
|
||||
help="comma-separated measure types to consider (default: all)",
|
||||
help="optional override: comma-separated measure types to consider. The "
|
||||
"Scenario's exclusions already drive this; the flag narrows it further.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--exclude-measures",
|
||||
default=None,
|
||||
help="optional override: comma-separated measure types to exclude on top "
|
||||
"of the Scenario's own exclusions (e.g. secondary_heating_removal, which "
|
||||
"the live catalogue does not yet stock)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--portfolio-id",
|
||||
|
|
@ -335,7 +419,9 @@ def main() -> None:
|
|||
geospatial = GeospatialS3Repository(_s3_parquet_reader(os.environ["DATA_BUCKET"]))
|
||||
solar_client = GoogleSolarApiClient(os.environ["GOOGLE_SOLAR_API_KEY"])
|
||||
engine = _engine()
|
||||
considered = _parse_measures(args.measures)
|
||||
cli_considered = _resolve_considered(
|
||||
_parse_measures(args.measures), _parse_measures(args.exclude_measures)
|
||||
)
|
||||
uprns = _uprns_for(engine, args.property_ids)
|
||||
# Landlord Overrides are read from property_overrides and folded onto the lodged
|
||||
# EPC to form the Effective EPC the calculator scores (ADR-0032).
|
||||
|
|
@ -350,6 +436,12 @@ def main() -> None:
|
|||
if args.scenario_id is not None
|
||||
else None
|
||||
)
|
||||
# The Scenario's own exclusions drive which measures the run considers; the
|
||||
# --measures/--exclude-measures flags are an optional override layered on top.
|
||||
considered = combine_considered_measures(
|
||||
scenario.considered_measures() if scenario is not None else None,
|
||||
cli_considered,
|
||||
)
|
||||
|
||||
target = (
|
||||
f"scenario {scenario.id} (band {scenario.goal_value})"
|
||||
|
|
@ -367,6 +459,10 @@ def main() -> None:
|
|||
csv_rows: list[str] = [
|
||||
"property_id,uprn,baseline_sap,post_sap,measures,measure_types,cost_of_works"
|
||||
]
|
||||
candidate_csv_rows: list[str] = [
|
||||
"property_id,uprn,surface,measure_type,cost_total,contingency_rate,"
|
||||
"selected,description"
|
||||
]
|
||||
|
||||
for property_id in args.property_ids:
|
||||
uprn = uprns.get(property_id)
|
||||
|
|
@ -408,6 +504,15 @@ def main() -> None:
|
|||
scenario=scenario,
|
||||
print_table=False,
|
||||
)
|
||||
# The full candidate menu (every Generator Option + its cost), so
|
||||
# measures the Optimiser did not select are still visible.
|
||||
candidates: list[Recommendation] = candidate_recommendations(
|
||||
epc,
|
||||
planning_restrictions=restrictions,
|
||||
solar_insights=solar_insights,
|
||||
considered_measures=considered,
|
||||
products=products,
|
||||
)
|
||||
if args.persist:
|
||||
assert scenario is not None # guaranteed by the --persist guard
|
||||
_persist(
|
||||
|
|
@ -436,7 +541,9 @@ def main() -> None:
|
|||
continue
|
||||
|
||||
measure_types = [m.measure_type for m in plan.measures]
|
||||
selected: set[MeasureType] = {m.measure_type for m in plan.measures}
|
||||
context = _context_summary(spatial, solar_insights)
|
||||
candidate_lines = _candidate_lines(candidates, selected)
|
||||
header = (
|
||||
f"=== Property {property_id} (uprn {uprn}) === "
|
||||
f"SAP {plan.baseline.sap_continuous:.1f} -> {plan.post_sap_continuous:.1f} "
|
||||
|
|
@ -444,6 +551,9 @@ def main() -> None:
|
|||
)
|
||||
print(header)
|
||||
print(format_plan_table(plan))
|
||||
print(f" candidate measures considered ({len(candidate_lines)} option(s)):")
|
||||
for candidate_line in candidate_lines:
|
||||
print(candidate_line)
|
||||
print()
|
||||
|
||||
md_lines.append(f"## Property {property_id} (uprn {uprn})\n")
|
||||
|
|
@ -452,19 +562,30 @@ def main() -> None:
|
|||
f"· {len(plan.measures)} measure(s) · cost £{plan.cost_of_works:,.0f} "
|
||||
f"· {context}\n"
|
||||
)
|
||||
md_lines.append("**Selected Plan**\n")
|
||||
md_lines.extend(_measure_summary(m) for m in plan.measures)
|
||||
md_lines.append("")
|
||||
md_lines.append("**All candidate measures (cost per measure)**\n")
|
||||
md_lines.extend(candidate_lines)
|
||||
md_lines.append("")
|
||||
csv_rows.append(
|
||||
f"{property_id},{uprn},{plan.baseline.sap_continuous:.2f},"
|
||||
f"{plan.post_sap_continuous:.2f},{len(plan.measures)},"
|
||||
f"{'|'.join(measure_types)},{plan.cost_of_works:.0f}"
|
||||
)
|
||||
candidate_csv_rows.extend(
|
||||
_candidate_csv_rows(property_id, uprn, candidates, selected)
|
||||
)
|
||||
|
||||
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")
|
||||
_CANDIDATES_CSV_PATH.write_text(
|
||||
"\n".join(candidate_csv_rows) + "\n", encoding="utf-8"
|
||||
)
|
||||
print(f"wrote {_MARKDOWN_PATH.resolve()}")
|
||||
print(f"wrote {_CSV_PATH.resolve()}")
|
||||
print(f"wrote {_CANDIDATES_CSV_PATH.resolve()}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
|
|
@ -8,7 +8,10 @@ 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.considered_measures import (
|
||||
combine_considered_measures,
|
||||
restrict_to_considered_measures,
|
||||
)
|
||||
from domain.modelling.measure_type import MeasureType
|
||||
from domain.modelling.recommendation import MeasureOption, Recommendation
|
||||
from domain.modelling.simulation import EpcSimulation
|
||||
|
|
@ -68,6 +71,18 @@ def test_drops_recommendations_with_no_allowed_option() -> None:
|
|||
assert surfaces == {"Heating & Hot Water", "Solar PV"}
|
||||
|
||||
|
||||
def test_combine_treats_none_as_all_and_intersects_two_allowlists() -> None:
|
||||
# Arrange
|
||||
a = frozenset({MeasureType.SOLAR_PV, MeasureType.LOFT_INSULATION})
|
||||
b = frozenset({MeasureType.SOLAR_PV, MeasureType.CAVITY_WALL_INSULATION})
|
||||
|
||||
# Act / Assert — None means "all", so it never narrows; two sets intersect.
|
||||
assert combine_considered_measures(None, b) == b
|
||||
assert combine_considered_measures(a, None) == a
|
||||
assert combine_considered_measures(None, None) is None
|
||||
assert combine_considered_measures(a, b) == frozenset({MeasureType.SOLAR_PV})
|
||||
|
||||
|
||||
def test_filters_options_within_a_kept_recommendation() -> None:
|
||||
# Arrange — HHRSH is allowed but the competing ASHP bundle is not.
|
||||
considered = frozenset(
|
||||
|
|
|
|||
|
|
@ -8,10 +8,14 @@ later slices. Detection + pricing only; impact is produced by scoring (ADR-0016)
|
|||
from datatypes.epc.domain.epc_property_data import EpcPropertyData
|
||||
from domain.geospatial.planning_restrictions import PlanningRestrictions
|
||||
from domain.modelling.generators.heating_recommendation import recommend_heating
|
||||
from domain.modelling.measure_type import MeasureType
|
||||
from domain.modelling.product import Product
|
||||
from domain.modelling.recommendation import Recommendation
|
||||
from domain.modelling.simulation import HeatingOverlay
|
||||
from repositories.product.product_repository import ProductRepository
|
||||
from repositories.product.product_repository import (
|
||||
ProductNotFound,
|
||||
ProductRepository,
|
||||
)
|
||||
from tests.domain.modelling._elmhurst_recommendation import (
|
||||
parse_recommendation_summary,
|
||||
)
|
||||
|
|
@ -170,6 +174,41 @@ def test_gas_boiler_house_yields_an_ashp_bundle() -> None:
|
|||
)
|
||||
|
||||
|
||||
class _StubProductsWithoutAshp(ProductRepository):
|
||||
"""A catalogue with no ASHP row. ASHP's cost is composed from the rate sheet
|
||||
(ADR-0025) and the catalogue row is read only for its id, so a missing row
|
||||
must not suppress the bundle — it just carries no material_id."""
|
||||
|
||||
def get(self, measure_type: str) -> Product:
|
||||
if measure_type == MeasureType.AIR_SOURCE_HEAT_PUMP:
|
||||
raise ProductNotFound(f"no active product for {measure_type!r}")
|
||||
return Product(
|
||||
measure_type=measure_type, unit_cost_per_m2=3500.0, contingency_rate=0.26
|
||||
)
|
||||
|
||||
|
||||
def test_ashp_bundle_offered_when_catalogue_lacks_an_ashp_product() -> None:
|
||||
# Arrange — a mains-gas house (ASHP-eligible) priced against a catalogue with
|
||||
# no ASHP row; ASHP is costed from the rate sheet, so the bundle must still
|
||||
# be offered, just without a material id.
|
||||
baseline: EpcPropertyData = _gas_boiler_house()
|
||||
|
||||
# Act
|
||||
recommendation: Recommendation | None = recommend_heating(
|
||||
baseline, _StubProductsWithoutAshp()
|
||||
)
|
||||
|
||||
# Assert — the ASHP bundle is still offered, carrying its composite cost and
|
||||
# no material id.
|
||||
assert recommendation is not None
|
||||
option = next(
|
||||
o for o in recommendation.options if o.measure_type == "air_source_heat_pump"
|
||||
)
|
||||
assert option.material_id is None
|
||||
assert option.cost is not None
|
||||
assert option.cost.total > 0.0
|
||||
|
||||
|
||||
def test_ashp_bundle_carries_the_composite_per_dwelling_cost() -> None:
|
||||
# Arrange — a mains-gas regular boiler with a cylinder (90 m2, 7 habitable
|
||||
# rooms): the ASHP reuses the existing wet system (ADR-0025).
|
||||
|
|
|
|||
43
tests/domain/modelling/test_scenario.py
Normal file
43
tests/domain/modelling/test_scenario.py
Normal file
|
|
@ -0,0 +1,43 @@
|
|||
"""The Scenario's measure scoping: its exclusions imply the allowlist the run
|
||||
considers (the live `scenario` table persists exclusions only — no inclusions)."""
|
||||
|
||||
from domain.modelling.measure_type import MeasureType
|
||||
from domain.modelling.scenario import Scenario
|
||||
|
||||
|
||||
def _scenario(exclusions: frozenset[MeasureType]) -> Scenario:
|
||||
return Scenario(
|
||||
id=1,
|
||||
goal="Increasing EPC",
|
||||
goal_value="C",
|
||||
budget=None,
|
||||
is_default=True,
|
||||
exclusions=exclusions,
|
||||
)
|
||||
|
||||
|
||||
def test_no_exclusions_considers_every_measure() -> None:
|
||||
# Arrange
|
||||
scenario = _scenario(frozenset())
|
||||
|
||||
# Act
|
||||
considered = scenario.considered_measures()
|
||||
|
||||
# Assert — None means "consider all" (the unrestricted default).
|
||||
assert considered is None
|
||||
|
||||
|
||||
def test_exclusions_imply_the_complement_allowlist() -> None:
|
||||
# Arrange — exclude solar PV and ASHP.
|
||||
scenario = _scenario(
|
||||
frozenset({MeasureType.SOLAR_PV, MeasureType.AIR_SOURCE_HEAT_PUMP})
|
||||
)
|
||||
|
||||
# Act
|
||||
considered = scenario.considered_measures()
|
||||
|
||||
# Assert — every modelled measure survives except the two excluded ones.
|
||||
assert considered is not None
|
||||
assert MeasureType.SOLAR_PV not in considered
|
||||
assert MeasureType.AIR_SOURCE_HEAT_PUMP not in considered
|
||||
assert considered == frozenset(MeasureType) - scenario.exclusions
|
||||
|
|
@ -10,9 +10,16 @@ from datatypes.epc.domain.epc import Epc
|
|||
from datatypes.epc.domain.epc_property_data import EpcPropertyData
|
||||
from domain.geospatial.planning_restrictions import PlanningRestrictions
|
||||
from domain.modelling.contingencies import contingency_rate
|
||||
from domain.modelling.measure_type import MeasureType
|
||||
from domain.modelling.scenario import Scenario
|
||||
from domain.modelling.generators.heating_recommendation import recommend_heating
|
||||
from domain.modelling.generators.solid_wall_recommendation import recommend_solid_wall
|
||||
from harness.console import DEFAULT_CATALOGUE, run_modelling, run_one
|
||||
from harness.console import (
|
||||
DEFAULT_CATALOGUE,
|
||||
candidate_recommendations,
|
||||
run_modelling,
|
||||
run_one,
|
||||
)
|
||||
from repositories.product.product_json_repository import ProductJsonRepository
|
||||
from tests.domain.modelling._elmhurst_recommendation import (
|
||||
parse_recommendation_summary,
|
||||
|
|
@ -224,6 +231,54 @@ def test_run_modelling_recommends_hhr_storage_for_an_electric_dwelling() -> None
|
|||
assert "air_source_heat_pump" in {m.measure_type for m in plan.measures}
|
||||
|
||||
|
||||
def test_candidate_recommendations_surface_unselected_options_with_cost() -> None:
|
||||
# Arrange — an electric dwelling whose heating Recommendation offers both an
|
||||
# ASHP and an HHR-storage bundle; the Optimiser selects only one of them.
|
||||
epc: EpcPropertyData = _electric_storage_lit_epc()
|
||||
|
||||
# Act — the full candidate menu (every Generator Option, pre-optimisation)
|
||||
# alongside the optimised Plan.
|
||||
candidates = candidate_recommendations(epc)
|
||||
plan = run_modelling(epc, goal_band="C", print_table=False)
|
||||
|
||||
# Assert — the menu carries every offered Option (so a measure the Plan did
|
||||
# not select, like the passed-over HHR bundle, is still inspectable), and
|
||||
# every Option carries a cost so "cost per measure" is always available.
|
||||
offered = {
|
||||
option.measure_type for rec in candidates for option in rec.options
|
||||
}
|
||||
selected = {measure.measure_type for measure in plan.measures}
|
||||
assert "high_heat_retention_storage_heaters" in offered
|
||||
assert "air_source_heat_pump" in offered
|
||||
assert offered - selected # at least one offered measure was not selected
|
||||
assert all(
|
||||
option.cost is not None for rec in candidates for option in rec.options
|
||||
)
|
||||
|
||||
|
||||
def test_run_modelling_honours_a_scenarios_exclusions() -> None:
|
||||
# Arrange — an electric dwelling whose optimised Plan normally leads with the
|
||||
# ASHP bundle; a Scenario that excludes ASHP must keep it out of the Plan.
|
||||
epc: EpcPropertyData = _electric_storage_lit_epc()
|
||||
scenario = Scenario(
|
||||
id=1,
|
||||
goal="Increasing EPC",
|
||||
goal_value="C",
|
||||
budget=None,
|
||||
is_default=True,
|
||||
exclusions=frozenset({MeasureType.AIR_SOURCE_HEAT_PUMP}),
|
||||
)
|
||||
|
||||
# Act — the orchestrator derives the allowlist from the Scenario's exclusions.
|
||||
plan = run_modelling(epc, scenario=scenario, print_table=False)
|
||||
|
||||
# Assert — ASHP is excluded; the Plan still improves the dwelling via other
|
||||
# measures (e.g. the HHR storage bundle).
|
||||
measure_types = {measure.measure_type for measure in plan.measures}
|
||||
assert MeasureType.AIR_SOURCE_HEAT_PUMP not in measure_types
|
||||
assert measure_types # a non-empty Plan still came back
|
||||
|
||||
|
||||
def test_sample_catalogue_prices_every_generator_measure_type() -> None:
|
||||
# Arrange — the default offline catalogue.
|
||||
products: ProductJsonRepository = ProductJsonRepository(DEFAULT_CATALOGUE)
|
||||
|
|
|
|||
|
|
@ -116,6 +116,53 @@ def test_candidate_recommendations_excludes_solar_without_potential() -> None:
|
|||
assert "Solar PV" not in {r.surface for r in recommendations}
|
||||
|
||||
|
||||
class _ProductsRaisingFor(ProductRepository):
|
||||
"""A catalogue that raises for one measure type — mirroring the live DB,
|
||||
whose ``material.type`` enum does not carry ``secondary_heating_removal``.
|
||||
Invoking that measure's generator would raise, so this proves an excluded
|
||||
generator is never run."""
|
||||
|
||||
def __init__(self, forbidden: MeasureType) -> None:
|
||||
self._forbidden = forbidden
|
||||
|
||||
def get(self, measure_type: str) -> Product:
|
||||
if measure_type == self._forbidden:
|
||||
raise ValueError(f"catalogue cannot represent {measure_type!r}")
|
||||
return Product(
|
||||
measure_type=measure_type,
|
||||
unit_cost_per_m2=0.0,
|
||||
contingency_rate=0.15,
|
||||
id=909,
|
||||
)
|
||||
|
||||
|
||||
def test_an_excluded_measures_generator_is_not_invoked() -> None:
|
||||
# Arrange — a dwelling with a lodged secondary heating system (so the
|
||||
# secondary-heating generator is eligible to fire) priced against a catalogue
|
||||
# that raises for that type, exactly as the live `material.type` enum does.
|
||||
epc = _eligible_house()
|
||||
epc.sap_heating.secondary_heating_type = 631
|
||||
allowlist = frozenset(MeasureType) - {MeasureType.SECONDARY_HEATING_REMOVAL}
|
||||
|
||||
# Act — excluding the measure must stop its generator running at all (it would
|
||||
# otherwise query the catalogue and raise).
|
||||
recommendations = _candidate_recommendations(
|
||||
epc,
|
||||
_ProductsRaisingFor(MeasureType.SECONDARY_HEATING_REMOVAL),
|
||||
PlanningRestrictions(),
|
||||
None,
|
||||
allowlist,
|
||||
)
|
||||
|
||||
# Assert — the run completes and no secondary-heating option leaks through.
|
||||
option_types = {
|
||||
option.measure_type
|
||||
for recommendation in recommendations
|
||||
for option in recommendation.options
|
||||
}
|
||||
assert MeasureType.SECONDARY_HEATING_REMOVAL not in option_types
|
||||
|
||||
|
||||
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.
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ import pytest
|
|||
from sqlalchemy import Engine
|
||||
from sqlmodel import Session
|
||||
|
||||
from domain.modelling.measure_type import MeasureType
|
||||
from domain.modelling.portfolio_goal import PortfolioGoal
|
||||
from domain.modelling.scenario import Scenario
|
||||
from infrastructure.postgres.modelling import ScenarioModel
|
||||
|
|
@ -67,6 +68,80 @@ def test_get_many_maps_live_scenario_rows_to_domain_in_input_order(
|
|||
)
|
||||
|
||||
|
||||
def test_get_many_parses_the_exclusions_array_into_measure_types(
|
||||
db_engine: Engine,
|
||||
) -> None:
|
||||
# Arrange — the live `exclusions` column is a Postgres text-array literal of
|
||||
# exact MeasureType values.
|
||||
with Session(db_engine) as session:
|
||||
session.add(
|
||||
ScenarioModel(
|
||||
id=7,
|
||||
goal=PortfolioGoal.INCREASING_EPC,
|
||||
goal_value="C",
|
||||
is_default=True,
|
||||
exclusions="{solar_pv,internal_wall_insulation}",
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
|
||||
# Act
|
||||
with Session(db_engine) as session:
|
||||
scenario: Scenario = ScenarioPostgresRepository(session).get_many([7])[0]
|
||||
|
||||
# Assert
|
||||
assert scenario.exclusions == frozenset(
|
||||
{MeasureType.SOLAR_PV, MeasureType.INTERNAL_WALL_INSULATION}
|
||||
)
|
||||
|
||||
|
||||
def test_get_many_treats_a_null_exclusions_column_as_no_exclusions(
|
||||
db_engine: Engine,
|
||||
) -> None:
|
||||
# Arrange
|
||||
with Session(db_engine) as session:
|
||||
session.add(
|
||||
ScenarioModel(
|
||||
id=7,
|
||||
goal=PortfolioGoal.INCREASING_EPC,
|
||||
goal_value="C",
|
||||
is_default=True,
|
||||
exclusions=None,
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
|
||||
# Act
|
||||
with Session(db_engine) as session:
|
||||
scenario: Scenario = ScenarioPostgresRepository(session).get_many([7])[0]
|
||||
|
||||
# Assert
|
||||
assert scenario.exclusions == frozenset()
|
||||
|
||||
|
||||
def test_get_many_raises_on_an_exclusion_that_is_not_a_measure_type(
|
||||
db_engine: Engine,
|
||||
) -> None:
|
||||
# Arrange — a legacy high-level category (`heating`) is not an exact
|
||||
# MeasureType value; exact-only resolution must reject it loudly.
|
||||
with Session(db_engine) as session:
|
||||
session.add(
|
||||
ScenarioModel(
|
||||
id=7,
|
||||
goal=PortfolioGoal.INCREASING_EPC,
|
||||
goal_value="C",
|
||||
is_default=True,
|
||||
exclusions="{heating}",
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
|
||||
# Act / Assert
|
||||
with Session(db_engine) as session:
|
||||
with pytest.raises(ValueError):
|
||||
ScenarioPostgresRepository(session).get_many([7])
|
||||
|
||||
|
||||
def test_get_many_raises_when_a_scenario_id_is_missing(db_engine: Engine) -> None:
|
||||
# Arrange
|
||||
with Session(db_engine) as session:
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue