Coverage for backend/django/core/auxiliary/services/result_summary/aggregation.py: 85%

1"""Reusable numeric aggregation helpers for MSS result-derived values.""" 

2 

3from __future__ import annotations 

4 

5from decimal import Decimal 

6from enum import StrEnum 

7from typing import Iterable 

8 

9 

10class NumericAggregateFunction(StrEnum): 

11 """Supported scalar reductions for schedule and result-summary values.""" 

12 

13 MIN = "min" 

14 MAX = "max" 

15 MEAN = "mean" 

16 PERCENTILE = "percentile" 

17 

18 

19def aggregate_values( 

20 values: Iterable[Decimal | float | int | str], 

21 aggregate_function: str, 

22 *, 

23 percentile: Decimal | float | int | str | None = None, 

24) -> Decimal: 

25 """Aggregate finite numeric values using the shared MSS reduction semantics.""" 

26 

27 numeric_values = _decimal_values(values) 

28 if not numeric_values: 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true

29 raise ValueError("At least one finite value is required.") 

30 function = NumericAggregateFunction(str(aggregate_function or "").strip().lower()) 

31 if function == NumericAggregateFunction.MIN: 

32 return min(numeric_values) 

33 if function == NumericAggregateFunction.MAX: 

34 return max(numeric_values) 

35 if function == NumericAggregateFunction.MEAN: 

36 return sum(numeric_values, Decimal("0")) / Decimal(len(numeric_values)) 

37 if function == NumericAggregateFunction.PERCENTILE: 37 ↛ 40line 37 didn't jump to line 40 because the condition on line 37 was always true

38 percentile_value = Decimal("50") if percentile is None else _decimal_value(percentile) 

39 return linear_percentile(numeric_values, percentile_value) 

40 raise ValueError(f"Unsupported aggregate function: {aggregate_function}") 

41 

42 

43def linear_percentile( 

44 values: Iterable[Decimal | float | int | str], 

45 percentile: Decimal | float | int | str, 

46) -> Decimal: 

47 """Return a percentile using linear interpolation between sorted values.""" 

48 

49 numeric_values = sorted(_decimal_values(values)) 

50 if not numeric_values: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true

51 raise ValueError("At least one finite value is required.") 

52 percentile_value = _decimal_value(percentile) 

53 if percentile_value < 0 or percentile_value > 100: 53 ↛ 54line 53 didn't jump to line 54 because the condition on line 53 was never true

54 raise ValueError("Percentile must be between 0 and 100.") 

55 if len(numeric_values) == 1: 

56 return numeric_values[0] 

57 rank = (percentile_value / Decimal("100")) * Decimal(len(numeric_values) - 1) 

58 lower_index = int(rank) 

59 upper_index = min(lower_index + 1, len(numeric_values) - 1) 

60 fraction = rank - Decimal(lower_index) 

61 lower = numeric_values[lower_index] 

62 upper = numeric_values[upper_index] 

63 return lower + ((upper - lower) * fraction) 

64 

65 

66def _decimal_values(values: Iterable[Decimal | float | int | str]) -> list[Decimal]: 

67 return [_decimal_value(value) for value in values] 

68 

69 

70def _decimal_value(value: Decimal | float | int | str) -> Decimal: 

71 decimal_value = Decimal(str(value)) 

72 if not decimal_value.is_finite(): 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true

73 raise ValueError("Aggregate values must be finite.") 

74 return decimal_value