Spaces:
Runtime error
Runtime error
File size: 4,000 Bytes
2eafbc4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 |
import time
from enum import Enum
from typing import Generator, Iterable, Optional, Tuple, Union
import numpy as np
from inference.core.interfaces.camera.entities import (
FrameID,
FrameTimestamp,
VideoFrame,
)
from inference.core.interfaces.camera.video_source import SourceProperties, VideoSource
MINIMAL_FPS = 0.01
class FPSLimiterStrategy(Enum):
DROP = "drop"
WAIT = "wait"
def get_video_frames_generator(
video: Union[VideoSource, str, int],
max_fps: Optional[Union[float, int]] = None,
limiter_strategy: Optional[FPSLimiterStrategy] = None,
) -> Generator[VideoFrame, None, None]:
"""
Util function to create a frames generator from `VideoSource` with possibility to
limit FPS of consumed frames and dictate what to do if frames are produced to fast.
Args:
video (Union[VideoSource, str, int]): Either instance of VideoSource or video reference accepted
by VideoSource.init(...)
max_fps (Optional[Union[float, int]]): value of maximum FPS rate of generated frames - can be used to limit
generation frequency
limiter_strategy (Optional[FPSLimiterStrategy]): strategy used to deal with frames decoding exceeding
limit of `max_fps`. By default - for files, in the interest of processing all frames -
generation will be awaited, for streams - frames will be dropped on the floor.
Returns: generator of `VideoFrame`
Example:
```python
for frame in get_video_frames_generator(
video="./some.mp4",
max_fps=50,
):
pass
```
"""
if issubclass(type(video), str) or issubclass(type(video), int):
video = VideoSource.init(
video_reference=video,
)
video.start()
if max_fps is None:
yield from video
return None
limiter_strategy = resolve_limiter_strategy(
explicitly_defined_strategy=limiter_strategy,
source_properties=video.describe_source().source_properties,
)
yield from limit_frame_rate(
frames_generator=video, max_fps=max_fps, strategy=limiter_strategy
)
def resolve_limiter_strategy(
explicitly_defined_strategy: Optional[FPSLimiterStrategy],
source_properties: Optional[SourceProperties],
) -> FPSLimiterStrategy:
if explicitly_defined_strategy is not None:
return explicitly_defined_strategy
limiter_strategy = FPSLimiterStrategy.DROP
if source_properties is not None and source_properties.is_file:
limiter_strategy = FPSLimiterStrategy.WAIT
return limiter_strategy
def limit_frame_rate(
frames_generator: Iterable[Tuple[FrameTimestamp, FrameID, np.ndarray]],
max_fps: Union[float, int],
strategy: FPSLimiterStrategy,
) -> Generator[Tuple[FrameTimestamp, FrameID, np.ndarray], None, None]:
rate_limiter = RateLimiter(desired_fps=max_fps)
for frame_data in frames_generator:
delay = rate_limiter.estimate_next_action_delay()
if delay <= 0.0:
rate_limiter.tick()
yield frame_data
continue
if strategy is FPSLimiterStrategy.WAIT:
time.sleep(delay)
rate_limiter.tick()
yield frame_data
class RateLimiter:
"""
Implements rate upper-bound rate limiting by ensuring estimate_next_tick_delay()
to be at min 1 / desired_fps, not letting the client obeying outcomes to exceed
assumed rate.
"""
def __init__(self, desired_fps: Union[float, int]):
self._desired_fps = max(desired_fps, MINIMAL_FPS)
self._last_tick: Optional[float] = None
def tick(self) -> None:
self._last_tick = time.monotonic()
def estimate_next_action_delay(self) -> float:
if self._last_tick is None:
return 0.0
desired_delay = 1 / self._desired_fps
time_since_last_tick = time.monotonic() - self._last_tick
return max(desired_delay - time_since_last_tick, 0.0)
|