Coverage for rivapy/tools/scheduler.py: 94%

1from typing import Union, Set, List 

2import numpy as np 

3import pandas as pd 

4import datetime as dt 

5import rivapy.tools.interfaces as interfaces 

6from rivapy.tools.datetime_grid import DateTimeGrid 

7from rivapy.tools.enums import EnergyTimeGridStructure as ets 

8from abc import abstractmethod 

9 

10 

11class SimpleSchedule(interfaces.FactoryObject): 

12 

13 def __init_subclass__(cls, **kwargs): 

14 super().__init_subclass__(**kwargs) 

15 if not hasattr(cls, "_name"): 

16 raise TypeError(f"Class {cls.__name__} must define a class attribute '_name' from the 'EnergyTimeGridStructure' Enum.") 

17 

18 def __init__( 

19 self, 

20 start: dt.datetime, 

21 end: dt.datetime, 

22 freq: str = "h", 

23 weekdays: Set[int] = None, 

24 hours: Set[int] = None, 

25 ignore_hours_for_weekdays: Set[int] = None, 

26 tz: str = None, 

27 ): 

28 """Simple schedule of fixed datetime points. 

29 

30 Args: 

31 start (dt.datetime): Start of schedule (including this timepoint). 

32 end (dt.datetime): End of schedule (excluding this timepoint). 

33 freq (str, optional): Frequency of timepoints. Defaults to 'h'. See documentation for pandas.date_range for further details on freq. 

34 weekdays (Set[int], optional): List of integers representing the weekdays where the schedule is defined. 

35 Integers according to datetime weekdays (0->Monay, 1->Tuesday,...,6->Sunday). 

36 If None, all weekdays are used. Defaults to None. 

37 hours (Set[int], optional): List of hours where schedule is defined. If None, all hours are included. Defaults to None. 

38 ignor_hours_for_weekdays (Set[int], optional): List of days for which the hours setting is ignored and each hour is considered where the schedule is defined. Defaults to None. 

39 tz (str or tzinfo): Time zone name for returning localized datetime points, for example ‘Asia/Hong_Kong’. 

40 By default, the resulting datetime points are timezone-naive. See documentation for pandas.date_range for further details on tz. 

41 Examples: 

42 

43 .. highlight:: python 

44 .. code-block:: python 

45 

46 >>> simple_schedule = SimpleSchedule(dt.datetime(2023,1,1), dt.datetime(2023,1,1,4,0,0), freq='h') 

47 >>> simple_schedule.get_schedule() 

48 [datetime(2023,1,1,0,0,0), datetime(2023,1,1,1,0,0), datetime(2023,1,1,2,0,0), datetime(2023,1,1,3,0,0)] 

49 

50 # We include only hours 2 and 3 into schedule 

51 >>> simple_schedule = SimpleSchedule(dt.datetime(2023,1,1), dt.datetime(2023,1,1,4,0,0), freq='h', hours=[2,3]) 

52 >>> simple_schedule.get_schedule() 

53 [datetime.datetime(2023, 1, 1, 2, 0), datetime.datetime(2023, 1, 1, 3, 0)] 

54 

55 # We restrict further to only mondays as weekdays included 

56 >>> simple_schedule = SimpleSchedule(dt.datetime(2023,1,1), dt.datetime(2023,1,2,4,0,0), freq='h', hours=[2,3], weekdays=[0]) 

57 >>> simple_schedule.get_schedule() 

58 [datetime.datetime(2023, 1, 2, 2, 0), datetime.datetime(2023, 1, 2, 3, 0)] 

59 """ 

60 self.start = start 

61 self.end = end 

62 self.freq = freq 

63 self.weekdays = weekdays 

64 self.hours = hours 

65 self.tz = tz 

66 self._df = None 

67 self.ignore_hours_for_weekdays = ignore_hours_for_weekdays 

68 

69 def _to_dict(self) -> dict: 

70 return { 

71 "start": self.start, 

72 "end": self.end, 

73 "freq": self.freq, 

74 "weekdays": self.weekdays, 

75 "hours": self.hours, 

76 "tz": self.tz, 

77 } 

78 

79 def get_schedule(self, refdate: dt.datetime = None) -> np.ndarray: 

80 """Return vector of datetime values belonging to the schedule. 

81 

82 Args: 

83 refdate (dt.datetime): All schedule dates are ignored before this reference date. If None, all schedule dates are returned. Defaults to None. 

84 

85 Returns: 

86 np.ndarray: Vector of all datetimepoints of the schedule. 

87 """ 

88 d_ = pd.date_range(self.start, self.end, freq=self.freq, tz=self.tz, inclusive="left").to_pydatetime() 

89 if self.weekdays is not None: 

90 d_ = [d for d in d_ if d.weekday() in self.weekdays] 

91 if self.hours is not None: 

92 if self.ignore_hours_for_weekdays is not None: 

93 d_ = [d for d in d_ if (d.hour in self.hours) or (d.weekday() in self.ignore_hours_for_weekdays)] 

94 else: 

95 d_ = [d for d in d_ if d.hour in self.hours] 

96 if refdate is not None: 

97 d_ = [d for d in d_ if d >= refdate] 

98 return d_ 

99 

100 def get_df(self) -> pd.DataFrame: 

101 if self._df is None: 

102 self._df = pd.DataFrame( 

103 {"dates": pd.date_range(self.start, self.end, freq=self.freq, tz=self.tz, inclusive="left").to_pydatetime()} 

104 ).reset_index() 

105 return self._df 

106 

107 def applies(self, dates: DateTimeGrid, index: bool) -> List[Union[bool, int]]: 

108 dates.dates 

109 

110 def get_params(self) -> dict: 

111 """Return all params as json serializable dictionary. 

112 

113 Returns: 

114 dict: Dictionary of all parameters. 

115 """ 

116 return { 

117 "start": self.start, 

118 "end": self.end, 

119 "freq": self.freq, 

120 "weekdays": self.weekdays, 

121 "hours": self.hours, 

122 "ignore_hours_for_weekdays": self.ignore_hours_for_weekdays, 

123 "tz": self.tz, 

124 } 

125 

126 @staticmethod 

127 def _create_sample(n_samples: int, seed: int = None, ref_date=None): 

128 if ref_date is None: 

129 ref_date = dt.datetime(1980, 1, 1) 

130 if seed is not None: 

131 np.random.seed(seed) 

132 result = [] 

133 for i in range(n_samples): 

134 start = ref_date + dt.timedelta(days=np.random.randint(0, 100)) 

135 end = start + +dt.timedelta(days=np.random.randint(5, 365)) 

136 result.append(SimpleSchedule(start=start, end=end)) 

137 return result 

138 

139 

140class BaseSchedule(SimpleSchedule): 

141 _name = ets.BASE 

142 

143 def __init__(self, start: dt.datetime, end: dt.datetime, tz: str = None): 

144 """Scheduler, which returns the base time grid between the start and end date times. 

145 

146 Args: 

147 start (dt.datetime): Start of schedule (including this timepoint). 

148 end (dt.datetime): End of schedule (excluding this timepoint). 

149 tz (str or tzinfo): Time zone name for returning localized datetime points, for example ‘Asia/Hong_Kong’. 

150 By default, the resulting datetime points are timezone-naive. See documentation for pandas.date_range for further details on tz. 

151 Examples: 

152 

153 .. highlight:: python 

154 .. code-block:: python 

155 

156 >>> base_schedule = BaseSchedule(dt.datetime(2023,1,5), dt.datetime(2023,1,6)) 

157 >>> base_schedule.get_schedule() 

158 

159 """ 

160 super().__init__( 

161 start=start, 

162 end=end, 

163 freq="h", 

164 hours=None, 

165 weekdays=None, 

166 tz=tz, 

167 ) 

168 

169 

170class PeakSchedule(SimpleSchedule): 

171 _name = ets.PEAK 

172 

173 def __init__(self, start: dt.datetime, end: dt.datetime, tz: str = None): 

174 """Scheduler, which returns the peak time grid between the start and end date times. 

175 

176 Args: 

177 start (dt.datetime): Start of schedule (including this timepoint). 

178 end (dt.datetime): End of schedule (excluding this timepoint). 

179 tz (str or tzinfo): Time zone name for returning localized datetime points, for example ‘Asia/Hong_Kong’. 

180 By default, the resulting datetime points are timezone-naive. See documentation for pandas.date_range for further details on tz. 

181 Examples: 

182 

183 .. highlight:: python 

184 .. code-block:: python 

185 

186 >>> peak_schedule = PeakSchedule(dt.datetime(2023,1,5), dt.datetime(2023,1,6)) 

187 >>> peak_schedule.get_schedule() 

188 [datetime.datetime(2023, 1, 5, 8, 0), 

189 datetime.datetime(2023, 1, 5, 9, 0), 

190 datetime.datetime(2023, 1, 5, 10, 0), 

191 datetime.datetime(2023, 1, 5, 11, 0), 

192 datetime.datetime(2023, 1, 5, 12, 0), 

193 datetime.datetime(2023, 1, 5, 13, 0), 

194 datetime.datetime(2023, 1, 5, 14, 0), 

195 datetime.datetime(2023, 1, 5, 15, 0), 

196 datetime.datetime(2023, 1, 5, 16, 0), 

197 datetime.datetime(2023, 1, 5, 17, 0), 

198 datetime.datetime(2023, 1, 5, 18, 0), 

199 datetime.datetime(2023, 1, 5, 19, 0)] 

200 """ 

201 super().__init__( 

202 start=start, 

203 end=end, 

204 freq="h", 

205 hours=[8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19], 

206 weekdays=[0, 1, 2, 3, 4], 

207 tz=tz, 

208 ) 

209 

210 

211class OffPeakSchedule(SimpleSchedule): 

212 _name = ets.OFFPEAK 

213 

214 def __init__(self, start: dt.datetime, end: dt.datetime, tz: str = None): 

215 """Scheduler, which returns the offpeak time grid between the start and end date times. 

216 

217 Args: 

218 start (dt.datetime): Start of schedule (including this timepoint). 

219 end (dt.datetime): End of schedule (excluding this timepoint). 

220 tz (str or tzinfo): Time zone name for returning localized datetime points, for example ‘Asia/Hong_Kong’. 

221 By default, the resulting datetime points are timezone-naive. See documentation for pandas.date_range for further details on tz. 

222 Examples: 

223 

224 .. highlight:: python 

225 .. code-block:: python 

226 

227 >>> offpeak_schedule = OffPeakSchedule(dt.datetime(2023,1,5), dt.datetime(2023,1,6)) 

228 >>> offpeak_schedule.get_schedule() 

229 [datetime.datetime(2023, 1, 5, 0, 0), 

230 datetime.datetime(2023, 1, 5, 1, 0), 

231 datetime.datetime(2023, 1, 5, 2, 0), 

232 datetime.datetime(2023, 1, 5, 3, 0), 

233 datetime.datetime(2023, 1, 5, 4, 0), 

234 datetime.datetime(2023, 1, 5, 5, 0), 

235 datetime.datetime(2023, 1, 5, 6, 0), 

236 datetime.datetime(2023, 1, 5, 7, 0), 

237 datetime.datetime(2023, 1, 5, 20, 0), 

238 datetime.datetime(2023, 1, 5, 21, 0), 

239 datetime.datetime(2023, 1, 5, 22, 0), 

240 datetime.datetime(2023, 1, 5, 23, 0)] 

241 """ 

242 super().__init__( 

243 start=start, 

244 end=end, 

245 freq="h", 

246 hours=[0, 1, 2, 3, 4, 5, 6, 7, 20, 21, 22, 23], 

247 ignore_hours_for_weekdays=[5, 6], 

248 tz=tz, 

249 ) 

250 

251 

252class GasSchedule(SimpleSchedule): 

253 _name = ets.BASE 

254 

255 def __init__(self, start: dt.datetime, end: dt.datetime, tz: str = None): 

256 """Scheduler, which returns the gas day time grid (from 6 am to 6 am) between the start and end date times. 

257 

258 Args: 

259 start (dt.datetime): Start of schedule (including this timepoint). 

260 end (dt.datetime): End of schedule (excluding this timepoint). 

261 tz (str or tzinfo): Time zone name for returning localized datetime points, for example ‘Asia/Hong_Kong’. 

262 By default, the resulting datetime points are timezone-naive. See documentation for pandas.date_range for further details on tz. 

263 Examples: 

264 

265 .. highlight:: python 

266 .. code-block:: python 

267 

268 >>> gas_schedule = GasSchedule(dt.datetime(2023,1,5), dt.datetime(2023,1,7)) 

269 >>> gas_schedule.get_schedule() 

270 [datetime.datetime(2023, 1, 5, 6, 0), datetime.datetime(2023, 1, 6, 6, 0)] 

271 """ 

272 super().__init__( 

273 start=start, 

274 end=end, 

275 freq="h", 

276 hours=[6], 

277 weekdays=[0, 1, 2, 3, 4, 5, 6], 

278 tz=tz, 

279 )