duck-preview/app/telemetry/telemetry_collector.py

"""Telemetry Collector — measures video pipeline performance metrics."""

from __future__ import annotations

import time
from collections import deque
from dataclasses import dataclass

import psutil
from PySide6.QtCore import QObject, QTimer, Signal
from PySide6.QtMultimedia import QVideoFrame

from app.config import TELEMETRY_UPDATE_INTERVAL_MS


@dataclass
class TelemetrySnapshot:
    """Immutable snapshot of current performance metrics."""

    fps: float
    frame_time_ms: float          # average inter-frame time in ms
    dropped_frames: int           # cumulative dropped frames detected
    cpu_percent: float            # overall CPU usage (0–100)
    memory_mb: float | None       # RSS memory usage in MB (optional)
    timestamp: float              # time.perf_counter() when snapshot was taken


class TelemetryCollector(QObject):
    """
    Frame subscriber that measures pipeline performance.

    Connect to FrameDispatcher:
        dispatcher.subscribe(collector.on_frame, drop_if_busy=False)

    Listen to metrics updates:
        collector.metrics_updated.connect(my_slot)
    """

    metrics_updated = Signal(object)  # emits TelemetrySnapshot

    def __init__(
        self,
        update_interval_ms: int = TELEMETRY_UPDATE_INTERVAL_MS,
        parent: QObject | None = None,
    ) -> None:
        super().__init__(parent)

        self._update_interval_ms = update_interval_ms

        # frame timing ring-buffer (last 120 samples)
        self._frame_times: deque[float] = deque(maxlen=120)
        self._last_frame_time: float = 0.0
        self._total_frames: int = 0
        self._dropped_frames: int = 0

        # FPS window — count frames in the last second
        self._fps_window: deque[float] = deque()  # timestamps of recent frames
        self._fps_window_size_s: float = 1.0

        # psutil process reference
        self._process = psutil.Process()

        # periodic snapshot timer
        self._timer = QTimer(self)
        self._timer.setInterval(update_interval_ms)
        self._timer.timeout.connect(self._emit_snapshot)
        self._timer.start()

        # latest snapshot (available synchronously)
        self._latest: TelemetrySnapshot = self._make_empty_snapshot()

    # ------------------------------------------------------------------
    # Frame subscriber callback
    # ------------------------------------------------------------------

    def on_frame(self, frame: QVideoFrame) -> None:
        """Called by FrameDispatcher for every frame. Must be fast."""
        now = time.perf_counter()

        # inter-frame time
        if self._last_frame_time > 0:
            delta = now - self._last_frame_time
            self._frame_times.append(delta)

            # drop detection: if delta > 2.5× the rolling average, count as drop
            if len(self._frame_times) >= 5:
                avg = sum(self._frame_times) / len(self._frame_times)
                if delta > avg * 2.5:
                    self._dropped_frames += 1

        self._last_frame_time = now
        self._total_frames += 1

        # FPS window
        self._fps_window.append(now)
        # prune old entries
        cutoff = now - self._fps_window_size_s
        while self._fps_window and self._fps_window[0] < cutoff:
            self._fps_window.popleft()

    # ------------------------------------------------------------------
    # Snapshot
    # ------------------------------------------------------------------

    def latest_snapshot(self) -> TelemetrySnapshot:
        """Return the most recently computed snapshot."""
        return self._latest

    def reset_counters(self) -> None:
        """Reset cumulative counters (e.g. after camera switch)."""
        self._frame_times.clear()
        self._fps_window.clear()
        self._last_frame_time = 0.0
        self._total_frames = 0
        self._dropped_frames = 0

    # ------------------------------------------------------------------
    # Internal
    # ------------------------------------------------------------------

    def _emit_snapshot(self) -> None:
        snapshot = self._compute_snapshot()
        self._latest = snapshot
        self.metrics_updated.emit(snapshot)

    def _compute_snapshot(self) -> TelemetrySnapshot:
        now = time.perf_counter()

        # FPS — prune stale entries before counting
        cutoff = now - self._fps_window_size_s
        while self._fps_window and self._fps_window[0] < cutoff:
            self._fps_window.popleft()
        fps = float(len(self._fps_window))  # frames in the last second

        # average frame time
        if self._frame_times:
            avg_frame_time_ms = (sum(self._frame_times) / len(self._frame_times)) * 1000.0
        else:
            avg_frame_time_ms = 0.0

        # CPU
        try:
            cpu = psutil.cpu_percent(interval=None)
        except Exception:
            cpu = 0.0

        # memory
        try:
            mem_mb = self._process.memory_info().rss / (1024 * 1024)
        except Exception:
            mem_mb = None

        return TelemetrySnapshot(
            fps=round(fps, 1),
            frame_time_ms=round(avg_frame_time_ms, 2),
            dropped_frames=self._dropped_frames,
            cpu_percent=round(cpu, 1),
            memory_mb=round(mem_mb, 1) if mem_mb is not None else None,
            timestamp=now,
        )

    @staticmethod
    def _make_empty_snapshot() -> TelemetrySnapshot:
        return TelemetrySnapshot(
            fps=0.0,
            frame_time_ms=0.0,
            dropped_frames=0,
            cpu_percent=0.0,
            memory_mb=None,
            timestamp=time.perf_counter(),
        )