Merge pull request #178 from IFCA/fix-CircularMean

Fix inefficient CircularMean implementation

Author: Jaime Céspedes Sisniega

frouros/utils/data_structures.py

@@ -1,7 +1,6 @@
 """Data structures module."""
 
-from typing import Optional, List, Union  # noqa: TYP001
-
+from typing import Any, Optional, List, Union
 
 import numpy as np  # type: ignore
 
@@ -115,20 +114,20 @@ def max_len(self, value: int) -> None:
         self._max_len = value
 
     @property
-    def queue(self) -> List[Optional[bool]]:
+    def queue(self) -> List[Optional[Any]]:
         """Queue property.
 
         :return: queue
-        :rtype: List[Optional[bool]]
+        :rtype: List[Optional[Any]]
         """
         return self._queue
 
     @queue.setter
-    def queue(self, value: List[Optional[bool]]) -> None:
+    def queue(self, value: List[Optional[Any]]) -> None:
         """Queue setter.
 
         :param value: value to be set
-        :type value: List[Optional[bool]]
+        :type value: List[Optional[Any]]
         :raises ValueError: Value error exception
         """
         if not isinstance(value, list):
@@ -151,10 +150,10 @@ def clear(self) -> None:
         self.last = -1
         self.queue = [None] * self.max_len
 
-    def dequeue(self) -> bool:
+    def dequeue(self) -> Any:
         """Dequeue oldest element.
 
-        :rtype: bool
+        :rtype: value: Any
         :raises EmptyQueue: Empty queue error exception
         """
         if self.is_empty():
@@ -164,17 +163,19 @@ def dequeue(self) -> bool:
         self.count -= 1
         return element  # type: ignore
 
-    def enqueue(self, value: Union[np.ndarray, float]) -> None:
+    def enqueue(self, value: Union[np.ndarray, int, float]) -> Optional[Any]:
         """Enqueue element/s.
 
         :param value: value to be enqueued
         :type value: Union[np.ndarray, float]
+        :return element: dequeued element
+        :rtype: Optional[Any]
         """
-        if self.is_full():
-            _ = self.dequeue()
+        element = self.dequeue() if self.is_full() else None
         self.last = (self.last + 1) % self.max_len
         self.queue[self.last] = value  # type: ignore
         self.count += 1
+        return element
 
     def is_empty(self) -> bool:
         """Check if queue is empty.
@@ -205,13 +206,13 @@ def maintain_last_element(self) -> None:
         self.first = self.last
         self.count = 1
 
-    def __getitem__(self, idx: int) -> float:
+    def __getitem__(self, idx: int) -> Any:
         """Get queue item by position.
 
         :param idx: position index
         :type idx: int
         :return: queue item
-        :rtype: float
+        :rtype: Any
         """
         return self.queue[idx]  # type: ignore
 

frouros/utils/stats.py

@@ -90,46 +90,33 @@ def update(self, value: Union[int, float]) -> None:
         if not isinstance(value, (int, float)):
             raise TypeError("value must be of type int or float.")
         self.num_values += 1
-        self.mean += (value - self.mean) / self.num_values
+        self.mean += self.incremental_op(
+            value=value,
+            element=self.mean,
+            size=self.num_values,
+        )
+
+    @staticmethod
+    def incremental_op(
+        value: Union[int, float],
+        element: Union[int, float],
+        size: int,
+    ) -> float:
+        """Incremental operation."""
+        return (value - element) / size
 
     def get(self) -> float:
         """Get method."""
         return self.mean
 
 
-class CircularMean(IncrementalStat):
+class CircularMean(Mean):
     """Circular mean class."""
 
     def __init__(self, size: int) -> None:
-        """Init method.
-
-        :param size: size of the circular mean
-        :type size: int
-        """
-        self.size = size
-        self.mean = 0.0
-        self.queue = CircularQueue(max_len=self.size)
-
-    @property
-    def size(self) -> int:
-        """Size property.
-
-        :return: size value
-        :rtype: int
-        """
-        return self._size
-
-    @size.setter
-    def size(self, value: int) -> None:
-        """Size setter.
-
-        :param value: value to be set
-        :type value: int
-        :raises ValueError: Value error exception
-        """
-        if value < 0:
-            raise ValueError("size must be greater of equal than 0.")
-        self._size = value
+        """Init method."""
+        super().__init__()
+        self.queue = CircularQueue(max_len=size)
 
     def update(self, value: Union[int, float]) -> None:
         """Update the mean value sequentially.
@@ -138,14 +125,15 @@ def update(self, value: Union[int, float]) -> None:
         :type value: int
         :raises TypeError: Type error exception
         """
-        # FIXME: Inefficient implementation  # pylint: disable=fixme
-        self.queue.enqueue(value=value)
-        queue = np.array(self.queue.queue)
-        self.mean = np.mean(queue[queue != np.array(None)])
-
-    def get(self) -> float:
-        """Get method."""
-        return self.mean
+        if not isinstance(value, (int, float)):
+            raise TypeError("value must be of type int or float.")
+        element = self.queue.enqueue(value=value)
+        self.num_values = len(self.queue)
+        self.mean += self.incremental_op(
+            value=value,
+            element=self.mean if element is None else element,
+            size=self.num_values,
+        )
 
 
 class EWMA(IncrementalStat):