Implement TimeHandler.get_time() and TimeHandler.get_date_as_string()

Author: felixschndr

source/abscence_handler.py

@@ -40,7 +40,7 @@ def _check_for_current_absence(self, absence_input: str) -> bool:
             if timestamp.tzinfo is None:
                 raise ValueError(f'"{absence_start_raw}" has no timezone information')
 
-        if absence_start < datetime.now(tz=TimeHandler.get_timezone()) < absence_end:
+        if absence_start < TimeHandler.get_time() < absence_end:
             return True
 
         return False

source/deprecated_sun_forecast_handler.py

@@ -14,14 +14,6 @@ class DeprecatedSunForecastHandler(LoggerMixin):
     def __init__(self):
         super().__init__()
 
-    @staticmethod
-    def _get_date_as_string() -> str:
-        """
-        Returns:
-            str: The current date formatted as 'YYYY-MM-DD'.
-        """
-        return datetime.datetime.now().strftime("%Y-%m-%d")
-
     def get_expected_solar_output_of_today(self) -> EnergyAmount:
         """
         Gets the expected solar energy output for the current day.
@@ -55,7 +47,7 @@ def get_expected_solar_output_of_today(self) -> EnergyAmount:
             data = response.json()
             self.log.trace(f"Retrieved data: {data}")
 
-            total_solar_forecast += data["result"][self._get_date_as_string()]
+            total_solar_forecast += data["result"][TimeHandler.get_date_as_string()]
 
         return total_solar_forecast
 
@@ -130,7 +122,7 @@ def _get_sunset_and_sunrise_with_offset(self) -> tuple[datetime, datetime]:
         Returns:
             tuple[datetime, datetime]: A tuple containing the adjusted sunrise and sunset times.
         """
-        date = datetime.datetime.now()
+        date = TimeHandler.get_time()
         sun = SunTimes(
             float(EnvironmentVariableGetter.get("LOCATION_LONGITUDE")),
             float(EnvironmentVariableGetter.get("LOCATION_LATITUDE")),

source/inverter_charge_controller.py

@@ -54,9 +54,7 @@ def start(self) -> None:
                     next_price_minimum, minimum_has_to_be_rechecked = self._do_iteration()
 
                 if minimum_has_to_be_rechecked:
-                    time_to_sleep_to = datetime.now(tz=self.timezone).replace(
-                        hour=14, minute=0, second=0, microsecond=0
-                    )
+                    time_to_sleep_to = TimeHandler.get_time().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...")
                     pause.until(time_to_sleep_to)
                     self.log.info("Waking up since the the price minimum has to re-checked")
@@ -98,7 +96,7 @@ def _do_iteration(self) -> tuple[datetime, bool]:  # FIXME: Find better name
             "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)
+        timestamp_now = TimeHandler.get_time()
 
         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}")
@@ -202,12 +200,12 @@ def _do_iteration(self) -> tuple[datetime, bool]:  # FIXME: Find better name
             required_state_of_charge = max_target_soc
 
         energy_bought_before_charging = self.sems_portal_api_handler.get_energy_buy()
-        timestamp_starting_to_charge = datetime.now(tz=self.timezone)
+        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)
 
-        timestamp_ending_to_charge = datetime.now(tz=self.timezone)
+        timestamp_ending_to_charge = TimeHandler.get_time()
 
         duration_to_wait_for_semsportal_update = timedelta(minutes=10)
         self.log.info(

source/main.py

@@ -1,6 +1,6 @@
 import signal
 import sys
-from datetime import datetime, timedelta
+from datetime import timedelta
 from types import FrameType
 
 from deprecated_sun_forecast_handler import DeprecatedSunForecastHandler
@@ -14,7 +14,7 @@ def log_solar_forecast(log_as_review: bool = False) -> None:
     sun_forecast_handler = SunForecastHandler()
     deprecated_sun_forecast_handler = DeprecatedSunForecastHandler()
 
-    now = datetime.now(tz=TimeHandler.get_timezone())
+    now = TimeHandler.get_time()
     start = now.replace(hour=5, minute=0, second=0, microsecond=0)
     end = now.replace(hour=23, minute=0, second=0, microsecond=0)
     if log_as_review:

source/sems_portal_api_handler.py

@@ -93,7 +93,7 @@ def _retrieve_energy_consumption_data(self) -> dict:
             "id": EnvironmentVariableGetter.get("SEMSPORTAL_POWERSTATION_ID"),
             "range": 2,
             "chartIndexId": "8",
-            "date": datetime.now().strftime("%Y-%m-%d"),
+            "date": TimeHandler.get_date_as_string(),
         }
 
         response = requests.post(url, headers=headers, json=payload, timeout=20)

source/sun_forecast_handler.py

@@ -40,7 +40,7 @@ def _get_debug_solar_output(self) -> EnergyAmount:
     def get_solar_output_in_timeframe(self, timestamp_start: datetime, timestamp_end: datetime) -> EnergyAmount:
         solar_data = []
 
-        now = datetime.now(tz=(TimeHandler.get_timezone())).replace(second=0, microsecond=0) - timedelta(
+        now = TimeHandler.get_time().replace(second=0, microsecond=0) - timedelta(
             seconds=1
         )  # Account for execution times of the program
         if timestamp_start >= now or timestamp_end >= now:
@@ -78,7 +78,7 @@ def get_solar_output_in_timeframe(self, timestamp_start: datetime, timestamp_end
 
 if __name__ == "__main__":
     new_solar_forecast_handler = SunForecastHandler()
-    start = datetime.now(tz=TimeHandler.get_timezone()).replace(hour=0, minute=0, second=0, microsecond=0)
+    start = TimeHandler.get_time().replace(hour=0, minute=0, second=0, microsecond=0)
     end = start + timedelta(days=1)
     # end = start + timedelta(hours=10)
     # start = datetime.now(tz=TimeHandler.get_timezone())

source/tibber_api_handler.py

@@ -169,7 +169,7 @@ def _remove_energy_rates_from_the_past(self, all_energy_rates: list[EnergyRate])
         Returns:
             A list of EnergyRate objects that have timestamps in the future relative to the beginning of the current hour.
         """
-        current_time = datetime.now(tz=all_energy_rates[0].timestamp.tzinfo)
+        current_time = TimeHandler.get_time()
         beginning_of_current_hour = current_time.replace(minute=0, second=0, microsecond=0)
 
         upcoming_energy_rates = [
@@ -285,17 +285,15 @@ def _check_if_minimum_is_at_end_of_day_and_energy_rates_of_tomorrow_are_unavaila
         """
 
         # We use 00:01 instead of 00:00 since the software runs just a few (milli)seconds after the start of the hour
-        end_of_day = (datetime.now(tz=TimeHandler.get_timezone()) + timedelta(days=1)).replace(
-            hour=0, minute=1, second=0, microsecond=0
-        )
+        end_of_day = (TimeHandler.get_time() + 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 = three_hours_before_end_of_day <= price_minimum.timestamp <= 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}"
         )
 
-        today = datetime.now().date()
+        today = TimeHandler.get_time().date()
         are_tomorrows_rates_unavailable = not all(rate.timestamp.date() == today for rate in upcoming_energy_rates)
         self.log.trace(f"The price rates for tomorrow are unavailable: {are_tomorrows_rates_unavailable}")
 

source/time_handler.py

@@ -1,4 +1,4 @@
-from datetime import datetime, time, timedelta
+from datetime import date, datetime, time, timedelta
 
 from dateutil.tz import tz, tzfile
 
@@ -95,3 +95,15 @@ def calculate_day_night_duration(
     @staticmethod
     def get_timezone() -> tzfile:
         return tz.gettz()
+
+    @staticmethod
+    def get_time() -> datetime:
+        return datetime.now(tz=(TimeHandler.get_timezone()))
+
+    @staticmethod
+    def get_date() -> date:
+        return TimeHandler.get_time().date()
+
+    @staticmethod
+    def get_date_as_string() -> str:
+        return TimeHandler.get_date().strftime("%Y-%m-%d")