40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from collections import deque
|
|
|
|
from PySide6.QtCore import QObject
|
|
from PySide6.QtMultimedia import QVideoFrame
|
|
|
|
|
|
class TelemetryCollector(QObject):
|
|
def __init__(self, parent: QObject | None = None) -> None:
|
|
super().__init__(parent)
|
|
self._timestamps: deque[float] = deque(maxlen=500)
|
|
|
|
def on_frame(self, frame: QVideoFrame) -> None:
|
|
self._timestamps.append(time.perf_counter())
|
|
|
|
def metrics(self) -> dict[str, float | int]:
|
|
if not self._timestamps:
|
|
return {"fps": 0, "frame_time_ms": 0.0, "frame_count": 0}
|
|
|
|
now = time.perf_counter()
|
|
cutoff = now - 1.0
|
|
recent = [t for t in self._timestamps if t >= cutoff]
|
|
fps = len(recent)
|
|
|
|
frame_time_ms = 0.0
|
|
if len(self._timestamps) >= 2:
|
|
diffs = [
|
|
self._timestamps[i] - self._timestamps[i - 1]
|
|
for i in range(1, len(self._timestamps))
|
|
]
|
|
frame_time_ms = (sum(diffs) / len(diffs)) * 1000 if diffs else 0.0
|
|
|
|
return {
|
|
"fps": fps,
|
|
"frame_time_ms": round(frame_time_ms, 2),
|
|
"frame_count": len(self._timestamps),
|
|
}
|