distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""RateAverages — reusable container for average computation.
2
3Ported from net.i2p.stat.RateAverages.
4"""
5
6import threading
7
8
9class RateAverages:
10 """Holds computed averages for a Rate. Thread-local reuse via get_temp()."""
11
12 _thread_local = threading.local()
13
14 def __init__(self) -> None:
15 self.average: float = 0.0
16 self.current: float = 0.0
17 self.last: float = 0.0
18 self.total_values: float = 0.0
19 self.total_event_count: int = 0
20
21 @classmethod
22 def get_temp(cls) -> "RateAverages":
23 """Get a thread-local reusable instance."""
24 if not hasattr(cls._thread_local, "instance"):
25 cls._thread_local.instance = RateAverages()
26 return cls._thread_local.instance