diff --git a/docs/webrtc-streaming.md b/docs/webrtc-streaming.md index 7cc64bf30e..0c299f5f75 100644 --- a/docs/webrtc-streaming.md +++ b/docs/webrtc-streaming.md @@ -138,6 +138,26 @@ session = client.webrtc.stream( ) ``` +### Processing a Video File by URL (server-side) + +The `POST /initialise_webrtc_worker` request accepts a `video_file_url` pointing at an HTTP(S) +video file (mp4, webm, ogv, ...). The server streams it directly with ffmpeg - no upload needed. +With `webrtc_realtime_processing: false` every frame is processed as fast as possible; +`max_fps` optionally caps the processed rate (frames are decimated with ffmpeg's `fps` filter, +and workflows see the effective rate). `max_fps` is ignored in realtime mode. + +```json +{ + "video_file_url": "https://example.com/video.mp4", + "webrtc_realtime_processing": false, + "max_fps": 5, + "data_output": ["*"] +} +``` + +The URL is validated against the same allow/block lists as image URL inputs +(`WHITELISTED_DESTINATIONS_FOR_URL_INPUT` / `BLACKLISTED_DESTINATIONS_FOR_URL_INPUT`). + ## Support For questions and support: diff --git a/inference/core/interfaces/webrtc_worker/entities.py b/inference/core/interfaces/webrtc_worker/entities.py index c02169acff..e4a8750ffb 100644 --- a/inference/core/interfaces/webrtc_worker/entities.py +++ b/inference/core/interfaces/webrtc_worker/entities.py @@ -47,6 +47,10 @@ class WebRTCWorkerRequest(BaseModel): declared_fps: Optional[float] = None rtsp_url: Optional[str] = None mjpeg_url: Optional[str] = None + # HTTP(S) video file streamed directly by ffmpeg - no upload needed + video_file_url: Optional[str] = None + # cap for video-file processing; unset = process all frames + max_fps: Optional[float] = Field(default=None, gt=0) processing_timeout: Optional[int] = WEBRTC_MODAL_FUNCTION_TIME_LIMIT processing_session_started: Optional[datetime.datetime] = None requested_plan: Optional[str] = "webrtc-gpu-small" diff --git a/inference/core/interfaces/webrtc_worker/modal.py b/inference/core/interfaces/webrtc_worker/modal.py index aa71a71a66..35b7b47d92 100644 --- a/inference/core/interfaces/webrtc_worker/modal.py +++ b/inference/core/interfaces/webrtc_worker/modal.py @@ -368,6 +368,8 @@ def rtc_peer_connection_modal( logger.info("data_output: %s", webrtc_request.data_output) logger.info("declared_fps: %s", webrtc_request.declared_fps) logger.info("rtsp_url: %s", webrtc_request.rtsp_url) + logger.info("video_file_url: %s", webrtc_request.video_file_url) + logger.info("max_fps: %s", webrtc_request.max_fps) logger.info("processing_timeout: %s", webrtc_request.processing_timeout) logger.info("requested_region: %s", webrtc_request.requested_region) logger.info("watchdog_timeout: %s", WEBRTC_MODAL_WATCHDOG_TIMEMOUT) @@ -458,6 +460,8 @@ def send_answer(obj: WebRTCWorkerResult): video_source = "realtime browser stream" if webrtc_request.rtsp_url: video_source = "rtsp" + elif webrtc_request.video_file_url: + video_source = "video_file_url" elif not webrtc_request.webrtc_realtime_processing: video_source = "buffered browser stream" else: diff --git a/inference/core/interfaces/webrtc_worker/sources/file.py b/inference/core/interfaces/webrtc_worker/sources/file.py index 9eb8987031..7395bd3470 100644 --- a/inference/core/interfaces/webrtc_worker/sources/file.py +++ b/inference/core/interfaces/webrtc_worker/sources/file.py @@ -13,7 +13,7 @@ from inference.core.interfaces.webrtc_worker.entities import VideoFileUploadState -def _decode_worker(filepath: str, frame_queue, stop_event): +def _decode_worker(filepath: str, frame_queue, stop_event, target_fps=None): """Decode video frames in a separate thread and put them on a queue. We decode in a background thread to avoid deadlocks. PyAV (the video decoder) @@ -32,9 +32,27 @@ def _decode_worker(filepath: str, frame_queue, stop_event): stream = container.streams.video[0] stream.thread_type = "AUTO" + # ffmpeg's fps filter decimates to target_fps; it duplicates frames when + # the target exceeds the source rate, so callers only set it below native + graph = None + if target_fps is not None: + graph = av.filter.Graph() + src = graph.add_buffer(template=stream) + fps_filter = graph.add("fps", f"fps={target_fps}") + sink = graph.add("buffersink") + src.link_to(fps_filter) + fps_filter.link_to(sink) + graph.configure() + for frame in container.decode(stream): if stop_event.is_set(): break + if graph is not None: + graph.push(frame) + try: + frame = graph.pull() + except av.error.BlockingIOError: + continue # frame decimated by the fps filter try: frame_queue.put(frame, timeout=300) frame_count += 1 @@ -48,6 +66,14 @@ def _decode_worker(filepath: str, frame_queue, stop_event): ) return + if graph is not None and not stop_event.is_set(): + graph.push(None) # flush the frame buffered inside the filter + try: + frame_queue.put(graph.pull(), timeout=300) + frame_count += 1 + except (av.error.EOFError, av.error.BlockingIOError): + pass + container.close() except Exception as e: logger.error("[DECODE_WORKER] Error at frame %d: %s", frame_count, e) @@ -72,14 +98,16 @@ class ThreadedVideoFileTrack(MediaStreamTrack): kind = "video" - def __init__(self, filepath: str, queue_size: int = 60): + def __init__( + self, filepath: str, queue_size: int = 60, target_fps: Optional[float] = None + ): # TODO: add parameter queue size in settings super().__init__() self._queue = queue.Queue(maxsize=queue_size) self._stop_event = threading.Event() self._decode_thread = threading.Thread( target=_decode_worker, - args=(filepath, self._queue, self._stop_event), + args=(filepath, self._queue, self._stop_event, target_fps), daemon=True, ) self._decode_thread.start() diff --git a/inference/core/interfaces/webrtc_worker/webrtc.py b/inference/core/interfaces/webrtc_worker/webrtc.py index eabe82afce..ca70abed0f 100644 --- a/inference/core/interfaces/webrtc_worker/webrtc.py +++ b/inference/core/interfaces/webrtc_worker/webrtc.py @@ -74,6 +74,7 @@ ) from inference.core.managers.base import ModelManager from inference.core.roboflow_api import get_workflow_specification +from inference.core.utils.image_utils import _validate_url_destination from inference.core.workflows.errors import WorkflowError, WorkflowSyntaxError from inference.core.workflows.execution_engine.entities.base import WorkflowImageData from inference.usage_tracking.collector import usage_collector @@ -851,6 +852,66 @@ async def recv(self): return new_frame +def _begin_video_file_processing( + video_processor, + video_source: str, + webrtc_request: WebRTCWorkerRequest, + should_send_video: bool, +) -> None: + """Start processing a video file - local path or HTTP URL (ffmpeg streams both).""" + video_processor._file_processing = True + logger.info( + "Video file processing: realtime=%s, source=%s", + webrtc_request.webrtc_realtime_processing, + video_source, + ) + + rotation = get_video_rotation(video_source) + rotation_code = get_cv2_rotation_code(rotation) + if rotation_code is not None: + logger.info("Video has %d° rotation, will correct", rotation) + + detected_fps = get_video_fps(video_source) + # the fps filter duplicates frames above native rate, so only cap below it + target_fps = None + if ( + webrtc_request.max_fps is not None + and not webrtc_request.webrtc_realtime_processing + and detected_fps is not None + and webrtc_request.max_fps < detected_fps + ): + target_fps = webrtc_request.max_fps + if detected_fps is not None: + # workflows must see the effective rate, or temporal analytics go wrong + video_processor._declared_fps = target_fps or detected_fps + logger.info( + "FPS detection: detected=%.2f, effective=%s", + detected_fps, + video_processor._declared_fps, + ) + else: + logger.warning( + "FPS detection failed, keeping default: %s", + video_processor._declared_fps, + ) + + if webrtc_request.webrtc_realtime_processing: + if webrtc_request.max_fps is not None: + logger.warning("max_fps is ignored when webrtc_realtime_processing=True") + # We are dealing with a live video stream, + player = MediaPlayer(video_source, loop=False) + player._throttle_playback = True + video_processor.set_track(track=player.video, rotation_code=rotation_code) + else: + # we are dealing with a video file, + track = ThreadedVideoFileTrack(video_source, target_fps=target_fps) + video_processor.set_track(track=track, rotation_code=rotation_code) + + if not should_send_video: + logger.info("Starting data-only processing for video file") + asyncio.create_task(video_processor.process_frames_data_only()) + + async def _wait_ice_complete(peer_connection: RTCPeerConnectionWithLoop, timeout=2.0): if peer_connection.iceGatheringState == "complete": logger.info("ICE gathering state already complete") @@ -1111,6 +1172,31 @@ async def init_rtc_peer_connection_with_loop( logger.info("Starting data-only processing for MJPEG stream") asyncio.create_task(video_processor.process_frames_data_only()) + elif webrtc_request.video_file_url: + video_file_url = _validate_url_destination(value=webrtc_request.video_file_url) + + async def _start_video_file_when_ready() -> None: + # the file is finite - starting before the connection is up would + # drop the outputs of every frame processed until the channel opens + while not terminate_event.is_set(): + data_channel_open = ( + video_processor.data_channel is not None + and video_processor.data_channel.readyState == "open" + ) + if peer_connection.connectionState == "connected" and ( + data_channel_open or should_send_video + ): + _begin_video_file_processing( + video_processor, + video_file_url, + webrtc_request, + should_send_video, + ) + return + await asyncio.sleep(0.05) + + asyncio.create_task(_start_video_file_when_ready()) + @peer_connection.on("track") def on_track(track: RemoteStreamTrack): logger.info("Track received from client") @@ -1231,50 +1317,10 @@ async def on_upload_message(message): None, process_video_upload_message, message, video_processor ) if video_path: - video_processor._file_processing = True - logger.info( - "Video upload complete, processing: realtime=%s, path=%s", - webrtc_request.webrtc_realtime_processing, - video_path, + _begin_video_file_processing( + video_processor, video_path, webrtc_request, should_send_video ) - rotation = get_video_rotation(video_path) - rotation_code = get_cv2_rotation_code(rotation) - if rotation_code is not None: - logger.info("Video has %d° rotation, will correct", rotation) - - detected_fps = get_video_fps(video_path) - if detected_fps is not None: - logger.info( - "FPS detection: detected=%.2f, previous=%s", - detected_fps, - video_processor._declared_fps, - ) - video_processor._declared_fps = detected_fps - else: - logger.warning( - "FPS detection failed, keeping default: %s", - video_processor._declared_fps, - ) - - if webrtc_request.webrtc_realtime_processing: - # We are dealing with a live video stream, - player = MediaPlayer(video_path, loop=False) - player._throttle_playback = True - video_processor.set_track( - track=player.video, rotation_code=rotation_code - ) - else: - # we are dealing with a video file, - track = ThreadedVideoFileTrack(video_path) - video_processor.set_track( - track=track, rotation_code=rotation_code - ) - - if not should_send_video: - logger.info("Starting data-only processing for video file") - asyncio.create_task(video_processor.process_frames_data_only()) - return # Handle inference control channel (bidirectional communication) diff --git a/tests/inference/unit_tests/core/interfaces/webrtc_worker/test_video_file_source.py b/tests/inference/unit_tests/core/interfaces/webrtc_worker/test_video_file_source.py new file mode 100644 index 0000000000..0f3f134616 --- /dev/null +++ b/tests/inference/unit_tests/core/interfaces/webrtc_worker/test_video_file_source.py @@ -0,0 +1,86 @@ +import queue +import threading + +import av +import numpy as np +import pytest +from pydantic import ValidationError + +from inference.core.interfaces.webrtc_worker.entities import WebRTCWorkerRequest +from inference.core.interfaces.webrtc_worker.sources.file import _decode_worker + +MINIMAL_REQUEST_PAYLOAD = { + "workflow_configuration": { + "type": "WorkflowConfiguration", + "workflow_specification": {}, + }, + "webrtc_offer": {"type": "offer", "sdp": "v=0"}, +} + + +def test_request_accepts_video_file_url_and_max_fps() -> None: + request = WebRTCWorkerRequest( + **MINIMAL_REQUEST_PAYLOAD, + video_file_url="https://example.com/video.mp4", + max_fps=5, + ) + + assert request.video_file_url == "https://example.com/video.mp4" + assert request.max_fps == 5 + + +def test_request_new_fields_default_to_none() -> None: + request = WebRTCWorkerRequest(**MINIMAL_REQUEST_PAYLOAD) + + assert request.video_file_url is None + assert request.max_fps is None + + +@pytest.mark.parametrize("max_fps", [0, -1]) +def test_request_rejects_non_positive_max_fps(max_fps: float) -> None: + with pytest.raises(ValidationError): + WebRTCWorkerRequest(**MINIMAL_REQUEST_PAYLOAD, max_fps=max_fps) + + +def _write_test_video(path: str, frames: int = 30, fps: int = 30) -> None: + container = av.open(path, mode="w") + stream = container.add_stream("h264", rate=fps) + stream.width = 64 + stream.height = 64 + stream.pix_fmt = "yuv420p" + for i in range(frames): + image = np.full((64, 64, 3), i % 255, dtype=np.uint8) + frame = av.VideoFrame.from_ndarray(image, format="rgb24") + for packet in stream.encode(frame): + container.mux(packet) + for packet in stream.encode(): + container.mux(packet) + container.close() + + +def _run_decode_worker(video_path: str, target_fps=None) -> int: + frame_queue = queue.Queue(maxsize=100) + _decode_worker(video_path, frame_queue, threading.Event(), target_fps) + decoded = 0 + while True: + item = frame_queue.get_nowait() + if item is None: + return decoded + assert not isinstance(item, dict), f"decode error: {item}" + decoded += 1 + + +def test_decode_worker_without_target_fps_yields_all_frames(tmp_path) -> None: + video_path = str(tmp_path / "video.mp4") + _write_test_video(video_path, frames=30, fps=30) + + assert _run_decode_worker(video_path) == 30 + + +def test_decode_worker_with_target_fps_decimates_frames(tmp_path) -> None: + video_path = str(tmp_path / "video.mp4") + _write_test_video(video_path, frames=30, fps=30) + + decoded = _run_decode_worker(video_path, target_fps=10) + + assert 9 <= decoded <= 11