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