Merge pull request #1778 from roboflow/new-upload-format

New upload format for webrtc file

Author: digaobarbosa

inference/core/env.py

@@ -782,6 +782,12 @@
 WEBRTC_MODAL_USAGE_QUOTA_ENABLED = str2bool(
     os.getenv("WEBRTC_MODAL_USAGE_QUOTA_ENABLED", "False")
 )
+WEBRTC_DATA_CHANNEL_BUFFER_DRAINING_DELAY = float(
+    os.getenv("WEBRTC_DATA_CHANNEL_BUFFER_DRAINING_DELAY", "0.1")
+)
+WEBRTC_DATA_CHANNEL_BUFFER_SIZE_LIMIT = int(
+    os.getenv("WEBRTC_DATA_CHANNEL_BUFFER_SIZE_LIMIT", str(1024 * 1024))  # 1MB
+)
 
 HTTP_API_SHARED_WORKFLOWS_THREAD_POOL_ENABLED = str2bool(
     os.getenv("HTTP_API_SHARED_WORKFLOWS_THREAD_POOL_ENABLED", "True")

inference/core/interfaces/webrtc_worker/entities.py

@@ -72,6 +72,7 @@ class WebRTCOutput(BaseModel):
     serialized_output_data: Optional[Dict[str, Any]] = None
     video_metadata: Optional[WebRTCVideoMetadata] = None
     errors: List[str] = Field(default_factory=list)
+    processing_complete: bool = False  # Signals end of video file processing
 
 
 class WebRTCWorkerResult(BaseModel):
@@ -93,3 +94,18 @@ class DataOutputMode(str, Enum):
     NONE = "none"  # None or [] -> no data sent
     ALL = "all"  # ["*"] -> send all (skip images)
     SPECIFIC = "specific"  # ["field1", "field2"] -> send only these
+
+
+# Video File Upload Protocol
+# Binary header: [chunk_index:u32][total_chunks:u32][payload]
+VIDEO_FILE_HEADER_SIZE = 8
+
+
+class VideoFileUploadState(str, Enum):
+    """State of video file upload."""
+
+    IDLE = "idle"
+    UPLOADING = "uploading"
+    COMPLETE = "complete"
+    PROCESSING = "processing"
+    ERROR = "error"

inference/core/interfaces/webrtc_worker/utils.py

@@ -1,6 +1,7 @@
 import ctypes
 import datetime
 import logging
+import struct
 import time
 from typing import Any, Dict, List, Optional, Tuple, Union
 
@@ -12,6 +13,7 @@
 from inference.core.env import DEBUG_WEBRTC_PROCESSING_LATENCY
 from inference.core.interfaces.camera.entities import VideoFrame as InferenceVideoFrame
 from inference.core.interfaces.stream.inference_pipeline import InferencePipeline
+from inference.core.interfaces.webrtc_worker.entities import VIDEO_FILE_HEADER_SIZE
 from inference.core.utils.roboflow import get_model_id_chunks
 from inference.core.workflows.execution_engine.entities.base import WorkflowImageData
 from inference.models.aliases import resolve_roboflow_model_alias
@@ -186,6 +188,19 @@ def workflow_contains_preloaded_model(
     return False
 
 
+# Video File Upload Protocol
+# Header: [chunk_index:u32][total_chunks:u32][payload]
+def parse_video_file_chunk(message: bytes) -> Tuple[int, int, bytes]:
+    """Parse video file chunk message.
+
+    Returns: (chunk_index, total_chunks, payload)
+    """
+    if len(message) < VIDEO_FILE_HEADER_SIZE:
+        raise ValueError(f"Message too short: {len(message)} bytes")
+    chunk_index, total_chunks = struct.unpack("<II", message[:8])
+    return chunk_index, total_chunks, message[8:]
+
+
 def warmup_cuda(
     max_retries: int = 10,
     retry_delay: float = 0.5,