Model/domain/modelling/scoring/overlay_applicator.py

"""The Overlay Applicator — folds an ordered set of Simulation Overlays onto
a baseline EpcPropertyData and returns a new one for the calculator.

Sequential fold: overlays are applied in order and a later overlay wins on a
field it shares with an earlier one. The baseline is never mutated; the
returned EpcPropertyData is throwaway (handed to the calculator for scoring,
then discarded). See ADR-0016.
"""

import copy
from dataclasses import fields
from typing import Final, Optional, Sequence

from datatypes.epc.domain.epc_property_data import (
    BuildingPartIdentifier,
    EpcPropertyData,
    SapBuildingPart,
    SapVentilation,
    SapWindow,
    WindowTransmissionDetails,
)
from domain.modelling.simulation import (
    EpcSimulation,
    GlazingOverlay,
    HeatingOverlay,
    LightingOverlay,
    SecondaryHeatingOverlay,
    SolarOverlay,
    VentilationOverlay,
    WindowOverlay,
)


# A Landlord Override's building part is a POSITIONAL index (0=main, 1=extension
# 1…, ADR-0004), translated to a `BuildingPartIdentifier` upstream. This recovers
# that position so an override can fall back onto the part the EPC actually models
# at that slot when the gov-API labelled it differently (e.g. lodged a 2nd part as
# `other` rather than `extension_1`).
_POSITION_BY_IDENTIFIER: Final[dict[BuildingPartIdentifier, int]] = {
    BuildingPartIdentifier.MAIN: 0,
    BuildingPartIdentifier.EXTENSION_1: 1,
    BuildingPartIdentifier.EXTENSION_2: 2,
    BuildingPartIdentifier.EXTENSION_3: 3,
    BuildingPartIdentifier.EXTENSION_4: 4,
}


def _resolve_part(
    epc: EpcPropertyData,
    parts_by_id: dict[BuildingPartIdentifier, SapBuildingPart],
    identifier: BuildingPartIdentifier,
) -> Optional[SapBuildingPart]:
    """The building part an overlay targets: the EPC's part with that identifier
    when present, else the part at the override's positional index (so a
    correction for `extension_1` still lands on the EPC's 2nd part even when the
    gov-API lodged it under a different label). ``None`` when the EPC models no
    part at that position."""
    part = parts_by_id.get(identifier)
    if part is not None:
        return part
    position = _POSITION_BY_IDENTIFIER.get(identifier)
    if position is None or position >= len(epc.sap_building_parts):
        return None
    return epc.sap_building_parts[position]


def apply_simulations(
    baseline: EpcPropertyData, simulations: Sequence[EpcSimulation]
) -> EpcPropertyData:
    """Return a copy of ``baseline`` with every Simulation Overlay's non-``None``
    fields written onto the building part it targets, applied in order. A
    whole-dwelling ``ventilation`` overlay folds onto ``sap_ventilation``
    (creating one if the baseline lodged none)."""
    result: EpcPropertyData = copy.deepcopy(baseline)
    parts_by_id = {part.identifier: part for part in result.sap_building_parts}

    for simulation in simulations:
        for identifier, overlay in simulation.building_parts.items():
            part = _resolve_part(result, parts_by_id, identifier)
            # No part at this position — the EPC models fewer parts than the
            # override's index. We have no geometry to model the missing part, so
            # skip it rather than crash the whole property's modelling.
            if part is None:
                continue
            for overlay_field in fields(overlay):
                value = getattr(overlay, overlay_field.name)
                if value is not None:
                    setattr(part, overlay_field.name, value)
        for index, window_overlay in simulation.windows.items():
            _fold_window(result.sap_windows[index], window_overlay)
        if simulation.ventilation is not None:
            result.sap_ventilation = _fold_ventilation(
                result.sap_ventilation, simulation.ventilation
            )
        if simulation.lighting is not None:
            _fold_lighting(result, simulation.lighting)
        if simulation.glazing is not None:
            _fold_glazing(result, simulation.glazing)
        if simulation.heating is not None:
            _fold_heating(result, simulation.heating)
        if simulation.secondary_heating is not None:
            _fold_secondary_heating(result, simulation.secondary_heating)
        if simulation.solar is not None:
            _fold_solar(result, simulation.solar)
        if simulation.property_type is not None:
            result.property_type = simulation.property_type
        if simulation.built_form is not None:
            result.built_form = simulation.built_form
        if simulation.percent_draughtproofed is not None:
            result.percent_draughtproofed = simulation.percent_draughtproofed

    return result


def _fold_secondary_heating(
    epc: EpcPropertyData, overlay: SecondaryHeatingOverlay
) -> None:
    """Strip the dwelling's lodged secondary heating system (ADR-0028) — the one
    fold that sets fields to *absent* rather than to a target state. Clears
    `secondary_heating_type` + `secondary_fuel_type` on `sap_heating`, so the
    calculator's Table 11 split routes 100% of space heating to the main (or, on
    an electric-storage main, re-forces the §A.2.2 default — a no-op the
    Optimiser de-selects)."""
    if not overlay.remove:
        return
    epc.sap_heating.secondary_heating_type = None
    epc.sap_heating.secondary_fuel_type = None


# `HeatingOverlay` fields grouped by the object they target — the deepest fold,
# spanning the primary `MainHeatingDetail`, `sap_heating`, the top-level
# `EpcPropertyData`, and `sap_energy_source` (ADR-0024).
_MAIN_HEATING_FIELDS: tuple[str, ...] = (
    "main_fuel_type",
    "heat_emitter_type",
    "main_heating_control",
    "sap_main_heating_code",
    "main_heating_index_number",
    "main_heating_category",
    "fan_flue_present",
    "boiler_flue_type",
)
_SAP_HEATING_FIELDS: tuple[str, ...] = (
    "water_heating_code",
    "water_heating_fuel",
    "cylinder_size",
    "cylinder_insulation_type",
    "cylinder_insulation_thickness_mm",
    "cylinder_thermostat",
    "immersion_heating_type",
)
_ENERGY_SOURCE_FIELDS: tuple[str, ...] = ("meter_type", "mains_gas")


def _is_off_peak_meter(meter_type: object) -> bool:
    """True iff the meter resolves to an off-peak Table 12a tariff (not the
    STANDARD single-rate column). Unparseable / absent meters count as not
    off-peak so a coherent override meter still applies to them."""
    from domain.sap10_calculator.exceptions import UnmappedSapCode
    from domain.sap10_calculator.tables.table_12a import (
        Tariff,
        tariff_from_meter_type,
    )

    try:
        return tariff_from_meter_type(meter_type) is not Tariff.STANDARD
    except UnmappedSapCode:
        return False


def _fold_heating(epc: EpcPropertyData, overlay: HeatingOverlay) -> None:
    """Write a `HeatingOverlay`'s non-``None`` fields onto the (copied) dwelling,
    routing each to its home: the primary ``main_heating_details[0]``, the
    ``sap_heating`` hot-water fields, the top-level ``has_hot_water_cylinder``,
    and the ``sap_energy_source`` meter/mains-gas fields. The bundle targets the
    primary system only (index 0)."""
    main = epc.sap_heating.main_heating_details[0]
    for field_name in _MAIN_HEATING_FIELDS:
        value = getattr(overlay, field_name)
        if value is not None:
            setattr(main, field_name, value)
    # `main_heating_index_number` (PCDB-resolved, e.g. a heat pump) and
    # `sap_main_heating_code` (Table 4a-resolved, e.g. storage heaters) are
    # mutually-exclusive efficiency anchors: a whole-system replacement to one
    # must clear the other, else a stale code from the old system wins the
    # calculator's dispatch (e.g. a gas-boiler code 104 left beside a heat-pump
    # index makes hot water use boiler efficiency, not the HP SCOP).
    if overlay.main_heating_index_number is not None:
        main.sap_main_heating_code = None
    elif overlay.sap_main_heating_code is not None:
        main.main_heating_index_number = None
    for field_name in _SAP_HEATING_FIELDS:
        value = getattr(overlay, field_name)
        if value is not None:
            setattr(epc.sap_heating, field_name, value)
    if overlay.has_hot_water_cylinder is not None:
        epc.has_hot_water_cylinder = overlay.has_hot_water_cylinder
    for field_name in _ENERGY_SOURCE_FIELDS:
        value = getattr(overlay, field_name)
        if value is None:
            continue
        # A landlord heating override's assumed meter (Dual for off-peak
        # systems) is a coherent default, not a re-metering of the cert: when it
        # opts in (`keep_existing_off_peak_meter`), don't downgrade a cert that
        # already lodges a MORE off-peak meter (e.g. a 24-hour all-low tariff) to
        # the overlay's 7-hour E7 — keep the cert's (more specific) one. Single/
        # unknown existing meters still receive the off-peak meter, and a switch
        # to single-rate still resets it (its desired value isn't off-peak).
        # Heating MEASURES leave the flag False — they re-meter for real
        # (Elmhurst re-lodges 18-hour → Dual on a storage install).
        if (
            field_name == "meter_type"
            and overlay.keep_existing_off_peak_meter
            and _is_off_peak_meter(value)
            and _is_off_peak_meter(epc.sap_energy_source.meter_type)
        ):
            continue
        setattr(epc.sap_energy_source, field_name, value)


# `SolarOverlay` fields all live on `sap_energy_source` (the home of the SAP
# Appendix M PV inputs) — the sixth overlay surface (ADR-0026).
_ENERGY_SOURCE_SOLAR_FIELDS: tuple[str, ...] = (
    "photovoltaic_arrays",
    "pv_diverter_present",
    "pv_connection",
    "is_dwelling_export_capable",
    "pv_batteries",
)


def _fold_solar(epc: EpcPropertyData, overlay: SolarOverlay) -> None:
    """Write a `SolarOverlay`'s non-``None`` fields onto the (copied) dwelling's
    ``sap_energy_source`` — the PV arrays, diverter, connection, export
    capability and battery a Solar PV Measure Option installs (ADR-0026). The
    arrays are an absolute target: they replace the dwelling's existing
    ``photovoltaic_arrays`` (empty for a non-PV dwelling)."""
    for field_name in _ENERGY_SOURCE_SOLAR_FIELDS:
        value = getattr(overlay, field_name)
        if value is not None:
            setattr(epc.sap_energy_source, field_name, value)


def _fold_lighting(epc: EpcPropertyData, overlay: LightingOverlay) -> None:
    """Write a `LightingOverlay`'s non-``None`` bulb counts onto the (copied)
    dwelling's top-level fields by name — the four counts live directly on
    `EpcPropertyData`, so the fold writes onto it, not a nested object."""
    for overlay_field in fields(overlay):
        value = getattr(overlay, overlay_field.name)
        if value is not None:
            setattr(epc, overlay_field.name, value)


def _fold_window(window: SapWindow, overlay: WindowOverlay) -> None:
    """Write a `WindowOverlay`'s non-``None`` fields onto a (copied) window:
    ``glazing_type`` and ``frame_factor`` flat on the window, ``u_value`` /
    ``solar_transmittance`` into its `WindowTransmissionDetails` (where the
    cascade reads them), starting a fresh one when the window lodged none."""
    if overlay.glazing_type is not None:
        window.glazing_type = overlay.glazing_type
    if overlay.frame_factor is not None:
        window.frame_factor = overlay.frame_factor
    if overlay.u_value is None and overlay.solar_transmittance is None:
        return
    details: Optional[WindowTransmissionDetails] = window.window_transmission_details
    if details is None:
        # data_source 1 = manufacturer-lodged (the case the cascade's per-window
        # U path keys on); both values must be present to start fresh.
        window.window_transmission_details = WindowTransmissionDetails(
            u_value=overlay.u_value if overlay.u_value is not None else 0.0,
            data_source=1,
            solar_transmittance=(
                overlay.solar_transmittance
                if overlay.solar_transmittance is not None
                else 0.0
            ),
        )
        return
    if overlay.u_value is not None:
        details.u_value = overlay.u_value
    if overlay.solar_transmittance is not None:
        details.solar_transmittance = overlay.solar_transmittance


def _fold_glazing(epc: EpcPropertyData, overlay: GlazingOverlay) -> None:
    """Expand a whole-dwelling `GlazingOverlay` across every window: set each
    window's `glazing_type` to the corrected SAP10 code AND clear its lodged
    transmission U, so `heat_transmission`'s Table-24 cascade re-derives U from
    the new type (the lodged U was for the old, mis-recorded glazing). A landlord
    glazing override carries no per-window geometry, so it applies uniformly —
    the expansion lives here because the baseline window list is known only at
    fold time."""
    if overlay.glazing_type is None:
        return
    for window in epc.sap_windows:
        window.glazing_type = overlay.glazing_type
        window.window_transmission_details = None


def _fold_ventilation(
    baseline: Optional[SapVentilation], overlay: VentilationOverlay
) -> SapVentilation:
    """Write the overlay's non-``None`` fields onto a (copied) ``SapVentilation``,
    starting a fresh one when the baseline lodged none."""
    folded: SapVentilation = (
        copy.deepcopy(baseline) if baseline is not None else SapVentilation()
    )
    for overlay_field in fields(overlay):
        value = getattr(overlay, overlay_field.name)
        if value is not None:
            setattr(folded, overlay_field.name, value)
    return folded