1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
|
#!/usr/bin/env python3
import math
from numbers import Number
from typing import Generic, Iterable, List, Optional, Tuple, TypeVar
T = TypeVar("T", bound=Number)
class SimpleHistogram(Generic[T]):
# Useful in defining wide open bottom/top bucket bounds:
POSITIVE_INFINITY = math.inf
NEGATIVE_INFINITY = -math.inf
def __init__(self, buckets: List[Tuple[T, T]]):
from math_utils import RunningMedian
self.buckets = {}
for start_end in buckets:
if self._get_bucket(start_end[0]) is not None:
raise Exception("Buckets overlap?!")
self.buckets[start_end] = 0
self.sigma = 0
self.median = RunningMedian()
self.maximum = None
self.minimum = None
self.count = 0
@staticmethod
def n_evenly_spaced_buckets(
min_bound: T,
max_bound: T,
n: int,
) -> List[Tuple[T, T]]:
ret = []
stride = int((max_bound - min_bound) / n)
if stride <= 0:
raise Exception("Min must be < Max")
for bucket_start in range(min_bound, max_bound, stride):
ret.append((bucket_start, bucket_start + stride))
return ret
def _get_bucket(self, item: T) -> Optional[Tuple[T, T]]:
for start_end in self.buckets:
if start_end[0] <= item < start_end[1]:
return start_end
return None
def add_item(self, item: T) -> bool:
bucket = self._get_bucket(item)
if bucket is None:
return False
self.count += 1
self.buckets[bucket] += 1
self.sigma += item
self.median.add_number(item)
if self.maximum is None or item > self.maximum:
self.maximum = item
if self.minimum is None or item < self.minimum:
self.minimum = item
return True
def add_items(self, lst: Iterable[T]) -> bool:
all_true = True
for item in lst:
all_true = all_true and self.add_item(item)
return all_true
def __repr__(self, label_formatter='%10s') -> str:
from text_utils import bar_graph
max_population: Optional[int] = None
for bucket in self.buckets:
pop = self.buckets[bucket]
if pop > 0:
last_bucket_start = bucket[0] # beginning of range
if max_population is None or pop > max_population:
max_population = pop # bucket with max items
txt = ""
if max_population is None:
return txt
for bucket in sorted(self.buckets, key=lambda x: x[0]):
pop = self.buckets[bucket]
start = bucket[0]
end = bucket[1]
bar = bar_graph(
(pop / max_population),
include_text=False,
width=58,
left_end="",
right_end="",
)
label = f'{label_formatter}..{label_formatter}' % (start, end)
txt += (
f'{label:20}: '
+ bar
+ f"({pop/self.count*100.0:5.2f}% n={pop})\n"
)
if start == last_bucket_start:
break
return txt
|
