X-Git-Url: https://gerrit.fd.io/r/gitweb?p=csit.git;a=blobdiff_plain;f=resources%2Flibraries%2Fpython%2Fjumpavg%2FBitCountingGroupList.py;h=468e79b2363c80d759aa8b36ca950bf288a7b555;hp=6a1c86baf2eefd67bb8ea5fedf9a3c7598bbab80;hb=4bfbd4d72ad53eb1694868c19640c8b4a17d32cb;hpb=c1b770bc71eda83468c0e2a97c851b831b76641b diff --git a/resources/libraries/python/jumpavg/BitCountingGroupList.py b/resources/libraries/python/jumpavg/BitCountingGroupList.py index 6a1c86baf2..468e79b236 100644 --- a/resources/libraries/python/jumpavg/BitCountingGroupList.py +++ b/resources/libraries/python/jumpavg/BitCountingGroupList.py @@ -13,13 +13,16 @@ """Module holding BitCountingGroupList class.""" -import copy +import collections +import dataclasses +import typing +from .AvgStdevStats import AvgStdevStats # Just for type hints. from .BitCountingGroup import BitCountingGroup -class BitCountingGroupList: - # TODO: Inherit from collections.abc.Sequence in Python 3. +@dataclasses.dataclass +class BitCountingGroupList(collections.abc.Sequence): """List of data groups which tracks overall bit count. The Sequence-like access is related to the list of groups, @@ -41,55 +44,27 @@ class BitCountingGroupList: recalculations if the bit count is not needed. """ - def __init__(self, group_list=None, bits_except_last=0.0, max_value=None): - """Set the internal state without any calculations. - - The group list argument is copied deeply, so it is not a problem - if the value object is mutated afterwards. + max_value: float + """Maximal sample value to base bits computation on.""" + group_list: typing.List[BitCountingGroup] = None + """List of groups to compose this group list. + Init also accepts None standing for an empty list. + This class takes ownership of the list, + so caller of init should clone their copy to avoid unexpected mutations. + """ + bits_except_last: float = 0.0 + """Partial sum of all but one group bits.""" - A "group" stands for an Iterable of runs, where "run" is either - a float value, or a stats-like object (only size, avg and stdev - are accessed). Run is a hypothetical abstract class, - defining it in Python 2 is too much hassle. + def __post_init__(self): + """Turn possible None into an empty list. It is not verified whether the user provided values are valid, - e.g. whether the cached bits values make sense. - - The max_value is required and immutable, - it is recommended the callers find their maximum beforehand. - - :param group_list: List of groups to compose this group list (or empty). - :param bits_except_last: Partial sum of all but one group bits. - :param max_value: Maximal sample value to base bits computation on. - :type group_list: Iterable[BitCountingGroup] - :type bits_except_last: float - :type max_value: float - """ - self.group_list = copy.deepcopy(group_list) if group_list else list() - self.bits_except_last = bits_except_last - self.max_value = max_value - - def __str__(self): - """Return string with human readable description of the group list. - - :returns: Readable description. - :rtype: str + e.g. whether the cached bits values (and bits_except_last) make sense. """ - return f"group_list={self.group_list} bits={self.bits}" - - def __repr__(self): - """Return string executable as Python constructor call. + if self.group_list is None: + self.group_list = list() - :returns: Executable constructor call. - :rtype: str - """ - return ( - f"BitCountingGroupList(group_list={self.group_list!r}" - f",bits_except_last={self.bits_except_last!r}" - f",max_value={self.max_value!r})" - ) - - def __getitem__(self, index): + def __getitem__(self, index: int) -> BitCountingGroup: """Return the group at the index. :param index: Index of the group to return. @@ -99,7 +74,7 @@ class BitCountingGroupList: """ return self.group_list[index] - def __len__(self): + def __len__(self) -> int: """Return the length of the group list. :returns: The Length of group_list. @@ -107,19 +82,44 @@ class BitCountingGroupList: """ return len(self.group_list) - def copy(self): + def copy(self) -> "BitCountingGroupList": """Return a new instance with copied internal state. :returns: The copied instance. :rtype: BitCountingGroupList """ return self.__class__( - group_list=self.group_list, bits_except_last=self.bits_except_last, - max_value=self.max_value + max_value=self.max_value, + group_list=[group.copy() for group in self.group_list], + bits_except_last=self.bits_except_last, + ) + + def copy_fast(self) -> "BitCountingGroupList": + """Return a new instance with minimaly copied internal state. + + The assumption here is that only the last group will ever be mutated + (in self, probably never in the return value), + so all the previous groups can be "copied by reference". + + :returns: The copied instance. + :rtype: BitCountingGroupList + """ + group_list = list(self.group_list) + if group_list: + group_list[-1] = group_list[-1].copy() + # Further speedup is possible by keeping the last group + # as a singly linked (from end) list, + # but for CSIT sample sizes, copy of whole Python list is faster. + # TODO: Implement linked list as an option + # for users with many samples. + return self.__class__( + max_value=self.max_value, + group_list=group_list, + bits_except_last=self.bits_except_last, ) @property - def bits(self): + def bits(self) -> float: """Return overall bit content of the group list. :returns: The overall information content in bits. @@ -130,12 +130,17 @@ class BitCountingGroupList: # TODO: Is it worth to cache the overall result? return self.bits_except_last + self.group_list[-1].bits - def append_group_of_runs(self, runs): + def append_group_of_runs( + self, + runs: typing.Union[ + BitCountingGroup, typing.List[typing.Union[float, AvgStdevStats]] + ], + ) -> "BitCountingGroupList": """Mutate to add a new group based on the runs, return self. - The argument is copied before adding to the group list, - so further edits do not affect the grup list. - The argument can also be a group, only runs from it are used. + The list argument is NOT copied before adding to the group list, + so further edits MAY not affect the grup list. + The list from BitCountingGroup is shallow copied though. :param runs: Runs to form the next group to be appended to self. :type runs: Union[Iterable[Run], BitCountingGroup] @@ -151,12 +156,15 @@ class BitCountingGroupList: new_group.cached_bits = None else: new_group = BitCountingGroup( - run_list=runs, max_value=self.max_value, prev_avg=prev_avg) + run_list=runs, max_value=self.max_value, prev_avg=prev_avg + ) self.bits_except_last = self.bits self.group_list.append(new_group) return self - def append_run_to_to_last_group(self, run): + def append_run_to_to_last_group( + self, run: typing.Union[float, AvgStdevStats] + ) -> "BitCountingGroupList": """Mutate to add new run at the end of the last group. Basically a one-liner, only returning group list instead of last group. @@ -170,7 +178,9 @@ class BitCountingGroupList: self.group_list[-1].append(run) return self - def extend_runs_to_last_group(self, runs): + def extend_runs_to_last_group( + self, runs: typing.Iterable[typing.Union[float, AvgStdevStats]] + ) -> "BitCountingGroupList": """Mutate to add new runs to the end of the last group. A faster alternative to appending runs one by one in a loop.