summaryrefslogtreecommitdiff
path: root/histogram.py
blob: 3796470d211ba27e1ce36cfbb6ea18957a9a892f (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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""A text-based simple histogram helper class."""

import math
from dataclasses import dataclass
from typing import Dict, Generic, Iterable, List, Optional, Tuple, TypeVar

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


@dataclass
class BucketDetails:
    """A collection of details about the internal histogram buckets."""

    num_populated_buckets: int = 0
    max_population: Optional[int] = None
    last_bucket_start: Optional[int] = None
    lowest_start: Optional[int] = None
    highest_end: Optional[int] = None
    max_label_width: Optional[int] = None


class SimpleHistogram(Generic[T]):
    """A simple histogram."""

    # 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 NumericPopulation

        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.stats: NumericPopulation = NumericPopulation()
        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.stats.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 get_bucket_details(self, label_formatter: str) -> BucketDetails:
        details = BucketDetails()
        for (start, end), pop in sorted(self.buckets.items(), key=lambda x: x[0]):
            if pop > 0:
                details.num_populated_buckets += 1
                details.last_bucket_start = start
                if details.max_population is None or pop > details.max_population:
                    details.max_population = pop
                if details.lowest_start is None or start < details.lowest_start:
                    details.lowest_start = start
                if details.highest_end is None or end > details.highest_end:
                    details.highest_end = end
                label = f'[{label_formatter}..{label_formatter}): ' % (start, end)
                label_width = len(label)
                if details.max_label_width is None or label_width > details.max_label_width:
                    details.max_label_width = label_width
        return details

    def __repr__(self, *, width: int = 80, label_formatter: str = '%d') -> str:
        from text_utils import bar_graph

        details = self.get_bucket_details(label_formatter)
        txt = ""
        if details.num_populated_buckets == 0:
            return txt
        assert details.max_label_width is not None
        assert details.lowest_start is not None
        assert details.highest_end is not None
        assert details.max_population is not None
        sigma_label = f'[{label_formatter}..{label_formatter}): ' % (
            details.lowest_start,
            details.highest_end,
        )
        if len(sigma_label) > details.max_label_width:
            details.max_label_width = len(sigma_label)
        bar_width = width - (details.max_label_width + 17)

        for (start, end), pop in sorted(self.buckets.items(), key=lambda x: x[0]):
            if start < details.lowest_start:
                continue
            label = f'[{label_formatter}..{label_formatter}): ' % (start, end)
            bar = bar_graph(
                (pop / details.max_population),
                include_text=False,
                width=bar_width,
                left_end="",
                right_end="",
            )
            txt += label.rjust(details.max_label_width)
            txt += bar
            txt += f"({pop/self.count*100.0:5.2f}% n={pop})\n"
            if start == details.last_bucket_start:
                break
        txt += '-' * width + '\n'
        txt += sigma_label.rjust(details.max_label_width)
        txt += ' ' * (bar_width - 2)
        txt += f'     pop(Σn)={self.count}\n'
        txt += ' ' * (bar_width + details.max_label_width - 2)
        txt += f'     mean(x̄)={self.stats.get_mean():.3f}\n'
        txt += ' ' * (bar_width + details.max_label_width - 2)
        txt += f' median(p50)={self.stats.get_median():.3f}\n'
        txt += ' ' * (bar_width + details.max_label_width - 2)
        txt += f'    mode(Mo)={self.stats.get_mode()[0]:.3f}\n'
        txt += ' ' * (bar_width + details.max_label_width - 2)
        txt += f'    stdev(σ)={self.stats.get_stdev():.3f}\n'
        txt += '\n'
        return txt