diff --git a/domain/modelling/generators/solar_recommendation.py b/domain/modelling/generators/solar_recommendation.py index ed97f863..842e5a99 100644 --- a/domain/modelling/generators/solar_recommendation.py +++ b/domain/modelling/generators/solar_recommendation.py @@ -13,6 +13,19 @@ selection, the overlay and `recommend_solar` land in later slices. from __future__ import annotations +from typing import Optional + +from datatypes.epc.domain.epc_property_data import ( + EpcPropertyData, + PhotovoltaicArray, + PvBatteries, + PvBattery, +) +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.recommendation import Cost, MeasureOption, Recommendation +from domain.modelling.simulation import EpcSimulation, SolarOverlay from domain.modelling.solar_potential import ( SolarPanelConfiguration, SolarPotential, @@ -21,6 +34,20 @@ from domain.modelling.solar_potential import ( from domain.sap10_calculator.rdsap.cert_to_inputs import ( pv_annual_solar_radiation_kwh_per_m2, ) +from repositories.product.product_repository import ProductRepository + +_SOLAR_SURFACE = "Solar PV" +_SOLAR_MEASURE_TYPE = "solar_pv" + +# The fixed, representative battery capacity for the with-battery variant +# (ADR-0026) — a flagged estimate (see the rate sheet), 5 kWh. +_BATTERY_CAPACITY_KWH = 5.0 +# Watts → kilowatts for peak-power. +_WATTS_PER_KW = 1000.0 +# The dwelling's PV connects to its own meter (the after-cert §19 "Connected to +# the dwelling's meter: Yes"). Non-load-bearing for the SAP cascade; carried for +# fidelity. 1 = connected, the modal install case. +_PV_CONNECTED_TO_DWELLING = 1 # A roof plane within this many degrees of due north (0°/360°, Google compass # convention) is dropped: it generates little and is not worth panelling. The @@ -145,3 +172,140 @@ def select_conservative_configs( ) unique = [unique[index] for index in sampled_indices] return tuple(sorted(unique, key=lambda c: c.panels_count)) + + +def _array_for_segment( + segment: SolarRoofSegment, panel_capacity_watts: float +) -> PhotovoltaicArray: + """Project a chosen roof segment into a SAP `PhotovoltaicArray`: peak power + from its panels, orientation/pitch from its geometry, and the + generation-calibrated overshading code (ADR-0026).""" + return PhotovoltaicArray( + peak_power=segment.panels_count * panel_capacity_watts / _WATTS_PER_KW, + pitch=segment.sap_pitch_code, + orientation=segment.sap_orientation, + overshading=segment_overshading_code(segment, panel_capacity_watts), + ) + + +def _solar_overlay( + config: SolarPanelConfiguration, + panel_capacity_watts: float, + has_cylinder: bool, + with_battery: bool, +) -> SolarOverlay: + """Build the `SolarOverlay` for one array config variant: one + `PhotovoltaicArray` per segment, export ensured, a diverter when the + dwelling has a cylinder, and a battery for the with-battery variant.""" + return SolarOverlay( + photovoltaic_arrays=[ + _array_for_segment(segment, panel_capacity_watts) + for segment in config.segments + ], + # App G4 routes surplus PV to the cylinder immersion; a combi has nothing + # to divert to, so leave the field unset (None) when there is no cylinder. + pv_diverter_present=True if has_cylinder else None, + pv_connection=_PV_CONNECTED_TO_DWELLING, + is_dwelling_export_capable=True, + pv_batteries=( + PvBatteries(pv_battery=PvBattery(battery_capacity=_BATTERY_CAPACITY_KWH)) + if with_battery + else None + ), + ) + + +def _option( + config: SolarPanelConfiguration, + panel_capacity_watts: float, + has_cylinder: bool, + with_battery: bool, + products: ProductRepository, +) -> MeasureOption: + """Assemble one competing Solar PV Measure Option for a config variant.""" + peak_power_kwp: float = config.panels_count * panel_capacity_watts / _WATTS_PER_KW + cost: Cost = Products().solar_bundle_cost( + SolarCostInputs( + peak_power_kwp=peak_power_kwp, + has_cylinder=has_cylinder, + has_battery=with_battery, + ) + ) + battery_suffix: str = " with a 5 kWh battery" if with_battery else "" + description: str = ( + f"Install a {peak_power_kwp:.1f} kWp roof-mounted solar PV array" + f"{battery_suffix}, ensuring an export meter" + ) + return MeasureOption( + measure_type=_SOLAR_MEASURE_TYPE, + description=description, + overlay=EpcSimulation( + solar=_solar_overlay( + config, panel_capacity_watts, has_cylinder, with_battery + ) + ), + cost=cost, + material_id=products.get(_SOLAR_MEASURE_TYPE).id, + ) + + +def recommend_solar( + epc: EpcPropertyData, + products: ProductRepository, + solar_potential: Optional[SolarPotential], + restrictions: PlanningRestrictions = PlanningRestrictions(), +) -> Optional[Recommendation]: + """Return a "Solar PV" Recommendation of competing whole-array Options — + up to five conservatively-sized configs × {no battery, battery} — for an + eligible dwelling with feasible Google solar potential, else None + (ADR-0026). A free Optimiser candidate; the Optimiser owns whether and at + what size to install it.""" + if solar_potential is None or not _solar_eligible(epc, restrictions): + return None + configs: tuple[SolarPanelConfiguration, ...] = select_conservative_configs( + solar_potential + ) + if not configs: + return None + has_cylinder: bool = bool(epc.has_hot_water_cylinder) + capacity: float = solar_potential.panel_capacity_watts + options: list[MeasureOption] = [ + _option(config, capacity, has_cylinder, with_battery, products) + for config in configs + for with_battery in (False, True) + ] + return Recommendation(surface=_SOLAR_SURFACE, options=tuple(options)) + + +def _solar_eligible( + epc: EpcPropertyData, restrictions: PlanningRestrictions +) -> bool: + """Solar PV suits a non-flat house/bungalow that is not fabric-protected and + has no existing PV (ADR-0026). Eligibility encodes only physical/legal + installability — the Optimiser owns the economics. A conservation area does + NOT block PV (offered, installed sympathetically); a listed/heritage + protection (`blocks_internal`) does — the same gate as ASHP.""" + if restrictions.blocks_internal: + return False + if not _is_house_or_bungalow(epc): + return False + return not _has_existing_pv(epc) + + +def _has_existing_pv(epc: EpcPropertyData) -> bool: + """Whether the dwelling already has PV — the *existing* arrays on the EPC + (existing-PV top-up is deferred), distinct from the Google potential.""" + arrays: Optional[list[PhotovoltaicArray]] = epc.sap_energy_source.photovoltaic_arrays + return bool(arrays) + + +def _is_house_or_bungalow(epc: EpcPropertyData) -> bool: + """Whether the dwelling is a house or bungalow (not a flat/maisonette). The + Elmhurst path lodges the name; the API path a stringified RdSAP code + (`PROPERTY_TYPE_LOOKUP`: 0 House, 1 Bungalow, 2 Flat, 3 Maisonette).""" + raw: str = (epc.property_type or "").strip() + if raw.lower() in ("house", "bungalow"): + return True + if raw.isdigit(): + return PROPERTY_TYPE_LOOKUP.get(int(raw)) in ("House", "Bungalow") + return False diff --git a/tests/domain/modelling/test_solar_recommendation.py b/tests/domain/modelling/test_solar_recommendation.py new file mode 100644 index 00000000..9d51b0d8 --- /dev/null +++ b/tests/domain/modelling/test_solar_recommendation.py @@ -0,0 +1,238 @@ +"""Behaviour of the Solar PV Recommendation Generator (ADR-0026): one "Solar +PV" Recommendation of competing whole-array Options — up to five +conservatively-sized configs × {no battery, battery} — built from a typed +`SolarPotential`. Detection + pricing only; impact is produced by scoring. +""" + +import json +from dataclasses import replace +from pathlib import Path +from typing import Any, Optional + +from datatypes.epc.domain.epc_property_data import EpcPropertyData, PhotovoltaicArray +from domain.geospatial.planning_restrictions import PlanningRestrictions +from domain.modelling.generators.solar_recommendation import recommend_solar +from domain.modelling.product import Product +from domain.modelling.recommendation import Recommendation +from domain.modelling.solar_potential import SolarPotential +from repositories.product.product_repository import ProductRepository +from tests.domain.modelling._elmhurst_recommendation import ( + parse_recommendation_summary, +) + +_FIXTURE: Path = ( + Path(__file__).resolve().parent + / "fixtures" + / "google_building_insights_001431.json" +) +_SOLAR_MEASURE_TYPE = "solar_pv" +_BATTERY_CAPACITY_KWH = 5.0 + + +class _StubProducts(ProductRepository): + """In-memory ProductRepository returning a fixed solar_pv catalogue row.""" + + def get(self, measure_type: str) -> Product: + return Product( + measure_type=measure_type, + unit_cost_per_m2=0.0, + contingency_rate=0.15, + id=909, + ) + + +def _solar_potential() -> SolarPotential: + with _FIXTURE.open(encoding="utf-8") as handle: + data: dict[str, Any] = json.load(handle) + return SolarPotential.from_building_insights(data) + + +def _eligible_house() -> EpcPropertyData: + """The solar 001431 before cert: a House with a hot-water cylinder, no + existing PV — solar-eligible.""" + return parse_recommendation_summary("solar_pv_001431_before.pdf") + + +def test_eligible_house_yields_a_solar_pv_recommendation_of_competing_options() -> None: + # Arrange — a house with feasible Google solar potential (5 conservative + # configs) and a cylinder. + baseline = _eligible_house() + + # Act + recommendation: Optional[Recommendation] = recommend_solar( + baseline, _StubProducts(), _solar_potential() + ) + + # Assert — one "Solar PV" Recommendation, 5 configs × {no battery, battery} + # = 10 competing Options, all measure_type solar_pv. + assert recommendation is not None + assert recommendation.surface == "Solar PV" + assert len(recommendation.options) == 10 + assert {o.measure_type for o in recommendation.options} == {_SOLAR_MEASURE_TYPE} + assert all(o.material_id == 909 for o in recommendation.options) + + +def test_each_option_overlay_installs_per_segment_arrays_and_ensures_export() -> None: + # Arrange + baseline = _eligible_house() + + # Act + recommendation = recommend_solar(baseline, _StubProducts(), _solar_potential()) + + # Assert — every option folds a SolarOverlay: one PhotovoltaicArray per + # config segment, export ensured, diverter set (the dwelling has a cylinder). + assert recommendation is not None + for option in recommendation.options: + overlay = option.overlay.solar + assert overlay is not None + assert overlay.is_dwelling_export_capable is True + assert overlay.pv_diverter_present is True + arrays = overlay.photovoltaic_arrays + assert arrays is not None and len(arrays) >= 1 + assert all(isinstance(a, PhotovoltaicArray) for a in arrays) + assert all(1 <= a.orientation <= 8 for a in arrays) + assert all(1 <= a.pitch <= 5 for a in arrays) + assert all(1 <= a.overshading <= 4 for a in arrays) + + +def test_smallest_config_array_peak_power_matches_panels_times_capacity() -> None: + # Arrange — the smallest conservative config is 4 panels × 400 W = 1.6 kWp + # on one SE plane (≈32° → pitch code 2), back-solved to a heavy-ish bucket. + baseline = _eligible_house() + + # Act + recommendation = recommend_solar(baseline, _StubProducts(), _solar_potential()) + + # Assert — find the no-battery option whose single array totals 1.6 kWp. + assert recommendation is not None + no_battery_arrays: list[list[PhotovoltaicArray]] = [] + for option in recommendation.options: + overlay = option.overlay.solar + assert overlay is not None + if overlay.pv_batteries is None and overlay.photovoltaic_arrays is not None: + no_battery_arrays.append(overlay.photovoltaic_arrays) + smallest = min( + no_battery_arrays, key=lambda arrays: sum(a.peak_power for a in arrays) + ) + assert len(smallest) == 1 + assert abs(smallest[0].peak_power - 1.6) <= 1e-9 + assert smallest[0].orientation == 4 # SE + assert smallest[0].pitch == 2 # ~32° → 30° + + +def test_battery_variant_adds_a_five_kwh_battery_and_costs_more() -> None: + # Arrange + baseline = _eligible_house() + + # Act + recommendation = recommend_solar(baseline, _StubProducts(), _solar_potential()) + + # Assert — for the same array size, the battery variant carries a 5 kWh + # battery and a higher cost than its no-battery twin. + assert recommendation is not None + by_size: dict[float, dict[bool, float]] = {} + for option in recommendation.options: + overlay = option.overlay.solar + assert overlay is not None and option.cost is not None + size = round(sum(a.peak_power for a in (overlay.photovoltaic_arrays or [])), 6) + has_battery = overlay.pv_batteries is not None + by_size.setdefault(size, {})[has_battery] = option.cost.total + if has_battery: + assert overlay.pv_batteries is not None + assert ( + abs( + overlay.pv_batteries.pv_battery.battery_capacity + - _BATTERY_CAPACITY_KWH + ) + <= 1e-9 + ) + for size, costs in by_size.items(): + assert costs[True] > costs[False], size + + +def test_combi_dwelling_gets_no_diverter() -> None: + # Arrange — the same house without a cylinder (a combi has nothing to divert + # surplus PV to), so the diverter field is left unset. + baseline = _eligible_house() + baseline.has_hot_water_cylinder = False + + # Act + recommendation = recommend_solar(baseline, _StubProducts(), _solar_potential()) + + # Assert + assert recommendation is not None + for option in recommendation.options: + assert option.overlay.solar is not None + assert option.overlay.solar.pv_diverter_present is None + + +def test_flat_is_not_eligible() -> None: + # Arrange — a flat needs building-level shared-roof coordination (deferred). + baseline = _eligible_house() + baseline.property_type = "Flat" + + # Act / Assert + assert recommend_solar(baseline, _StubProducts(), _solar_potential()) is None + + +def test_listed_building_blocks_solar() -> None: + # Arrange — a listed building protects the fabric (blocks_internal). + baseline = _eligible_house() + + # Act / Assert + assert ( + recommend_solar( + baseline, + _StubProducts(), + _solar_potential(), + PlanningRestrictions(is_listed=True), + ) + is None + ) + + +def test_conservation_area_does_not_block_solar() -> None: + # Arrange — a conservation area blocks external work generally, but PV is + # offered (installed sympathetically) — same gate as ASHP, not blocks_external. + baseline = _eligible_house() + + # Act + recommendation = recommend_solar( + baseline, + _StubProducts(), + _solar_potential(), + PlanningRestrictions(in_conservation_area=True), + ) + + # Assert + assert recommendation is not None + assert len(recommendation.options) == 10 + + +def test_existing_pv_dwelling_is_not_eligible() -> None: + # Arrange — a dwelling that already has PV (existing-PV top-up is deferred). + baseline = _eligible_house() + baseline.sap_energy_source.photovoltaic_arrays = [ + PhotovoltaicArray(peak_power=2.0, pitch=2, orientation=5, overshading=1) + ] + + # Act / Assert + assert recommend_solar(baseline, _StubProducts(), _solar_potential()) is None + + +def test_no_solar_potential_yields_no_recommendation() -> None: + # Arrange — no Google solar data (or no feasible config) → no recommendation. + baseline = _eligible_house() + + # Act / Assert + assert recommend_solar(baseline, _StubProducts(), None) is None + + +def test_infeasible_potential_yields_no_recommendation() -> None: + # Arrange — a potential whose only config faces due north (dropped → empty). + baseline = _eligible_house() + potential = _solar_potential() + north_only = replace(potential, configurations=()) + + # Act / Assert + assert recommend_solar(baseline, _StubProducts(), north_only) is None