Enhance price minimum handling to consider end-of-day rates

Refactor methods to return a tuple with next price minimum timestamp and a flag for rechecking due to incomplete next-day rates. Update inverter charge controller logic to adjust waiting times and energy usage estimates accordingly.

Author: felixschndr

source/inverter_charge_controller.py

@@ -47,14 +47,22 @@ def start(self) -> None:
         while True:
             try:
                 if first_iteration:
-                    next_price_minimum = self.tibber_api_handler.get_timestamp_of_next_price_minimum(first_iteration)
+                    next_price_minimum, minimum_has_to_be_rechecked = (
+                        self.tibber_api_handler.get_timestamp_of_next_price_minimum(first_iteration)
+                    )
                     first_iteration = False
                 else:
-                    next_price_minimum = self._do_iteration()
+                    next_price_minimum, minimum_has_to_be_rechecked = self._do_iteration()
+
+                time_to_sleep_to = next_price_minimum
+                if minimum_has_to_be_rechecked:
+                    time_to_sleep_to = time_to_sleep_to.replace(hour=14, minute=0, second=0, microsecond=0)
+                    self.log.info(f"The price minimum has to re-checked at {time_to_sleep_to}. Waiting until then...")
+                else:
+                    self.log.info(f"The next price minimum is at {next_price_minimum}. Waiting until then...")
 
-                self.log.info(f"The next price minimum is at {next_price_minimum}. Waiting until then...")
                 self._write_newlines_to_log_file()
-                pause.until(next_price_minimum)
+                pause.until(time_to_sleep_to)
 
             except (ClientError, RequestException) as e:
                 self.log.exception(f"An exception occurred while trying to fetch data from a different system: {e}")
@@ -66,7 +74,7 @@ def start(self) -> None:
                 self.log.critical("Exiting now...")
                 exit(1)
 
-    def _do_iteration(self) -> datetime:  # FIXME: Find better name
+    def _do_iteration(self) -> tuple[datetime, bool]:  # FIXME: Find better name
         """
         Computes the optimal charging strategy for an inverter until the next price minimum.
 
@@ -80,16 +88,23 @@ def _do_iteration(self) -> datetime:  # FIXME: Find better name
 
         Returns:
             datetime: The timestamp of the next price minimum.
+            minimum_has_to_be_rechecked: Whether the price minimum has to be re-checked since not all the prices were
+                available yet.
         """
         self.log.info(
             "Waiting is over, now is the a price minimum. Checking what has to be done to reach the next minimum..."
         )
 
         timestamp_now = datetime.now(tz=self.timezone)
 
-        next_price_minimum = self.tibber_api_handler.get_timestamp_of_next_price_minimum()
+        next_price_minimum, minimum_has_to_be_rechecked = self.tibber_api_handler.get_timestamp_of_next_price_minimum()
         self.log.info(f"The next price minimum is at {next_price_minimum}")
 
+        if minimum_has_to_be_rechecked:
+            self.log.info(
+                "The price minimum has to be re-checked since it is at the end of a day and the price rates for tomorrow are unavailable"
+            )
+
         expected_power_harvested_till_next_minimum = self.sun_forecast_handler.get_solar_output_in_timeframe(
             timestamp_now, next_price_minimum
         )
@@ -111,6 +126,15 @@ def _do_iteration(self) -> datetime:  # FIXME: Find better name
             expected_energy_usage_till_next_minimum = self.sems_portal_api_handler.estimate_energy_usage_in_timeframe(
                 timestamp_now, next_price_minimum
             )
+
+        if minimum_has_to_be_rechecked:
+            power_usage_factor = 1.15
+            self.log.info(
+                "Since the real price minimum is unknown at the moment the expected power usage "
+                + f"({expected_energy_usage_till_next_minimum}) is increased by {(power_usage_factor - 1 ) * 100} %"
+            )
+            expected_energy_usage_till_next_minimum = expected_energy_usage_till_next_minimum * power_usage_factor
+
         self.log.info(
             f"The total expected energy usage till the next price minimum is {expected_energy_usage_till_next_minimum}"
         )
@@ -140,6 +164,7 @@ def _do_iteration(self) -> datetime:  # FIXME: Find better name
             "current energy in battery": current_energy_in_battery,
             "target min state of charge": target_min_state_of_charge,
             "energy to be in battery when reaching next minimum": energy_to_be_in_battery_when_reaching_next_minimum,
+            "minimum_has_to_be_rechecked": minimum_has_to_be_rechecked,
         }
         self.log.debug(f"Summary of energy values: {summary_of_energy_vales}")
 
@@ -151,7 +176,7 @@ def _do_iteration(self) -> datetime:  # FIXME: Find better name
         )
         if excess_energy.watt_hours > 0:
             self.log.info(f"There is {excess_energy} of excess energy, thus there is no need to charge")
-            return next_price_minimum
+            return next_price_minimum, minimum_has_to_be_rechecked
 
         missing_energy = excess_energy * -1
         self.log.info(f"There is a need to charge {missing_energy}")
@@ -181,7 +206,7 @@ def _do_iteration(self) -> datetime:  # FIXME: Find better name
             timestamp_starting_to_charge, timestamp_ending_to_charge, energy_bought
         )
 
-        return next_price_minimum
+        return next_price_minimum, minimum_has_to_be_rechecked
 
     def _charge_inverter(self, target_state_of_charge: int) -> None:
         """

source/tibber_api_handler.py

@@ -1,5 +1,6 @@
-from datetime import datetime
+from datetime import datetime, timedelta
 
+from dateutil import tz
 from energy_amount import EnergyRate
 from environment_variable_getter import EnvironmentVariableGetter
 from gql import Client, gql
@@ -18,7 +19,7 @@ def __init__(self):
         self.client = Client(transport=transport, fetch_schema_from_transport=True)
         self.maximum_threshold = 0.03  # in €
 
-    def get_timestamp_of_next_price_minimum(self, first_iteration: bool = False) -> datetime:
+    def get_timestamp_of_next_price_minimum(self, first_iteration: bool = False) -> tuple[datetime, bool]:
         """
         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.
@@ -46,6 +47,8 @@ def get_timestamp_of_next_price_minimum(self, first_iteration: bool = False) ->
 
         Returns:
             datetime: The timestamp of the next minimum energy rate.
+            minimum_has_to_be_rechecked: Whether the price minimum has to be re-checked since not all the prices were
+                available yet.
         """
         self.log.debug("Finding the price minimum...")
         api_result = self._fetch_upcoming_prices_from_api()
@@ -58,7 +61,13 @@ def get_timestamp_of_next_price_minimum(self, first_iteration: bool = False) ->
             energy_rates_between_first_and_second_maximum
         )
 
-        return minimum_of_energy_rates.timestamp
+        minimum_has_to_be_rechecked = (
+            self._check_if_minimum_is_at_end_of_day_and_energy_rates_of_tomorrow_are_unavailable(
+                minimum_of_energy_rates, upcoming_energy_rates
+            )
+        )
+
+        return minimum_of_energy_rates.timestamp, minimum_has_to_be_rechecked
 
     def _fetch_upcoming_prices_from_api(self) -> dict:
         """
@@ -212,3 +221,48 @@ def get_global_minimum_of_energy_rates(self, energy_rates_till_maximum: list[Ene
             f"Found {global_minimum_of_energy_rates} to be the global minimum of the energy rates between the first and second maximum"
         )
         return global_minimum_of_energy_rates
+
+    def _check_if_minimum_is_at_end_of_day_and_energy_rates_of_tomorrow_are_unavailable(
+        self, price_minimum: EnergyRate, upcoming_energy_rates: list[EnergyRate]
+    ) -> bool:
+        """
+        This method determines whether the timestamp of the `price_minimum` falls within the last 3 hours of the current
+        day and checks if there are no energy rates available for the subsequent day.
+        This is done since the price rates of the next day are only available after ~ 02:00 PM. If the price rates of
+        the next day are unavailable while determining the price minimum, it is likely that the price minimum is just
+        the last rate of the day but not actually the minimum.
+        In this case we have to check in later (after ~ 02:00 PM) to re-request the prices from the Tibber API to get
+        the values of the next day.
+
+        Args:
+            price_minimum (EnergyRate): The energy rate with the minimum price.
+            upcoming_energy_rates (list[EnergyRate]): List of upcoming energy rates.
+
+        Returns:
+            bool: True if the price minimum is in the last 3 hours of the current day and there are no rates for
+                the next day, otherwise False.
+        """
+
+        # We use 00:01 instead of 00:00 since the software runs just a few (milli)seconds after the start of the hour
+        timezone = tz.gettz()
+        end_of_day = (datetime.now(tz=timezone) + timedelta(days=1)).replace(hour=0, minute=1, second=0, microsecond=0)
+        three_hours_before_end_of_day = end_of_day - timedelta(hours=3)
+
+        is_price_minimum_near_end_of_day = price_minimum.timestamp >= three_hours_before_end_of_day
+        self.log.trace(
+            f"The price minimum {price_minimum.timestamp} is at the end of the day: {is_price_minimum_near_end_of_day}"
+        )
+
+        are_tomorrows_rates_unavailable = all(
+            rate.timestamp.date() == price_minimum.timestamp.date() for rate in upcoming_energy_rates
+        )
+        self.log.trace(f"The price rates for tomorrow are unavailable: {are_tomorrows_rates_unavailable}")
+
+        return is_price_minimum_near_end_of_day and are_tomorrows_rates_unavailable
+
+
+if __name__ == "__main__":
+    api_handler = TibberAPIHandler()
+    timestamp, minimum_has_to_be_rechecked = api_handler.get_timestamp_of_next_price_minimum(first_iteration=True)
+    print(f"The timestamp of the next minimum energy rate is {timestamp}")
+    print(f"The minimum has to be re-checked: {minimum_has_to_be_rechecked}")