Class MovingAverage

pymongo/read_preferences.py:614–637

Tracks an exponentially-weighted moving average.

class MovingAverage:
    """Tracks an exponentially-weighted moving average."""

    average: Optional[float]

    def __init__(self) -> None:
        self.average = None

    def add_sample(self, sample: float) -> None:
        if sample < 0:
            raise ValueError(f"duration cannot be negative {sample}")
        if self.average is None:
            self.average = sample
        else:
            # The Server Selection Spec requires an exponentially weighted
            # average with alpha = 0.2.
            self.average = 0.8 * self.average + 0.2 * sample

    def get(self) -> Optional[float]:
        """Get the calculated average, or None if no samples yet."""
        return self.average

    def reset(self) -> None:
        self.average = None