Cache upcoming energy rates in an iteration

Author: felixschndr

source/inverter_charge_controller.py

@@ -58,7 +58,7 @@ def start(self) -> None:
         while True:
             try:
                 if first_iteration:
-                    self.next_price_minimum = self.tibber_api_handler.get_next_price_minimum(first_iteration)
+                    self.next_price_minimum = self.tibber_api_handler.get_next_price_minimum(first_iteration=True)
                     first_iteration = False
                 else:
                     self._do_iteration()
@@ -75,7 +75,7 @@ def start(self) -> None:
                     pause.until(time_to_sleep_to)
                     self.write_newlines_to_log_file()
                     self.log.info("Waking up since the the price minimum has to re-checked")
-                    self.next_price_minimum = self.tibber_api_handler.get_next_price_minimum(True)
+                    self.next_price_minimum = self.tibber_api_handler.get_next_price_minimum(first_iteration=True)
 
                 self.sems_portal_api_handler.write_values_to_database()
 
@@ -464,13 +464,25 @@ def _write_energy_buy_statistics_to_database(
     In this case, previous values from the same iteration are saved and used when the failed API call is retried.
     """
 
+    def _get_upcoming_energy_rates(self) -> list[EnergyRate]:
+        cache_key = "upcoming_energy_rates"
+        upcoming_energy_rates = self._get_value_from_cache_if_exists(cache_key)
+        if upcoming_energy_rates:
+            return upcoming_energy_rates
+
+        upcoming_energy_rates = self.tibber_api_handler.get_upcoming_energy_rates()
+        self._set_cache_key(cache_key, upcoming_energy_rates)
+        return upcoming_energy_rates
+
     def _get_next_price_minimum(self) -> EnergyRate:
         cache_key = "next_price_minimum"
         next_price_minimum = self._get_value_from_cache_if_exists(cache_key)
         if next_price_minimum:
             return next_price_minimum
 
-        next_price_minimum = self.tibber_api_handler.get_next_price_minimum()
+        next_price_minimum = self.tibber_api_handler.get_next_price_minimum(
+            upcoming_energy_rates=self._get_upcoming_energy_rates()
+        )
         self._set_cache_key(cache_key, next_price_minimum)
         return next_price_minimum
 

source/tibber_api_handler.py

@@ -23,9 +23,9 @@ def __init__(self):
 
         self.database_handler = DatabaseHandler("energy_prices")
 
-        self.upcoming_energy_rates_cache = []
-
-    def get_next_price_minimum(self, first_iteration: bool = False) -> EnergyRate:
+    def get_next_price_minimum(
+        self, first_iteration: bool = False, upcoming_energy_rates: list[EnergyRate] = None
+    ) -> EnergyRate:
         """
         This method performs a series of operations to determine the most cost-effective time to charge by analyzing
         upcoming energy rates retrieved from the Tibber API and returns its timestamp.
@@ -55,11 +55,8 @@ def get_next_price_minimum(self, first_iteration: bool = False) -> EnergyRate:
             EnergyRate: The next price minimum energy rate.
         """
         self.log.debug("Finding the price minimum...")
-        api_result = self._fetch_upcoming_prices_from_api()
-        all_energy_rates = self._extract_energy_rates_from_api_response(api_result)
-        self.write_energy_rates_to_database(all_energy_rates)
-        upcoming_energy_rates = self._remove_energy_rates_from_the_past(all_energy_rates)
-        self.upcoming_energy_rates_cache = upcoming_energy_rates
+        if upcoming_energy_rates is None:
+            upcoming_energy_rates = self.get_upcoming_energy_rates()
         if first_iteration and not self._check_if_next_three_prices_are_greater_than_current_one(
             upcoming_energy_rates
         ):
@@ -85,6 +82,13 @@ def get_next_price_minimum(self, first_iteration: bool = False) -> EnergyRate:
 
         return minimum_of_energy_rates
 
+    def get_upcoming_energy_rates(self) -> list[EnergyRate]:
+        self.log.debug("Fetching the upcoming energy rates from the Tibber API...")
+        api_result = self._fetch_upcoming_prices_from_api()
+        all_energy_rates = self._extract_energy_rates_from_api_response(api_result)
+        self.write_energy_rates_to_database(all_energy_rates)
+        return self._remove_energy_rates_from_the_past(all_energy_rates)
+
     @staticmethod
     def _check_if_next_three_prices_are_greater_than_current_one(all_upcoming_energy_rates: list[EnergyRate]) -> bool:
         """