From ff468b175dcfc9ae0485392ce787f34de1a04327 Mon Sep 17 00:00:00 2001 From: Rodrigo Barbosa Date: Mon, 31 Aug 2026 14:19:13 -0300 Subject: [PATCH] feat(webrtc): add video_file_url and max_fps to WebRTC worker Allow creating a WebRTC session that processes a video file directly from an HTTP(S) URL - ffmpeg streams it, so no data-channel upload is needed. With webrtc_realtime_processing=false every frame is processed as fast as possible; max_fps optionally caps the rate via ffmpeg's fps filter, and workflows see the effective (capped) fps so temporal analytics stay correct. The upload path and the new URL path share the same processing logic (_begin_video_file_processing). URL processing is deferred until the peer connection and data channel are ready, since a finite file processed at full speed would otherwise finish before outputs can reach the client. URL destinations are validated with the existing SSRF guard used for image URL inputs. --- docs/webrtc-streaming.md | 20 +++ .../core/interfaces/webrtc_worker/entities.py | 4 + .../core/interfaces/webrtc_worker/modal.py | 4 + .../interfaces/webrtc_worker/sources/file.py | 34 ++++- .../core/interfaces/webrtc_worker/webrtc.py | 130 ++++++++++++------ .../webrtc_worker/test_video_file_source.py | 86 ++++++++++++ 6 files changed, 233 insertions(+), 45 deletions(-) create mode 100644 tests/inference/unit_tests/core/interfaces/webrtc_worker/test_video_file_source.py 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