118 lines
3.6 KiB
Python
118 lines
3.6 KiB
Python
"""
|
|
Statistics tracking for debugging.
|
|
"""
|
|
|
|
import time
|
|
import logging
|
|
from typing import List
|
|
from dataclasses import dataclass, field
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class StatsTracker:
|
|
"""
|
|
Track processing statistics.
|
|
|
|
Metrics:
|
|
- Frame count
|
|
- Detection count
|
|
- Processing FPS
|
|
- Detection latency
|
|
"""
|
|
|
|
total_frames: int = 0
|
|
total_detections: int = 0
|
|
|
|
_frame_times: List[float] = field(default_factory=list)
|
|
_detection_times: List[float] = field(default_factory=list)
|
|
_frame_start: float = 0
|
|
_detection_start: float = 0
|
|
|
|
_max_history: int = 100
|
|
|
|
def start_frame(self) -> None:
|
|
"""Mark start of frame processing."""
|
|
self._frame_start = time.time()
|
|
self._detection_start = time.time()
|
|
|
|
def end_detection(self, count: int = 0) -> None:
|
|
"""Mark end of detection, record time and count."""
|
|
detection_time = time.time() - self._detection_start
|
|
self._detection_times.append(detection_time)
|
|
|
|
if len(self._detection_times) > self._max_history:
|
|
self._detection_times.pop(0)
|
|
|
|
self.total_detections += count
|
|
|
|
def end_frame(self) -> None:
|
|
"""Mark end of frame processing."""
|
|
frame_time = time.time() - self._frame_start
|
|
self._frame_times.append(frame_time)
|
|
|
|
if len(self._frame_times) > self._max_history:
|
|
self._frame_times.pop(0)
|
|
|
|
self.total_frames += 1
|
|
|
|
@property
|
|
def average_fps(self) -> float:
|
|
"""Calculate average FPS."""
|
|
if not self._frame_times:
|
|
return 0.0
|
|
avg_time = sum(self._frame_times) / len(self._frame_times)
|
|
return 1.0 / avg_time if avg_time > 0 else 0.0
|
|
|
|
@property
|
|
def current_fps(self) -> float:
|
|
"""Get current FPS based on recent frames."""
|
|
if len(self._frame_times) < 2:
|
|
return 0.0
|
|
recent = self._frame_times[-10:]
|
|
avg_time = sum(recent) / len(recent)
|
|
return 1.0 / avg_time if avg_time > 0 else 0.0
|
|
|
|
@property
|
|
def average_detection_time(self) -> float:
|
|
"""Get average detection time in milliseconds."""
|
|
if not self._detection_times:
|
|
return 0.0
|
|
return (sum(self._detection_times) / len(self._detection_times)) * 1000
|
|
|
|
@property
|
|
def detections_per_frame(self) -> float:
|
|
"""Get average detections per frame."""
|
|
if self.total_frames == 0:
|
|
return 0.0
|
|
return self.total_detections / self.total_frames
|
|
|
|
def log(self, camera_name: str = "") -> None:
|
|
"""Log current statistics."""
|
|
prefix = f"[{camera_name}] " if camera_name else ""
|
|
logger.info(
|
|
f"{prefix}Stats: frames={self.total_frames}, "
|
|
f"detections={self.total_detections}, "
|
|
f"fps={self.current_fps:.1f}, "
|
|
f"det_time={self.average_detection_time:.1f}ms"
|
|
)
|
|
|
|
def reset(self) -> None:
|
|
"""Reset all statistics."""
|
|
self.total_frames = 0
|
|
self.total_detections = 0
|
|
self._frame_times.clear()
|
|
self._detection_times.clear()
|
|
|
|
def summary(self) -> dict:
|
|
"""Get statistics summary as dict."""
|
|
return {
|
|
'total_frames': self.total_frames,
|
|
'total_detections': self.total_detections,
|
|
'average_fps': round(self.average_fps, 2),
|
|
'current_fps': round(self.current_fps, 2),
|
|
'avg_detection_time_ms': round(self.average_detection_time, 2),
|
|
'detections_per_frame': round(self.detections_per_frame, 2),
|
|
}
|