Move FpsMonitor from utils to video module

This commit relocates the FpsMonitor class from the utils module to the video module.
This commit is contained in:
SkalskiP 2023-10-07 00:24:20 +02:00
parent c5c37e4a6c
commit c5e1211c74
6 changed files with 64 additions and 44 deletions

View File

@ -1,3 +0,0 @@
## list_files_with_extensions
:::supervision.utils.fps.FpsMonitor

View File

@ -6,6 +6,10 @@
:::supervision.utils.video.VideoSink
## FPSMonitor
:::supervision.utils.video.FPSMonitor
## get_video_frames_generator
:::supervision.utils.video.get_video_frames_generator
@ -13,3 +17,5 @@
## process_video
:::supervision.utils.video.process_video

View File

@ -49,7 +49,6 @@ nav:
- Image: utils/image.md
- Notebook: utils/notebook.md
- File: utils/file.md
- FPS: utils/fps.md
- Changelog: changelog.md
theme:

View File

@ -44,7 +44,6 @@ from supervision.geometry.utils import get_polygon_center
from supervision.metrics.detection import ConfusionMatrix, MeanAveragePrecision
from supervision.tracker.byte_tracker.core import ByteTrack
from supervision.utils.file import list_files_with_extensions
from supervision.utils.fps import FpsMonitor
from supervision.utils.image import ImageSink, crop_image
from supervision.utils.notebook import plot_image, plot_images_grid
from supervision.utils.video import (
@ -52,4 +51,5 @@ from supervision.utils.video import (
VideoSink,
get_video_frames_generator,
process_video,
FPSMonitor
)

View File

@ -1,34 +1 @@
import time
from collections import deque
class FpsMonitor:
def __init__(self, sample_size: int = 30):
"""
Initialize class to measure Frame per second for latency benchmark.
Args:
sample_size (int): Maximum observation to measure latency benchmark
Examples:
```python
>>> import supervision as sv
>>> fps_monitor = sv.FpsMonitor()
>>> while True:
... (...)
... fps_monitor.tick()
... fps = fps_monitor()
```
"""
self.all_timestamps = deque(maxlen=sample_size)
def __call__(self) -> float:
if not self.all_timestamps:
return 0.0
taken_time = self.all_timestamps[-1] - self.all_timestamps[0]
return (len(self.all_timestamps)) / taken_time if taken_time != 0 else 0.0
def tick(self) -> None:
self.all_timestamps.append(time.monotonic())
def reset(self) -> None:
self.all_timestamps.clear()

View File

@ -1,5 +1,7 @@
from __future__ import annotations
import time
from collections import deque
from dataclasses import dataclass
from typing import Callable, Generator, Optional, Tuple
@ -71,13 +73,11 @@ class VideoSink:
```python
>>> import supervision as sv
>>> video_info = sv.VideoInfo.from_video_path(video_path='source_video.mp4')
>>> video_info = sv.VideoInfo.from_video_path('source.mp4')
>>> frames_generator = get_video_frames_generator('source.mp4')
>>> with sv.VideoSink(target_path='target_video.mp4',
... video_info=video_info,
codec='H264') as sink:
... for frame in get_video_frames_generator(source_path='source_video.mp4',
... stride=2):
>>> with sv.VideoSink(target_path='target.mp4', video_info=video_info) as sink:
... for frame in frames_generator:
... sink.write_frame(frame=frame)
```
"""
@ -202,3 +202,54 @@ def process_video(
):
result_frame = callback(frame, index)
sink.write_frame(frame=result_frame)
class FPSMonitor:
"""
A class for monitoring frames per second (FPS) to benchmark latency.
"""
def __init__(self, sample_size: int = 30):
"""
Args:
sample_size (int): The maximum number of observations for latency
benchmarking.
Examples:
```python
>>> import supervision as sv
>>> frames_generator = sv.get_video_frames_generator('source.mp4')
>>> fps_monitor = sv.FPSMonitor()
>>> for frame in frames_generator:
... # your processing code here
... fps_monitor.tick()
... fps = fps_monitor()
```
"""
self.all_timestamps = deque(maxlen=sample_size)
def __call__(self) -> float:
"""
Computes and returns the average FPS based on the stored time stamps.
Returns:
float: The average FPS. Returns 0.0 if no time stamps are stored.
"""
if not self.all_timestamps:
return 0.0
taken_time = self.all_timestamps[-1] - self.all_timestamps[0]
return (len(self.all_timestamps)) / taken_time if taken_time != 0 else 0.0
def tick(self) -> None:
"""
Adds a new time stamp to the deque for FPS calculation.
"""
self.all_timestamps.append(time.monotonic())
def reset(self) -> None:
"""
Clears all the time stamps from the deque.
"""
self.all_timestamps.clear()