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

1""" 

2masked_accumulations.py is for accumulation algorithms using a mask-based approach 

3for missing values. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import TYPE_CHECKING 

9 

10import numpy as np 

11 

12if TYPE_CHECKING: 

13 from collections.abc import Callable 

14 

15 from pandas._typing import npt 

16 

17 

18def _cum_func( 

19 func: Callable, 

20 values: np.ndarray, 

21 mask: npt.NDArray[np.bool_], 

22 *, 

23 skipna: bool = True, 

24) -> tuple[np.ndarray, npt.NDArray[np.bool_]]: 

25 """ 

26 Accumulations for 1D masked array. 

27 

28 We will modify values in place to replace NAs with the appropriate fill value. 

29 

30 Parameters 

31 ---------- 

32 func : np.cumsum, np.cumprod, np.maximum.accumulate, np.minimum.accumulate 

33 values : np.ndarray 

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

35 operation). 

36 mask : np.ndarray 

37 Boolean numpy array (True values indicate missing values). 

38 skipna : bool, default True 

39 Whether to skip NA. 

40 """ 

41 dtype_info: np.iinfo | np.finfo 

42 if values.dtype.kind == "f": 

43 dtype_info = np.finfo(values.dtype.type) 

44 elif values.dtype.kind in "iu": 

45 dtype_info = np.iinfo(values.dtype.type) 

46 elif values.dtype.kind == "b": 

47 # Max value of bool is 1, but since we are setting into a boolean 

48 # array, 255 is fine as well. Min value has to be 0 when setting 

49 # into the boolean array. 

50 dtype_info = np.iinfo(np.uint8) 

51 else: 

52 raise NotImplementedError( 

53 f"No masked accumulation defined for dtype {values.dtype.type}" 

54 ) 

55 try: 

56 fill_value = { 

57 np.cumprod: 1, 

58 np.maximum.accumulate: dtype_info.min, 

59 np.cumsum: 0, 

60 np.minimum.accumulate: dtype_info.max, 

61 }[func] 

62 except KeyError as err: 

63 raise NotImplementedError( 

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

65 ) from err 

66 

67 values[mask] = fill_value 

68 

69 if not skipna: 

70 mask = np.maximum.accumulate(mask) 

71 

72 values = func(values) 

73 return values, mask 

74 

75 

76def cumsum( 

77 values: np.ndarray, mask: npt.NDArray[np.bool_], *, skipna: bool = True 

78) -> tuple[np.ndarray, npt.NDArray[np.bool_]]: 

79 return _cum_func(np.cumsum, values, mask, skipna=skipna) 

80 

81 

82def cumprod( 

83 values: np.ndarray, mask: npt.NDArray[np.bool_], *, skipna: bool = True 

84) -> tuple[np.ndarray, npt.NDArray[np.bool_]]: 

85 return _cum_func(np.cumprod, values, mask, skipna=skipna) 

86 

87 

88def cummin( 

89 values: np.ndarray, mask: npt.NDArray[np.bool_], *, skipna: bool = True 

90) -> tuple[np.ndarray, npt.NDArray[np.bool_]]: 

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

92 

93 

94def cummax( 

95 values: np.ndarray, mask: npt.NDArray[np.bool_], *, skipna: bool = True 

96) -> tuple[np.ndarray, npt.NDArray[np.bool_]]: 

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