summaryrefslogtreecommitdiff
path: root/histogram.py
blob: cea8b766e3f88a3e197cbdff128539f199fdc83a (plain)
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
#!/usr/bin/env python3

import math
from numbers import Number
from typing import Dict, Generic, Iterable, List, Optional, Tuple, TypeVar

T = TypeVar("T", int, float)
Bound = int
Count = int


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[Bound, Bound]]):
        from math_utils import RunningMedian

        self.buckets: Dict[Tuple[Bound, Bound], Count] = {}
        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: float = 0.0
        self.median: RunningMedian = RunningMedian()
        self.maximum: Optional[T] = None
        self.minimum: Optional[T] = None
        self.count: Count = 0

    @staticmethod
    def n_evenly_spaced_buckets(
        min_bound: T,
        max_bound: T,
        n: int,
    ) -> List[Tuple[int, int]]:
        ret: List[Tuple[int, int]] = []
        stride = int((max_bound - min_bound) / n)
        if stride <= 0:
            raise Exception("Min must be < Max")
        imax = math.ceil(max_bound)
        imin = math.floor(min_bound)
        for bucket_start in range(imin, imax, stride):
            ret.append((bucket_start, bucket_start + stride))
        return ret

    def _get_bucket(self, item: T) -> Optional[Tuple[int, int]]:
        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 = '%d') -> str:
        from text_utils import bar_graph

        txt = ""
        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
        if len(self.buckets) == 0 or max_population is None:
            return txt

        max_label_width: Optional[int] = None
        lowest_start: Optional[int] = None
        highest_end: Optional[int] = 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
            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
        assert max_label_width is not None
        assert lowest_start is not None
        assert highest_end is not None

        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