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) 

39 

40# constants: 

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

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

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

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

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

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

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

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

49 

50 

51@dataclass(frozen=True) 

52class PeriodAnchors: 

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

54 

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

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

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

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

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

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

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

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

63 

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

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

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

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

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

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

70 

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

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

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

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

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

76 

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

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

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

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

81 

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

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

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

85 

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

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

88 

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

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

91 

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

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

94 

95 @classmethod 

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

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

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

99 return cls(**kwargs) 

100 

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

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

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

104 

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

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

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

108 Examples: 

109 Default hourly anchor is midnight 

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

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

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

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

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

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

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

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

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

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

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

121 """ 

122 

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

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

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

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

127 new_month: int = total_month % 12 + 1 

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

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

130 

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

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

133 new_year: int = dt.year + years 

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

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

136 

137 if duration_amount == 0: 

138 return dt 

139 

140 period: timedelta | None = None 

141 anchor: datetime 

142 daily_base: datetime 

143 last_day: int 

144 if duration_unit == "millisecondly": 

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

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

147 period = timedelta(milliseconds=duration_amount) 

148 

149 elif duration_unit == "secondly": 

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

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

152 period = timedelta(seconds=duration_amount) 

153 

154 elif duration_unit == "minutely": 

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

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

157 period = timedelta(minutes=duration_amount) 

158 

159 elif duration_unit == "hourly": 

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

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

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

163 period = timedelta(hours=duration_amount) 

164 

165 elif duration_unit == "daily": 

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

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

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

169 period = timedelta(days=duration_amount) 

170 

171 elif duration_unit == "weekly": 

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

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

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

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

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

177 anchor = anchor - timedelta(days=diff_days) 

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

179 period = timedelta(weeks=duration_amount) 

180 

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

182 delta: timedelta = dt - anchor 

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

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

185 remainder: int = delta_micros % period_micros 

186 if remainder == 0: 

187 return dt 

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

189 

190 elif duration_unit == "monthly": 

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

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

193 month=self.monthly_month, 

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

195 hour=self.monthly_hour, 

196 minute=self.monthly_minute, 

197 second=self.monthly_second, 

198 microsecond=0, 

199 ) 

200 if anchor > dt: 

201 anchor = add_months(anchor, -duration_amount) 

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

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

204 if anchor_boundary < dt: 

205 anchor_boundary = add_months(anchor_boundary, duration_amount) 

206 return anchor_boundary 

207 

208 elif duration_unit == "yearly": 

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

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

211 period_start_year: int = dt.year - year_offset 

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

213 anchor = dt.replace( 

214 year=period_start_year, 

215 month=self.yearly_month, 

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

217 hour=self.yearly_hour, 

218 minute=self.yearly_minute, 

219 second=self.yearly_second, 

220 microsecond=0, 

221 ) 

222 if anchor < dt: 

223 return add_years(anchor, duration_amount) 

224 return anchor 

225 

226 else: 

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