GSYE-667: Move solar api clients into gsy-framework

gsy_framework/solar_api_clients/api_client_base.py

@@ -0,0 +1,75 @@
+import abc
+from dataclasses import dataclass
+from datetime import timedelta
+from typing import Dict, Any
+
+from pendulum import DateTime
+from gsy_framework.read_user_profile import resample_hourly_energy_profile
+
+
+@dataclass
+class PvApiParameters:
+    """
+    Parameters that contains all PV settings that are needed for external PV profile APIs
+        longitude: longitude of the PV location
+        latitude: latitude of the PV location
+        capacity_kW: base pv production in kW that is used to scale the data on the API side
+        azimuth: orientation of the PV (cardinal direction)
+        tilt: tilt of the PV panel WRT to the vertical axis (e.g. inclination of the roof)
+    """
+    latitude: float
+    longitude: float
+    capacity_kW: float
+    azimuth: float
+    tilt: float
+
+
+class SolarAPIClientBase(abc.ABC):
+    """Baseclass for all solar api clients."""
+
+    def __init__(self, api_url: str):
+        self.api_url = api_url
+
+    def get_solar_energy_profile(
+            self, request_parameters: PvApiParameters, start_date: DateTime, end_date: DateTime,
+            slot_length: timedelta) -> Dict[DateTime, float]:
+        """Request energy profile from external solar API.
+
+        return: Dictionary of raw data including a time series of energy production with
+                resolution of 1 hour.
+        """
+        raw_data = self._request_raw_solar_energy_data(
+            request_parameters,
+            self._get_corresponding_historical_time_stamp(start_date),
+            self._get_corresponding_historical_time_stamp(end_date))
+        energy_profile = self._create_time_series_from_solar_profile(
+            raw_data, start_date.year, end_date.year)
+        resampled_profile = resample_hourly_energy_profile(
+            energy_profile, slot_length, end_date - start_date, start_date)
+        return resampled_profile
+
+    @staticmethod
+    @abc.abstractmethod
+    def _get_corresponding_historical_time_stamp(input_datetime: DateTime) -> DateTime:
+        """
+        Return the corresponding historical time stamp that is different for each API client.
+        Due to the different data availability of different sources, the requested year
+        needs to be altered in order to request historical data that is available. Leap years have
+        to be considered.
+        """
+
+    @abc.abstractmethod
+    def _request_raw_solar_energy_data(self, request_parameters: PvApiParameters,
+                                       start_date: DateTime, end_date: DateTime) -> Any:
+        """
+        Perform the actual request to the API and return raw data in the API specific format.
+        """
+
+    @staticmethod
+    @abc.abstractmethod
+    def _create_time_series_from_solar_profile(request_data: Dict,
+                                               out_start_year: int,
+                                               out_end_year: int) -> Dict[DateTime, float]:
+        """
+        Convert the API specific data into a time series dict that can be digested by gsy-web.
+        """

gsy_framework/solar_api_clients/solar_api_client_active.py

@@ -0,0 +1,101 @@
+# pylint: disable=invalid-name
+import logging
+from typing import Dict
+
+import requests
+from pendulum import from_format, DateTime, today
+from gsy_framework.constants_limits import TIME_ZONE
+
+from gsy_framework.solar_api_clients.api_client_base import SolarAPIClientBase, PvApiParameters
+
+
+class SolarAPIClientActiveException(Exception):
+    """Exception for active solar api client"""
+
+
+TIME_OUTPUT_FORMAT = "YYYY-MM-DDTHH:mm:ss+00:00"
+TIME_INPUT_FORMAT = "YYYY-MM-DD"
+NORMAL_YEAR = 2019
+LEAP_YEAR = 2020
+FIRST_AVAILABLE_YEAR = 1979
+
+
+class SolarAPIClientActive(SolarAPIClientBase):
+    """ETL for energy data from active solar API."""
+
+    @staticmethod
+    def _get_corresponding_historical_time_stamp(input_datetime: DateTime) -> DateTime:
+        """
+        Substitute the year of the input_datetime with the leap/normal years used as reference.
+        """
+        if ((input_datetime.year >= today().year) or
+                (input_datetime.year < FIRST_AVAILABLE_YEAR)):
+            # if the user request pseudo-future data, use historical years
+            request_year = (LEAP_YEAR
+                            if input_datetime.is_leap_year() else NORMAL_YEAR)
+            output_datetime = input_datetime.set(year=request_year)
+        else:
+            output_datetime = input_datetime
+        return output_datetime
+
+    def _request_raw_solar_energy_data(self, request_parameters: PvApiParameters,
+                                       start_date: DateTime,
+                                       end_date: DateTime) -> Dict:
+        """Request energy profile from API.
+
+        return: Dictionary of raw data including a time series of energy production with
+                resolution of 1 hour.
+        """
+        payload = {"longitude": request_parameters.longitude,
+                   "latitude": request_parameters.latitude,
+                   "capacity_kw": request_parameters.capacity_kW,
+                   "azimuth": request_parameters.azimuth,
+                   "tilt": request_parameters.tilt,
+                   "frequency": "1h",  # never change this in order to receive energy values
+                   "start_date": start_date.format(TIME_INPUT_FORMAT),
+                   "end_date": end_date.format(TIME_INPUT_FORMAT)
+                   }
+
+        try:
+            response = requests.get(self.api_url, params=payload,
+                                    timeout=30)
+        except requests.exceptions.RequestException as ex:
+            error_message = ("An error happened when requesting solar energy profile from "
+                             f"active solar API: {ex}")
+            logging.error(error_message)
+            raise SolarAPIClientActiveException(error_message) from ex
+
+        if response.status_code != 200:
+            error_message = ("An error happened when requesting solar energy profile from "
+                             f"active solar API: status_code = {response.status_code}")
+            raise SolarAPIClientActiveException(error_message)
+
+        return response.json()
+
+    @staticmethod
+    def _create_time_series_from_solar_profile(request_data: Dict,
+                                               out_start_year: int,
+                                               out_end_year: int) -> Dict[DateTime, float]:
+        """
+        Convert time stamps back to DateTime and change requested years back to the buffer values.
+        """
+        try:
+            out_dict = {}
+            for index, energy in enumerate(request_data["Physical_Forecast"]):
+                time_key = from_format(request_data["valid_datetime"][index],
+                                       TIME_OUTPUT_FORMAT, tz=TIME_ZONE)
+                if time_key.year == LEAP_YEAR:
+                    out_year = out_start_year
+                elif time_key.year == NORMAL_YEAR:
+                    out_year = out_end_year
+                else:
+                    error_message = f"Unexpected year value for {time_key}"
+                    logging.error(error_message)
+                    assert False, error_message
+                out_dict[time_key.set(year=out_year)] = energy
+            return out_dict
+
+        except Exception as error:
+            raise SolarAPIClientActiveException("Something went wrong while converting "
+                                                "active solar API data to GSy-compatible: "
+                                                f"{error}.") from error

gsy_framework/solar_api_clients/solar_api_client_backup.py

@@ -0,0 +1,74 @@
+# pylint: disable=invalid-name
+from typing import Dict, Any
+
+import pendulum
+import requests
+from pendulum import DateTime
+from gsy_framework.constants_limits import TIME_ZONE
+
+from gsy_framework.solar_api_clients.api_client_base import SolarAPIClientBase, PvApiParameters
+
+LEAP_YEAR = 2016
+NORMAL_YEAR = 2015
+TIME_OUTPUT_FORMAT = "YYYYMMDD:HHmm"
+
+
+class SolarAPIClientBackupException(Exception):
+    """Exception for backup solar api client"""
+
+
+class SolarAPIClientBackup(SolarAPIClientBase):
+    """ETL for energy data from backup solar API."""
+
+    @staticmethod
+    def _get_corresponding_historical_time_stamp(input_datetime: DateTime) -> DateTime:
+
+        request_year = (LEAP_YEAR
+                        if input_datetime.is_leap_year() else NORMAL_YEAR)
+        output_datetime = input_datetime.set(year=request_year)
+        return output_datetime
+
+    def _request_raw_solar_energy_data(
+            self, request_parameters: PvApiParameters, start_date: DateTime, end_date: DateTime
+    ) -> Any:
+        try:
+            params = {"lat": request_parameters.latitude,
+                      "lon": request_parameters.longitude,
+                      "outputformat": "json",
+                      "angle": request_parameters.tilt,
+                      "aspect": request_parameters.azimuth - 180,
+                      "pvcalculation": 1,
+                      "pvtechchoice": "crystSi",
+                      "mountingplace": "free",
+                      "trackingtype": 0,
+                      "components": 0,
+                      "usehorizon": 1,
+                      "optimalangles": 0,
+                      "optimalinclination": 0,
+                      "loss": 0,
+                      "peakpower": request_parameters.capacity_kW,
+                      "startyear": start_date.year,
+                      "endyear": end_date.year}
+
+            res = requests.get(self.api_url, params=params, timeout=30)
+            return res.json()
+
+        except Exception as ex:
+            raise SolarAPIClientBackupException from ex
+
+    @staticmethod
+    def _create_time_series_from_solar_profile(request_data: Dict,
+                                               out_start_year: int,
+                                               out_end_year: int) -> Dict[DateTime, float]:
+        out_dict = {}
+        for dp in request_data["outputs"]["hourly"]:
+            time_key = pendulum.from_format(dp["time"], TIME_OUTPUT_FORMAT, tz=TIME_ZONE)
+            if time_key.year == LEAP_YEAR:
+                out_year = out_start_year
+            elif time_key.year == NORMAL_YEAR:
+                out_year = out_end_year
+            else:
+                error_message = f"Unexpected year value for {time_key}"
+                assert False, error_message
+            out_dict[time_key.set(year=out_year, minute=0)] = dp["P"] / 1000
+        return out_dict