Coverage for backend/django/core/auxiliary/services/result_summary/stats.py: 87%

1from collections import Counter 

2from math import isfinite 

3from typing import NamedTuple 

4 

5from core.auxiliary.services.result_summary.aggregation import aggregate_values 

6from core.auxiliary.services.result_summary.contracts import ( 

7 ResultMetricMatchKind, 

8 ResultMetricRowTarget, 

9 ResultMetricTargets, 

10 ResultStats, 

11) 

12 

13 

14class FiniteResultValuePoint(NamedTuple): 

15 row_index: int 

16 value: float 

17 

18 

19def finite_values(values) -> list[FiniteResultValuePoint]: 

20 finite_result_values = [] 

21 for point in values: 

22 if point.value is None: 22 ↛ 23line 22 didn't jump to line 23 because the condition on line 22 was never true

23 continue 

24 try: 

25 numeric_value = float(point.value) 

26 except (TypeError, ValueError): 

27 continue 

28 if isfinite(numeric_value): 28 ↛ 21line 28 didn't jump to line 21 because the condition on line 28 was always true

29 finite_result_values.append( 

30 FiniteResultValuePoint( 

31 row_index=point.row_index, 

32 value=numeric_value, 

33 ) 

34 ) 

35 return finite_result_values 

36 

37 

38def format_metric_value(value: float) -> str: 

39 return f"{value:.4g}" 

40 

41 

42def result_stats( 

43 finite_result_values: list[FiniteResultValuePoint], 

44 total_value_count: int, 

45) -> ResultStats: 

46 omitted_count = total_value_count - len(finite_result_values) 

47 

48 if not finite_result_values: 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true

49 return ResultStats( 

50 count=0, 

51 omitted_count=omitted_count, 

52 mean=None, 

53 median=None, 

54 mode=None, 

55 mode_label="Unavailable", 

56 min=None, 

57 max=None, 

58 ) 

59 

60 numeric_values = [point.value for point in finite_result_values] 

61 median = float(aggregate_values(numeric_values, "percentile", percentile=50)) 

62 

63 frequencies = Counter(numeric_values) 

64 highest_frequency = max(frequencies.values()) 

65 modes = sorted( 

66 value 

67 for value, frequency in frequencies.items() 

68 if frequency == highest_frequency 

69 ) 

70 if highest_frequency == 1: 

71 mode = None 

72 mode_label = "No repeated value" 

73 elif len(modes) == 1: 

74 mode = modes[0] 

75 mode_label = format_metric_value(mode) 

76 else: 

77 mode = None 

78 mode_label = "Multiple repeated values" 

79 

80 return ResultStats( 

81 count=len(finite_result_values), 

82 omitted_count=omitted_count, 

83 mean=float(aggregate_values(numeric_values, "mean")), 

84 median=median, 

85 mode=mode, 

86 mode_label=mode_label, 

87 min=float(aggregate_values(numeric_values, "min")), 

88 max=float(aggregate_values(numeric_values, "max")), 

89 ) 

90 

91 

92def metric_targets( 

93 finite_result_values: list[FiniteResultValuePoint], 

94 stats: ResultStats, 

95) -> ResultMetricTargets: 

96 return ResultMetricTargets( 

97 first=_first_target(finite_result_values), 

98 mean=_closest_target(finite_result_values, stats.mean), 

99 median=_target_for_metric(finite_result_values, stats.median), 

100 mode=_target_for_metric(finite_result_values, stats.mode), 

101 min=_target_for_metric(finite_result_values, stats.min), 

102 max=_target_for_metric(finite_result_values, stats.max), 

103 ) 

104 

105 

106def _first_target( 

107 finite_result_values: list[FiniteResultValuePoint], 

108) -> ResultMetricRowTarget | None: 

109 if not finite_result_values: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true

110 return None 

111 first_point = finite_result_values[0] 

112 return ResultMetricRowTarget( 

113 row_index=first_point.row_index, 

114 value=first_point.value, 

115 target_value=first_point.value, 

116 match_count=1, 

117 match_kind=ResultMetricMatchKind.first, 

118 ) 

119 

120 

121def _target_for_metric( 

122 finite_result_values: list[FiniteResultValuePoint], 

123 target_value: float | None, 

124) -> ResultMetricRowTarget | None: 

125 if target_value is None: 

126 return None 

127 exact_target = _exact_target(finite_result_values, target_value) 

128 return exact_target or _closest_target(finite_result_values, target_value) 

129 

130 

131def _exact_target( 

132 finite_result_values: list[FiniteResultValuePoint], 

133 target_value: float, 

134) -> ResultMetricRowTarget | None: 

135 matching_points = [ 

136 point for point in finite_result_values if point.value == target_value 

137 ] 

138 if not matching_points: 

139 return None 

140 first_point = matching_points[0] 

141 return ResultMetricRowTarget( 

142 row_index=first_point.row_index, 

143 value=first_point.value, 

144 target_value=target_value, 

145 match_count=len(matching_points), 

146 match_kind=ResultMetricMatchKind.exact, 

147 ) 

148 

149 

150def _closest_target( 

151 finite_result_values: list[FiniteResultValuePoint], 

152 target_value: float | None, 

153) -> ResultMetricRowTarget | None: 

154 if target_value is None or not finite_result_values: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true

155 return None 

156 

157 closest_distance = min( 

158 abs(point.value - target_value) for point in finite_result_values 

159 ) 

160 closest_points = [ 

161 point 

162 for point in finite_result_values 

163 if abs(point.value - target_value) == closest_distance 

164 ] 

165 first_point = closest_points[0] 

166 return ResultMetricRowTarget( 

167 row_index=first_point.row_index, 

168 value=first_point.value, 

169 target_value=target_value, 

170 match_count=len(closest_points), 

171 match_kind=ResultMetricMatchKind.closest, 

172 )