resources/libraries/python/jumpavg/bit_counting_group_list.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 BitCountingGroupList class."""
  15
  16 import collections
  17 import dataclasses
  18 import typing
  19
  20 from .avg_stdev_stats import AvgStdevStats  # Just for type hints.
  21 from .bit_counting_group 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     unit: float = 1.0
  50     """Typical resolution of the values."""
  51     group_list: typing.List[BitCountingGroup] = None
  52     """List of groups to compose this group list.
  53     Init also accepts None standing for an empty list.
  54     This class takes ownership of the list,
  55     so caller of init should clone their copy to avoid unexpected mutations.
  56     """
  57     bits_except_last: float = 0.0
  58     """Partial sum of all but one group bits."""
  59
  60     def __post_init__(self):
  61         """Turn possible None into an empty list.
  62
  63         It is not verified whether the user provided values are valid,
  64         e.g. whether the cached bits values (and bits_except_last) make sense.
  65         """
  66         if self.group_list is None:
  67             self.group_list = []
  68
  69     def __getitem__(self, index: int) -> BitCountingGroup:
  70         """Return the group at the index.
  71
  72         :param index: Index of the group to return.
  73         :type index: int
  74         :returns: The group at the index.
  75         :rtype: BitCountingGroup
  76         """
  77         return self.group_list[index]
  78
  79     def __len__(self) -> int:
  80         """Return the length of the group list.
  81
  82         :returns: The Length of group_list.
  83         :rtype: int
  84         """
  85         return len(self.group_list)
  86
  87     def copy(self) -> "BitCountingGroupList":
  88         """Return a new instance with copied internal state.
  89
  90         :returns: The copied instance.
  91         :rtype: BitCountingGroupList
  92         """
  93         return self.__class__(
  94             max_value=self.max_value,
  95             unit=self.unit,
  96             group_list=[group.copy() for group in self.group_list],
  97             bits_except_last=self.bits_except_last,
  98         )
  99
 100     def copy_fast(self) -> "BitCountingGroupList":
 101         """Return a new instance with minimaly copied internal state.
 102
 103         The assumption here is that only the last group will ever be mutated
 104         (in self, probably never in the return value),
 105         so all the previous groups can be "copied by reference".
 106
 107         :returns: The copied instance.
 108         :rtype: BitCountingGroupList
 109         """
 110         group_list = list(self.group_list)
 111         if group_list:
 112             group_list[-1] = group_list[-1].copy()
 113             # Further speedup is possible by keeping the last group
 114             # as a singly linked (from end) list,
 115             # but for CSIT sample sizes, copy of whole Python list is faster.
 116             # TODO: Implement linked list as an option
 117             # for users with many samples.
 118         return self.__class__(
 119             max_value=self.max_value,
 120             unit=self.unit,
 121             group_list=group_list,
 122             bits_except_last=self.bits_except_last,
 123         )
 124
 125     @property
 126     def bits(self) -> float:
 127         """Return overall bit content of the group list.
 128
 129         :returns: The overall information content in bits.
 130         :rtype: float
 131         """
 132         if not self.group_list:
 133             return 0.0
 134         # TODO: Is it worth to cache the overall result?
 135         return self.bits_except_last + self.group_list[-1].bits
 136
 137     def append_group_of_runs(
 138         self,
 139         runs: typing.Union[
 140             BitCountingGroup, typing.List[typing.Union[float, AvgStdevStats]]
 141         ],
 142     ) -> "BitCountingGroupList":
 143         """Mutate to add a new group based on the runs, return self.
 144
 145         The list argument is NOT copied before adding to the group list,
 146         so further edits MAY not affect the grup list.
 147         The list from BitCountingGroup is shallow copied though.
 148
 149         :param runs: Runs to form the next group to be appended to self.
 150         :type runs: Union[Iterable[Run], BitCountingGroup]
 151         :returns: The updated self.
 152         :rtype: BitCountingGroupList
 153         """
 154         prev_avg = self.group_list[-1].stats.avg if self.group_list else None
 155         if isinstance(runs, BitCountingGroup):
 156             # It is faster to avoid stats recalculation.
 157             new_group = runs.copy()
 158             new_group.max_value = self.max_value
 159             # Unit is common.
 160             new_group.prev_avg = prev_avg
 161             new_group.cached_bits = None
 162         else:
 163             new_group = BitCountingGroup(
 164                 run_list=runs,
 165                 max_value=self.max_value,
 166                 unit=self.unit,
 167                 prev_avg=prev_avg,
 168             )
 169         self.bits_except_last = self.bits
 170         self.group_list.append(new_group)
 171         return self
 172
 173     def append_run_to_to_last_group(
 174         self, run: typing.Union[float, AvgStdevStats]
 175     ) -> "BitCountingGroupList":
 176         """Mutate to add new run at the end of the last group.
 177
 178         Basically a one-liner, only returning group list instead of last group.
 179
 180         :param run: The run value to add to the last group.
 181         :type run: Run
 182         :returns: The updated self.
 183         :rtype: BitCountingGroupList
 184         :raises IndexError: If group list is empty, no last group to add to.
 185         """
 186         self.group_list[-1].append(run)
 187         return self
 188
 189     def extend_runs_to_last_group(
 190         self, runs: typing.Iterable[typing.Union[float, AvgStdevStats]]
 191     ) -> "BitCountingGroupList":
 192         """Mutate to add new runs to the end of the last group.
 193
 194         A faster alternative to appending runs one by one in a loop.
 195
 196         :param runs: The runs to add to the last group.
 197         :type runs: Iterable[Run]
 198         :returns: The updated self
 199         :rtype: BitCountingGroupList
 200         :raises IndexError: If group list is empty, no last group to add to.
 201         """
 202         self.group_list[-1].extend(runs)
 203         return self