distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Stats collector — records and queries time-series statistics.
2
3Replaces Java I2P's RateStat/FrequencyStat/StatManager with a
4simple in-memory collector. Stats are exposed via the REST API
5and rendered as Plotly.js charts.
6
7Ported from net.i2p.stat.RateStat / StatManager.
8"""
9
10from __future__ import annotations
11
12import time
13from collections import deque
14from dataclasses import dataclass, field
15
16
17@dataclass
18class Sample:
19 """A single stat sample."""
20 value: float
21 timestamp: float = field(default_factory=time.monotonic)
22
23
24class StatsCollector:
25 """Collects named statistics with time-series values."""
26
27 def __init__(self, max_samples: int = 1000) -> None:
28 self._max_samples = max_samples
29 self._stats: dict[str, deque[Sample]] = {}
30
31 def record(self, name: str, value: float) -> None:
32 """Record a sample for a named stat."""
33 if name not in self._stats:
34 self._stats[name] = deque(maxlen=self._max_samples)
35 self._stats[name].append(Sample(value=value))
36
37 def get(self, name: str) -> list[Sample]:
38 """Get all samples for a named stat."""
39 return list(self._stats.get(name, []))
40
41 def average(self, name: str) -> float:
42 """Get the average value for a named stat."""
43 samples: deque[Sample] | list[Sample] = self._stats.get(name, [])
44 if not samples:
45 return 0.0
46 return sum(s.value for s in samples) / len(samples)
47
48 def latest(self, name: str) -> float | None:
49 """Get the most recent value for a named stat."""
50 samples = self._stats.get(name)
51 if not samples:
52 return None
53 return samples[-1].value
54
55 def summary(self) -> dict[str, dict]:
56 """Get a summary of all stats."""
57 result = {}
58 for name, samples in self._stats.items():
59 values = [s.value for s in samples]
60 result[name] = {
61 "count": len(values),
62 "average": sum(values) / len(values) if values else 0.0,
63 "latest": values[-1] if values else None,
64 "min": min(values) if values else None,
65 "max": max(values) if values else None,
66 }
67 return result