resources/libraries/python/jumpavg/BitCountingStats.py

   1 # Copyright (c) 2021 Cisco and/or its affiliates.
   2 # Licensed under the Apache License, Version 2.0 (the "License");
   3 # you may not use this file except in compliance with the License.
   4 # You may obtain a copy of the License at:
   5 #
   6 #     http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS,
  10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11 # See the License for the specific language governing permissions and
  12 # limitations under the License.
  13
  14 """Module holding BitCountingStats class."""
  15
  16 import math
  17
  18 from .AvgStdevStats import AvgStdevStats
  19
  20
  21 class BitCountingStats(AvgStdevStats):
  22     """Class for statistics which include information content of a group.
  23
  24     The information content is based on an assumption that the data
  25     consists of independent random values from a normal distribution.
  26
  27     Instances are only statistics, the data itself is stored elsewhere.
  28
  29     The coding needs to know the previous average, and a maximal value
  30     so both values are required as inputs.
  31
  32     This is a subclass of AvgStdevStats, even though all methods are overriden.
  33     Only for_runs method calls the parent implementation, without using super().
  34     """
  35
  36     def __init__(
  37             self, size=0, avg=None, stdev=0.0, max_value=None, prev_avg=None):
  38         """Construct the stats object by computing from the values needed.
  39
  40         The values are not sanitized, faulty callers can cause math errors.
  41
  42         The None values are allowed for stats for zero size data,
  43         but such stats can report arbitrary avg and max_value.
  44         Stats for nonzero size data cannot contain None,
  45         else ValueError is raised.
  46
  47         The max_value needs to be numeric for nonzero size,
  48         but its relations to avg and prev_avg are not examined.
  49
  50         The bit count is not real, as that would depend on numeric precision
  51         (number of significant bits in values).
  52         The difference is assumed to be constant per value,
  53         which is consistent with Gauss distribution
  54         (but not with floating point mechanic).
  55         The hope is the difference will have
  56         no real impact on the classification procedure.
  57
  58         :param size: Number of values participating in this group.
  59         :param avg: Population average of the participating sample values.
  60         :param stdev: Population standard deviation of the sample values.
  61         :param max_value: Maximal expected value.
  62             TODO: This might be more optimal,
  63             but max-invariant algorithm will be nicer.
  64         :param prev_avg: Population average of the previous group.
  65             If None, no previous average is taken into account.
  66             If not None, the given previous average is used to discourage
  67             consecutive groups with similar averages
  68             (opposite triangle distribution is assumed).
  69         :type avg: float
  70         :type size: int
  71         :type stdev: float
  72         :type max_value: Union[float, NoneType]
  73         :type prev_avg: Union[float, NoneType]
  74         """
  75         self.avg = avg
  76         self.size = size
  77         self.stdev = stdev
  78         self.max_value = max_value
  79         self.prev_avg = prev_avg
  80         # Zero size should in principle have non-zero bits (coding zero size),
  81         # but zero allows users to add empty groups without affecting bits.
  82         self.bits = 0.0
  83         if self.size < 1:
  84             return
  85         if avg is None:
  86             raise ValueError(f"Avg is None: {self!r}")
  87         if max_value is None or max_value <= 0.0:
  88             raise ValueError(f"Invalid max value: {self!r}")
  89         # Length of the sequence must be also counted in bits,
  90         # otherwise the message would not be decodable.
  91         # Model: probability of k samples is 1/k - 1/(k+1) == 1/k/(k+1)
  92         # This is compatible with zero size leading to zero bits.
  93         self.bits += math.log(size * (size + 1), 2)
  94         if prev_avg is None:
  95             # Avg is considered to be uniformly distributed
  96             # from zero to max_value.
  97             self.bits += math.log(max_value + 1.0, 2)
  98         else:
  99             # Opposite triangle distribution with minimum.
 100             self.bits += math.log(
 101                 max_value * (max_value + 1) / (abs(avg - prev_avg) + 1), 2)
 102         if self.size < 2:
 103             return
 104         # Stdev is considered to be uniformly distributed
 105         # from zero to max_value. That is quite a bad expectation,
 106         # but resilient to negative samples etc.
 107         self.bits += math.log(max_value + 1.0, 2)
 108         # Now we know the samples lie on sphere in size-1 dimensions.
 109         # So it is (size-2)-sphere, with radius^2 == stdev^2 * size.
 110         # https://en.wikipedia.org/wiki/N-sphere
 111         sphere_area_ln = math.log(2) + math.log(math.pi) * ((size - 1) / 2.0)
 112         sphere_area_ln -= math.lgamma((size - 1) / 2.0)
 113         sphere_area_ln += math.log(stdev + 1.0) * (size - 2)
 114         sphere_area_ln += math.log(size) * ((size - 2) / 2.0)
 115         self.bits += sphere_area_ln / math.log(2)
 116
 117     def __str__(self):
 118         """Return string with human readable description of the group.
 119
 120         :returns: Readable description.
 121         :rtype: str
 122         """
 123         return (
 124             f"size={self.size} avg={self.avg} stdev={self.stdev}"
 125             f" bits={self.bits}"
 126         )
 127
 128     def __repr__(self):
 129         """Return string executable as Python constructor call.
 130
 131         :returns: Executable constructor call.
 132         :rtype: str
 133         """
 134         return (
 135             f"BitCountingStats(size={self.size!r},avg={self.avg!r}"
 136             f",stdev={self.stdev!r},max_value={self.max_value!r}"
 137             f",prev_avg={self.prev_avg!r})"
 138         )
 139
 140     @classmethod
 141     def for_runs(cls, runs, max_value=None, prev_avg=None):
 142         """Return new stats instance describing the sequence of runs.
 143
 144         If you want to append data to existing stats object,
 145         you can simply use the stats object as the first run.
 146
 147         Instead of a verb, "for" is used to start this method name,
 148         to signify the result contains less information than the input data.
 149
 150         The two optional values can come from outside of the runs provided.
 151
 152         The max_value cannot be None for non-zero size data.
 153         The implementation does not check if no datapoint exceeds max_value.
 154
 155         TODO: Document the behavior for zero size result.
 156
 157         :param runs: Sequence of data to describe by the new metadata.
 158         :param max_value: Maximal expected value.
 159         :param prev_avg: Population average of the previous group, if any.
 160         :type runs: Iterable[Union[float, AvgStdevStats]]
 161         :type max_value: Union[float, NoneType]
 162         :type prev_avg: Union[float, NoneType]
 163         :returns: The new stats instance.
 164         :rtype: cls
 165         """
 166         asd = AvgStdevStats.for_runs(runs)
 167         ret_obj = cls(size=asd.size, avg=asd.avg, stdev=asd.stdev,
 168                       max_value=max_value, prev_avg=prev_avg)
 169         return ret_obj