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, energy_consumption, required_quality="MEDIUM", is_building=False):
        """
        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 energy_consumption: The energy consumption of the building/unit associated to the longitude and latitude.
        :param required_quality: The required quality of the data (default is "MEDIUM").
        :param is_building: Whether the energy consumption is for a building or a unit.
        :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.roof_area = self.insights_data["solarPotential"]["wholeRoofStats"]['areaMeters2']
        self.floor_area = self.insights_data["solarPotential"]["wholeRoofStats"]['groundAreaMeters2']
        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(energy_consumption=energy_consumption, is_building=is_building)

    @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 geometric sum
        """

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

    def optimise_solar_configuration(self, energy_consumption, is_building=False):
        """
        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()
        # If we look at the building level, we don't include any projects fewer than 10 panels, otherwise the
        # minimum is 4
        min_panels = 10 if is_building else 4
        panel_performance = panel_performance[panel_performance["n_panels"] >= min_panels]

        panel_performance["initial_ac_kwh_per_year"] = panel_performance["yearly_dc_energy"] * self.dc_to_ac_rate

        # 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
        )

        # Now that we know the lifetime cnsumption of ac kwh, we can estimate the roi
        roi_results = []
        for _, panel_config in panel_performance.iterrows():
            lifetime_ac_kwh = panel_config["lifetime_ac_kwh"]
            lifetime_energy_consumption = energy_consumption * self.installation_life_span

            if lifetime_ac_kwh < lifetime_energy_consumption:
                # We estimate the amount of electricity generated, based on the price cap
                generation_value = lifetime_ac_kwh * AnnualBillSavings.ELECTRICITY_PRICE_CAP
                roi = generation_value / panel_config["total_cost"]
                generation_deficit = lifetime_energy_consumption - lifetime_ac_kwh
            else:
                # We now have a surplus of energy, which we can sell back to the grid
                surplus = lifetime_ac_kwh - lifetime_energy_consumption
                surplus_value = surplus * AnnualBillSavings.ELECTRICITY_EXPORT_PAYMENT
                generation_value = lifetime_energy_consumption * AnnualBillSavings.ELECTRICITY_PRICE_CAP
                roi = (generation_value + surplus_value) / panel_config["total_cost"]
                generation_deficit = surplus_value

            # Generation deficit tells us how much more energy we need to meet the generation demand.
            roi_results.append(
                {
                    "n_panels": panel_config["n_panels"],
                    "roi": roi,
                    "generation_value": generation_value,
                    "generation_deficit": generation_deficit
                }
            )

        roi_results = pd.DataFrame(roi_results)

        panel_performance = panel_performance.merge(
            roi_results, how="left", on="n_panels"
        )

        # We prioritise maximal roi, then minimal geneartion deficit, then maximal generation value (if there is still
        # a tie). Ideally, we want the best roi over the lifetime of the solar panels, but we also want to ensure that
        # we can meet the energy demands of the building.
        panel_performance = panel_performance.sort_values(
            ["roi", "generation_deficit", "generation_value"], ascending=[False, True, False]
        )

        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