Implement behavior for when the current price minimum is less expensive than the next price minimum

Author: felixschndr

source/inverter_charge_controller.py

@@ -80,7 +80,6 @@ def _start(self) -> None:
                     next_price_minimum = self.tibber_api_handler.get_next_price_minimum(first_iteration)
                     first_iteration = False
                 else:
-                    self.write_newlines_to_log_file()
                     next_price_minimum = self._do_iteration(next_price_minimum)
 
                 if next_price_minimum.has_to_be_rechecked:
@@ -124,13 +123,19 @@ def _do_iteration(self, current_energy_rate: EnergyRate) -> EnergyRate:
          - Sets the maximum charging duration of the current energy rate.
          - Gets the current state of the inverter.
          - Gets the average power consumption.
-         - Calculates the estimated minimum state of charge until the next price minimum.
-         - If the estimated minimum state of charge is lower than the target minimum state of charge, the method
-           calculates the required state of charge to reach the target minimum state of charge and initiates charging.
+         - Calculates the estimated minimum and maximum state of charge until the next price minimum.
+         - If the next price minimum is higher than the current one
+            - If the estimated minimum state of charge is higher than the target minimum state of charge no charging is
+              initiated.
+            - Else the method calculates the required state of charge to reach the target minimum state of charge and
+              start charging.
+         - Else the method calculates how much power can be bought such that no energy of the sun is wasted.
+           "Wasted" being the energy is sold instead of used to charge the battery.
 
         Returns:
             EnergyRate: The next price minimum energy rate.
         """
+        self.write_newlines_to_log_file()
         self.log.info(
             "Waiting is over, now is the a price minimum. Checking what has to be done to reach the next minimum..."
         )
@@ -145,77 +150,89 @@ def _do_iteration(self, current_energy_rate: EnergyRate) -> EnergyRate:
             f"{current_energy_rate.maximum_charging_duration}"
         )
 
-        if next_price_minimum > current_energy_rate:
-            # Information is unused at the moment
-            self.log.info("The price of the upcoming minimum is higher than the current energy rate")
-
         current_state_of_charge = self.inverter.get_state_of_charge()
         self.log.info(f"The battery is currently is at {current_state_of_charge}")
 
         average_power_consumption = self._get_average_power_consumption()
         self.log.info(f"The average power consumption is {average_power_consumption}")
 
-        minimum_of_soc_until_next_price_minimum = self._get_minimum_of_soc_until_next_price_minimum(
-            next_price_minimum.timestamp,
-            average_power_consumption,
-            current_state_of_charge,
-            next_price_minimum.has_to_be_rechecked,
-        )
-        self.log.info(
-            f"The expected minimum of state of charge until the next price minimum with the current state of charge is "
-            f"{minimum_of_soc_until_next_price_minimum}"
-        )
-
         target_min_soc = StateOfCharge.from_percentage(
             int(EnvironmentVariableGetter.get("INVERTER_TARGET_MIN_STATE_OF_CHARGE", 15))
         )
         self.log.info(f"The battery shall be at least at {target_min_soc} at all times")
 
+        target_max_soc = StateOfCharge.from_percentage(
+            int(EnvironmentVariableGetter.get("INVERTER_TARGET_MAX_STATE_OF_CHARGE", 95))
+        )
+        self.log.info(f"The battery shall be at most be charged up to {target_max_soc}")
+
+        minimum_of_soc_until_next_price_minimum, maximum_of_soc_until_next_price_minimum = (
+            self.sun_forecast_handler.calculate_min_and_max_of_soc_in_timeframe(
+                timestamp_now,
+                next_price_minimum.timestamp,
+                average_power_consumption,
+                current_state_of_charge,
+                next_price_minimum.has_to_be_rechecked,
+            )
+        )
+        self.log.info(
+            f"The expected minimum of state of charge until the next price minimum with the current state of charge is "
+            f"{minimum_of_soc_until_next_price_minimum}, "
+            f"the expected maximum is {maximum_of_soc_until_next_price_minimum}"
+        )
+
         summary_of_energy_vales = {
             "timestamp now": str(timestamp_now),
             "next price minimum": next_price_minimum,
-            "minimum_has_to_be_rechecked": next_price_minimum.has_to_be_rechecked,
+            "next price minimum has to be rechecked": next_price_minimum.has_to_be_rechecked,
             "maximum charging duration": current_energy_rate.format_maximum_charging_duration(),
             "current state of charge": current_state_of_charge,
             "average power consumption": average_power_consumption.watts,
-            "minimum of soc until next price minimum": minimum_of_soc_until_next_price_minimum,
             "target min soc": target_min_soc,
+            "target max soc": target_max_soc,
+            "minimum of soc until next price minimum": minimum_of_soc_until_next_price_minimum,
+            "maximum of soc until next price minimum": maximum_of_soc_until_next_price_minimum,
         }
         self.log.debug(f"Summary of energy values: {summary_of_energy_vales}")
 
-        if minimum_of_soc_until_next_price_minimum > target_min_soc:
+        if current_energy_rate > next_price_minimum:
             self.log.info(
-                "The expected minimum state of charge until the next price minimum without additional charging "
-                "is higher than the target minimum state of charge --> There is no need to charge"
+                f"The price of the current minimum ({next_price_minimum.rate} ct/kWh) is higher than the one of "
+                f"the upcoming minimum ({current_energy_rate.rate} ct/kWh) "
+                "--> Will only charge the battery to reach the next price minimum"
             )
-            self.iteration_cache = {}
-            return next_price_minimum
+            if minimum_of_soc_until_next_price_minimum > target_min_soc:
+                self.log.info(
+                    "The expected minimum state of charge until the next price minimum without additional charging "
+                    "is higher than the target minimum state of charge --> There is no need to charge"
+                )
+                self.iteration_cache = {}
+                return next_price_minimum
 
-        required_state_of_charge = current_state_of_charge + (target_min_soc - minimum_of_soc_until_next_price_minimum)
-        self.log.info(f"There is a need to charge to {required_state_of_charge} (from {current_state_of_charge})")
-        maximum_possible_state_of_charge = StateOfCharge.from_percentage(100)
-        if required_state_of_charge > maximum_possible_state_of_charge:
+            charging_target_soc = current_state_of_charge + (target_min_soc - minimum_of_soc_until_next_price_minimum)
+        else:
             self.log.info(
-                "The target state of charge is higher than possible "
-                f"--> Setting it to {maximum_possible_state_of_charge}"
+                f"The price of the upcoming minimum ({next_price_minimum.rate} ct/kWh) is higher than the one of "
+                f"the current minimum ({current_energy_rate.rate} ct/kWh)"
+                "--> Will charge as much as possible without wasting any energy of the sun"
+            )
+            charging_target_soc = current_state_of_charge + (
+                StateOfCharge.from_percentage(100) - maximum_of_soc_until_next_price_minimum
             )
-            required_state_of_charge = maximum_possible_state_of_charge
+        self.log.info(f"The calculated target state of charge is {charging_target_soc}")
 
-        max_target_soc = StateOfCharge.from_percentage(
-            int(EnvironmentVariableGetter.get("INVERTER_TARGET_MAX_STATE_OF_CHARGE", 95))
-        )
-        if required_state_of_charge > max_target_soc:
+        if charging_target_soc > target_max_soc:
             self.log.info(
                 "The target state of charge is more than the maximum allowed charge set in the environment "
-                f"--> Setting it to {max_target_soc}"
+                f"--> Setting it to {target_max_soc}"
             )
-            required_state_of_charge = max_target_soc
+            charging_target_soc = target_max_soc
 
         energy_bought_before_charging = self.sems_portal_api_handler.get_energy_buy()
         timestamp_starting_to_charge = TimeHandler.get_time()
         self.log.debug(f"The amount of energy bought before charging is {energy_bought_before_charging}")
 
-        self._charge_inverter(required_state_of_charge, current_energy_rate.maximum_charging_duration)
+        self._charge_inverter(charging_target_soc, current_energy_rate.maximum_charging_duration)
 
         timestamp_ending_to_charge = TimeHandler.get_time()
 
@@ -408,30 +425,6 @@ def _get_average_power_consumption(self) -> Power:
         self._set_cache_key(cache_key, average_power_consumption)
         return average_power_consumption
 
-    def _get_minimum_of_soc_until_next_price_minimum(
-        self,
-        next_price_minimum_timestamp: datetime,
-        average_power_consumption: Power,
-        current_soc: StateOfCharge,
-        minimum_has_to_rechecked: bool,
-    ) -> StateOfCharge:
-        cache_key = "minimum_of_soc_until_next_price_minimum"
-        minimum_of_soc_until_next_price_minimum = self._get_value_from_cache_if_exists(cache_key)
-        if minimum_of_soc_until_next_price_minimum:
-            return minimum_of_soc_until_next_price_minimum
-
-        minimum_of_soc_until_next_price_minimum, _ = (
-            self.sun_forecast_handler.calculate_minimum_of_soc_and_power_generation_in_timeframe(
-                TimeHandler.get_time(),
-                next_price_minimum_timestamp,
-                average_power_consumption,
-                current_soc,
-                minimum_has_to_rechecked,
-            )
-        )
-        self._set_cache_key(cache_key, minimum_of_soc_until_next_price_minimum)
-        return minimum_of_soc_until_next_price_minimum
-
     def _get_value_from_cache_if_exists(self, cache_key: str) -> Optional[Any]:
         if cache_key not in self.iteration_cache.keys():
             return None

source/sun_forecast_handler.py

@@ -22,16 +22,16 @@ def __init__(self):
 
         self.database_handler = DatabaseHandler("solar_forecast")
 
-    def calculate_minimum_of_soc_and_power_generation_in_timeframe(
+    def calculate_min_and_max_of_soc_in_timeframe(
         self,
         timeframe_start: datetime,
         timeframe_end: datetime,
         average_power_usage: Power,
         starting_soc: StateOfCharge,
-        minimum_has_to_rechecked: bool = False,
-    ) -> tuple[StateOfCharge, EnergyAmount]:
+        minimum_has_to_rechecked: bool,
+    ) -> tuple[StateOfCharge, StateOfCharge]:
         """
-        Calculates the minimum state of charge (SOC) and total power generation within a specified timeframe.
+        Calculates the minimum state of charge (SOC) and maximum state of charge within a specified timeframe.
         It considers average power usage, initial SOC and optionally adjusts for higher power usage in cases where the
         pricing for the next day is unavailable. This function uses solar data and iteratively computes power usage and
         generation for subintervals within the timeframe.
@@ -41,12 +41,12 @@ def calculate_minimum_of_soc_and_power_generation_in_timeframe(
             timeframe_end: The ending timestamp of the timeframe.
             average_power_usage: The average power consumption over the timeframe.
             starting_soc: The battery's state of charge at the beginning of the timeframe.
-            minimum_has_to_rechecked (optional): Whether to increase the power usage by POWER_USAGE_INCREASE_FACTOR
+            minimum_has_to_rechecked: Whether to increase the power usage by POWER_USAGE_INCREASE_FACTOR
 
         Returns:
             A tuple containing:
                 - The minimum state of charge observed during the timeframe.
-                - The total amount of power generated within the timeframe.
+                - The maximum state of charge observed during the timeframe.
         """
         self.log.debug(
             "Calculating the estimated minimum of state of charge and power generation in the timeframe "
@@ -66,8 +66,9 @@ def calculate_minimum_of_soc_and_power_generation_in_timeframe(
         current_timeframe_start = timeframe_start
         soc_after_current_timeframe = starting_soc
         minimum_soc = starting_soc
-        total_power_usage = EnergyAmount(0)
-        total_power_generation = EnergyAmount(0)
+        maximum_soc = starting_soc
+        total_energy_used = EnergyAmount(0)
+        total_energy_harvested = EnergyAmount(0)
 
         first_iteration = True
         while True:
@@ -83,40 +84,46 @@ def calculate_minimum_of_soc_and_power_generation_in_timeframe(
 
             current_timeframe_end = current_timeframe_start + current_timeframe_duration
 
-            power_usage_during_timeframe = self._calculate_energy_usage_in_timeframe(
+            energy_usage_during_timeframe = self._calculate_energy_usage_in_timeframe(
                 current_timeframe_start, current_timeframe_duration, average_power_usage, power_usage_increase_factor
             )
-            total_power_usage += power_usage_during_timeframe
-            power_generation_during_timeframe = self._get_energy_produced_in_timeframe_from_solar_data(
+            total_energy_used += energy_usage_during_timeframe
+            energy_harvested_during_timeframe = self._get_energy_harvested_in_timeframe_from_solar_data(
                 current_timeframe_end, current_timeframe_duration, solar_data
             )
-            total_power_generation += power_generation_during_timeframe
+            total_energy_harvested += energy_harvested_during_timeframe
             soc_after_current_timeframe = StateOfCharge(
-                soc_after_current_timeframe.absolute - power_usage_during_timeframe + power_generation_during_timeframe
+                soc_after_current_timeframe.absolute
+                - energy_usage_during_timeframe
+                + energy_harvested_during_timeframe
             )
 
             if soc_after_current_timeframe < minimum_soc:
                 log_text = " (new minimum)"
                 minimum_soc = soc_after_current_timeframe
+            elif soc_after_current_timeframe > maximum_soc:
+                log_text = " (new maximum)"
+                maximum_soc = soc_after_current_timeframe
             else:
                 log_text = ""
             self.log.debug(
                 f"{current_timeframe_end}"
                 f" - estimated SOC: {soc_after_current_timeframe}{log_text}"
-                f" - expected power usage: {power_usage_during_timeframe}"
-                f" - expected power generation: {power_generation_during_timeframe}"
+                f" - expected energy used: {energy_usage_during_timeframe}"
+                f" - expected energy harvested: {energy_harvested_during_timeframe}"
             )
             current_timeframe_start += current_timeframe_duration
 
             if current_timeframe_start >= timeframe_end:
                 break
 
         self.log.debug(
-            f"From {timeframe_start} to {timeframe_end} the expected minimum of state of charge is {minimum_soc}, the "
-            f"expected total amount of power usage is {total_power_usage} and the expected total amount of power "
-            f"generated is {total_power_generation}"
+            f"From {timeframe_start} to {timeframe_end} the expected minimum of state of charge is {minimum_soc}, "
+            f"the expected maximum of state of charge is {maximum_soc}, "
+            f"the expected total amount of energy used is {total_energy_used} "
+            f"and the expected total amount of energy harvested is {total_energy_harvested}"
         )
-        return minimum_soc, total_power_generation
+        return minimum_soc, maximum_soc
 
     def retrieve_solar_data(self, timeframe_start: datetime, timeframe_end: datetime) -> dict[str, Power]:
         """
@@ -303,7 +310,7 @@ def _calculate_energy_usage_in_timeframe(
             return average_power_usage * factor_energy_usage_during_the_day
         return average_power_usage * factor_energy_usage_during_the_night
 
-    def _get_energy_produced_in_timeframe_from_solar_data(
+    def _get_energy_harvested_in_timeframe_from_solar_data(
         self, timeframe_end: datetime, timeframe_duration: timedelta, solar_data: dict[str, Power]
     ) -> EnergyAmount:
         """