Eliminated the INVERTER_MAX_TARGET_SOC restriction from the environment and charging logic. Updated the inverter charging process to account for consecutive cheap energy rates, allowing for extended charging times if favorable pricing persists.

Author: felixschndr

.env.example

@@ -13,7 +13,6 @@ TIBBER_API_TOKEN=
 INVERTER_HOSTNAME=
 INVERTER_BATTERY_CAPACITY=
 INVERTER_TARGET_MIN_STATE_OF_CHARGE=
-INVERTER_MAX_TARGET_SOC=
 
 SEMSPORTAL_USERNAME=
 SEMSPORTAL_PASSWORD=

README.md

@@ -134,7 +134,6 @@ When the current time is a price minimum, the program wakes up and does the foll
    | `INVERTER_HOSTNAME`                                                                             | The hostname or IP of the inverter.                                                                                                                                                                                                                                                 | -                            | [`inverter.mydomain.com`, `192.168.5.10`, ...]                                                                               |
    | `INVERTER_BATTERY_CAPACITY`                                                                     | The capacity of the battery in watt hours without any separators.                                                                                                                                                                                                                   | -                            | A number, typically between `3000` and `15000`                                                                               |
    | `INVERTER_TARGET_MIN_STATE_OF_CHARGE`                                                           | The state of charge the battery shall have when reaching the next minimum as a buffer.                                                                                                                                                                                              | `20`                         | A number between `0` and `100`, typically between `0` and `40`                                                               |
-   | `INVERTER_MAX_TARGET_SOC`                                                                       | The maximum state of charge the inverter will charge to since the last few percent take a long time to charge.                                                                                                                                                                      | `90`                         | A number between `0` and `100`, typically between `80` and `100`                                                             |
    | `SEMSPORTAL_USERNAME`                                                                           | The username to login into the SEMSPortal.                                                                                                                                                                                                                                          | -                            | A string, example: `[email protected]`                                                                                       |
    | `SEMSPORTAL_PASSWORD`                                                                           | The password to login into the SEMSPortal.                                                                                                                                                                                                                                          | -                            | A string, example: `my-secret-password`                                                                                      |
    | `SEMSPORTAL_POWERSTATION_ID`                                                                    | The ID of the inverter in the SEMSPortal. This can be found at the end of the URL in the browser after logging in.                                                                                                                                                                  | -                            | A string, example: `aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee`                                                                    |

source/inverter_charge_controller.py

@@ -78,7 +78,7 @@ def _start(self) -> None:
                 self.sems_portal_api_handler.write_values_to_database()
 
                 if first_iteration:
-                    next_price_minimum = self.tibber_api_handler.get_next_price_minimum(first_iteration)
+                    next_price_minimum, _ = self.tibber_api_handler.get_next_price_minimum(first_iteration)
                     first_iteration = False
                 else:
                     next_price_minimum = self._do_iteration(next_price_minimum.rate)
@@ -94,7 +94,7 @@ def _start(self) -> None:
                     )
                     pause.until(time_to_sleep_to)
                     self.log.info("Waking up since the the price minimum has to re-checked")
-                    next_price_minimum = self.tibber_api_handler.get_next_price_minimum(True)
+                    next_price_minimum, _ = self.tibber_api_handler.get_next_price_minimum(True)
 
                 self.log.info(f"The next price minimum is at {next_price_minimum.timestamp}. Waiting until then...")
 
@@ -132,10 +132,11 @@ def _do_iteration(self, current_energy_rate: float) -> EnergyRate:
 
         timestamp_now = TimeHandler.get_time()
 
-        next_price_minimum = self.tibber_api_handler.get_next_price_minimum()
+        next_price_minimum, consecutive_energy_rate_is_cheap = self.tibber_api_handler.get_next_price_minimum()
         self.log.info(f"The next price minimum is at {next_price_minimum}")
 
         if next_price_minimum.rate > 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")
 
         if next_price_minimum.is_minimum_that_has_to_be_rechecked:
@@ -203,6 +204,7 @@ def _do_iteration(self, current_energy_rate: float) -> EnergyRate:
         summary_of_energy_vales = {
             "timestamp now": str(timestamp_now),
             "next price minimum": str(next_price_minimum.timestamp),
+            "consecutive_energy_rate_is_cheap": consecutive_energy_rate_is_cheap,
             "expected power harvested till next minimum": expected_power_harvested_till_next_minimum,
             "expected energy usage till next minimum": expected_energy_usage_till_next_minimum,
             "current state of charge": current_state_of_charge,
@@ -232,20 +234,11 @@ def _do_iteration(self, current_energy_rate: float) -> EnergyRate:
         )
         self.log.info(f"Need to charge to {required_state_of_charge} %")
 
-        max_target_soc_environment_variable = "INVERTER_MAX_TARGET_SOC"
-        max_target_soc = int(EnvironmentVariableGetter.get(max_target_soc_environment_variable, 90))
-        if required_state_of_charge > max_target_soc:
-            self.log.info(
-                "The target state of charge is more than the maximum allowed charge set in the environment "
-                + f'("{max_target_soc_environment_variable}") --> Setting it to {max_target_soc} %'
-            )
-            required_state_of_charge = max_target_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)
+        self._charge_inverter(required_state_of_charge, consecutive_energy_rate_is_cheap)
 
         timestamp_ending_to_charge = TimeHandler.get_time()
 
@@ -268,22 +261,31 @@ def _do_iteration(self, current_energy_rate: float) -> EnergyRate:
 
         return next_price_minimum
 
-    def _charge_inverter(self, target_state_of_charge: int) -> None:
+    def _charge_inverter(self, target_state_of_charge: int, consecutive_energy_rate_is_cheap: bool) -> None:
         """
         Charges the inverter until a given state of charge is reached.
         Checks every few minutes the current state of charge and compares to the target value.
+        Charges the inverter for a maximum of one hour. If consecutive_energy_rate_is_cheap is True it charges for a
+        maximum of two hours.
+        --> Stops when the target state of charge or the maximum charge time is reached (whichever comes first)
 
         Args:
             target_state_of_charge: The desired state of charge percentage to reach before stopping the charging process.
         """
         charging_progress_check_interval = timedelta(minutes=5)
         dry_run = EnvironmentVariableGetter.get(name_of_variable="DRY_RUN", default_value=True)
 
+        maximum_end_charging_time = TimeHandler.get_time().replace(minute=0, second=0) + timedelta(hours=1)
+        if consecutive_energy_rate_is_cheap:
+            maximum_end_charging_time += timedelta(hours=1)
+
         self.log.info("Starting to charge")
         self.inverter.set_operation_mode(OperationMode.ECO_CHARGE)
 
         self.log.info(
-            f"Set the inverter to charge, the target state of charge is {target_state_of_charge} %. Checking the charging progress every {charging_progress_check_interval}..."
+            f"Set the inverter to charge, the target state of charge is {target_state_of_charge} %. "
+            + f"The maximum end charging time is {maximum_end_charging_time.strftime("%H:%M:%S")}. "
+            + f"Checking the charging progress every {charging_progress_check_interval}..."
         )
 
         error_counter = 0
@@ -326,6 +328,14 @@ def _charge_inverter(self, target_state_of_charge: int) -> None:
                 self.inverter.set_operation_mode(OperationMode.GENERAL)
                 break
 
+            if TimeHandler.get_time() > maximum_end_charging_time:
+                self.log.info(
+                    f"The maximum end charging time of {maximum_end_charging_time} has been reached "
+                    + "--> Stopping the charging process"
+                )
+                self.inverter.set_operation_mode(OperationMode.GENERAL)
+                break
+
             self.log.debug(
                 f"Charging is still ongoing (current: {current_state_of_charge} %, target: >= {target_state_of_charge} %) --> Waiting for another {charging_progress_check_interval}..."
             )

source/tibber_api_handler.py

@@ -22,7 +22,7 @@ def __init__(self):
 
         self.database_handler = DatabaseHandler("energy_prices")
 
-    def get_next_price_minimum(self, first_iteration: bool = False) -> EnergyRate:
+    def get_next_price_minimum(self, first_iteration: bool = False) -> [EnergyRate, 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.
@@ -49,6 +49,8 @@ def get_next_price_minimum(self, first_iteration: bool = False) -> EnergyRate:
 
         Returns:
             EnergyRate: The next price minimum energy rate.
+            bool: Whether the energy rate after the minimum is at most self.maximum_threshold more expensive than the
+            minimum. This is used to determine the maximum charging time.
         """
         self.log.debug("Finding the price minimum...")
         api_result = self._fetch_upcoming_prices_from_api()
@@ -78,8 +80,51 @@ def get_next_price_minimum(self, first_iteration: bool = False) -> EnergyRate:
         ):
             minimum_of_energy_rates.is_minimum_that_has_to_be_rechecked = True
 
+        energy_rate_after_minimum = self._get_energy_rate_after_minimum(minimum_of_energy_rates, upcoming_energy_rates)
+        energy_rate_after_minimum_is_below_threshold = self._is_consecutive_energy_rate_below_threshold(
+            minimum_of_energy_rates, energy_rate_after_minimum
+        )
+
+        return minimum_of_energy_rates, energy_rate_after_minimum_is_below_threshold
+
+    def _get_energy_rate_after_minimum(
+        self, minimum_of_energy_rates: EnergyRate, upcoming_energy_rates: list[EnergyRate]
+    ) -> EnergyRate:
+        """
+        Determines the energy rate that immediately follows the minimum energy rate in the list of upcoming energy rates.
+        If the minimum energy rate is the last element in the list, the same minimum value is returned.
+
+        Args:
+            minimum_of_energy_rates: The minimum energy rate in the list of energy rates.
+            upcoming_energy_rates: A list of energy rates where the minimum energy rate resides.
+
+        Returns:
+            The energy rate that comes directly after the provided minimum energy rate within the list, or the minimum
+            energy rate itself if it is the last one in the list.
+        """
+        index = upcoming_energy_rates.index(minimum_of_energy_rates)
+        if index + 1 < len(upcoming_energy_rates):
+            return upcoming_energy_rates[index + 1]
+
+        self.log.debug("Can't get the energy rate after the minimum since the minimum is the last one in the list")
         return minimum_of_energy_rates
 
+    def _is_consecutive_energy_rate_below_threshold(
+        self, first_energy_rate: EnergyRate, second_energy_rate: EnergyRate
+    ) -> bool:
+        """
+        Checks if the difference between two consecutive energy rates is below the defined threshold.
+
+        Args:
+            first_energy_rate: The first energy rate object to compare.
+            second_energy_rate: The second energy rate object to compare.
+
+        Returns:
+            bool: True if the difference between the rates of the two energy rate objects is less than or
+            equal to the `maximum_threshold`, otherwise False.
+        """
+        return second_energy_rate.rate - first_energy_rate.rate <= self.maximum_threshold
+
     @staticmethod
     def _check_if_next_three_prices_are_greater_than_current_one(all_upcoming_energy_rates: list[EnergyRate]) -> bool:
         """