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_YEAR: Final = {"min": 1, "max": 9999, "help": None} 

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

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

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

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

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

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

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

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

51 

52 

53@dataclass(frozen=True) 

54@final 

55class PeriodAnchors: 

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

57 

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

59 # yearly: Count from anchor.yearly_* vars 

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

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

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

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

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

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

66 

67 # monthly: Count months from anchor.monthly_* vars 

68 monthly_year: int = field(default=2025, metadata=METADATA_YEAR) # 1 <= x <= 9999 

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

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

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

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

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

74 

75 # weekly: Count weeks from the first weekly_weekday on or after January 1 of weekly_year 

76 weekly_year: int = field(default=2025, metadata=METADATA_YEAR) # 1 <= x <= 9999 

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

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

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

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

81 

82 # daily: Count days from January 1 of daily_year 

83 daily_year: int = field(default=2025, metadata=METADATA_YEAR) # 1 <= x <= 9999 

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

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

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

87 

88 # hourly: Anchor(dt) = midnight of dt + anchor.hourly_* vars 

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

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

91 

92 # minutely: Anchor(dt) = midnight of dt + anchor.minutely_* vars 

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

94 

95 # secondly: Anchor(dt) = midnight of dt + anchor.secondly_* vars 

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

97 

98 # millisecondly: Anchor(dt) = midnight of dt + anchor.millisecondly_* vars 

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

100 

101 @classmethod 

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

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

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

105 return cls(**kwargs) 

106 

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

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

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

110 

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

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

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

114 Examples: 

115 Default hourly anchor is midnight 

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

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

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

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

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

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

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

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

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

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

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

127 """ 

128 

129 if duration_amount == 0: 

130 return dt 

131 

132 assert duration_amount > 0 

133 period: timedelta 

134 anchor: datetime 

135 if duration_unit == "millisecondly": 

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

137 period = timedelta(milliseconds=duration_amount) 

138 

139 elif duration_unit == "secondly": 

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

141 period = timedelta(seconds=duration_amount) 

142 

143 elif duration_unit == "minutely": 

144 anchor = dt.replace(hour=0, minute=0, second=self.minutely_second, microsecond=0) 

145 period = timedelta(minutes=duration_amount) 

146 

147 elif duration_unit == "hourly": 

148 anchor = dt.replace(hour=0, minute=self.hourly_minute, second=self.hourly_second, microsecond=0) 

149 period = timedelta(hours=duration_amount) 

150 

151 elif duration_unit == "daily": 

152 anchor = dt.replace( 

153 year=self.daily_year, 

154 month=1, 

155 day=1, 

156 hour=self.daily_hour, 

157 minute=self.daily_minute, 

158 second=self.daily_second, 

159 microsecond=0, 

160 ) 

161 period = timedelta(days=duration_amount) 

162 

163 elif duration_unit == "weekly": 

164 anchor = dt.replace( 

165 year=self.weekly_year, 

166 month=1, 

167 day=1, 

168 hour=self.weekly_hour, 

169 minute=self.weekly_minute, 

170 second=self.weekly_second, 

171 microsecond=0, 

172 ) 

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

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

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

176 anchor = anchor + timedelta(days=diff_days) 

177 period = timedelta(weeks=duration_amount) 

178 

179 elif duration_unit == "monthly": 

180 months_since_anchor: int = (dt.year - self.monthly_year) * 12 + dt.month - self.monthly_month 

181 offset_months: int = -months_since_anchor % duration_amount 

182 anchor = self._add_months(dt, offset_months) 

183 if anchor < dt: 

184 anchor = self._add_months(dt, offset_months + duration_amount) 

185 return anchor 

186 

187 elif duration_unit == "yearly": 

188 offset_years: int = (self.yearly_year - dt.year) % duration_amount 

189 anchor = self._add_years(dt, offset_years) 

190 if anchor < dt: 

191 anchor = self._add_years(dt, offset_years + duration_amount) 

192 return anchor 

193 

194 else: 

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

196 

197 # "millisecondly", "secondly", "minutely", "hourly", "daily", "weekly" 

198 delta: timedelta = dt - anchor 

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

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

201 remainder: int = delta_micros % period_micros 

202 if remainder == 0: 

203 return dt 

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

205 

206 def _add_months(self, dt: datetime, months: int) -> datetime: 

207 """Build a boundary offset from dt's month, clamping the configured day only against the target month's end.""" 

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

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

210 new_month: int = total_month % 12 + 1 

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

212 return dt.replace( 

213 year=new_year, 

214 month=new_month, 

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

216 hour=self.monthly_hour, 

217 minute=self.monthly_minute, 

218 second=self.monthly_second, 

219 microsecond=0, 

220 ) 

221 

222 def _add_years(self, dt: datetime, years: int) -> datetime: 

223 """Build a boundary offset from dt's year using yearly anchors, clamping the configured day to the target month's end.""" 

224 new_year: int = dt.year + years 

225 last_day: int = calendar.monthrange(new_year, self.yearly_month)[1] # last valid day of the target month 

226 return dt.replace( 

227 year=new_year, 

228 month=self.yearly_month, 

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

230 hour=self.yearly_hour, 

231 minute=self.yearly_minute, 

232 second=self.yearly_second, 

233 microsecond=0, 

234 )