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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
#!/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, *, width: int = 80, label_formatter: str = None) -> 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
max_label_width = None
lowest_start = None
highest_end = None
for bucket in sorted(self.buckets, key=lambda x: x[0]):
start = bucket[0]
if lowest_start is None:
lowest_start = start
end = bucket[1]
if highest_end is None or end > highest_end:
highest_end = end
if label_formatter is None:
if type(start) == int and type(end) == int:
label_formatter = '%d'
elif type(start) == float and type(end) == float:
label_formatter = '%.2f'
else:
label_formatter = '%s'
label = f'[{label_formatter}..{label_formatter}): ' % (start, end)
label_width = len(label)
if max_label_width is None or label_width > max_label_width:
max_label_width = label_width
if start == last_bucket_start:
break
sigma_label = f'[{label_formatter}..{label_formatter}): ' % (
lowest_start,
highest_end,
)
if len(sigma_label) > max_label_width:
max_label_width = len(sigma_label)
bar_width = width - (max_label_width + 16)
for bucket in sorted(self.buckets, key=lambda x: x[0]):
start = bucket[0]
end = bucket[1]
label = f'[{label_formatter}..{label_formatter}): ' % (start, end)
pop = self.buckets[bucket]
bar = bar_graph(
(pop / max_population),
include_text=False,
width=bar_width,
left_end="",
right_end="",
)
txt += label.rjust(max_label_width)
txt += bar
txt += f"({pop/self.count*100.0:5.2f}% n={pop})\n"
if start == last_bucket_start:
break
txt += '-' * width + '\n'
txt += sigma_label.rjust(max_label_width)
txt += ' ' * (bar_width - 2)
txt += f'Σ=(100.00% n={self.count})\n'
return txt
|
