"""In-memory request statistics. All counters are safe to mutate without locks because asyncio is single-threaded cooperative multitasking. """ from __future__ import annotations import time from collections import deque from dataclasses import dataclass, field _start_time: float = time.monotonic() @dataclass class EndpointStats: """Per-endpoint request counters and latency tracking.""" requests: int = 0 successes: int = 0 failures: int = 0 total_ms: float = 0.0 _latencies: deque[float] = field(default_factory=lambda: deque(maxlen=1000)) def record_success(self, elapsed_ms: float) -> None: """Record a successful request.""" self.successes += 1 self.total_ms += elapsed_ms self._latencies.append(elapsed_ms) def record_failure(self, elapsed_ms: float) -> None: """Record a failed request.""" self.failures += 1 self.total_ms += elapsed_ms self._latencies.append(elapsed_ms) @property def avg_ms(self) -> float: """Average latency in milliseconds.""" total = self.successes + self.failures return self.total_ms / total if total > 0 else 0.0 @property def p95_ms(self) -> float: """95th percentile latency in milliseconds.""" if not self._latencies: return 0.0 sorted_lat = sorted(self._latencies) idx = int(len(sorted_lat) * 0.95) return sorted_lat[min(idx, len(sorted_lat) - 1)] @property def success_rate(self) -> float: """Success rate as a percentage.""" total = self.successes + self.failures return (self.successes / total * 100) if total > 0 else 0.0 render = EndpointStats() screenshot = EndpointStats() def uptime_seconds() -> float: """Seconds since the process started.""" return time.monotonic() - _start_time def format_uptime() -> str: """Human-readable uptime string.""" secs = int(uptime_seconds()) days, secs = divmod(secs, 86400) hours, secs = divmod(secs, 3600) mins, secs = divmod(secs, 60) parts = [] if days: parts.append(f"{days}d") if hours: parts.append(f"{hours}h") if mins: parts.append(f"{mins}m") parts.append(f"{secs}s") return " ".join(parts)