Skip to content

Commit

Permalink
Replaced the tuple-based return structure with a property in the `Ene…
Browse files Browse the repository at this point in the history
…rgyRate` class to indicate if the price minimum requires rechecking.
  • Loading branch information
felixschndr committed Dec 10, 2024
1 parent 8fc1b38 commit 1dfc428
Show file tree
Hide file tree
Showing 3 changed files with 17 additions and 22 deletions.
1 change: 1 addition & 0 deletions source/energy_amount.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ def from_kilo_watts(kilo_watts: float) -> Power:
class EnergyRate:
rate: float
timestamp: datetime
is_minimum_that_has_to_be_rechecked: bool = False

def __repr__(self):
return f"{self.rate} € at {self.timestamp}"
Expand Down
28 changes: 12 additions & 16 deletions source/inverter_charge_controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,22 +47,20 @@ def start(self) -> None:
while True:
try:
if first_iteration:
next_price_minimum, minimum_has_to_be_rechecked = 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, minimum_has_to_be_rechecked = self._do_iteration(next_price_minimum.rate)
next_price_minimum = self._do_iteration(next_price_minimum.rate)

if minimum_has_to_be_rechecked:
if next_price_minimum.is_minimum_that_has_to_be_rechecked:
now = TimeHandler.get_time()
time_to_sleep_to = now.replace(hour=14, minute=0, second=0, microsecond=0)
if now > time_to_sleep_to:
time_to_sleep_to += timedelta(days=1)
self.log.info(f"The price minimum has to re-checked at {time_to_sleep_to}. Waiting until then...")
self.log.info(f"The price minimum has to re-checked --> Waiting until {time_to_sleep_to}...")
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...")

Expand All @@ -79,7 +77,7 @@ def start(self) -> None:
self.log.critical("Exiting now...")
exit(1)

def _do_iteration(self, current_energy_rate: float) -> tuple[EnergyRate, bool]: # FIXME: Find better name
def _do_iteration(self, current_energy_rate: float) -> EnergyRate:
"""
Computes the optimal charging strategy for an inverter until the next price minimum.
Expand All @@ -93,22 +91,20 @@ def _do_iteration(self, current_energy_rate: float) -> tuple[EnergyRate, bool]:
Returns:
EnergyRate: The next price 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.info(
"Waiting is over, now is the a price minimum. Checking what has to be done to reach the next minimum..."
)

timestamp_now = TimeHandler.get_time()

next_price_minimum, minimum_has_to_be_rechecked = self.tibber_api_handler.get_next_price_minimum()
next_price_minimum = 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:
self.log.info("The price of the upcoming minimum is higher than the current energy rate")

if minimum_has_to_be_rechecked:
if next_price_minimum.is_minimum_that_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"
)
Expand All @@ -135,7 +131,7 @@ def _do_iteration(self, current_energy_rate: float) -> tuple[EnergyRate, bool]:
timestamp_now, next_price_minimum.timestamp
)

if minimum_has_to_be_rechecked:
if next_price_minimum.is_minimum_that_has_to_be_rechecked:
power_usage_increase_factor = 20
self.log.info(
"Since the real price minimum is unknown at the moment the expected power usage "
Expand Down Expand Up @@ -172,7 +168,7 @@ def _do_iteration(self, current_energy_rate: float) -> tuple[EnergyRate, bool]:
"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,
"minimum_has_to_be_rechecked": next_price_minimum.is_minimum_that_has_to_be_rechecked,
}
self.log.debug(f"Summary of energy values: {summary_of_energy_vales}")

Expand All @@ -184,7 +180,7 @@ def _do_iteration(self, current_energy_rate: float) -> tuple[EnergyRate, bool]:
)
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, minimum_has_to_be_rechecked
return next_price_minimum

missing_energy = excess_energy * -1
self.log.info(f"There is a need to charge {missing_energy}")
Expand Down Expand Up @@ -229,7 +225,7 @@ def _do_iteration(self, current_energy_rate: float) -> tuple[EnergyRate, bool]:
timestamp_starting_to_charge, timestamp_ending_to_charge, energy_bought
)

return next_price_minimum, minimum_has_to_be_rechecked
return next_price_minimum

def _charge_inverter(self, target_state_of_charge: int) -> None:
"""
Expand Down
10 changes: 4 additions & 6 deletions source/tibber_api_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ def __init__(self):
self.client = Client(transport=transport, fetch_schema_from_transport=True)
self.maximum_threshold = 0.03 # in €

def get_next_price_minimum(self, first_iteration: bool = False) -> tuple[EnergyRate, bool]:
def get_next_price_minimum(self, first_iteration: bool = False) -> 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.
Expand All @@ -34,7 +34,6 @@ def get_next_price_minimum(self, first_iteration: bool = False) -> tuple[EnergyR
TLDR: Introduce a maximum threshold to better identify real maximum energy rates, preventing minor fluctuations
from being misinterpreted as maxima.
Steps:
1. Fetches the upcoming energy prices from the API.
2. Extracts energy rates from the API response.
Expand All @@ -47,8 +46,6 @@ def get_next_price_minimum(self, first_iteration: bool = False) -> tuple[EnergyR
Returns:
EnergyRate: The next price 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()
Expand All @@ -72,9 +69,10 @@ def get_next_price_minimum(self, first_iteration: bool = False) -> tuple[EnergyR
energy_rates_between_first_and_second_maximum
)

minimum_has_to_be_rechecked = self._check_if_minimum_is_at_end_of_day(minimum_of_energy_rates)
if self._check_if_minimum_is_at_end_of_day(minimum_of_energy_rates):
minimum_of_energy_rates.is_minimum_that_has_to_be_rechecked = True

return minimum_of_energy_rates, minimum_has_to_be_rechecked
return minimum_of_energy_rates

@staticmethod
def _check_if_next_three_prices_are_greater_than_current_one(all_upcoming_energy_rates: list[EnergyRate]) -> bool:
Expand Down

0 comments on commit 1dfc428

Please sign in to comment.