"""A counter's previous sample, to turn the next one into a per-second rate. Scrapers that read monotonic counters (SNMP octets, Prometheus *_total) store the rate, not the running total. Keys are tuples whose first element is the owner (a device, a target), so forgetting an owner drops all of its counters at once. """ from __future__ import annotations from datetime import datetime class CounterRates: """In memory: after a restart the first sample of each counter gives no rate and the second does. A sample lower than the one before (the source restarted, or a 32-bit counter wrapped) gives no rate rather than a huge negative or wrapped one.""" def __init__(self) -> None: self._last: dict[tuple, tuple[float, datetime]] = {} def rate(self, key: tuple, value: float, at: datetime) -> float | None: prev = self._last.get(key) self._last[key] = (value, at) if prev is None: return None seconds = (at - prev[1]).total_seconds() if seconds <= 0 or value < prev[0]: return None return (value - prev[0]) / seconds def forget(self, owner) -> None: for key in [k for k in self._last if k[0] == owner]: del self._last[key]