feat(jumpavg): speed up, use Python 3.8 features
[csit.git] / resources / libraries / python / jumpavg / BitCountingGroupList.py
index 6a1c86b..468e79b 100644 (file)
 
 """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.