Coverage for bzfs_main/period_anchors.py: 100%

1# Copyright 2024 Wolfgang Hoschek AT mac DOT com 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# 

15"""Utility that snaps datetimes to calendar periods. 

16 

17Anchors specify offsets within yearly, monthly and smaller cycles. These values are used by 

18``round_datetime_up_to_duration_multiple`` to snap datetimes to the next boundary. Keeping anchors in a dataclass simplifies 

19argument handling and makes the rounding logic reusable. 

20""" 

21 

22from __future__ import annotations 

23import argparse 

24import calendar 

25import dataclasses 

26from dataclasses import dataclass, field 

27from datetime import datetime, timedelta 

28 

29# constants: 

30METADATA_MONTH = {"min": 1, "max": 12, "help": "The month within a year"} 

31METADATA_WEEKDAY = {"min": 0, "max": 6, "help": "The weekday within a week: 0=Sunday, 1=Monday, ..., 6=Saturday"} 

32METADATA_DAY = {"min": 1, "max": 31, "help": "The day within a month"} 

33METADATA_HOUR = {"min": 0, "max": 23, "help": "The hour within a day"} 

34METADATA_MINUTE = {"min": 0, "max": 59, "help": "The minute within an hour"} 

35METADATA_SECOND = {"min": 0, "max": 59, "help": "The second within a minute"} 

36METADATA_MILLISECOND = {"min": 0, "max": 999, "help": "The millisecond within a second"} 

37METADATA_MICROSECOND = {"min": 0, "max": 999, "help": "The microsecond within a millisecond"} 

38 

39 

40@dataclass(frozen=True) 

41class PeriodAnchors: 

42 """Anchor offsets used to round datetimes up to periodic boundaries.""" 

43 

44 # The anchors for a given duration unit are computed as follows: 

45 # yearly: Anchor(dt) = latest T where T <= dt and T == Start of January 1 of dt + anchor.yearly_* vars 

46 yearly_year: int = field(default=2025, metadata={"min": 1, "max": 9999, "help": "The anchor year of multi-year periods"}) 

47 yearly_month: int = field(default=1, metadata=METADATA_MONTH) # 1 <= x <= 12 

48 yearly_monthday: int = field(default=1, metadata=METADATA_DAY) # 1 <= x <= 31 

49 yearly_hour: int = field(default=0, metadata=METADATA_HOUR) # 0 <= x <= 23 

50 yearly_minute: int = field(default=0, metadata=METADATA_MINUTE) # 0 <= x <= 59 

51 yearly_second: int = field(default=0, metadata=METADATA_SECOND) # 0 <= x <= 59 

52 

53 # monthly: Anchor(dt) = latest T where T <= dt && T == Start of first day of month of dt + anchor.monthly_* vars 

54 monthly_month: int = field(default=1, metadata={"min": 1, "max": 12, "help": "The anchor month of multi-month periods"}) 

55 monthly_monthday: int = field(default=1, metadata=METADATA_DAY) # 1 <= x <= 31 

56 monthly_hour: int = field(default=0, metadata=METADATA_HOUR) # 0 <= x <= 23 

57 monthly_minute: int = field(default=0, metadata=METADATA_MINUTE) # 0 <= x <= 59 

58 monthly_second: int = field(default=0, metadata=METADATA_SECOND) # 0 <= x <= 59 

59 

60 # weekly: Anchor(dt) = latest T where T <= dt && T == Latest midnight from Sunday to Monday of dt + anchor.weekly_* vars 

61 weekly_weekday: int = field(default=0, metadata=METADATA_WEEKDAY) # 0 <= x <= 7 

62 weekly_hour: int = field(default=0, metadata=METADATA_HOUR) # 0 <= x <= 23 

63 weekly_minute: int = field(default=0, metadata=METADATA_MINUTE) # 0 <= x <= 59 

64 weekly_second: int = field(default=0, metadata=METADATA_SECOND) # 0 <= x <= 59 

65 

66 # daily: Anchor(dt) = latest T where T <= dt && T == Latest midnight of dt + anchor.daily_* vars 

67 daily_hour: int = field(default=0, metadata=METADATA_HOUR) # 0 <= x <= 23 

68 daily_minute: int = field(default=0, metadata=METADATA_MINUTE) # 0 <= x <= 59 

69 daily_second: int = field(default=0, metadata=METADATA_SECOND) # 0 <= x <= 59 

70 

71 # hourly: Anchor(dt) = latest T where T <= dt && T == Latest midnight of dt + anchor.hourly_* vars 

72 hourly_minute: int = field(default=0, metadata=METADATA_MINUTE) # 0 <= x <= 59 

73 hourly_second: int = field(default=0, metadata=METADATA_SECOND) # 0 <= x <= 59 

74 

75 # minutely: Anchor(dt) = latest T where T <= dt && T == Latest midnight of dt + anchor.minutely_* vars 

76 minutely_second: int = field(default=0, metadata=METADATA_SECOND) # 0 <= x <= 59 

77 

78 # secondly: Anchor(dt) = latest T where T <= dt && T == Latest midnight of dt + anchor.secondly_* vars 

79 secondly_millisecond: int = field(default=0, metadata=METADATA_MILLISECOND) # 0 <= x <= 999 

80 

81 # secondly: Anchor(dt) = latest T where T <= dt && T == Latest midnight of dt + anchor.millisecondly_* vars 

82 millisecondly_microsecond: int = field(default=0, metadata=METADATA_MICROSECOND) # 0 <= x <= 999 

83 

84 @staticmethod 

85 def parse(args: argparse.Namespace) -> PeriodAnchors: 

86 """Creates a ``PeriodAnchors`` instance from parsed CLI arguments.""" 

87 kwargs: dict[str, int] = {f.name: getattr(args, f.name) for f in dataclasses.fields(PeriodAnchors)} 

88 return PeriodAnchors(**kwargs) 

89 

90 

91def round_datetime_up_to_duration_multiple( 

92 dt: datetime, duration_amount: int, duration_unit: str, anchors: PeriodAnchors 

93) -> datetime: 

94 """Given a timezone-aware datetime and a duration, returns a datetime (in the same timezone) that is greater than or 

95 equal to dt, and rounded up (ceiled) and snapped to an anchor plus a multiple of the duration. 

96 

97 The snapping is done relative to the anchors object and the rules defined therein. 

98 Supported units: "millisecondly", "secondly", "minutely", "hourly", "daily", "weekly", "monthly", "yearly". 

99 If dt is already exactly on a boundary (i.e. exactly on a multiple), it is returned unchanged. 

100 Examples: 

101 Default hourly anchor is midnight 

102 14:00:00, 1 hours --> 14:00:00 

103 14:05:01, 1 hours --> 15:00:00 

104 15:05:01, 1 hours --> 16:00:00 

105 16:05:01, 1 hours --> 17:00:00 

106 23:55:01, 1 hours --> 00:00:00 on the next day 

107 14:05:01, 2 hours --> 16:00:00 

108 15:00:00, 2 hours --> 16:00:00 

109 15:05:01, 2 hours --> 16:00:00 

110 16:00:00, 2 hours --> 16:00:00 

111 16:05:01, 2 hours --> 18:00:00 

112 23:55:01, 2 hours --> 00:00:00 on the next day 

113 """ 

114 

115 def add_months(dt: datetime, months: int) -> datetime: 

116 """Returns ``dt`` plus ``months`` with day clamped to month's end.""" 

117 total_month: int = dt.month - 1 + months 

118 new_year: int = dt.year + total_month // 12 

119 new_month: int = total_month % 12 + 1 

120 last_day: int = calendar.monthrange(new_year, new_month)[1] # last valid day of the current month 

121 return dt.replace(year=new_year, month=new_month, day=min(dt.day, last_day)) 

122 

123 def add_years(dt: datetime, years: int) -> datetime: 

124 """Returns ``dt`` plus ``years`` with day clamped to month's end.""" 

125 new_year: int = dt.year + years 

126 last_day: int = calendar.monthrange(new_year, dt.month)[1] # last valid day of the current month 

127 return dt.replace(year=new_year, day=min(dt.day, last_day)) 

128 

129 if duration_amount == 0: 

130 return dt 

131 

132 period: timedelta | None = None 

133 anchor: datetime 

134 daily_base: datetime 

135 last_day: int 

136 if duration_unit == "millisecondly": 

137 anchor = dt.replace(hour=0, minute=0, second=0, microsecond=anchors.millisecondly_microsecond) 

138 anchor = anchor if anchor <= dt else anchor - timedelta(milliseconds=1) 

139 period = timedelta(milliseconds=duration_amount) 

140 

141 elif duration_unit == "secondly": 

142 anchor = dt.replace(hour=0, minute=0, second=0, microsecond=anchors.secondly_millisecond * 1000) 

143 anchor = anchor if anchor <= dt else anchor - timedelta(seconds=1) 

144 period = timedelta(seconds=duration_amount) 

145 

146 elif duration_unit == "minutely": 

147 anchor = dt.replace(second=anchors.minutely_second, microsecond=0) 

148 anchor = anchor if anchor <= dt else anchor - timedelta(minutes=1) 

149 period = timedelta(minutes=duration_amount) 

150 

151 elif duration_unit == "hourly": 

152 daily_base = dt.replace(hour=0, minute=0, second=0, microsecond=0) 

153 anchor = daily_base + timedelta(minutes=anchors.hourly_minute, seconds=anchors.hourly_second) 

154 anchor = anchor if anchor <= dt else anchor - timedelta(days=1) 

155 period = timedelta(hours=duration_amount) 

156 

157 elif duration_unit == "daily": 

158 daily_base = dt.replace(hour=0, minute=0, second=0, microsecond=0) 

159 anchor = daily_base + timedelta(hours=anchors.daily_hour, minutes=anchors.daily_minute, seconds=anchors.daily_second) 

160 anchor = anchor if anchor <= dt else anchor - timedelta(days=1) 

161 period = timedelta(days=duration_amount) 

162 

163 elif duration_unit == "weekly": 

164 daily_base = dt.replace(hour=0, minute=0, second=0, microsecond=0) 

165 anchor = daily_base + timedelta( 

166 hours=anchors.weekly_hour, minutes=anchors.weekly_minute, seconds=anchors.weekly_second 

167 ) 

168 # Convert cron weekday (0=Sunday, 1=Monday, ..., 6=Saturday) to Python's weekday (0=Monday, ..., 6=Sunday) 

169 target_py_weekday: int = (anchors.weekly_weekday - 1) % 7 

170 diff_days: int = (anchor.weekday() - target_py_weekday) % 7 

171 anchor = anchor - timedelta(days=diff_days) 

172 anchor = anchor if anchor <= dt else anchor - timedelta(weeks=1) 

173 period = timedelta(weeks=duration_amount) 

174 

175 if period is not None: # "millisecondly", "secondly", "minutely", "hourly", "daily", "weekly" 

176 delta: timedelta = dt - anchor 

177 period_micros: int = (period.days * 86400 + period.seconds) * 1_000_000 + period.microseconds 

178 delta_micros: int = (delta.days * 86400 + delta.seconds) * 1_000_000 + delta.microseconds 

179 remainder: int = delta_micros % period_micros 

180 if remainder == 0: 

181 return dt 

182 return dt + timedelta(microseconds=period_micros - remainder) 

183 

184 elif duration_unit == "monthly": 

185 last_day = calendar.monthrange(dt.year, dt.month)[1] # last valid day of the current month 

186 anchor = dt.replace( # Compute the base anchor for the month ensuring the day is valid 

187 month=anchors.monthly_month, 

188 day=min(anchors.monthly_monthday, last_day), 

189 hour=anchors.monthly_hour, 

190 minute=anchors.monthly_minute, 

191 second=anchors.monthly_second, 

192 microsecond=0, 

193 ) 

194 if anchor > dt: 

195 anchor = add_months(anchor, -duration_amount) 

196 diff_months: int = (dt.year - anchor.year) * 12 + (dt.month - anchor.month) 

197 anchor_boundary: datetime = add_months(anchor, duration_amount * (diff_months // duration_amount)) 

198 if anchor_boundary < dt: 

199 anchor_boundary = add_months(anchor_boundary, duration_amount) 

200 return anchor_boundary 

201 

202 elif duration_unit == "yearly": 

203 # Calculate the start of the cycle period that `dt` falls into. 

204 year_offset: int = (dt.year - anchors.yearly_year) % duration_amount 

205 period_start_year: int = dt.year - year_offset 

206 last_day = calendar.monthrange(period_start_year, anchors.yearly_month)[1] # last valid day of the month 

207 anchor = dt.replace( 

208 year=period_start_year, 

209 month=anchors.yearly_month, 

210 day=min(anchors.yearly_monthday, last_day), 

211 hour=anchors.yearly_hour, 

212 minute=anchors.yearly_minute, 

213 second=anchors.yearly_second, 

214 microsecond=0, 

215 ) 

216 if anchor < dt: 

217 return add_years(anchor, duration_amount) 

218 return anchor 

219 

220 else: 

221 raise ValueError(f"Unsupported duration unit: {duration_unit}")