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 copy
  17
  18 from .BitCountingGroup import BitCountingGroup
  19
  20
  21 class BitCountingGroupList:
  22     # TODO: Inherit from collections.abc.Sequence in Python 3.
  23     """List of data groups which tracks overall bit count.
  24
  25     The Sequence-like access is related to the list of groups,
  26     for example group_list[0] returns the first group in the list.
  27     Writable list-like methods are not implemented.
  28
  29     The overall bit count is the sum of bit counts of each group.
  30     Group is a sequence of data samples accompanied by their stats.
  31     Different partitioning of data samples into the groups
  32     results in different overall bit count.
  33     This can be used to group samples in various contexts.
  34
  35     As the group bit count depends on previous average
  36     and overall maximal value, order of groups is important.
  37     Having the logic encapsulated here spares the caller
  38     the effort to pass averages around.
  39
  40     The data can be only added, and there is some logic to skip
  41     recalculations if the bit count is not needed.
  42     """
  43
  44     def __init__(self, group_list=None, bits_except_last=0.0, max_value=None):
  45         """Set the internal state without any calculations.
  46
  47         The group list argument is copied deeply, so it is not a problem
  48         if the value object is mutated afterwards.
  49
  50         A "group" stands for an Iterable of runs, where "run" is either
  51         a float value, or a stats-like object (only size, avg and stdev
  52         are accessed). Run is a hypothetical abstract class,
  53         defining it in Python 2 is too much hassle.
  54
  55         It is not verified whether the user provided values are valid,
  56         e.g. whether the cached bits values make sense.
  57
  58         The max_value is required and immutable,
  59         it is recommended the callers find their maximum beforehand.
  60
  61         :param group_list: List of groups to compose this group list (or empty).
  62         :param bits_except_last: Partial sum of all but one group bits.
  63         :param max_value: Maximal sample value to base bits computation on.
  64         :type group_list: Iterable[BitCountingGroup]
  65         :type bits_except_last: float
  66         :type max_value: float
  67         """
  68         self.group_list = copy.deepcopy(group_list) if group_list else list()
  69         self.bits_except_last = bits_except_last
  70         self.max_value = max_value
  71
  72     def __str__(self):
  73         """Return string with human readable description of the group list.
  74
  75         :returns: Readable description.
  76         :rtype: str
  77         """
  78         return f"group_list={self.group_list} bits={self.bits}"
  79
  80     def __repr__(self):
  81         """Return string executable as Python constructor call.
  82
  83         :returns: Executable constructor call.
  84         :rtype: str
  85         """
  86         return (
  87             f"BitCountingGroupList(group_list={self.group_list!r}"
  88             f",bits_except_last={self.bits_except_last!r}"
  89             f",max_value={self.max_value!r})"
  90         )
  91
  92     def __getitem__(self, index):
  93         """Return the group at the index.
  94
  95         :param index: Index of the group to return.
  96         :type index: int
  97         :returns: The group at the index.
  98         :rtype: BitCountingGroup
  99         """
 100         return self.group_list[index]
 101
 102     def __len__(self):
 103         """Return the length of the group list.
 104
 105         :returns: The Length of group_list.
 106         :rtype: int
 107         """
 108         return len(self.group_list)
 109
 110     def copy(self):
 111         """Return a new instance with copied internal state.
 112
 113         :returns: The copied instance.
 114         :rtype: BitCountingGroupList
 115         """
 116         return self.__class__(
 117             group_list=self.group_list, bits_except_last=self.bits_except_last,
 118             max_value=self.max_value
 119         )
 120
 121     @property
 122     def bits(self):
 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(self, runs):
 134         """Mutate to add a new group based on the runs, return self.
 135
 136         The argument is copied before adding to the group list,
 137         so further edits do not affect the grup list.
 138         The argument can also be a group, only runs from it are used.
 139
 140         :param runs: Runs to form the next group to be appended to self.
 141         :type runs: Union[Iterable[Run], BitCountingGroup]
 142         :returns: The updated self.
 143         :rtype: BitCountingGroupList
 144         """
 145         prev_avg = self.group_list[-1].stats.avg if self.group_list else None
 146         if isinstance(runs, BitCountingGroup):
 147             # It is faster to avoid stats recalculation.
 148             new_group = runs.copy()
 149             new_group.max_value = self.max_value
 150             new_group.prev_avg = prev_avg
 151             new_group.cached_bits = None
 152         else:
 153             new_group = BitCountingGroup(
 154                 run_list=runs, max_value=self.max_value, prev_avg=prev_avg)
 155         self.bits_except_last = self.bits
 156         self.group_list.append(new_group)
 157         return self
 158
 159     def append_run_to_to_last_group(self, run):
 160         """Mutate to add new run at the end of the last group.
 161
 162         Basically a one-liner, only returning group list instead of last group.
 163
 164         :param run: The run value to add to the last group.
 165         :type run: Run
 166         :returns: The updated self.
 167         :rtype: BitCountingGroupList
 168         :raises IndexError: If group list is empty, no last group to add to.
 169         """
 170         self.group_list[-1].append(run)
 171         return self
 172
 173     def extend_runs_to_last_group(self, runs):
 174         """Mutate to add new runs to the end of the last group.
 175
 176         A faster alternative to appending runs one by one in a loop.
 177
 178         :param runs: The runs to add to the last group.
 179         :type runs: Iterable[Run]
 180         :returns: The updated self
 181         :rtype: BitCountingGroupList
 182         :raises IndexError: If group list is empty, no last group to add to.
 183         """
 184         self.group_list[-1].extend(runs)
 185         return self