"""Stats collector — records and queries time-series statistics. Replaces Java I2P's RateStat/FrequencyStat/StatManager with a simple in-memory collector. Stats are exposed via the REST API and rendered as Plotly.js charts. Ported from net.i2p.stat.RateStat / StatManager. """ from __future__ import annotations import time from collections import deque from dataclasses import dataclass, field @dataclass class Sample: """A single stat sample.""" value: float timestamp: float = field(default_factory=time.monotonic) class StatsCollector: """Collects named statistics with time-series values.""" def __init__(self, max_samples: int = 1000) -> None: self._max_samples = max_samples self._stats: dict[str, deque[Sample]] = {} def record(self, name: str, value: float) -> None: """Record a sample for a named stat.""" if name not in self._stats: self._stats[name] = deque(maxlen=self._max_samples) self._stats[name].append(Sample(value=value)) def get(self, name: str) -> list[Sample]: """Get all samples for a named stat.""" return list(self._stats.get(name, [])) def average(self, name: str) -> float: """Get the average value for a named stat.""" samples: deque[Sample] | list[Sample] = self._stats.get(name, []) if not samples: return 0.0 return sum(s.value for s in samples) / len(samples) def latest(self, name: str) -> float | None: """Get the most recent value for a named stat.""" samples = self._stats.get(name) if not samples: return None return samples[-1].value def summary(self) -> dict[str, dict]: """Get a summary of all stats.""" result = {} for name, samples in self._stats.items(): values = [s.value for s in samples] result[name] = { "count": len(values), "average": sum(values) / len(values) if values else 0.0, "latest": values[-1] if values else None, "min": min(values) if values else None, "max": max(values) if values else None, } return result