admin / Syanpse-Vanguard

"""Lightweight in-process throughput meter for the ingest receive rate.

A sliding window of (timestamp, count) samples; EPS is the events in the window
divided by the window length (a smoothed trailing average). Cheap enough to call
on the ingest hot path — record() is O(1) amortized."""
from __future__ import annotations

import time
from collections import deque


class EpsMeter:
    def __init__(self, window_s: float = 5.0):
        self.window_s = window_s
        self._samples: deque[tuple[float, int]] = deque()
        self._total = 0

    def record(self, n: int) -> None:
        now = time.monotonic()
        self._samples.append((now, n))
        self._total += n
        self._prune(now)

    def _prune(self, now: float) -> None:
        cutoff = now - self.window_s
        while self._samples and self._samples[0][0] < cutoff:
            self._samples.popleft()

    def eps(self) -> float:
        now = time.monotonic()
        self._prune(now)
        return sum(c for _, c in self._samples) / self.window_s

    @property
    def total(self) -> int:
        return self._total

    def snapshot(self) -> dict:
        return {"eps": round(self.eps(), 2), "received_total": self._total,
                "window_s": self.window_s}