feat(perf): Expand instrumentation with context manager and extended metrics

This commit is contained in:
2026-05-06 14:30:22 -04:00
parent 022c39888c
commit 23c1e21661
5 changed files with 240 additions and 136 deletions
+30
View File
@@ -62,6 +62,18 @@ from collections import deque
_instance: Optional[PerformanceMonitor] = None
class PerformanceScope:
"""Helper class for PerformanceMonitor.scope() context manager."""
def __init__(self, monitor: PerformanceMonitor, name: str) -> None:
self.monitor = monitor
self.name = name
def __enter__(self) -> PerformanceScope:
self.monitor.start_component(self.name)
return self
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
self.monitor.end_component(self.name)
def get_monitor() -> PerformanceMonitor:
global _instance
if _instance is None:
@@ -90,6 +102,9 @@ class PerformanceMonitor:
self._component_starts: dict[str, float] = {}
self._component_timings: dict[str, float] = {}
self._component_counts: dict[str, int] = {}
self._component_max: dict[str, float] = {}
self._component_min: dict[str, float] = {}
# Rolling history and running sums for O(1) average calculation
# deques are thread-safe for appends and pops.
@@ -192,6 +207,11 @@ class PerformanceMonitor:
elapsed = (now - start) * 1000
with self._lock:
self._component_timings[name] = elapsed
self._component_counts[name] = self._component_counts.get(name, 0) + 1
if name not in self._component_max or elapsed > self._component_max[name]:
self._component_max[name] = elapsed
if name not in self._component_min or elapsed < self._component_min[name]:
self._component_min[name] = elapsed
self._add_to_history(f'comp_{name}', elapsed)
def get_metrics(self) -> dict[str, float]:
@@ -203,6 +223,9 @@ class PerformanceMonitor:
ilag = self._input_lag_ms
last_calc_fps = self._last_calculated_fps
timings_snapshot = dict(self._component_timings)
counts_snapshot = dict(self._component_counts)
max_snapshot = dict(self._component_max)
min_snapshot = dict(self._component_min)
metrics = {
'fps': fps,
@@ -217,6 +240,9 @@ class PerformanceMonitor:
for name, elapsed in timings_snapshot.items():
metrics[f'time_{name}_ms'] = elapsed
metrics[f'time_{name}_ms_avg'] = self._get_avg(f'comp_{name}')
metrics[f'count_{name}'] = float(counts_snapshot.get(name, 0))
metrics[f'max_{name}_ms'] = max_snapshot.get(name, 0.0)
metrics[f'min_{name}_ms'] = min_snapshot.get(name, 0.0)
return metrics
def get_history(self, key: str) -> List[float]:
@@ -228,6 +254,10 @@ class PerformanceMonitor:
return list(self._history[f'comp_{key}'])
return []
def scope(self, name: str) -> PerformanceScope:
"""Returns a context manager for timing a component."""
return PerformanceScope(self, name)
def stop(self) -> None:
self._stop_event.set()
if self._cpu_thread.is_alive():