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 ( 

23 annotations, 

24) 

25import argparse 

26import calendar 

27import dataclasses 

28from dataclasses import ( 

29 dataclass, 

30 field, 

31) 

32from datetime import ( 

33 datetime, 

34 timedelta, 

35) 

36from typing import ( 

37 Final, 

38 final, 

39) 

40 

41# constants: 

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

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

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

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

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

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

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

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

50 

51 

52@dataclass(frozen=True) 

53@final 

54class PeriodAnchors: 

55 """Anchor offsets used to round datetimes up to periodic boundaries; Immutable.""" 

56 

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

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

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

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

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

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

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

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

65 

66 # monthly: Anchor(dt) = latest T <= dt at phase month (monthly_month) + anchor.monthly_* vars (day clamped; multi-month) 

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

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

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

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

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

72 

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

74 weekly_weekday: int = field(default=0, metadata=METADATA_WEEKDAY) # 0 <= x <= 6 (0=Sunday, ..., 6=Saturday) 

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

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

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

78 

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

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

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

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

83 

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

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

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

87 

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

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

90 

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

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

93 

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

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

96 

97 @classmethod 

98 def parse(cls, args: argparse.Namespace) -> PeriodAnchors: 

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

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

101 return cls(**kwargs) 

102 

103 def round_datetime_up_to_duration_multiple(self, dt: datetime, duration_amount: int, duration_unit: str) -> datetime: 

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

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

106 

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

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

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

110 Examples: 

111 Default hourly anchor is midnight 

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

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

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

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

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

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

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

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

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

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

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

123 """ 

124 

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

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

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

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

129 new_month: int = total_month % 12 + 1 

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

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

132 

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

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

135 new_year: int = dt.year + years 

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

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

138 

139 if duration_amount == 0: 

140 return dt 

141 

142 period: timedelta | None = None 

143 anchor: datetime 

144 daily_base: datetime 

145 last_day: int 

146 if duration_unit == "millisecondly": 

147 anchor = dt.replace(hour=0, minute=0, second=0, microsecond=self.millisecondly_microsecond) 

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

149 period = timedelta(milliseconds=duration_amount) 

150 

151 elif duration_unit == "secondly": 

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

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

154 period = timedelta(seconds=duration_amount) 

155 

156 elif duration_unit == "minutely": 

157 anchor = dt.replace(second=self.minutely_second, microsecond=0) 

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

159 period = timedelta(minutes=duration_amount) 

160 

161 elif duration_unit == "hourly": 

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

163 anchor = daily_base + timedelta(minutes=self.hourly_minute, seconds=self.hourly_second) 

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

165 period = timedelta(hours=duration_amount) 

166 

167 elif duration_unit == "daily": 

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

169 anchor = daily_base + timedelta(hours=self.daily_hour, minutes=self.daily_minute, seconds=self.daily_second) 

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

171 period = timedelta(days=duration_amount) 

172 

173 elif duration_unit == "weekly": 

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

175 anchor = daily_base + timedelta(hours=self.weekly_hour, minutes=self.weekly_minute, seconds=self.weekly_second) 

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

177 target_py_weekday: int = (self.weekly_weekday - 1) % 7 

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

179 anchor = anchor - timedelta(days=diff_days) 

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

181 period = timedelta(weeks=duration_amount) 

182 

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

184 delta: timedelta = dt - anchor 

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

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

187 remainder: int = delta_micros % period_micros 

188 if remainder == 0: 

189 return dt 

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

191 

192 elif duration_unit == "monthly": 

193 last_day = calendar.monthrange(dt.year, self.monthly_month)[1] # last valid day of the anchor month 

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

195 month=self.monthly_month, 

196 day=min(self.monthly_monthday, last_day), 

197 hour=self.monthly_hour, 

198 minute=self.monthly_minute, 

199 second=self.monthly_second, 

200 microsecond=0, 

201 ) 

202 if anchor > dt: 

203 anchor = add_months(anchor, -duration_amount) 

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

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

206 if anchor_boundary < dt: 

207 anchor_boundary = add_months(anchor_boundary, duration_amount) 

208 return anchor_boundary 

209 

210 elif duration_unit == "yearly": 

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

212 year_offset: int = (dt.year - self.yearly_year) % duration_amount 

213 period_start_year: int = dt.year - year_offset 

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

215 anchor = dt.replace( 

216 year=period_start_year, 

217 month=self.yearly_month, 

218 day=min(self.yearly_monthday, last_day), 

219 hour=self.yearly_hour, 

220 minute=self.yearly_minute, 

221 second=self.yearly_second, 

222 microsecond=0, 

223 ) 

224 if anchor < dt: 

225 return add_years(anchor, duration_amount) 

226 return anchor 

227 

228 else: 

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