tapo-weather-control/scheduler.py at main · dimatteoluca/tapo-weather-control · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
import datetime, schedule, sys
from astral.sun import sun  # pip install astral
from astral.location import LocationInfo
try:
    from TapoWeatherControl import *
    from utils.log_setup import *
    from utils.time_functions import wait_until
    from utils.time_functions import wait_until_next_hour
except ImportError:         # handle the case when the direct import fails, likely when executing through an external file
    from .TapoWeatherControl import *
    from .utils.log_setup import *
    from .utils.time_functions import wait_until
    from .utils.time_functions import wait_until_next_hour

# Configuration values
first_hour = int(os.getenv("FIRST_HOUR"))

# Global variables
sunset_time = None

# Get the sunset time for the specified coordinates
def set_sunset_time(latitude, longitude):
    global sunset_time
    try:
        if not (-90 <= float(latitude) <= 90 and -180 <= float(longitude) <= 180):  # check the validity of coordinates
            raise ValueError("Invalid coordinates")
        city = LocationInfo(latitude, longitude)
        sun_info = sun(city.observer, date=datetime.date.today())
        sunset_time = sun_info['sunset']
    except ValueError as ve:
        logger_tw.error(f"Invalid coordinates: {str(ve)}")
        # Terminate execution in case of invalid coordinates
        sys.exit(1)
    except Exception as e:
        logger_tw.error(f"Error occurred while getting sunset time: {str(e)}")
        logger_tw.warning("Continuing execution with default sunset time.")
        sunset_time = datetime.datetime.now().replace(hour=18, minute=40, second=0)

# Get the range of hours in which the script needs to be executed
def get_target_range():
    global sunset_time
    last_hour = sunset_time.hour - 1
    current_hour = datetime.datetime.now().hour
    if current_hour > first_hour and current_hour <= last_hour:
        target_range = range(current_hour, last_hour)
    else:
        target_range = range(first_hour - 1, last_hour)
    return target_range

# Schedule for 1 minute before every hour in the target range
def schedule_action():
    try:
        target_range = get_target_range()
        for hour in target_range:
            schedule.every().day.at(f"{hour:02d}:59").do(start_control)
    except Exception as e:
        logger_tw.error(f"Error occurred while scheduling action: {str(e)}")

def reschedule_action():
    schedule.clear()
    schedule_action()

def main_loop():
    global sunset_time
    try:
        logger_tw.info("First hour is set on:    %s", datetime.datetime.strptime(str(first_hour), "%H").strftime("%H:%M"))
        set_sunset_time(latitude, longitude)
        logger_tw.info(f"Today's sunset time is:  {sunset_time.strftime('%H:%M')}")
        last_hour = sunset_time.hour - 1
        logger_tw.info("So today's last hour is: %s", datetime.datetime.strptime(str(last_hour), "%H").strftime("%H:%M"))

        current_hour = datetime.datetime.now().hour
        if current_hour <= last_hour:
            reschedule_action()
            if current_hour >= first_hour:
                start_control()
                wait_until_next_hour()
            else:
                wait_until([first_hour, 0, 0])  # wait until first_hour

        current_hour = datetime.datetime.now().hour
        while current_hour < last_hour:
            # Check if there are pending scheduled events and execute them
            schedule.run_pending()
            wait_until_next_hour()
            current_hour = datetime.datetime.now().hour

        wait_until([1, 0, 0])          # wait until 1AM

    except KeyboardInterrupt:
        logger_tw.info("Program interrupted by user.")
        sys.exit(0)
    except Exception as e:
        logger_tw.error(f"Error occurred in the main loop: {str(e)}")
        # Terminate execution in case of critical errors
        sys.exit(1)

if __name__ == "__main__":
    while True:
        main_loop()