Coverage for scheduler/util.py: 100%
43 statements
« prev ^ index » next coverage.py v7.6.10, created at 2025-01-21 13:55 +0000
1"""
2Collection of datetime and trigger related utility functions.
4Author: Jendrik A. Potyka, Fabian A. Preiss
5"""
6from __future__ import annotations
8import datetime as dt
9from typing import Optional
11from scheduler.base.definition import JobType
12from scheduler.error import SchedulerError
13from scheduler.trigger.core import Weekday
16def days_to_weekday(wkdy_src: int, wkdy_dest: int) -> int:
17 """
18 Calculate the days to a specific destination weekday.
20 Notes
21 -----
22 Weekday enumeration based on
23 the `datetime` standard library.
25 Parameters
26 ----------
27 wkdy_src : int
28 Source :class:`~scheduler.util.Weekday` integer representation.
29 wkdy_dest : int
30 Destination :class:`~scheduler.util.Weekday` integer representation.
32 Returns
33 -------
34 int
35 Days to the destination :class:`~scheduler.util.Weekday`.
36 """
37 if not (0 <= wkdy_src <= 6 and 0 <= wkdy_dest <= 6):
38 raise SchedulerError("Weekday enumeration interval: [0,6] <=> [Monday, Sunday]")
40 return (wkdy_dest - wkdy_src - 1) % 7 + 1
43def next_daily_occurrence(now: dt.datetime, target_time: dt.time) -> dt.datetime:
44 """
45 Estimate the next daily occurrence of a given time.
47 .. warning:: Both arguments are expected to have the same tzinfo, no internal checks.
49 Parameters
50 ----------
51 now : datetime.datetime
52 `datetime.datetime` object of today
53 target_time : datetime.time
54 Desired `datetime.time`.
56 Returns
57 -------
58 datetime.datetime
59 Next `datetime.datetime` object with the desired time.
60 """
61 target = now.replace(
62 hour=target_time.hour,
63 minute=target_time.minute,
64 second=target_time.second,
65 microsecond=target_time.microsecond,
66 )
67 if (target - now).total_seconds() <= 0:
68 target = target + dt.timedelta(days=1)
69 return target
72def next_hourly_occurrence(now: dt.datetime, target_time: dt.time) -> dt.datetime:
73 """
74 Estimate the next hourly occurrence of a given time.
76 .. warning:: Both arguments are expected to have the same tzinfo, no internal checks.
78 Parameters
79 ----------
80 now : datetime.datetime
81 `datetime.datetime` object of today
82 target_time : datetime.time
83 Desired `datetime.time`.
85 Returns
86 -------
87 datetime.datetime
88 Next `datetime.datetime` object with the desired time.
89 """
90 target = now.replace(
91 minute=target_time.minute,
92 second=target_time.second,
93 microsecond=target_time.microsecond,
94 )
95 if (target - now).total_seconds() <= 0:
96 target = target + dt.timedelta(hours=1)
97 return target
100def next_minutely_occurrence(now: dt.datetime, target_time: dt.time) -> dt.datetime:
101 """
102 Estimate the next weekly occurrence of a given time.
104 .. warning:: Both arguments are expected to have the same tzinfo, no internal checks.
106 Parameters
107 ----------
108 now : datetime.datetime
109 `datetime.datetime` object of today
110 target_time : datetime.time
111 Desired `datetime.time`.
113 Returns
114 -------
115 datetime.datetime
116 Next `datetime.datetime` object with the desired time.
117 """
118 target = now.replace(
119 second=target_time.second,
120 microsecond=target_time.microsecond,
121 )
122 if (target - now).total_seconds() <= 0:
123 return target + dt.timedelta(minutes=1)
124 return target
127def next_weekday_time_occurrence(
128 now: dt.datetime, weekday: Weekday, target_time: dt.time
129) -> dt.datetime:
130 """
131 Estimate the next occurrence of a given weekday and time.
133 .. warning:: Arguments `now` and `target_time` are expected to have the same tzinfo,
134 no internal checks.
136 Parameters
137 ----------
138 now : datetime.datetime
139 `datetime.datetime` object of today
140 weekday : Weekday
141 Desired :class:`~scheduler.util.Weekday`.
142 target_time : datetime.time
143 Desired `datetime.time`.
145 Returns
146 -------
147 datetime.datetime
148 Next `datetime.datetime` object with the desired weekday and time.
149 """
150 days = days_to_weekday(now.weekday(), weekday.value)
151 if days == 7:
152 candidate = next_daily_occurrence(now, target_time)
153 if candidate.date() == now.date():
154 return candidate
156 delta = dt.timedelta(days=days)
157 target = now.replace(
158 hour=target_time.hour,
159 minute=target_time.minute,
160 second=target_time.second,
161 microsecond=target_time.microsecond,
162 )
163 return target + delta
166JOB_NEXT_DAYLIKE_MAPPING = {
167 JobType.MINUTELY: next_minutely_occurrence,
168 JobType.HOURLY: next_hourly_occurrence,
169 JobType.DAILY: next_daily_occurrence,
170}
173def are_times_unique(
174 timelist: list[dt.time],
175) -> bool:
176 r"""
177 Check if list contains distinct `datetime.time`\ s.
179 Parameters
180 ----------
181 timelist : list[datetime.time]
182 List of time objects.
184 Returns
185 -------
186 boolean
187 ``True`` if list entries are not equivalent with tzinfo offset.
188 """
189 ref = dt.datetime(year=1970, month=1, day=1)
190 collection = {
191 ref.replace(
192 hour=time.hour,
193 minute=time.minute,
194 second=time.second,
195 microsecond=time.microsecond,
196 )
197 + (time.utcoffset() or dt.timedelta())
198 for time in timelist
199 }
200 return len(collection) == len(timelist)
203def are_weekday_times_unique(weekday_list: list[Weekday], tzinfo: Optional[dt.tzinfo]) -> bool:
204 """
205 Check if list contains distinct weekday times.
207 .. warning:: Both arguments are expected to be either timezone aware or not
208 - no internal checks.
210 Parameters
211 ----------
212 weekday_list : list[Weekday]
213 List of weekday objects.
215 Returns
216 -------
217 boolean
218 ``True`` if list entries are not equivalent with timezone offset.
219 """
220 ref = dt.datetime(year=1970, month=1, day=1, tzinfo=tzinfo)
221 collection = {
222 next_weekday_time_occurrence(ref.astimezone(day.time.tzinfo), day, day.time)
223 for day in weekday_list
224 }
225 return len(collection) == len(weekday_list)