235 lines
6.9 KiB
Python
235 lines
6.9 KiB
Python
"""
|
|
Performance Monitor - Real-time FPS, frame time, and CPU usage tracking.
|
|
|
|
This module provides the PerformanceMonitor singleton class for tracking
|
|
application performance metrics with efficient O(1) moving averages.
|
|
|
|
Key Features:
|
|
- FPS and frame time tracking with rolling history
|
|
- CPU percentage monitoring via background thread
|
|
- Per-component timing with start_component() / end_component()
|
|
- Efficient moving average using deque + running sum
|
|
- Thread-safe metric collection
|
|
|
|
Usage:
|
|
perf = get_monitor()
|
|
perf.enabled = True
|
|
|
|
# In render loop:
|
|
perf.start_frame()
|
|
perf.start_component('panel_a')
|
|
# ... render panel A ...
|
|
perf.end_component('panel_a')
|
|
perf.end_frame()
|
|
|
|
# Get metrics:
|
|
metrics = perf.get_metrics()
|
|
fps = metrics['fps']
|
|
avg_frame_time = metrics['frame_time_ms_avg']
|
|
|
|
Metrics Available:
|
|
- fps: Instantaneous frames per second
|
|
- fps_avg: Rolling average FPS
|
|
- last_frame_time_ms: Last frame duration in milliseconds
|
|
- frame_time_ms_avg: Rolling average frame time
|
|
- cpu_percent: Current CPU usage
|
|
- cpu_percent_avg: Rolling average CPU usage
|
|
- input_lag_ms: Input latency estimate
|
|
- time_<component>_ms: Per-component timing
|
|
- time_<component>_ms_avg: Per-component rolling average
|
|
|
|
Thread Safety:
|
|
- All public methods are thread-safe
|
|
- Uses threading.Lock for state mutations
|
|
- Background CPU thread polls every 1 second
|
|
|
|
Configuration:
|
|
- history_size: Number of samples for rolling averages (default: 300)
|
|
- sample_interval: Minimum time between history samples (default: 100ms)
|
|
|
|
Integration:
|
|
- Instantiated as singleton via get_monitor()
|
|
- Used by gui_2.py for Diagnostics Panel
|
|
- Exposed via Hook API at /api/performance
|
|
"""
|
|
from __future__ import annotations
|
|
import time
|
|
import psutil
|
|
import threading
|
|
from typing import Any, Optional, Callable, Dict, List
|
|
from collections import deque
|
|
|
|
_instance: Optional[PerformanceMonitor] = None
|
|
|
|
|
|
def get_monitor() -> PerformanceMonitor:
|
|
global _instance
|
|
if _instance is None:
|
|
_instance = PerformanceMonitor()
|
|
return _instance
|
|
|
|
class PerformanceMonitor:
|
|
"""
|
|
Tracks application performance metrics like FPS, frame time, and CPU usage.
|
|
Supports thread-safe tracking for individual components with efficient moving averages.
|
|
"""
|
|
def __init__(self, history_size: int = 300) -> None:
|
|
self.enabled: bool = False
|
|
self.history_size = history_size
|
|
self._lock = threading.Lock()
|
|
|
|
self._start_time: Optional[float] = None
|
|
self._last_frame_start_time: float = 0.0
|
|
self._last_frame_time: float = 0.0
|
|
self._fps: float = 0.0
|
|
self._last_calculated_fps: float = 0.0
|
|
self._frame_count: int = 0
|
|
self._fps_timer: float = 0.0
|
|
self._cpu_percent: float = 0.0
|
|
self._input_lag_ms: float = 0.0
|
|
|
|
self._component_starts: dict[str, float] = {}
|
|
self._component_timings: dict[str, float] = {}
|
|
|
|
# Rolling history and running sums for O(1) average calculation
|
|
# deques are thread-safe for appends and pops.
|
|
self._history: Dict[str, deque[float]] = {}
|
|
self._history_sums: Dict[str, float] = {}
|
|
|
|
# For slowing down graph updates
|
|
self._last_sample_time = 0.0
|
|
self._sample_interval = 0.1 # 100ms
|
|
|
|
# Thread for CPU monitoring
|
|
self._stop_event = threading.Event()
|
|
self._cpu_thread = threading.Thread(target=self._monitor_cpu, daemon=True)
|
|
self._cpu_thread.start()
|
|
|
|
def _monitor_cpu(self) -> None:
|
|
while not self._stop_event.is_set():
|
|
try:
|
|
val = psutil.cpu_percent(interval=None)
|
|
with self._lock:
|
|
self._cpu_percent = val
|
|
except Exception:
|
|
pass
|
|
time.sleep(1.0)
|
|
|
|
def _add_to_history(self, key: str, value: float) -> None:
|
|
"""Thread-safe O(1) history update."""
|
|
with self._lock:
|
|
if key not in self._history:
|
|
self._history[key] = deque(maxlen=self.history_size)
|
|
self._history_sums[key] = 0.0
|
|
|
|
h = self._history[key]
|
|
if len(h) == self.history_size:
|
|
removed = h[0] # peek left
|
|
self._history_sums[key] -= removed
|
|
|
|
self._history_sums[key] += value
|
|
h.append(value)
|
|
|
|
def _get_avg(self, key: str) -> float:
|
|
"""Thread-safe O(1) average retrieval."""
|
|
with self._lock:
|
|
h = self._history.get(key)
|
|
if not h or len(h) == 0:
|
|
return 0.0
|
|
return self._history_sums[key] / len(h)
|
|
|
|
def start_frame(self) -> None:
|
|
now = time.time()
|
|
with self._lock:
|
|
if self._last_frame_start_time > 0:
|
|
dt = now - self._last_frame_start_time
|
|
if dt > 0:
|
|
self._fps = 1.0 / dt
|
|
self._last_frame_start_time = now
|
|
self._start_time = now
|
|
self._frame_count += 1
|
|
|
|
def end_frame(self) -> None:
|
|
if self._start_time is None:
|
|
return
|
|
now = time.time()
|
|
elapsed = now - self._start_time
|
|
frame_time_ms = elapsed * 1000
|
|
|
|
with self._lock:
|
|
self._last_frame_time = frame_time_ms
|
|
cpu = self._cpu_percent
|
|
ilag = self._input_lag_ms
|
|
fps = self._fps
|
|
|
|
# Slow down history sampling for core metrics
|
|
if now - self._last_sample_time >= self._sample_interval:
|
|
self._last_sample_time = now
|
|
self._add_to_history('frame_time_ms', frame_time_ms)
|
|
self._add_to_history('cpu_percent', cpu)
|
|
self._add_to_history('input_lag_ms', ilag)
|
|
self._add_to_history('fps', fps)
|
|
|
|
self._fps_timer += elapsed
|
|
if self._fps_timer >= 1.0:
|
|
with self._lock:
|
|
self._last_calculated_fps = self._frame_count / self._fps_timer
|
|
self._frame_count = 0
|
|
self._fps_timer = 0.0
|
|
|
|
def start_component(self, name: str) -> None:
|
|
if not self.enabled: return
|
|
now = time.time()
|
|
with self._lock:
|
|
self._component_starts[name] = now
|
|
|
|
def end_component(self, name: str) -> None:
|
|
if not self.enabled: return
|
|
now = time.time()
|
|
with self._lock:
|
|
start = self._component_starts.pop(name, None)
|
|
if start is not None:
|
|
elapsed = (now - start) * 1000
|
|
with self._lock:
|
|
self._component_timings[name] = elapsed
|
|
self._add_to_history(f'comp_{name}', elapsed)
|
|
|
|
def get_metrics(self) -> dict[str, float]:
|
|
"""Returns current metrics and their moving averages. Thread-safe."""
|
|
with self._lock:
|
|
fps = self._fps
|
|
last_ft = self._last_frame_time
|
|
cpu = self._cpu_percent
|
|
ilag = self._input_lag_ms
|
|
last_calc_fps = self._last_calculated_fps
|
|
timings_snapshot = dict(self._component_timings)
|
|
|
|
metrics = {
|
|
'fps': fps,
|
|
'fps_avg': self._get_avg('fps'),
|
|
'last_frame_time_ms': last_ft,
|
|
'frame_time_ms_avg': self._get_avg('frame_time_ms'),
|
|
'cpu_percent': cpu,
|
|
'cpu_percent_avg': self._get_avg('cpu_percent'),
|
|
'input_lag_ms': ilag,
|
|
'input_lag_ms_avg': self._get_avg('input_lag_ms')
|
|
}
|
|
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}')
|
|
return metrics
|
|
|
|
def get_history(self, key: str) -> List[float]:
|
|
"""Returns a snapshot of the full history buffer for a specific metric key."""
|
|
with self._lock:
|
|
if key in self._history:
|
|
return list(self._history[key])
|
|
if f'comp_{key}' in self._history:
|
|
return list(self._history[f'comp_{key}'])
|
|
return []
|
|
|
|
def stop(self) -> None:
|
|
self._stop_event.set()
|
|
if self._cpu_thread.is_alive():
|
|
self._cpu_thread.join(timeout=2.0)
|