feat(webrtc): process video files by URL with optional max_fps cap

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:

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"

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:

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()

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)

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