import time from dataclasses import dataclass from datetime import datetime from enum import Enum from queue import Empty, Queue from threading import Event, Lock, Thread from typing import Any, Callable, List, Optional, Protocol, Union import cv2 import supervision as sv from inference.core import logger from inference.core.env import ( DEFAULT_ADAPTIVE_MODE_READER_PACE_TOLERANCE, DEFAULT_ADAPTIVE_MODE_STREAM_PACE_TOLERANCE, DEFAULT_BUFFER_SIZE, DEFAULT_MAXIMUM_ADAPTIVE_FRAMES_DROPPED_IN_ROW, DEFAULT_MINIMUM_ADAPTIVE_MODE_SAMPLES, ) from inference.core.interfaces.camera.entities import ( StatusUpdate, UpdateSeverity, VideoFrame, ) from inference.core.interfaces.camera.exceptions import ( EndOfStreamError, SourceConnectionError, StreamOperationNotAllowedError, ) VIDEO_SOURCE_CONTEXT = "video_source" VIDEO_CONSUMER_CONTEXT = "video_consumer" SOURCE_STATE_UPDATE_EVENT = "SOURCE_STATE_UPDATE" SOURCE_ERROR_EVENT = "SOURCE_ERROR" FRAME_CAPTURED_EVENT = "FRAME_CAPTURED" FRAME_DROPPED_EVENT = "FRAME_DROPPED" FRAME_CONSUMED_EVENT = "FRAME_CONSUMED" VIDEO_CONSUMPTION_STARTED_EVENT = "VIDEO_CONSUMPTION_STARTED" VIDEO_CONSUMPTION_FINISHED_EVENT = "VIDEO_CONSUMPTION_FINISHED" class StreamState(Enum): NOT_STARTED = "NOT_STARTED" INITIALISING = "INITIALISING" RESTARTING = "RESTARTING" RUNNING = "RUNNING" PAUSED = "PAUSED" MUTED = "MUTED" TERMINATING = "TERMINATING" ENDED = "ENDED" ERROR = "ERROR" START_ELIGIBLE_STATES = { StreamState.NOT_STARTED, StreamState.RESTARTING, StreamState.ENDED, } PAUSE_ELIGIBLE_STATES = {StreamState.RUNNING} MUTE_ELIGIBLE_STATES = {StreamState.RUNNING} RESUME_ELIGIBLE_STATES = {StreamState.PAUSED, StreamState.MUTED} TERMINATE_ELIGIBLE_STATES = { StreamState.MUTED, StreamState.RUNNING, StreamState.PAUSED, StreamState.RESTARTING, StreamState.ENDED, StreamState.ERROR, } RESTART_ELIGIBLE_STATES = { StreamState.MUTED, StreamState.RUNNING, StreamState.PAUSED, StreamState.ENDED, StreamState.ERROR, } class BufferFillingStrategy(Enum): WAIT = "WAIT" DROP_OLDEST = "DROP_OLDEST" ADAPTIVE_DROP_OLDEST = "ADAPTIVE_DROP_OLDEST" DROP_LATEST = "DROP_LATEST" ADAPTIVE_DROP_LATEST = "ADAPTIVE_DROP_LATEST" ADAPTIVE_STRATEGIES = { BufferFillingStrategy.ADAPTIVE_DROP_LATEST, BufferFillingStrategy.ADAPTIVE_DROP_OLDEST, } DROP_OLDEST_STRATEGIES = { BufferFillingStrategy.DROP_OLDEST, BufferFillingStrategy.ADAPTIVE_DROP_OLDEST, } class BufferConsumptionStrategy(Enum): LAZY = "LAZY" EAGER = "EAGER" @dataclass(frozen=True) class SourceProperties: width: int height: int total_frames: int is_file: bool fps: float @dataclass(frozen=True) class SourceMetadata: source_properties: Optional[SourceProperties] source_reference: str buffer_size: int state: StreamState buffer_filling_strategy: Optional[BufferFillingStrategy] buffer_consumption_strategy: Optional[BufferConsumptionStrategy] class VideoSourceMethod(Protocol): def __call__(self, video_source: "VideoSource", *args, **kwargs) -> None: ... def lock_state_transition( method: VideoSourceMethod, ) -> Callable[["VideoSource"], None]: def locked_executor(video_source: "VideoSource", *args, **kwargs) -> None: with video_source._state_change_lock: return method(video_source, *args, **kwargs) return locked_executor class VideoSource: @classmethod def init( cls, video_reference: Union[str, int], buffer_size: int = DEFAULT_BUFFER_SIZE, status_update_handlers: Optional[List[Callable[[StatusUpdate], None]]] = None, buffer_filling_strategy: Optional[BufferFillingStrategy] = None, buffer_consumption_strategy: Optional[BufferConsumptionStrategy] = None, adaptive_mode_stream_pace_tolerance: float = DEFAULT_ADAPTIVE_MODE_STREAM_PACE_TOLERANCE, adaptive_mode_reader_pace_tolerance: float = DEFAULT_ADAPTIVE_MODE_READER_PACE_TOLERANCE, minimum_adaptive_mode_samples: int = DEFAULT_MINIMUM_ADAPTIVE_MODE_SAMPLES, maximum_adaptive_frames_dropped_in_row: int = DEFAULT_MAXIMUM_ADAPTIVE_FRAMES_DROPPED_IN_ROW, ): """ This class is meant to represent abstraction over video sources - both video files and on-line streams that are possible to be consumed and used by other components of `inference` library. Before digging into details of the class behaviour, it is advised to familiarise with the following concepts and implementation assumptions: 1. Video file can be accessed from local (or remote) storage by the consumer in a pace dictated by its processing capabilities. If processing is faster than the frame rate of video, operations may be executed in a time shorter than the time of video playback. In the opposite case - consumer may freely decode and process frames in its own pace, without risk for failures due to temporal dependencies of processing - this is classical offline processing example. 2. Video streams, on the other hand, usually need to be consumed in a pace near to their frame-rate - in other words - this is on-line processing example. Consumer being faster than incoming stream frames cannot utilise its resources to the full extent as not-yet-delivered data would be needed. Slow consumer, however, may not be able to process everything on time and to keep up with the pace of stream - some frames would need to be dropped. Otherwise - over time, consumer could go out of sync with the stream causing decoding failures or unpredictable behavior. To fit those two types of video sources, `VideoSource` introduces the concept of buffered decoding of video stream (like at the YouTube - player buffers some frames that are soon to be displayed). The way on how buffer is filled and consumed dictates the behavior of `VideoSource`. Starting from `BufferFillingStrategy` - we have 3 basic options: * WAIT: in case of slow video consumption, when buffer is full - `VideoSource` will wait for the empty spot in buffer before next frame will be processed - this is suitable in cases when we want to ensure EACH FRAME of the video to be processed * DROP_OLDEST: when buffer is full, the frame that sits there for the longest time will be dropped - this is suitable for cases when we want to process the most recent frames possible * DROP_LATEST: when buffer is full, the newly decoded frame is dropped - useful in cases when it is expected to have processing performance drops, but we would like to consume portions of video that are locally smooth - but this is probably the least common use-case. On top of that - there are two ADAPTIVE strategies: ADAPTIVE_DROP_OLDEST and ADAPTIVE_DROP_LATEST, which are equivalent to DROP_OLDEST and DROP_LATEST with adaptive decoding feature enabled. The notion of that mode will be described later. Naturally, decoded frames must also be consumed. `VideoSource` provides a handy interface for reading a video source frames by a SINGLE consumer. Consumption strategy can also be dictated via `BufferConsumptionStrategy`: * LAZY - consume all the frames from decoding buffer one-by-one * EAGER - at each readout - take all frames already buffered, drop all of them apart from the most recent In consequence - there are various combinations of `BufferFillingStrategy` and `BufferConsumptionStrategy`. The most popular would be: * `BufferFillingStrategy.WAIT` and `BufferConsumptionStrategy.LAZY` - to always decode and process each and every frame of the source (useful while processing video files - and default behaviour enforced by `inference` if there is no explicit configuration) * `BufferFillingStrategy.DROP_OLDEST` and `BufferConsumptionStrategy.EAGER` - to always process the most recent frames of source (useful while processing video streams when low latency [real-time experience] is required - ADAPTIVE version of this is default for streams) ADAPTIVE strategies were introduced to handle corner-cases, when consumer hardware is not capable to consume video stream and process frames at the same time (for instance - Nvidia Jetson devices running processing against hi-res streams with high FPS ratio). It acts with buffer in nearly the same way as `DROP_OLDEST` and `DROP_LATEST` strategies, but there are two more conditions that may influence frame drop: * announced rate of source - which in fact dictate the pace of frames grabbing from incoming stream that MUST be met by consumer to avoid strange decoding issues causing decoder to fail - if the pace of frame grabbing deviates too much - decoding will be postponed, and frames dropped to grab next ones sooner * consumption rate - in resource constraints environment, not only decoding is problematic from the performance perspective - but also heavy processing. If consumer is not quick enough - allocating more useful resources for decoding frames that may never be processed is a waste. That's why - if decoding happens more frequently than consumption of frame - ADAPTIVE mode causes decoding to be done in a slower pace and more frames are just grabbed and dropped on the floor. ADAPTIVE mode increases latency slightly, but may be the only way to operate in some cases. Behaviour of adaptive mode, including the maximum acceptable deviations of frames grabbing pace from source, reader pace and maximum number of consecutive frames dropped in ADAPTIVE mode are configurable by clients, with reasonable defaults being set. `VideoSource` emits events regarding its activity - which can be intercepted by custom handlers. Take into account that they are always executed in context of thread invoking them (and should be fast to complete, otherwise may block the flow of stream consumption). All errors raised will be emitted as logger warnings only. `VideoSource` implementation is naturally multithreading, with different thread decoding video and different one consuming it and manipulating source state. Implementation of user interface is thread-safe, although stream it is meant to be consumed by a single thread only. ENV variables involved: * VIDEO_SOURCE_BUFFER_SIZE - default: 64 * VIDEO_SOURCE_ADAPTIVE_MODE_STREAM_PACE_TOLERANCE - default: 0.1 * VIDEO_SOURCE_ADAPTIVE_MODE_READER_PACE_TOLERANCE - default: 5.0 * VIDEO_SOURCE_MINIMUM_ADAPTIVE_MODE_SAMPLES - default: 10 * VIDEO_SOURCE_MAXIMUM_ADAPTIVE_FRAMES_DROPPED_IN_ROW - default: 16 As an `inference` user, please use .init() method instead of constructor to instantiate objects. Args: video_reference (Union[str, int]): Either str with file or stream reference, or int representing device ID buffer_size (int): size of decoding buffer status_update_handlers (Optional[List[Callable[[StatusUpdate], None]]]): List of handlers for status updates buffer_filling_strategy (Optional[BufferFillingStrategy]): Settings for buffer filling strategy - if not given - automatic choice regarding source type will be applied buffer_consumption_strategy (Optional[BufferConsumptionStrategy]): Settings for buffer consumption strategy, if not given - automatic choice regarding source type will be applied adaptive_mode_stream_pace_tolerance (float): Maximum deviation between frames grabbing pace and stream pace that will not trigger adaptive mode frame drop adaptive_mode_reader_pace_tolerance (float): Maximum deviation between decoding pace and stream consumption pace that will not trigger adaptive mode frame drop minimum_adaptive_mode_samples (int): Minimal number of frames to be used to establish actual pace of processing, before adaptive mode can drop any frame maximum_adaptive_frames_dropped_in_row (int): Maximum number of frames dropped in row due to application of adaptive strategy Returns: Instance of `VideoSource` class """ frames_buffer = Queue(maxsize=buffer_size) if status_update_handlers is None: status_update_handlers = [] video_consumer = VideoConsumer.init( buffer_filling_strategy=buffer_filling_strategy, adaptive_mode_stream_pace_tolerance=adaptive_mode_stream_pace_tolerance, adaptive_mode_reader_pace_tolerance=adaptive_mode_reader_pace_tolerance, minimum_adaptive_mode_samples=minimum_adaptive_mode_samples, maximum_adaptive_frames_dropped_in_row=maximum_adaptive_frames_dropped_in_row, status_update_handlers=status_update_handlers, ) return cls( stream_reference=video_reference, frames_buffer=frames_buffer, status_update_handlers=status_update_handlers, buffer_consumption_strategy=buffer_consumption_strategy, video_consumer=video_consumer, ) def __init__( self, stream_reference: Union[str, int], frames_buffer: Queue, status_update_handlers: List[Callable[[StatusUpdate], None]], buffer_consumption_strategy: Optional[BufferConsumptionStrategy], video_consumer: "VideoConsumer", ): self._stream_reference = stream_reference self._video: Optional[cv2.VideoCapture] = None self._source_properties: Optional[SourceProperties] = None self._frames_buffer = frames_buffer self._status_update_handlers = status_update_handlers self._buffer_consumption_strategy = buffer_consumption_strategy self._video_consumer = video_consumer self._state = StreamState.NOT_STARTED self._playback_allowed = Event() self._frames_buffering_allowed = True self._stream_consumption_thread: Optional[Thread] = None self._state_change_lock = Lock() @lock_state_transition def restart(self, wait_on_frames_consumption: bool = True) -> None: """ Method to restart source consumption. Eligible to be used in states: [MUTED, RUNNING, PAUSED, ENDED, ERROR]. End state: * INITIALISING - that should change into RUNNING once first frame is ready to be grabbed * ERROR - if it was not possible to connect with source Thread safe - only one transition of states possible at the time. Args: wait_on_frames_consumption (bool): Flag telling if all frames from buffer must be consumed before completion of this operation. Returns: None Throws: * StreamOperationNotAllowedError: if executed in context of incorrect state of the source * SourceConnectionError: if source cannot be connected """ if self._state not in RESTART_ELIGIBLE_STATES: raise StreamOperationNotAllowedError( f"Could not RESTART stream in state: {self._state}" ) self._restart(wait_on_frames_consumption=wait_on_frames_consumption) @lock_state_transition def start(self) -> None: """ Method to be used to start source consumption. Eligible to be used in states: [NOT_STARTED, ENDED, (RESTARTING - which is internal state only)] End state: * INITIALISING - that should change into RUNNING once first frame is ready to be grabbed * ERROR - if it was not possible to connect with source Thread safe - only one transition of states possible at the time. Returns: None Throws: * StreamOperationNotAllowedError: if executed in context of incorrect state of the source * SourceConnectionError: if source cannot be connected """ if self._state not in START_ELIGIBLE_STATES: raise StreamOperationNotAllowedError( f"Could not START stream in state: {self._state}" ) self._start() @lock_state_transition def terminate(self, wait_on_frames_consumption: bool = True) -> None: """ Method to be used to terminate source consumption. Eligible to be used in states: [MUTED, RUNNING, PAUSED, ENDED, ERROR, (RESTARTING - which is internal state only)] End state: * ENDED - indicating success of the process * ERROR - if error with processing occurred Must be used to properly dispose resources at the end. Thread safe - only one transition of states possible at the time. Args: wait_on_frames_consumption (bool): Flag telling if all frames from buffer must be consumed before completion of this operation. Returns: None Throws: * StreamOperationNotAllowedError: if executed in context of incorrect state of the source """ if self._state not in TERMINATE_ELIGIBLE_STATES: raise StreamOperationNotAllowedError( f"Could not TERMINATE stream in state: {self._state}" ) self._terminate(wait_on_frames_consumption=wait_on_frames_consumption) @lock_state_transition def pause(self) -> None: """ Method to be used to pause source consumption. During pause - no new frames are consumed. Used on on-line streams for too long may cause stream disconnection. Eligible to be used in states: [RUNNING] End state: * PAUSED Thread safe - only one transition of states possible at the time. Returns: None Throws: * StreamOperationNotAllowedError: if executed in context of incorrect state of the source """ if self._state not in PAUSE_ELIGIBLE_STATES: raise StreamOperationNotAllowedError( f"Could not PAUSE stream in state: {self._state}" ) self._pause() @lock_state_transition def mute(self) -> None: """ Method to be used to mute source consumption. Muting is an equivalent of pause for stream - where frames grabbing is not put on hold, just new frames decoding and buffering is not allowed - causing intermediate frames to be dropped. May be also used against files, although arguably less useful. Eligible to be used in states: [RUNNING] End state: * MUTED Thread safe - only one transition of states possible at the time. Returns: None Throws: * StreamOperationNotAllowedError: if executed in context of incorrect state of the source """ if self._state not in MUTE_ELIGIBLE_STATES: raise StreamOperationNotAllowedError( f"Could not MUTE stream in state: {self._state}" ) self._mute() @lock_state_transition def resume(self) -> None: """ Method to recover from pause or mute into running state. [PAUSED, MUTED] End state: * RUNNING Thread safe - only one transition of states possible at the time. Returns: None Throws: * StreamOperationNotAllowedError: if executed in context of incorrect state of the source """ if self._state not in RESUME_ELIGIBLE_STATES: raise StreamOperationNotAllowedError( f"Could not RESUME stream in state: {self._state}" ) self._resume() def get_state(self) -> StreamState: """ Method to get current state of the `VideoSource` Returns: StreamState """ return self._state def frame_ready(self) -> bool: """ Method to check if decoded frame is ready for consumer Returns: boolean flag indicating frame readiness """ return not self._frames_buffer.empty() def read_frame(self) -> VideoFrame: """ Method to be used by the consumer to get decoded source frame. Returns: VideoFrame object with decoded frame and its metadata. Throws: * EndOfStreamError: when trying to get the frame from closed source. """ if self._buffer_consumption_strategy is BufferConsumptionStrategy.EAGER: video_frame: Optional[VideoFrame] = purge_queue( queue=self._frames_buffer, on_successful_read=self._video_consumer.notify_frame_consumed, ) else: video_frame: Optional[VideoFrame] = self._frames_buffer.get() self._frames_buffer.task_done() self._video_consumer.notify_frame_consumed() if video_frame is None: raise EndOfStreamError( "Attempted to retrieve frame from stream that already ended." ) send_video_source_status_update( severity=UpdateSeverity.DEBUG, event_type=FRAME_CONSUMED_EVENT, payload={ "frame_timestamp": video_frame.frame_timestamp, "frame_id": video_frame.frame_id, }, status_update_handlers=self._status_update_handlers, ) return video_frame def describe_source(self) -> SourceMetadata: return SourceMetadata( source_properties=self._source_properties, source_reference=self._stream_reference, buffer_size=self._frames_buffer.maxsize, state=self._state, buffer_filling_strategy=self._video_consumer.buffer_filling_strategy, buffer_consumption_strategy=self._buffer_consumption_strategy, ) def _restart(self, wait_on_frames_consumption: bool = True) -> None: self._terminate(wait_on_frames_consumption=wait_on_frames_consumption) self._change_state(target_state=StreamState.RESTARTING) self._playback_allowed = Event() self._frames_buffering_allowed = True self._video: Optional[cv2.VideoCapture] = None self._source_properties: Optional[SourceProperties] = None self._start() def _start(self) -> None: self._change_state(target_state=StreamState.INITIALISING) self._video = cv2.VideoCapture(self._stream_reference) if not self._video.isOpened(): self._change_state(target_state=StreamState.ERROR) raise SourceConnectionError( f"Cannot connect to video source under reference: {self._stream_reference}" ) self._source_properties = discover_source_properties(stream=self._video) self._video_consumer.reset(source_properties=self._source_properties) if self._source_properties.is_file: self._set_file_mode_consumption_strategies() else: self._set_stream_mode_consumption_strategies() self._playback_allowed.set() self._stream_consumption_thread = Thread(target=self._consume_video) self._stream_consumption_thread.start() def _terminate(self, wait_on_frames_consumption: bool) -> None: if self._state in RESUME_ELIGIBLE_STATES: self._resume() previous_state = self._state self._change_state(target_state=StreamState.TERMINATING) if self._stream_consumption_thread is not None: self._stream_consumption_thread.join() if wait_on_frames_consumption: self._frames_buffer.join() if previous_state is not StreamState.ERROR: self._change_state(target_state=StreamState.ENDED) def _pause(self) -> None: self._playback_allowed.clear() self._change_state(target_state=StreamState.PAUSED) def _mute(self) -> None: self._frames_buffering_allowed = False self._change_state(target_state=StreamState.MUTED) def _resume(self) -> None: previous_state = self._state self._change_state(target_state=StreamState.RUNNING) if previous_state is StreamState.PAUSED: self._video_consumer.reset_stream_consumption_pace() self._playback_allowed.set() if previous_state is StreamState.MUTED: self._frames_buffering_allowed = True def _set_file_mode_consumption_strategies(self) -> None: if self._buffer_consumption_strategy is None: self._buffer_consumption_strategy = BufferConsumptionStrategy.LAZY def _set_stream_mode_consumption_strategies(self) -> None: if self._buffer_consumption_strategy is None: self._buffer_consumption_strategy = BufferConsumptionStrategy.EAGER def _consume_video(self) -> None: send_video_source_status_update( severity=UpdateSeverity.INFO, event_type=VIDEO_CONSUMPTION_STARTED_EVENT, status_update_handlers=self._status_update_handlers, ) logger.info(f"Video consumption started") try: self._change_state(target_state=StreamState.RUNNING) declared_source_fps = None if self._source_properties is not None: declared_source_fps = self._source_properties.fps while self._video.isOpened(): if self._state is StreamState.TERMINATING: break self._playback_allowed.wait() success = self._video_consumer.consume_frame( video=self._video, declared_source_fps=declared_source_fps, buffer=self._frames_buffer, frames_buffering_allowed=self._frames_buffering_allowed, ) if not success: break self._frames_buffer.put(None) self._video.release() self._change_state(target_state=StreamState.ENDED) send_video_source_status_update( severity=UpdateSeverity.INFO, event_type=VIDEO_CONSUMPTION_FINISHED_EVENT, status_update_handlers=self._status_update_handlers, ) logger.info(f"Video consumption finished") except Exception as error: self._change_state(target_state=StreamState.ERROR) payload = { "error_type": error.__class__.__name__, "error_message": str(error), "error_context": "stream_consumer_thread", } send_video_source_status_update( severity=UpdateSeverity.ERROR, event_type=SOURCE_ERROR_EVENT, payload=payload, status_update_handlers=self._status_update_handlers, ) logger.exception("Encountered error in video consumption thread") def _change_state(self, target_state: StreamState) -> None: payload = { "previous_state": self._state, "new_state": target_state, } self._state = target_state send_video_source_status_update( severity=UpdateSeverity.INFO, event_type=SOURCE_STATE_UPDATE_EVENT, payload=payload, status_update_handlers=self._status_update_handlers, ) def __iter__(self) -> "VideoSource": return self def __next__(self) -> VideoFrame: """ Method allowing to use `VideoSource` convenient to read frames Returns: VideoFrame Example: ```python source = VideoSource.init(video_reference="./some.mp4") source.start() for frame in source: pass ``` """ try: return self.read_frame() except EndOfStreamError: raise StopIteration() class VideoConsumer: """ This class should be consumed as part of internal implementation. It provides abstraction around stream consumption strategies. """ @classmethod def init( cls, buffer_filling_strategy: Optional[BufferFillingStrategy], adaptive_mode_stream_pace_tolerance: float, adaptive_mode_reader_pace_tolerance: float, minimum_adaptive_mode_samples: int, maximum_adaptive_frames_dropped_in_row: int, status_update_handlers: List[Callable[[StatusUpdate], None]], ) -> "VideoConsumer": minimum_adaptive_mode_samples = max(minimum_adaptive_mode_samples, 2) reader_pace_monitor = sv.FPSMonitor( sample_size=10 * minimum_adaptive_mode_samples ) stream_consumption_pace_monitor = sv.FPSMonitor( sample_size=10 * minimum_adaptive_mode_samples ) decoding_pace_monitor = sv.FPSMonitor( sample_size=10 * minimum_adaptive_mode_samples ) return cls( buffer_filling_strategy=buffer_filling_strategy, adaptive_mode_stream_pace_tolerance=adaptive_mode_stream_pace_tolerance, adaptive_mode_reader_pace_tolerance=adaptive_mode_reader_pace_tolerance, minimum_adaptive_mode_samples=minimum_adaptive_mode_samples, maximum_adaptive_frames_dropped_in_row=maximum_adaptive_frames_dropped_in_row, status_update_handlers=status_update_handlers, reader_pace_monitor=reader_pace_monitor, stream_consumption_pace_monitor=stream_consumption_pace_monitor, decoding_pace_monitor=decoding_pace_monitor, ) def __init__( self, buffer_filling_strategy: Optional[BufferFillingStrategy], adaptive_mode_stream_pace_tolerance: float, adaptive_mode_reader_pace_tolerance: float, minimum_adaptive_mode_samples: int, maximum_adaptive_frames_dropped_in_row: int, status_update_handlers: List[Callable[[StatusUpdate], None]], reader_pace_monitor: sv.FPSMonitor, stream_consumption_pace_monitor: sv.FPSMonitor, decoding_pace_monitor: sv.FPSMonitor, ): self._buffer_filling_strategy = buffer_filling_strategy self._frame_counter = 0 self._adaptive_mode_stream_pace_tolerance = adaptive_mode_stream_pace_tolerance self._adaptive_mode_reader_pace_tolerance = adaptive_mode_reader_pace_tolerance self._minimum_adaptive_mode_samples = minimum_adaptive_mode_samples self._maximum_adaptive_frames_dropped_in_row = ( maximum_adaptive_frames_dropped_in_row ) self._adaptive_frames_dropped_in_row = 0 self._reader_pace_monitor = reader_pace_monitor self._stream_consumption_pace_monitor = stream_consumption_pace_monitor self._decoding_pace_monitor = decoding_pace_monitor self._status_update_handlers = status_update_handlers @property def buffer_filling_strategy(self) -> Optional[BufferFillingStrategy]: return self._buffer_filling_strategy def reset(self, source_properties: SourceProperties) -> None: if source_properties.is_file: self._set_file_mode_buffering_strategies() else: self._set_stream_mode_buffering_strategies() self._reader_pace_monitor.reset() self.reset_stream_consumption_pace() self._decoding_pace_monitor.reset() self._adaptive_frames_dropped_in_row = 0 def reset_stream_consumption_pace(self) -> None: self._stream_consumption_pace_monitor.reset() def notify_frame_consumed(self) -> None: self._reader_pace_monitor.tick() def consume_frame( self, video: cv2.VideoCapture, declared_source_fps: Optional[float], buffer: Queue, frames_buffering_allowed: bool, ) -> bool: frame_timestamp = datetime.now() success = video.grab() self._stream_consumption_pace_monitor.tick() if not success: return False self._frame_counter += 1 send_video_source_status_update( severity=UpdateSeverity.DEBUG, event_type=FRAME_CAPTURED_EVENT, payload={ "frame_timestamp": frame_timestamp, "frame_id": self._frame_counter, }, status_update_handlers=self._status_update_handlers, ) return self._consume_stream_frame( video=video, declared_source_fps=declared_source_fps, frame_timestamp=frame_timestamp, buffer=buffer, frames_buffering_allowed=frames_buffering_allowed, ) def _set_file_mode_buffering_strategies(self) -> None: if self._buffer_filling_strategy is None: self._buffer_filling_strategy = BufferFillingStrategy.WAIT def _set_stream_mode_buffering_strategies(self) -> None: if self._buffer_filling_strategy is None: self._buffer_filling_strategy = BufferFillingStrategy.ADAPTIVE_DROP_OLDEST def _consume_stream_frame( self, video: cv2.VideoCapture, declared_source_fps: Optional[float], frame_timestamp: datetime, buffer: Queue, frames_buffering_allowed: bool, ) -> bool: """ Returns: boolean flag with success status """ if not frames_buffering_allowed: send_frame_drop_update( frame_timestamp=frame_timestamp, frame_id=self._frame_counter, cause="Buffering not allowed at the moment", status_update_handlers=self._status_update_handlers, ) return True if self._frame_should_be_adaptively_dropped( declared_source_fps=declared_source_fps ): self._adaptive_frames_dropped_in_row += 1 send_frame_drop_update( frame_timestamp=frame_timestamp, frame_id=self._frame_counter, cause="ADAPTIVE strategy", status_update_handlers=self._status_update_handlers, ) return True self._adaptive_frames_dropped_in_row = 0 if ( not buffer.full() or self._buffer_filling_strategy is BufferFillingStrategy.WAIT ): return decode_video_frame_to_buffer( frame_timestamp=frame_timestamp, frame_id=self._frame_counter, video=video, buffer=buffer, decoding_pace_monitor=self._decoding_pace_monitor, ) if self._buffer_filling_strategy in DROP_OLDEST_STRATEGIES: return self._process_stream_frame_dropping_oldest( frame_timestamp=frame_timestamp, video=video, buffer=buffer, ) send_frame_drop_update( frame_timestamp=frame_timestamp, frame_id=self._frame_counter, cause="DROP_LATEST strategy", status_update_handlers=self._status_update_handlers, ) return True def _frame_should_be_adaptively_dropped( self, declared_source_fps: Optional[float] ) -> bool: if self._buffer_filling_strategy not in ADAPTIVE_STRATEGIES: return False if ( self._adaptive_frames_dropped_in_row >= self._maximum_adaptive_frames_dropped_in_row ): return False if ( len(self._stream_consumption_pace_monitor.all_timestamps) <= self._minimum_adaptive_mode_samples ): # not enough observations return False stream_consumption_pace = self._stream_consumption_pace_monitor() announced_stream_fps = stream_consumption_pace if declared_source_fps is not None and declared_source_fps > 0: announced_stream_fps = declared_source_fps if ( announced_stream_fps - stream_consumption_pace > self._adaptive_mode_stream_pace_tolerance ): # cannot keep up with stream emission return True if ( len(self._reader_pace_monitor.all_timestamps) <= self._minimum_adaptive_mode_samples ) or ( len(self._decoding_pace_monitor.all_timestamps) <= self._minimum_adaptive_mode_samples ): # not enough observations return False actual_reader_pace = get_fps_if_tick_happens_now( fps_monitor=self._reader_pace_monitor ) decoding_pace = self._decoding_pace_monitor() if ( decoding_pace - actual_reader_pace > self._adaptive_mode_reader_pace_tolerance ): # we are too fast for the reader - time to save compute on decoding return True return False def _process_stream_frame_dropping_oldest( self, frame_timestamp: datetime, video: cv2.VideoCapture, buffer: Queue, ) -> bool: drop_single_frame_from_buffer( buffer=buffer, cause="DROP_OLDEST strategy", status_update_handlers=self._status_update_handlers, ) return decode_video_frame_to_buffer( frame_timestamp=frame_timestamp, frame_id=self._frame_counter, video=video, buffer=buffer, decoding_pace_monitor=self._decoding_pace_monitor, ) def discover_source_properties(stream: cv2.VideoCapture) -> SourceProperties: width = int(stream.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(stream.get(cv2.CAP_PROP_FRAME_HEIGHT)) fps = stream.get(cv2.CAP_PROP_FPS) total_frames = int(stream.get(cv2.CAP_PROP_FRAME_COUNT)) return SourceProperties( width=width, height=height, total_frames=total_frames, is_file=total_frames > 0, fps=fps, ) def purge_queue( queue: Queue, wait_on_empty: bool = True, on_successful_read: Callable[[], None] = lambda: None, ) -> Optional[Any]: result = None if queue.empty() and wait_on_empty: result = queue.get() queue.task_done() on_successful_read() while not queue.empty(): result = queue.get() queue.task_done() on_successful_read() return result def drop_single_frame_from_buffer( buffer: Queue, cause: str, status_update_handlers: List[Callable[[StatusUpdate], None]], ) -> None: try: video_frame = buffer.get_nowait() buffer.task_done() send_frame_drop_update( frame_timestamp=video_frame.frame_timestamp, frame_id=video_frame.frame_id, cause=cause, status_update_handlers=status_update_handlers, ) except Empty: # buffer may be emptied in the meantime, hence we ignore Empty pass def send_frame_drop_update( frame_timestamp: datetime, frame_id: int, cause: str, status_update_handlers: List[Callable[[StatusUpdate], None]], ) -> None: send_video_source_status_update( severity=UpdateSeverity.DEBUG, event_type=FRAME_DROPPED_EVENT, payload={ "frame_timestamp": frame_timestamp, "frame_id": frame_id, "cause": cause, }, status_update_handlers=status_update_handlers, sub_context=VIDEO_CONSUMER_CONTEXT, ) def send_video_source_status_update( severity: UpdateSeverity, event_type: str, status_update_handlers: List[Callable[[StatusUpdate], None]], sub_context: Optional[str] = None, payload: Optional[dict] = None, ) -> None: if payload is None: payload = {} context = VIDEO_SOURCE_CONTEXT if sub_context is not None: context = f"{context}.{sub_context}" status_update = StatusUpdate( timestamp=datetime.now(), severity=severity, event_type=event_type, payload=payload, context=context, ) for handler in status_update_handlers: try: handler(status_update) except Exception as error: logger.warning(f"Could not execute handler update. Cause: {error}") def decode_video_frame_to_buffer( frame_timestamp: datetime, frame_id: int, video: cv2.VideoCapture, buffer: Queue, decoding_pace_monitor: sv.FPSMonitor, ) -> bool: success, image = video.retrieve() if not success: return False decoding_pace_monitor.tick() video_frame = VideoFrame( image=image, frame_id=frame_id, frame_timestamp=frame_timestamp ) buffer.put(video_frame) return True def get_fps_if_tick_happens_now(fps_monitor: sv.FPSMonitor) -> float: if len(fps_monitor.all_timestamps) == 0: return 0.0 min_reader_timestamp = fps_monitor.all_timestamps[0] now = time.monotonic() reader_taken_time = now - min_reader_timestamp return (len(fps_monitor.all_timestamps) + 1) / reader_taken_time