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:
6 # http://www.apache.org/licenses/LICENSE-2.0
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.
14 """Module holding BitCountingGroupList class."""
20 from .avg_stdev_stats import AvgStdevStats # Just for type hints.
21 from .bit_counting_group import BitCountingGroup
24 @dataclasses.dataclass
25 class BitCountingGroupList(collections.abc.Sequence):
26 """List of data groups which tracks overall bit count.
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.
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.
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.
43 The data can be only added, and there is some logic to skip
44 recalculations if the bit count is not needed.
48 """Maximal sample value to base bits computation on."""
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.
57 bits_except_last: float = 0.0
58 """Partial sum of all but one group bits."""
60 def __post_init__(self):
61 """Turn possible None into an empty list.
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.
66 if self.group_list is None:
69 def __getitem__(self, index: int) -> BitCountingGroup:
70 """Return the group at the index.
72 :param index: Index of the group to return.
74 :returns: The group at the index.
75 :rtype: BitCountingGroup
77 return self.group_list[index]
79 def __len__(self) -> int:
80 """Return the length of the group list.
82 :returns: The Length of group_list.
85 return len(self.group_list)
87 def copy(self) -> "BitCountingGroupList":
88 """Return a new instance with copied internal state.
90 :returns: The copied instance.
91 :rtype: BitCountingGroupList
93 return self.__class__(
94 max_value=self.max_value,
96 group_list=[group.copy() for group in self.group_list],
97 bits_except_last=self.bits_except_last,
100 def copy_fast(self) -> "BitCountingGroupList":
101 """Return a new instance with minimaly copied internal state.
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".
107 :returns: The copied instance.
108 :rtype: BitCountingGroupList
110 group_list = list(self.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,
121 group_list=group_list,
122 bits_except_last=self.bits_except_last,
126 def bits(self) -> float:
127 """Return overall bit content of the group list.
129 :returns: The overall information content in bits.
132 if not self.group_list:
134 # TODO: Is it worth to cache the overall result?
135 return self.bits_except_last + self.group_list[-1].bits
137 def append_group_of_runs(
140 BitCountingGroup, typing.List[typing.Union[float, AvgStdevStats]]
142 ) -> "BitCountingGroupList":
143 """Mutate to add a new group based on the runs, return self.
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.
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
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
160 new_group.prev_avg = prev_avg
161 new_group.cached_bits = None
163 new_group = BitCountingGroup(
165 max_value=self.max_value,
169 self.bits_except_last = self.bits
170 self.group_list.append(new_group)
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.
178 Basically a one-liner, only returning group list instead of last group.
180 :param run: The run value to add to the last group.
182 :returns: The updated self.
183 :rtype: BitCountingGroupList
184 :raises IndexError: If group list is empty, no last group to add to.
186 self.group_list[-1].append(run)
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.
194 A faster alternative to appending runs one by one in a loop.
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.
202 self.group_list[-1].extend(runs)