resources/libraries/python/jumpavg/bit_counting_stats.py

   1 # Copyright (c) 2023 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 dataclasses
  17 import math
  18 import typing
  19
  20 from .avg_stdev_stats import AvgStdevStats
  21
  22
  23 @dataclasses.dataclass
  24 class BitCountingStats(AvgStdevStats):
  25     """Class for statistics which include information content of a group.
  26
  27     The information content is based on an assumption that the data
  28     consists of independent random values from a normal distribution.
  29
  30     Instances are only statistics, the data itself is stored elsewhere.
  31
  32     The coding needs to know the previous average, and a maximal value
  33     so both values are required as inputs.
  34
  35     This is a subclass of AvgStdevStats, even though all methods are overriden.
  36     Only for_runs method calls the parent implementation, without using super().
  37     """
  38
  39     max_value: float = None
  40     """Maximal sample value (real or estimated).
  41     Default value is there just for argument ordering reasons,
  42     leaving None leads to exceptions."""
  43     unit: float = 1.0
  44     """Typical resolution of the values."""
  45     prev_avg: typing.Optional[float] = None
  46     """Population average of the previous group (if any)."""
  47     bits: float = None
  48     """The computed information content of the group.
  49     It is formally an argument to init function, just to keep repr string
  50     a valid call. ut the init value is ignored and always recomputed.
  51     """
  52
  53     def __post_init__(self):
  54         """Construct the stats object by computing from the values needed.
  55
  56         The None values are allowed for stats for zero size data,
  57         but such stats can report arbitrary avg and max_value.
  58         Stats for nonzero size data cannot contain None,
  59         else ValueError is raised.
  60
  61         The max_value needs to be numeric for nonzero size,
  62         but its relations to avg and prev_avg are not examined.
  63
  64         The bit count is not real, as that would depend on numeric precision
  65         (number of significant bits in values).
  66         The difference is assumed to be constant per value,
  67         which is consistent with Gauss distribution
  68         (but not with floating point mechanic).
  69         The hope is the difference will have
  70         no real impact on the classification procedure.
  71         """
  72         # Zero size should in principle have non-zero bits (coding zero size),
  73         # but zero allows users to add empty groups without affecting bits.
  74         self.bits = 0.0
  75         if self.size < 1:
  76             return
  77         if self.max_value <= 0.0:
  78             raise ValueError(f"Invalid max value: {self!r}")
  79         max_value = self.max_value / self.unit
  80         avg = self.avg / self.unit
  81         # Length of the sequence must be also counted in bits,
  82         # otherwise the message would not be decodable.
  83         # Model: probability of k samples is 1/k - 1/(k+1) == 1/k/(k+1)
  84         # This is compatible with zero size leading to zero bits.
  85         self.bits += math.log(self.size * (self.size + 1), 2)
  86         if self.prev_avg is None:
  87             # Avg is considered to be uniformly distributed
  88             # from zero to max_value.
  89             self.bits += math.log(max_value + 1, 2)
  90         else:
  91             # Opposite triangle distribution with minimum.
  92             prev_avg = self.prev_avg / self.unit
  93             norm = prev_avg * prev_avg
  94             norm -= (prev_avg - 1) * max_value
  95             norm += max_value * max_value / 2
  96             self.bits -= math.log((abs(avg - prev_avg) + 1) / norm, 2)
  97         if self.size < 2:
  98             return
  99         stdev = self.stdev / self.unit
 100         # Stdev can be anything between zero and max value.
 101         # For size==2, sphere surface is 2 points regardless of radius,
 102         # we need to penalize large stdev already when encoding the stdev.
 103         # The simplest way is to use the same distribution as with size...
 104         self.bits += math.log((stdev + 1) * (stdev + 2), 2)
 105         # .. just with added normalization from the max value cut-off.
 106         self.bits += math.log(1 - 1 / (max_value + 2), 2)
 107         # Now we know the samples lie on sphere in size-1 dimensions.
 108         # So it is (size-2)-sphere, with radius^2 == stdev^2 * size.
 109         # https://en.wikipedia.org/wiki/N-sphere
 110         sphere_area_ln = math.log(2)
 111         sphere_area_ln += math.log(math.pi) * ((self.size - 1) / 2)
 112         sphere_area_ln -= math.lgamma((self.size - 1) / 2)
 113         sphere_area_ln += math.log(stdev + 1) * (self.size - 2)
 114         sphere_area_ln += math.log(self.size) * ((self.size - 2) / 2)
 115         self.bits += sphere_area_ln / math.log(2)
 116
 117     @classmethod
 118     def for_runs_and_params(
 119         cls,
 120         runs: typing.Iterable[typing.Union[float, AvgStdevStats]],
 121         max_value: float,
 122         unit: float = 1.0,
 123         prev_avg: typing.Optional[float] = None,
 124     ):
 125         """Return new stats instance describing the sequence of runs.
 126
 127         If you want to append data to existing stats object,
 128         you can simply use the stats object as the first run.
 129
 130         Instead of a verb, "for" is used to start this method name,
 131         to signify the result contains less information than the input data.
 132
 133         The two optional values can come from outside of the runs provided.
 134
 135         The max_value cannot be None for non-zero size data.
 136         The implementation does not check if no datapoint exceeds max_value.
 137
 138         TODO: Document the behavior for zero size result.
 139
 140         :param runs: Sequence of data to describe by the new metadata.
 141         :param max_value: Maximal expected value.
 142         :param unit: Typical resolution of the values.
 143         :param prev_avg: Population average of the previous group, if any.
 144         :type runs: Iterable[Union[float, AvgStdevStats]]
 145         :type max_value: Union[float, NoneType]
 146         :type unit: float
 147         :type prev_avg: Union[float, NoneType]
 148         :returns: The new stats instance.
 149         :rtype: cls
 150         """
 151         asd = AvgStdevStats.for_runs(runs)
 152         ret_obj = cls(
 153             size=asd.size,
 154             avg=asd.avg,
 155             stdev=asd.stdev,
 156             max_value=max_value,
 157             unit=unit,
 158             prev_avg=prev_avg,
 159         )
 160         return ret_obj