Shift the datetime of segment If there are None (which indicates unbounded index) in the segment, this method will return None. Parameters ---------- seg : datetime segment step : int rolling step rtype : str
(self, seg: tuple, step: int, rtype=SHIFT_SD)
| 241 | return index + step |
| 242 | |
| 243 | def shift(self, seg: tuple, step: int, rtype=SHIFT_SD) -> tuple: |
| 244 | """ |
| 245 | Shift the datetime of segment |
| 246 | |
| 247 | If there are None (which indicates unbounded index) in the segment, this method will return None. |
| 248 | |
| 249 | Parameters |
| 250 | ---------- |
| 251 | seg : |
| 252 | datetime segment |
| 253 | step : int |
| 254 | rolling step |
| 255 | rtype : str |
| 256 | rolling type ("sliding" or "expanding") |
| 257 | |
| 258 | Returns |
| 259 | -------- |
| 260 | tuple: new segment |
| 261 | |
| 262 | Raises |
| 263 | ------ |
| 264 | KeyError: |
| 265 | shift will raise error if the index(both start and end) is out of self.cal |
| 266 | """ |
| 267 | if isinstance(seg, tuple): |
| 268 | start_idx, end_idx = self.align_idx(seg[0], tp_type="start"), self.align_idx(seg[1], tp_type="end") |
| 269 | if rtype == self.SHIFT_SD: |
| 270 | start_idx = self._add_step(start_idx, step) |
| 271 | end_idx = self._add_step(end_idx, step) |
| 272 | elif rtype == self.SHIFT_EX: |
| 273 | end_idx = self._add_step(end_idx, step) |
| 274 | else: |
| 275 | raise NotImplementedError(f"This type of input is not supported") |
| 276 | if start_idx is not None and start_idx > len(self.cals): |
| 277 | raise KeyError("The segment is out of valid calendar") |
| 278 | return self.get(start_idx), self.get(end_idx) |
| 279 | else: |
| 280 | raise NotImplementedError(f"This type of input is not supported") |
| 281 | |
| 282 | |
| 283 | def replace_task_handler_with_cache(task: dict, cache_dir: Union[str, Path] = ".") -> dict: |