Model/backend/apis/GoogleSolarApi.py

import pandas as pd
import numpy as np
from recommendations.Costs import MCS_SOLAR_PV_COST_DATA
from backend.ml_models.AnnualBillSavings import AnnualBillSavings
import requests
from functools import lru_cache
import time


class GoogleSolarApi:
    NORTH_FACING_AZIMUTH_RANGE = (-30, 30)

    # Conservative estimate of the proportion of electricity that will be consumed, whereas the rest will
    # be exported
    SOLAR_CONSUMPTION_PROPORTION = 0.5

    # These are variables, described in the documentation for cost analysis for non-us locations, seen here
    # https://developers.google.com/maps/documentation/solar/calculate-costs-non-us
    # We use the default figures that the API uses for US locations

    # The factor by which the cost of electricity increases annually. The Solar API uses 1.022 (2.2% annual increase)
    # for US locations.
    cost_increase_factor = 1.022

    # The efficiency at which an inverter converts the DC electricity that is produced by the solar panels to the AC
    # electricity that is used in a household. The Solar API uses 85% for US locations. We use 0.95.5 which is the
    # middle value of the 93-98% range, cited by Sunsave:
    # https://www.sunsave.energy/solar-panels-advice/system-size/inverters
    dc_to_ac_rate = 0.955

    # The Solar API uses 1.04 (4% annual increase) for US locations
    discount_rate = 1.04

    # How much the efficiency of the solar panels declines each year. The Solar API uses 0.995 (0.5% annual decrease)
    # for US locations
    efficiency_depreciation_factor = 0.995

    # The expected lifespan of the solar installation. The Solar API uses 20 years. Adjust this value as needed for
    # your area
    installation_life_span = 20

    def __init__(self, api_key, max_retries=5):
        """
        Initialize the GoogleSolarApi class with the provided API key and maximum retries.

        :param api_key: The API key to authenticate requests to the Google Solar API.
        :param max_retries: The maximum number of retries for the API request (default is 5).
        """
        self.api_key = api_key
        self.max_retries = max_retries
        self.base_url = "https://solar.googleapis.com/v1"

        self.insights_data = None
        self.roof_segments = []

        # property attributes:
        self.floor_area = None
        self.roof_area = None
        self.roof_segment_indexes = None
        self.panel_area = None
        self.panel_wattage = None
        self.panel_performance = None

    def get_building_insights(self, longitude, latitude, required_quality="MEDIUM", max_retries=None):
        """
        Make an API request to retrieve building insights based on the given longitude and latitude, with retry
        mechanism.

        :param longitude: The longitude of the location.
        :param latitude: The latitude of the location.
        :param required_quality: The required quality of the data (default is "MEDIUM").
        :param max_retries: The maximum number of retries for the API request (default is None, which uses the
        instance's max_retries).
        :return: The JSON response containing the building insights data.
        """
        if max_retries is None:
            max_retries = self.max_retries

        insights_url = f"{self.base_url}/buildingInsights:findClosest"
        params = {
            'location.latitude': f'{latitude:.5f}',
            'location.longitude': f'{longitude:.5f}',
            'requiredQuality': required_quality,
            'key': self.api_key
        }

        attempt = 0
        while attempt < max_retries:
            try:
                response = requests.get(insights_url, params=params)
                response.raise_for_status()  # Raise an error for bad status codes
                return response.json()
            except requests.exceptions.RequestException as e:
                attempt += 1
                print(f"Attempt {attempt} failed: {e}")
                time.sleep(2 ** attempt)  # Exponential backoff
                if attempt >= max_retries:
                    raise

    @lru_cache(maxsize=128)
    def get(self, longitude, latitude, required_quality="MEDIUM"):
        """
        Wrapper function that calls get_building_insights and extracts roof segments, with caching.

        :param longitude: The longitude of the location.
        :param latitude: The latitude of the location.
        :param required_quality: The required quality of the data (default is "MEDIUM").
        :return: The JSON response containing the building insights data.
        """

        self.insights_data = self.get_building_insights(longitude, latitude, required_quality)

        # Extract key data from the insights response
        self.roof_segments = self.insights_data["solarPotential"].get('roofSegmentStats', [])
        self.floor_area = self.insights_data["solarPotential"]["wholeRoofStats"]['groundAreaMeters2']
        self.roof_area = self.insights_data["solarPotential"]["wholeRoofStats"]['areaMeters2']
        self.panel_area = (
            self.insights_data["solarPotential"]["panelHeightMeters"] *
            self.insights_data["solarPotential"]["panelWidthMeters"]
        )
        self.panel_wattage = self.insights_data["solarPotential"]["panelCapacityWatts"]
        if self.panel_wattage != 400:
            # In the API documentation, it claims that the default output is 250W, however we've only seen 400W, so if
            # we get anything other than 400W, we'll need to adjust the calculations in the output. For this, we should
            # refer to https://developers.google.com/maps/documentation/solar/calculate-costs-non-us
            # Where the documentation explains how to adjust the yearlyEnergyDcKwh figures.
            # It should be straightforward, but I'd rather see an actual instance of this happening
            raise NotImplementedError("Panel wattage is not 400W - implement me")

        # Automatically exclude north-facing segments
        self.exclude_north_facing_segments()

        self.roof_segment_indexes = [segment['segmentIndex'] for segment in self.roof_segments]

        # We now start finding the solar panel configurations
        self.optimise_solar_configuration()

    @staticmethod
    def lifetime_production_ac_kwh(
        row,
        efficiency_depreciation_factor,
        installation_life_span
    ):
        """
        Mimics the function described in the Google Solar API documentation, presenting the lifetime production
        AC KWH as a geometri sum
        """

        return (
            row["initial_ac_kwh_per_year"] *
            (1 - pow(
                efficiency_depreciation_factor,
                installation_life_span)) /
            (1 - efficiency_depreciation_factor))

    @staticmethod
    def annualUtilityBillEstimate(
        yearlyKWhEnergyConsumption,
        initialAcKwhPerYear,
        efficiencyDepreciationFactor,
        year,
        costIncreaseFactor,
        discountRate):
        """
        Implements the bill costing model for esimating annual bill
        :param yearlyKWhEnergyConsumption:
        :param initialAcKwhPerYear:
        :param efficiencyDepreciationFactor:
        :param year:
        :param costIncreaseFactor:
        :param discountRate:
        :return:
        """

        return (
            billCostModel(
                yearlyKWhEnergyConsumption -
                annualProduction(
                    initialAcKwhPerYear,
                    efficiencyDepreciationFactor,
                    year)) *
            pow(costIncreaseFactor, year) /
            pow(discountRate, year))

    def lifetimeUtilityBill(
        yearlyKWhEnergyConsumption,
        initialAcKwhPerYear,
        efficiencyDepreciationFactor,
        installationLifeSpan,
        costIncreaseFactor,
        discountRate):
        bill = [0] * installationLifeSpan
        for year in range(installationLifeSpan):
            bill[year] = annualUtilityBillEstimate(
                yearlyKWhEnergyConsumption,
                initialAcKwhPerYear,
                efficiencyDepreciationFactor,
                year,
                costIncreaseFactor,
                discountRate)
        return bill

    def estimate_solar_costs(self, panel_performance):
        """
        This method implements the recommended costing approach, to estimate the ROI of a solar panel
        configuration, as described in the Google Solar API documentation
        :param panel_performance: dataframe containing the solar panel array configuration and energy generation data
        :return:
        """

        # we now estiamte the financial benefits of solar panels for the household, using the framework described
        # by the Google Solar API
        # 1) Convert Solar Energy AD production from the DC production
        panel_performance["initial_ac_kwh_per_year"] = panel_performance["yearly_dc_energy"] * self.dc_to_ac_rate

        # This is just a benchmark figure, based on the national figure. This doesn't not respect the fact that a
        # property could be 100% electric
        average_electricity_consumption

        # Remove anything where the total ac energy is less than half of the array wattage
        panel_performance = panel_performance[
            (panel_performance["initial_ac_kwh_per_year"] / panel_performance["array_warrage"]) >= 0.5
            ]

        # 2) Calculate the liftime solar energy production
        panel_performance['lifetime_ac_kwh'] = panel_performance.apply(
            self.lifetime_production_ac_kwh,
            axis=1,
            efficiency_depreciation_factor=self.efficiency_depreciation_factor,
            installation_life_span=self.installation_life_span
        )

        # TODO: Complete the rest of the solar model

    def optimise_solar_configuration(self):
        """
        Optimise the solar panel configuration for the building.
        :return:
        """

        # Remove any north facing roof segments
        panel_performance = []
        for config in self.insights_data["solarPotential"]["solarPanelConfigs"]:
            roof_segment_summaries = config["roofSegmentSummaries"]
            # Filter on just the segments in self.roof_segment_indexes
            roof_segment_summaries = [
                segment for segment in roof_segment_summaries if segment["segmentIndex"] in self.roof_segment_indexes
            ]

            roi_summary = []
            for segment in roof_segment_summaries:
                wattage = segment["panelsCount"] * self.insights_data["solarPotential"]["panelCapacityWatts"]
                generated_dc_energy = segment["yearlyEnergyDcKwh"]
                ratio = generated_dc_energy / wattage
                cost = MCS_SOLAR_PV_COST_DATA["average_cost_per_kwh"] * (generated_dc_energy / 1000)
                roi_summary.append(
                    {
                        "segmentIndex": segment["segmentIndex"],
                        "wattage": wattage,
                        "generated_dc_energy": generated_dc_energy,
                        "ratio": ratio,
                        "n_panels": segment["panelsCount"],
                        "cost": cost,
                        "panneled_roof_area": self.panel_area * int(segment["panelsCount"])
                    }
                )

            roi_summary = pd.DataFrame(roi_summary)

            weighted_ratio = np.average(
                roi_summary["ratio"].values, weights=roi_summary["generated_dc_energy"].values
            )
            total_cost = roi_summary["cost"].sum()
            yearly_dc_energy = roi_summary["generated_dc_energy"].sum()

            panel_performance.append(
                {
                    "n_panels": roi_summary["n_panels"].sum(),
                    "yearly_dc_energy": yearly_dc_energy,
                    "total_cost": total_cost,
                    "weighted_ratio": weighted_ratio,
                    "panneled_roof_area": roi_summary["panneled_roof_area"].sum(),
                    "array_warrage": roi_summary["n_panels"].sum() * self.panel_wattage
                }
            )

        panel_performance = pd.DataFrame(panel_performance)
        # We can have duplicate configurations
        panel_performance = panel_performance.drop_duplicates()
        # Ensure more than 4 panels
        panel_performance = panel_performance[panel_performance["n_panels"] >= 4]

        self.estimate_solar_costs()

        # This first bracket is the value of the energy bill savings
        panel_performance["bill_savings"] = (
            self.SOLAR_CONSUMPTION_PROPORTION *
            panel_performance["total_energy"] *
            AnnualBillSavings.ELECTRICITY_PRICE_CAP
        )
        # This is the amount of energy exported
        panel_performance["export_value"] = (
            (1 - self.SOLAR_CONSUMPTION_PROPORTION) *
            panel_performance["total_energy"] *
            AnnualBillSavings.ELECTRICITY_EXPORT_PAYMENT
        )
        panel_performance["energy_value"] = panel_performance["bill_savings"] + panel_performance["export_value"]
        panel_performance["payback_years"] = panel_performance["total_cost"] / panel_performance["energy_value"]

        panel_performance = panel_performance.sort_values("weighted_ratio", ascending=False)
        # TODO: Finish this!!

        panel_performance["roof_area_percentage"] = panel_performance["panneled_roof_area"] / self.roof_area

        self.panel_performance = panel_performance

    def exclude_north_facing_segments(self):
        """
        Filter out any north-facing roof segments from the roof_segments attribute.

        North-facing segments are defined as those with an azimuth between -30 and 30 degrees.
        """

        filtered_segments = []
        for segment_index, segment in enumerate(self.roof_segments):
            segment["segmentIndex"] = segment_index
            # Check if the segment is north-facing
            if self.NORTH_FACING_AZIMUTH_RANGE[0] <= segment['azimuthDegrees'] <= self.NORTH_FACING_AZIMUTH_RANGE[1]:
                continue

            filtered_segments.append(segment)

        self.roof_segments = filtered_segments