Coverage for /usr/local/lib/python3.12/site-packages/prefect/utilities/math.py: 24%

1import math 1 ctx1a

2import random 1 ctx1a

3 

4 

5def poisson_interval( 1 ctx1a

6 average_interval: float, lower: float = 0, upper: float = 1 

7) -> float: 

8 """ 

9 Generates an "inter-arrival time" for a Poisson process. 

10 

11 Draws a random variable from an exponential distribution using the inverse-CDF 

12 method. Can optionally be passed a lower and upper bound between (0, 1] to clamp 

13 the potential output values. 

14 """ 

15 

16 # note that we ensure the argument to the logarithm is stabilized to prevent 

17 # calling log(0), which results in a DomainError 

18 return -math.log(max(1 - random.uniform(lower, upper), 1e-10)) * average_interval 

19 

20 

21def exponential_cdf(x: float, average_interval: float) -> float: 1 ctx1a

22 ld = 1 / average_interval 

23 return 1 - math.exp(-ld * x) 

24 

25 

26def lower_clamp_multiple(k: float) -> float: 1 ctx1a

27 """ 

28 Computes a lower clamp multiple that can be used to bound a random variate drawn 

29 from an exponential distribution. 

30 

31 Given an upper clamp multiple `k` (and corresponding upper bound k * average_interval), 

32 this function computes a lower clamp multiple `c` (corresponding to a lower bound 

33 c * average_interval) where the probability mass between the lower bound and the 

34 median is equal to the probability mass between the median and the upper bound. 

35 """ 

36 if k >= 50: 

37 # return 0 for large values of `k` to prevent numerical overflow 

38 return 0.0 

39 

40 return math.log(max(2**k / (2**k - 1), 1e-10), 2) 

41 

42 

43def clamped_poisson_interval( 1 ctx1a

44 average_interval: float, clamping_factor: float = 0.3 

45) -> float: 

46 """ 

47 Bounds Poisson "inter-arrival times" to a range defined by the clamping factor. 

48 

49 The upper bound for this random variate is: average_interval * (1 + clamping_factor). 

50 A lower bound is picked so that the average interval remains approximately fixed. 

51 """ 

52 if clamping_factor <= 0: 

53 raise ValueError("`clamping_factor` must be >= 0.") 

54 

55 upper_clamp_multiple = 1 + clamping_factor 

56 upper_bound = average_interval * upper_clamp_multiple 

57 lower_bound = max(0, average_interval * lower_clamp_multiple(upper_clamp_multiple)) 

58 

59 upper_rv = exponential_cdf(upper_bound, average_interval) 

60 lower_rv = exponential_cdf(lower_bound, average_interval) 

61 return poisson_interval(average_interval, lower_rv, upper_rv) 

62 

63 

64def bounded_poisson_interval(lower_bound: float, upper_bound: float) -> float: 1 ctx1a

65 """ 

66 Bounds Poisson "inter-arrival times" to a range. 

67 

68 Unlike `clamped_poisson_interval` this does not take a target average interval. 

69 Instead, the interval is predetermined and the average is calculated as their 

70 midpoint. This allows Poisson intervals to be used in cases where a lower bound 

71 must be enforced. 

72 """ 

73 average = (float(lower_bound) + float(upper_bound)) / 2.0 

74 upper_rv = exponential_cdf(upper_bound, average) 

75 lower_rv = exponential_cdf(lower_bound, average) 

76 return poisson_interval(average, lower_rv, upper_rv)