resources/libraries/python/jumpavg/BitCountingGroup.py

   1 # Copyright (c) 2022 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 BitCountingGroup class."""
  15
  16 import collections
  17 import dataclasses
  18 import typing
  19
  20 from .AvgStdevStats import AvgStdevStats
  21 from .BitCountingStats import BitCountingStats
  22
  23
  24 @dataclasses.dataclass
  25 class BitCountingGroup(collections.abc.Sequence):
  26     """Group of runs which tracks bit count in an efficient manner.
  27
  28     This class contains methods that mutate the internal state,
  29     use copy() method to save the previous state.
  30
  31     The Sequence-like access is related to the list of runs,
  32     for example group[0] returns the first run in the list.
  33     Writable list-like methods are not implemented.
  34
  35     As the group bit count depends on previous average
  36     and overall maximal value, those values are assumed
  37     to be known beforehand (and immutable).
  38
  39     As the caller is allowed to divide runs into groups in any way,
  40     a method to add a single run in an efficient manner is provided.
  41     """
  42
  43     run_list: typing.List[typing.Union[float, AvgStdevStats]]
  44     """List of run to compose into this group.
  45     The init call takes ownership of the list,
  46     so the caller should clone it to avoid unexpected muations."""
  47     max_value: float
  48     """Maximal sample value to expect."""
  49     comment: str = "unknown"
  50     """Any string giving more info, e.g. "regression"."""
  51     prev_avg: typing.Optional[float] = None
  52     """Average of the previous group, if any."""
  53     stats: AvgStdevStats = None
  54     """Stats object used for computing bits.
  55     Almost always recomputed, except when non-None in init."""
  56     cached_bits: typing.Optional[float] = None
  57     """Cached value of information content.
  58     Noned on edit, recomputed if needed and None."""
  59
  60     def __post_init__(self):
  61         """Recompute stats is None.
  62
  63         It is not verified whether the user provided values are valid,
  64         e.g. whether the stats and bits values reflect the runs.
  65         """
  66         if self.stats is None:
  67             self.stats = AvgStdevStats.for_runs(self.run_list)
  68
  69     @property
  70     def bits(self) -> float:
  71         """Return overall bit content of the group list.
  72
  73         If not cached, compute from stats and cache.
  74
  75         :returns: The overall information content in bits.
  76         :rtype: float
  77         """
  78         if self.cached_bits is None:
  79             self.cached_bits = BitCountingStats.for_runs(
  80                 [self.stats], self.max_value, self.prev_avg
  81             ).bits
  82         return self.cached_bits
  83
  84     def __getitem__(self, index: int) -> typing.Union[float, AvgStdevStats]:
  85         """Return the run at the index.
  86
  87         :param index: Index of the run to return.
  88         :type index: int
  89         :returns: The run at the index.
  90         :rtype: typing.Union[float, AvgStdevStats]
  91         """
  92         return self.run_list[index]
  93
  94     def __len__(self) -> int:
  95         """Return the number of runs in the group.
  96
  97         :returns: The Length of run_list.
  98         :rtype: int
  99         """
 100         return len(self.run_list)
 101
 102     def copy(self) -> "BitCountingGroup":
 103         """Return a new instance with copied internal state.
 104
 105         Stats are preserved to avoid re-computation.
 106         As both float and AvgStdevStats are effectively immutable,
 107         only a shallow copy of the runs list is performed.
 108
 109         :returns: The copied instance.
 110         :rtype: BitCountingGroup
 111         """
 112         stats = AvgStdevStats.for_runs([self.stats])
 113         return self.__class__(
 114             run_list=list(self.run_list),
 115             stats=stats,
 116             cached_bits=self.cached_bits,
 117             max_value=self.max_value,
 118             prev_avg=self.prev_avg,
 119             comment=self.comment,
 120         )
 121
 122     def append(
 123         self, run: typing.Union[float, AvgStdevStats]
 124     ) -> "BitCountingGroup":
 125         """Mutate to add the new run, return self.
 126
 127         Stats are updated, but old bits value is deleted from cache.
 128
 129         :param run: The run value to add to the group.
 130         :type value: typing.Union[float, AvgStdevStats]
 131         :returns: The updated self.
 132         :rtype: BitCountingGroup
 133         """
 134         self.run_list.append(run)
 135         self.stats = AvgStdevStats.for_runs([self.stats, run])
 136         self.cached_bits = None
 137         return self
 138
 139     def extend(
 140         self, runs: typing.Iterable[typing.Union[float, AvgStdevStats]]
 141     ) -> "BitCountingGroup":
 142         """Mutate to add the new runs, return self.
 143
 144         This is saves small amount of computation
 145         compared to adding runs one by one in a loop.
 146
 147         Stats are updated, but old bits value is deleted from cache.
 148
 149         :param runs: The runs to add to the group.
 150         :type value: typing.Iterable[typing.Union[float, AvgStdevStats]]
 151         :returns: The updated self.
 152         :rtype: BitCountingGroup
 153         """
 154         self.run_list.extend(runs)
 155         self.stats = AvgStdevStats.for_runs([self.stats] + runs)
 156         self.cached_bits = None
 157         return self