distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""RateStat — named multi-period rate coordinator.
2
3Ported from net.i2p.stat.RateStat.
4"""
5
6from typing import List, Optional
7
8from i2p_stat.rate import Rate
9
10
11class RateStat:
12 """A named statistic containing Rate objects for multiple time periods."""
13
14 def __init__(
15 self,
16 name: str,
17 description: str,
18 group: str,
19 periods: List[int],
20 ) -> None:
21 self._name = name
22 self._description = description
23 self._group = group
24 self._rates = [Rate(p) for p in sorted(periods)]
25 for r in self._rates:
26 r.set_rate_stat(self)
27
28 @property
29 def name(self) -> str:
30 return self._name
31
32 @property
33 def group_name(self) -> str:
34 return self._group
35
36 @property
37 def description(self) -> str:
38 return self._description
39
40 def get_periods(self) -> List[int]:
41 return [r.period for r in self._rates]
42
43 def get_rate(self, period: int) -> Optional[Rate]:
44 for r in self._rates:
45 if r.period == period:
46 return r
47 return None
48
49 def contains_rate(self, period: int) -> bool:
50 return self.get_rate(period) is not None
51
52 def add_data(self, value: int, event_duration: int = 0) -> None:
53 for r in self._rates:
54 r.add_data(value, event_duration)
55
56 def coalesce_stats(self) -> None:
57 for r in self._rates:
58 r.coalesce()
59
60 def get_lifetime_average_value(self) -> float:
61 if self._rates:
62 return self._rates[0].get_lifetime_average_value()
63 return 0.0
64
65 def get_lifetime_event_count(self) -> int:
66 if self._rates:
67 return self._rates[0].get_lifetime_event_count()
68 return 0