Model/recommendations/SolarPvRecommendations.py

import numpy as np
import pandas as pd
import backend.app.assumptions as assumptions

from recommendations.Costs import Costs
from recommendations.recommendation_utils import override_costs, estimate_pitched_roof_area


class SolarPvRecommendations:
    # For domestic properties, we don't recommend a solar PV system with wattage outside of these
    # bounds
    MAX_SYSTEM_WATTAGE = 6000
    MIN_SYSTEM_WATTAGE = 1000

    # the maximum area of root we allow to be covered in solar panels for our recommendations.
    MAX_ROOF_AREA_PERCENTAGE = 0.7

    SAP_POINTS_PER_5_PERCENT_ROOF_COVERAGE = 1

    BACKUP_PANEL_PERFORMANCE = pd.DataFrame(
        [
            {
                "n_panels": 4,
                "array_wattage": 1600,
                "initial_ac_kwh_per_year": assumptions.MEDIAN_WATTAGE_TO_AC * 1600,
                "panneled_roof_area": 4 * assumptions.RDSAP_AREA_PER_PANEL
            },
            {
                "n_panels": 8,
                "array_warrage": 3200,
                "initial_ac_kwh_per_year": assumptions.MEDIAN_WATTAGE_TO_AC * 3200,
                "panneled_roof_area": 8 * assumptions.RDSAP_AREA_PER_PANEL
            },
        ]
    )

    PANEL_SIZES = [400, 435, 440, 445]

    def __init__(self, property_instance, materials: list):
        """
        :param property_instance: Instance of the Property class, for the home associated to property_id
        """

        self.property = property_instance
        self.costs = Costs(self.property)

        self.recommendation = []

        self.panels_products = [
            material for material in materials if material["type"] == "solar_pv"
        ]

        self.scaffolding_options = [
            material for material in materials if material["type"] == "scaffolding"
        ]

    @staticmethod
    def trim_solar_wattage_options(scenarios_with_wattage):
        # Initialize the list with the first element, assuming the list is not empty
        trimmed_list = [scenarios_with_wattage[0]]

        # Iterate over the list starting from the second element
        for scenario in scenarios_with_wattage[1:]:
            # Compare the second element (index 1) of the current tuple with the last tuple in the trimmed list
            if scenario[1] > trimmed_list[-1][1]:
                trimmed_list.append(scenario)

        return trimmed_list

    def recommend_building_analysis(self, phase):
        """
        This recommendation approach handles the case of producing solar PV recommendations at the building level,
        across multiple flats. For these recommendations, we don't include the battery option since it's impractical
        from a space perspective.
        :return:
        """

        panel_performance = self.property.solar_panel_configuration["panel_performance"]
        total_roof_area = (
            self.property.solar_panel_configuration["insights_data"]["solarPotential"]["wholeRoofStats"]["areaMeters2"]
        )
        n_units = self.property.solar_panel_configuration["n_units"]

        # At a building level, we take a single configuration so that all properties a guaranteed to use
        # the same configuration
        best_configurations = panel_performance.head(1).reset_index(drop=True)

        for rank, recommendation_config in best_configurations.iterrows():
            # If we dont have the panneled_roof_area in the recommendation_config we calculate it
            if recommendation_config.get("panneled_roof_area", None):
                # We spread the coverage across the individual units
                roof_coverage_percent = round(
                    ((recommendation_config["panneled_roof_area"] / total_roof_area) * 100) / n_units
                )
            else:
                raise Exception("IMPLEMENT ME")

            # We get solar PV options
            solar_product = [x for x in self.panels_products if x["id"] == recommendation_config["solar_product_id"]]
            if not solar_product:
                raise NotImplementedError(
                    f"Solar product with id {recommendation_config['solar_product_id']} not found in "
                    "panels_products"
                )
            solar_product = solar_product[0]

            n_floors = (
                self.property.number_of_storeys["number_of_storeys"] if
                self.property.number_of_storeys["number_of_storeys"] is not None else 3
            )

            total_cost = self.costs.solar_pv(
                solar_product=solar_product,
                scaffolding_options=self.scaffolding_options,
                n_floors=n_floors,
            )["total"]

            kw = np.floor(recommendation_config["array_wattage"] / 100) / 10
            # Default to a weeks work for a team of 3 people doing 8 hour days
            labour_days = 5
            labour_hours = 3 * 8 * labour_days

            description = (f"Install a {kw} kilowatt-peak (kWp) solar photovoltaic (PV) panel system on the roof "
                           "of the building")

            initial_ac_kwh_per_year = recommendation_config["initial_ac_kwh_per_year"]

            self.recommendation.append(
                {
                    "phase": phase,
                    "parts": [],
                    "type": "solar_pv",
                    "measure_type": "solar_pv",
                    "description": description,
                    "starting_u_value": None,
                    "new_u_value": None,
                    "sap_points": None,
                    "already_installed": False,
                    "total": total_cost,
                    "labour_days": labour_days,
                    "labour_hours": labour_hours,
                    # This is required for simulating the SAP impact. solar_pv_percentage is between 0 & 1 so we scale
                    # back up here
                    "photo_supply": roof_coverage_percent,
                    "has_battery": False,
                    "simulation_config": {
                        "photo_supply_ending": roof_coverage_percent
                    },
                    "initial_ac_kwh_per_year": initial_ac_kwh_per_year,
                    "description_simulation": {"photo-supply": roof_coverage_percent},
                    "rank": rank,  # Rank is used to get the representative recommendation - rank 0 will be chosen
                    "innovation_rate": solar_product["innovation_rate"],
                }
            )

    def _get_available_products(self, n_panels):
        """
        Utility function to get the available solar PV products based on the number of panels
        :param n_panels:
        :return:
        """
        available_products = []
        for panel_size in self.PANEL_SIZES:
            system_size = (n_panels * panel_size) / 1000
            prods = [
                x for x in self.panels_products if abs(x["size"] - system_size) < 0.01
            ]
            for x in prods:
                x["panel_size"] = panel_size
            available_products.extend(prods)

        return available_products

    def recommend(self, phase):
        """
        We check if a property is potentially suitable for solar PV based on the following criteria:
        - The property is a house or bungalow
        - The property has a flat or pitched roof
        - The property does not have existing solar pv
        :return:
        """

        if not self.property.is_solar_pv_valid():
            return

        # If we have a buiilding level analysis, we implement separate logic
        if self.property.building_id is not None:
            self.recommend_building_analysis(phase)
            return

        non_invasive_recommendation = next(
            (r for r in self.property.non_invasive_recommendations if r["type"] == "solar_pv"), {"suitable": True}
        )

        # We allow for the non-invasive recommendation to be that solar PV is not suitable
        if not non_invasive_recommendation["suitable"]:
            return

        if non_invasive_recommendation.get("array_wattage") is not None:

            if self.property.roof["is_flat"]:
                roof_area = self.property.insulation_floor_area
            else:
                roof_area = estimate_pitched_roof_area(floor_area=self.property.insulation_floor_area, )
            solar_configurations = pd.DataFrame(
                [
                    {
                        "array_wattage": non_invasive_recommendation["array_wattage"],
                        "initial_ac_kwh_per_year": non_invasive_recommendation["initial_ac_kwh_per_year"],
                        "panneled_roof_area": non_invasive_recommendation["panneled_roof_area"]
                    }
                ]
            )
        else:
            # TODO: There may be some instances where we don't want to use the solar API so we should cover for them
            panel_performance = self.property.solar_panel_configuration["panel_performance"].copy()
            # We don't allow for more than 70% of the roof to be covered
            panel_performance = panel_performance[
                panel_performance["panneled_roof_area"] / self.property.roof_area <= self.MAX_ROOF_AREA_PERCENTAGE
                ]

            roof_area = self.property.roof_area
            solar_configurations = panel_performance.head(6).reset_index(drop=True)

        # We combine each of these configurations with estimates with and without a battery
        for rank, recommendation_config in solar_configurations.iterrows():

            n_panels = recommendation_config["n_panels"]

            available_products = self._get_available_products(n_panels)

            # Given the available products in the database, we product the possible array of recommendations
            for solar_pv_product in available_products:

                # we take the paneled roof area and this tells us the roof coverage, based on 400W panels
                # We then look at the equivalent for larger panels, which will produce more energy in the same area

                paneled_roof_area = recommendation_config["panneled_roof_area"]

                roof_coverage_percent = round(
                    ((paneled_roof_area / 400) * solar_pv_product["panel_size"]) / roof_area * 100
                )
                # We round up to the nearest 5
                roof_coverage_percent = np.ceil(roof_coverage_percent / 5) * 5

                # Note roof_coverage_percent is based on 400 watt panels, so we need to scale it up based on
                # largest panels that will produce more energy in the same area

                # Typically, we've observed that every 5% of additional roof coverage will result in at least
                # an additional 1 SAP points (though often 2 points) Given this, we can add a reasonable minimum
                # for the number of SAP points we might expect. We've observed that for some cases where properties
                # are hitting the higher SAP scores (e.g. EPC A and above), the model can sometimes under-predict
                # the number of SAP points. This appears to be due to a relatively small number of properties
                # actually achieving the upper echelons of EPC rating. This can be the case if we're simulating a
                # whole house retrofit where the home is getting complete insulation, a heat pump and solar panels.
                # Because panels are the final recommendation, they are often the measure that takes the home
                # into the medium to high EPC A ranges and so because of a lack of training data, this means that
                # we might sometime under-predict. This minimum is intended to try and reduce the negative impact
                # of this. This minimum is used in Recommendations.calculate_recommendation_impact
                minimum_sap_points = (roof_coverage_percent / 5) * self.SAP_POINTS_PER_5_PERCENT_ROOF_COVERAGE

                cost_result = self.costs.solar_pv(
                    solar_product=solar_pv_product,
                    scaffolding_options=self.scaffolding_options,
                    n_floors=self.property.number_of_floors
                )
                description = f"Install a {solar_pv_product['description']}"

                if self.property.in_conservation_area:
                    description += " Property is in a consevation area - please check with local planning authority."

                already_installed = "solar_pv" in self.property.already_installed
                if already_installed:
                    cost_result = override_costs(cost_result)

                self.recommendation.append(
                    {
                        "phase": phase,
                        "parts": [solar_pv_product],
                        "type": "solar_pv",
                        "measure_type": "solar_pv",
                        "description": description,
                        "starting_u_value": None,
                        "new_u_value": None,
                        "sap_points": minimum_sap_points,
                        "already_installed": already_installed,
                        **cost_result,
                        "has_battery": solar_pv_product["includes_battery"],
                        "simulation_config": {
                            "photo_supply_ending": roof_coverage_percent
                        },
                        "initial_ac_kwh_per_year": recommendation_config["initial_ac_kwh_per_year"],
                        "description_simulation": {"photo-supply": roof_coverage_percent},
                        "innovation_rate": solar_pv_product["innovation_rate"],
                    }
                )