Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pandas/core/array_algos/datetimelike_accumulations.py: 32%

1""" 

2datetimelke_accumulations.py is for accumulations of datetimelike extension arrays 

3""" 

4 

5from __future__ import annotations 

6 

7from typing import TYPE_CHECKING 

8 

9import numpy as np 

10 

11from pandas._libs import iNaT 

12from pandas._libs.tslibs import OutOfBoundsTimedelta 

13 

14from pandas.core.dtypes.missing import isna 

15 

16if TYPE_CHECKING: 

17 from collections.abc import Callable 

18 

19 

20def _check_cumsum_overflow( 

21 values: np.ndarray, result: np.ndarray, mask: np.ndarray 

22) -> None: 

23 """ 

24 Raise if a running total wrapped around or landed on the NaT sentinel. 

25 

26 Parameters 

27 ---------- 

28 values : np.ndarray[int64] 

29 The addends, with NA positions already zeroed out. 

30 result : np.ndarray[int64] 

31 The running totals, before NA positions are set to NaT. 

32 mask : np.ndarray[bool] 

33 Positions whose result will be NaT regardless, and so are exempt. 

34 """ 

35 # Whether each running total is greater than the one before it, taking the 

36 # total before the first entry to be zero. 

37 stepped_up = np.empty(result.shape, dtype=bool) 

38 np.greater(result[:1], 0, out=stepped_up[:1]) 

39 np.greater(result[1:], result[:-1], out=stepped_up[1:]) 

40 

41 # Absent a signed wrap, the total goes up exactly when the addend is 

42 # positive; a wrap flips the direction of the step. 

43 invalid = (values > 0) != stepped_up 

44 # A total of exactly int64.min does not wrap, but is indistinguishable 

45 # from NaT once stored. 

46 invalid |= result == iNaT 

47 invalid &= ~mask 

48 

49 if invalid.any(): 

50 raise OutOfBoundsTimedelta("overflow in timedelta operation") 

51 

52 

53def _cum_func( 

54 func: Callable[..., np.ndarray], 

55 values: np.ndarray, 

56 *, 

57 skipna: bool = True, 

58) -> np.ndarray: 

59 """ 

60 Accumulations for 1D datetimelike arrays. 

61 

62 Parameters 

63 ---------- 

64 func : np.cumsum, np.maximum.accumulate, np.minimum.accumulate 

65 values : np.ndarray 

66 Numpy array with the values (can be of any dtype that support the 

67 operation). Values is changed is modified inplace. 

68 skipna : bool, default True 

69 Whether to skip NA. 

70 """ 

71 try: 

72 fill_value = { 

73 np.maximum.accumulate: np.iinfo(np.int64).min, 

74 np.cumsum: 0, 

75 np.minimum.accumulate: np.iinfo(np.int64).max, 

76 }[func] 

77 except KeyError as err: 

78 raise ValueError( 

79 f"No accumulation for {func} implemented on BaseMaskedArray" 

80 ) from err 

81 

82 mask = isna(values) 

83 y = values.view("i8") 

84 y[mask] = fill_value 

85 

86 if not skipna: 

87 mask = np.maximum.accumulate(mask) 

88 

89 # GH 57956 

90 result = func(y, axis=0) 

91 if func is np.cumsum: 

92 # GH#66551: cummin/cummax cannot leave the range, cumsum can 

93 _check_cumsum_overflow(y, result, mask) 

94 result[mask] = iNaT 

95 

96 if values.dtype.kind in "mM": 

97 return result.view(values.dtype.base) 

98 return result 

99 

100 

101def cumsum(values: np.ndarray, *, skipna: bool = True) -> np.ndarray: 

102 return _cum_func(np.cumsum, values, skipna=skipna) 

103 

104 

105def cummin(values: np.ndarray, *, skipna: bool = True) -> np.ndarray: 

106 return _cum_func(np.minimum.accumulate, values, skipna=skipna) 

107 

108 

109def cummax(values: np.ndarray, *, skipna: bool = True) -> np.ndarray: 

110 return _cum_func(np.maximum.accumulate, values, skipna=skipna)