resources/libraries/python/jumpavg/BitCountingGroupList.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 BitCountingGroupList class."""
  15
  16 import collections
  17 import dataclasses
  18 import typing
  19
  20 from .AvgStdevStats import AvgStdevStats  # Just for type hints.
  21 from .BitCountingGroup import BitCountingGroup
  22
  23
  24 @dataclasses.dataclass
  25 class BitCountingGroupList(collections.abc.Sequence):
  26     """List of data groups which tracks overall bit count.
  27
  28     The Sequence-like access is related to the list of groups,
  29     for example group_list[0] returns the first group in the list.
  30     Writable list-like methods are not implemented.
  31
  32     The overall bit count is the sum of bit counts of each group.
  33     Group is a sequence of data samples accompanied by their stats.
  34     Different partitioning of data samples into the groups
  35     results in different overall bit count.
  36     This can be used to group samples in various contexts.
  37
  38     As the group bit count depends on previous average
  39     and overall maximal value, order of groups is important.
  40     Having the logic encapsulated here spares the caller
  41     the effort to pass averages around.
  42
  43     The data can be only added, and there is some logic to skip
  44     recalculations if the bit count is not needed.
  45     """
  46
  47     max_value: float
  48     """Maximal sample value to base bits computation on."""
  49     group_list: typing.List[BitCountingGroup] = None
  50     """List of groups to compose this group list.
  51     Init also accepts None standing for an empty list.
  52     This class takes ownership of the list,
  53     so caller of init should clone their copy to avoid unexpected mutations.
  54     """
  55     bits_except_last: float = 0.0
  56     """Partial sum of all but one group bits."""
  57
  58     def __post_init__(self):
  59         """Turn possible None into an empty list.
  60
  61         It is not verified whether the user provided values are valid,
  62         e.g. whether the cached bits values (and bits_except_last) make sense.
  63         """
  64         if self.group_list is None:
  65             self.group_list = list()
  66
  67     def __getitem__(self, index: int) -> BitCountingGroup:
  68         """Return the group at the index.
  69
  70         :param index: Index of the group to return.
  71         :type index: int
  72         :returns: The group at the index.
  73         :rtype: BitCountingGroup
  74         """
  75         return self.group_list[index]
  76
  77     def __len__(self) -> int:
  78         """Return the length of the group list.
  79
  80         :returns: The Length of group_list.
  81         :rtype: int
  82         """
  83         return len(self.group_list)
  84
  85     def copy(self) -> "BitCountingGroupList":
  86         """Return a new instance with copied internal state.
  87
  88         :returns: The copied instance.
  89         :rtype: BitCountingGroupList
  90         """
  91         return self.__class__(
  92             max_value=self.max_value,
  93             group_list=[group.copy() for group in self.group_list],
  94             bits_except_last=self.bits_except_last,
  95         )
  96
  97     def copy_fast(self) -> "BitCountingGroupList":
  98         """Return a new instance with minimaly copied internal state.
  99
 100         The assumption here is that only the last group will ever be mutated
 101         (in self, probably never in the return value),
 102         so all the previous groups can be "copied by reference".
 103
 104         :returns: The copied instance.
 105         :rtype: BitCountingGroupList
 106         """
 107         group_list = list(self.group_list)
 108         if group_list:
 109             group_list[-1] = group_list[-1].copy()
 110             # Further speedup is possible by keeping the last group
 111             # as a singly linked (from end) list,
 112             # but for CSIT sample sizes, copy of whole Python list is faster.
 113             # TODO: Implement linked list as an option
 114             # for users with many samples.
 115         return self.__class__(
 116             max_value=self.max_value,
 117             group_list=group_list,
 118             bits_except_last=self.bits_except_last,
 119         )
 120
 121     @property
 122     def bits(self) -> float:
 123         """Return overall bit content of the group list.
 124
 125         :returns: The overall information content in bits.
 126         :rtype: float
 127         """
 128         if not self.group_list:
 129             return 0.0
 130         # TODO: Is it worth to cache the overall result?
 131         return self.bits_except_last + self.group_list[-1].bits
 132
 133     def append_group_of_runs(
 134         self,
 135         runs: typing.Union[
 136             BitCountingGroup, typing.List[typing.Union[float, AvgStdevStats]]
 137         ],
 138     ) -> "BitCountingGroupList":
 139         """Mutate to add a new group based on the runs, return self.
 140
 141         The list argument is NOT copied before adding to the group list,
 142         so further edits MAY not affect the grup list.
 143         The list from BitCountingGroup is shallow copied though.
 144
 145         :param runs: Runs to form the next group to be appended to self.
 146         :type runs: Union[Iterable[Run], BitCountingGroup]
 147         :returns: The updated self.
 148         :rtype: BitCountingGroupList
 149         """
 150         prev_avg = self.group_list[-1].stats.avg if self.group_list else None
 151         if isinstance(runs, BitCountingGroup):
 152             # It is faster to avoid stats recalculation.
 153             new_group = runs.copy()
 154             new_group.max_value = self.max_value
 155             new_group.prev_avg = prev_avg
 156             new_group.cached_bits = None
 157         else:
 158             new_group = BitCountingGroup(
 159                 run_list=runs, max_value=self.max_value, prev_avg=prev_avg
 160             )
 161         self.bits_except_last = self.bits
 162         self.group_list.append(new_group)
 163         return self
 164
 165     def append_run_to_to_last_group(
 166         self, run: typing.Union[float, AvgStdevStats]
 167     ) -> "BitCountingGroupList":
 168         """Mutate to add new run at the end of the last group.
 169
 170         Basically a one-liner, only returning group list instead of last group.
 171
 172         :param run: The run value to add to the last group.
 173         :type run: Run
 174         :returns: The updated self.
 175         :rtype: BitCountingGroupList
 176         :raises IndexError: If group list is empty, no last group to add to.
 177         """
 178         self.group_list[-1].append(run)
 179         return self
 180
 181     def extend_runs_to_last_group(
 182         self, runs: typing.Iterable[typing.Union[float, AvgStdevStats]]
 183     ) -> "BitCountingGroupList":
 184         """Mutate to add new runs to the end of the last group.
 185
 186         A faster alternative to appending runs one by one in a loop.
 187
 188         :param runs: The runs to add to the last group.
 189         :type runs: Iterable[Run]
 190         :returns: The updated self
 191         :rtype: BitCountingGroupList
 192         :raises IndexError: If group list is empty, no last group to add to.
 193         """
 194         self.group_list[-1].extend(runs)
 195         return self