Coverage for src/sankey_cashflow/utils.py: 89%

1import logging 

2import re 

3from os import path 

4from typing import Optional 

5 

6import pandas as pd 

7from numpy import float64, isnan 

8from pandas._libs.tslibs import nattype 

9 

10# Shared package-level logger. Submodules use `logging.getLogger(__name__)` and propagate up to 

11# this handler rather than each attaching their own - AppSettings' --verbose handling 

12# (logger.setLevel(...)) relies on there being exactly one logger/handler pair for the whole 

13# package. 

14logger = logging.getLogger("sankey_cashflow") 

15_console_handler = logging.StreamHandler() 

16# TODO: override this with params (note: root logger level will also need to be changed) 

17_console_handler.setLevel(logging.WARNING) 

18_console_handler.name = "console" 

19logger.addHandler(_console_handler) 

20 

21 

22def is_null(obj: any) -> bool: 

23 # Just using numpy.isnan() will throw errors for types that cannot be coerced to float64. 

24 # Could also use a try...catch 

25 # use as a general purpose null/none/NaN catch 

26 if obj is None: 

27 return True 

28 if type(obj) is float64 and isnan(obj): 

29 return True 

30 if type(obj) is nattype.NaTType: 

31 return True 

32 obj_as_str = None 

33 try: 

34 obj_as_str = str(obj) 

35 except Exception: 

36 pass 

37 if obj_as_str in ["None", "none", "NaN", "nan", "Null", "null"]: 

38 return True 

39 return False 

40 

41 

42def is_empty(obj: any, nonzero: Optional[bool] = False) -> bool: 

43 # Check for null, nan, none, etc as well as empty string. Optionally check for zero values. 

44 # Swallow errors casting to values 

45 if is_null(obj): 

46 return True 

47 obj_as_str = None 

48 try: 

49 obj_as_str = str(obj) 

50 except Exception: 

51 pass 

52 if obj_as_str == "": 

53 return True 

54 if nonzero: 

55 obj_as_float = None # int() truncates values like 0.25 to 0 

56 try: 

57 obj_as_float = float(obj) 

58 except Exception: 

59 pass 

60 if obj_as_float == 0: 

61 return True 

62 return False 

63 

64 

65def df_date_filter(df, start_date, end_date): 

66 # Filter a dataframe by date 

67 if is_empty(start_date) and is_empty(end_date): 

68 return df 

69 if is_empty(start_date): 

70 df = df[df["Date"] <= end_date] 

71 elif is_empty(end_date): 

72 df = df[df["Date"] >= start_date] 

73 else: 

74 df = df[(df["Date"] >= start_date) & (df["Date"] <= end_date)] 

75 df = df.reset_index(drop=True) 

76 if len(df) == 0: 

77 # TODO: this will error if start_date or end_date are not dates 

78 raise Exception(f"Supplied date range ({start_date.date()} - {end_date.date()}) does not contain any \ 

79 transactions!") 

80 return df 

81 

82 

83def save_report(report_data, basename): 

84 dtnow = pd.Timestamp.today() 

85 fname = f"{basename}-{dtnow}.txt" 

86 if path.isfile(fname): 

87 raise Exception(f"File named {fname} already exists!") 

88 with open(fname, 'wt') as f: 

89 f.write(report_data) 

90 

91 

92def validate_date_string(input, allow_empty=False): 

93 # Pandas will accept YYYY-MM-DD or MM/DD/YYYY 

94 if is_empty(input, True): 

95 if allow_empty: 

96 return True 

97 else: 

98 raise Exception("Date string cannot be empty!") 

99 match_obj = re.match(r"^([\d]{4})-([\d]{1,2})-([\d]{1,2})$", input) 

100 if match_obj: 

101 match_year = int(match_obj.groups()[0]) 

102 match_month = int(match_obj.groups()[1]) 

103 match_day = int(match_obj.groups()[2]) 

104 else: 

105 match_obj = re.match(r"^([\d]{1,2})/([\d]{1,2})/([\d]{4})$", input) 

106 if match_obj: 

107 match_year = int(match_obj.groups()[2]) 

108 match_month = int(match_obj.groups()[0]) 

109 match_day = int(match_obj.groups()[1]) 

110 if match_obj: 

111 if not (1900 < match_year < 2100): # Update if doing historical work! 

112 logger.warning(f"Supplied year doesn\t look right: {match_year}") 

113 return False 

114 if not (1 <= match_month <= 12): 

115 logger.warning(f"Invalid month value: {match_month}") 

116 return False 

117 if not (1 <= match_day <= 31): 

118 logger.warning(f"Invalid day value: {match_day}") 

119 return False 

120 return True 

121 return False 

122 

123 

124def normalize_amounts(df_row): 

125 for atype in ["Amount", "Sales Tax", "Tips"]: 

126 val = df_row[atype] 

127 if is_empty(val): 

128 continue 

129 try: 

130 val = float(val) 

131 except ValueError: 

132 if '$' in val or ',' in val: 

133 val = float(val.replace('$', '').replace(',', '')) 

134 df_row[atype] = val 

135 return df_row