feat(perf): Implement core PerformanceMonitor for telemetry collection
This commit is contained in:
58
performance_monitor.py
Normal file
58
performance_monitor.py
Normal file
@@ -0,0 +1,58 @@
|
||||
import time
|
||||
import psutil
|
||||
import threading
|
||||
|
||||
class PerformanceMonitor:
|
||||
def __init__(self):
|
||||
self._start_time = None
|
||||
self._last_frame_time = 0.0
|
||||
self._fps = 0.0
|
||||
self._frame_count = 0
|
||||
self._fps_last_time = time.time()
|
||||
self._process = psutil.Process()
|
||||
self._cpu_usage = 0.0
|
||||
self._cpu_lock = threading.Lock()
|
||||
|
||||
# Start CPU usage monitoring thread
|
||||
self._stop_event = threading.Event()
|
||||
self._cpu_thread = threading.Thread(target=self._monitor_cpu, daemon=True)
|
||||
self._cpu_thread.start()
|
||||
|
||||
def _monitor_cpu(self):
|
||||
while not self._stop_event.is_set():
|
||||
# psutil.cpu_percent is better than process.cpu_percent for real-time
|
||||
usage = self._process.cpu_percent(interval=1.0)
|
||||
with self._cpu_lock:
|
||||
self._cpu_usage = usage
|
||||
time.sleep(0.1)
|
||||
|
||||
def start_frame(self):
|
||||
self._start_time = time.time()
|
||||
|
||||
def end_frame(self):
|
||||
if self._start_time is None:
|
||||
return
|
||||
|
||||
end_time = time.time()
|
||||
self._last_frame_time = (end_time - self._start_time) * 1000.0
|
||||
self._frame_count += 1
|
||||
|
||||
elapsed_since_fps = end_time - self._fps_last_time
|
||||
if elapsed_since_fps >= 1.0:
|
||||
self._fps = self._frame_count / elapsed_since_fps
|
||||
self._frame_count = 0
|
||||
self._fps_last_time = end_time
|
||||
|
||||
def get_metrics(self):
|
||||
with self._cpu_lock:
|
||||
cpu_usage = self._cpu_usage
|
||||
|
||||
return {
|
||||
'last_frame_time_ms': self._last_frame_time,
|
||||
'fps': self._fps,
|
||||
'cpu_percent': cpu_usage
|
||||
}
|
||||
|
||||
def stop(self):
|
||||
self._stop_event.set()
|
||||
self._cpu_thread.join(timeout=2.0)
|
||||
@@ -8,7 +8,8 @@ dependencies = [
|
||||
"imgui-bundle",
|
||||
"google-genai",
|
||||
"anthropic",
|
||||
"tomli-w"
|
||||
"tomli-w",
|
||||
"psutil>=7.2.2",
|
||||
]
|
||||
|
||||
[dependency-groups]
|
||||
|
||||
27
tests/test_performance_monitor.py
Normal file
27
tests/test_performance_monitor.py
Normal file
@@ -0,0 +1,27 @@
|
||||
import unittest
|
||||
import time
|
||||
from performance_monitor import PerformanceMonitor
|
||||
|
||||
class TestPerformanceMonitor(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.monitor = PerformanceMonitor()
|
||||
|
||||
def test_frame_time_collection(self):
|
||||
# Simulate frames for 1.1 seconds to trigger FPS calculation
|
||||
start = time.time()
|
||||
while time.time() - start < 1.1:
|
||||
self.monitor.start_frame()
|
||||
time.sleep(0.01) # ~100 FPS
|
||||
self.monitor.end_frame()
|
||||
|
||||
metrics = self.monitor.get_metrics()
|
||||
self.assertAlmostEqual(metrics['last_frame_time_ms'], 10, delta=10)
|
||||
self.assertGreater(metrics['fps'], 0)
|
||||
|
||||
def test_cpu_usage_collection(self):
|
||||
metrics = self.monitor.get_metrics()
|
||||
self.assertIn('cpu_percent', metrics)
|
||||
self.assertIsInstance(metrics['cpu_percent'], float)
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user