X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=resources%2Flibraries%2Fpython%2Fjumpavg%2FBitCountingGroup.py;h=f1bdc502fd06be51f99c9f10b7158dbbb9cbe0f7;hb=HEAD;hp=48bea086f467ecac3316f6a410237836df9a8a09;hpb=4bfbd4d72ad53eb1694868c19640c8b4a17d32cb;p=csit.git diff --git a/resources/libraries/python/jumpavg/BitCountingGroup.py b/resources/libraries/python/jumpavg/BitCountingGroup.py deleted file mode 100644 index 48bea086f4..0000000000 --- a/resources/libraries/python/jumpavg/BitCountingGroup.py +++ /dev/null @@ -1,157 +0,0 @@ -# Copyright (c) 2022 Cisco and/or its affiliates. -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at: -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""Module holding BitCountingGroup class.""" - -import collections -import dataclasses -import typing - -from .AvgStdevStats import AvgStdevStats -from .BitCountingStats import BitCountingStats - - -@dataclasses.dataclass -class BitCountingGroup(collections.abc.Sequence): - """Group of runs which tracks bit count in an efficient manner. - - This class contains methods that mutate the internal state, - use copy() method to save the previous state. - - The Sequence-like access is related to the list of runs, - for example group[0] returns the first run in the list. - Writable list-like methods are not implemented. - - As the group bit count depends on previous average - and overall maximal value, those values are assumed - to be known beforehand (and immutable). - - As the caller is allowed to divide runs into groups in any way, - a method to add a single run in an efficient manner is provided. - """ - - run_list: typing.List[typing.Union[float, AvgStdevStats]] - """List of run to compose into this group. - The init call takes ownership of the list, - so the caller should clone it to avoid unexpected muations.""" - max_value: float - """Maximal sample value to expect.""" - comment: str = "unknown" - """Any string giving more info, e.g. "regression".""" - prev_avg: typing.Optional[float] = None - """Average of the previous group, if any.""" - stats: AvgStdevStats = None - """Stats object used for computing bits. - Almost always recomputed, except when non-None in init.""" - cached_bits: typing.Optional[float] = None - """Cached value of information content. - Noned on edit, recomputed if needed and None.""" - - def __post_init__(self): - """Recompute stats is None. - - It is not verified whether the user provided values are valid, - e.g. whether the stats and bits values reflect the runs. - """ - if self.stats is None: - self.stats = AvgStdevStats.for_runs(self.run_list) - - @property - def bits(self) -> float: - """Return overall bit content of the group list. - - If not cached, compute from stats and cache. - - :returns: The overall information content in bits. - :rtype: float - """ - if self.cached_bits is None: - self.cached_bits = BitCountingStats.for_runs( - [self.stats], self.max_value, self.prev_avg - ).bits - return self.cached_bits - - def __getitem__(self, index: int) -> typing.Union[float, AvgStdevStats]: - """Return the run at the index. - - :param index: Index of the run to return. - :type index: int - :returns: The run at the index. - :rtype: typing.Union[float, AvgStdevStats] - """ - return self.run_list[index] - - def __len__(self) -> int: - """Return the number of runs in the group. - - :returns: The Length of run_list. - :rtype: int - """ - return len(self.run_list) - - def copy(self) -> "BitCountingGroup": - """Return a new instance with copied internal state. - - Stats are preserved to avoid re-computation. - As both float and AvgStdevStats are effectively immutable, - only a shallow copy of the runs list is performed. - - :returns: The copied instance. - :rtype: BitCountingGroup - """ - stats = AvgStdevStats.for_runs([self.stats]) - return self.__class__( - run_list=list(self.run_list), - stats=stats, - cached_bits=self.cached_bits, - max_value=self.max_value, - prev_avg=self.prev_avg, - comment=self.comment, - ) - - def append( - self, run: typing.Union[float, AvgStdevStats] - ) -> "BitCountingGroup": - """Mutate to add the new run, return self. - - Stats are updated, but old bits value is deleted from cache. - - :param run: The run value to add to the group. - :type value: typing.Union[float, AvgStdevStats] - :returns: The updated self. - :rtype: BitCountingGroup - """ - self.run_list.append(run) - self.stats = AvgStdevStats.for_runs([self.stats, run]) - self.cached_bits = None - return self - - def extend( - self, runs: typing.Iterable[typing.Union[float, AvgStdevStats]] - ) -> "BitCountingGroup": - """Mutate to add the new runs, return self. - - This is saves small amount of computation - compared to adding runs one by one in a loop. - - Stats are updated, but old bits value is deleted from cache. - - :param runs: The runs to add to the group. - :type value: typing.Iterable[typing.Union[float, AvgStdevStats]] - :returns: The updated self. - :rtype: BitCountingGroup - """ - self.run_list.extend(runs) - self.stats = AvgStdevStats.for_runs([self.stats] + runs) - self.cached_bits = None - return self