Webrtc

ChunkReassembler

Helper to reassemble chunked binary messages.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

class ChunkReassembler:
    """Helper to reassemble chunked binary messages."""

    def __init__(self):
        self._chunks: Dict[int, Dict[int, bytes]] = (
            {}
        )  # {frame_id: {chunk_index: data}}
        self._total: Dict[int, int] = {}  # {frame_id: total_chunks}

    def add_chunk(
        self, frame_id: int, chunk_index: int, total_chunks: int, chunk_data: bytes
    ) -> Optional[bytes]:
        """Add a chunk and return complete payload if all chunks received.

        Returns:
            Complete reassembled payload bytes if all chunks received, None otherwise.
        """
        # Initialize buffers for new frame
        if frame_id not in self._chunks:
            self._chunks[frame_id] = {}
            self._total[frame_id] = total_chunks

        # Store chunk
        self._chunks[frame_id][chunk_index] = chunk_data

        # Check if all chunks received
        if len(self._chunks[frame_id]) >= total_chunks:
            # Reassemble in order
            complete_payload = b"".join(
                self._chunks[frame_id][i] for i in range(total_chunks)
            )

            # Clean up
            del self._chunks[frame_id]
            del self._total[frame_id]

            return complete_payload

        return None

add_chunk(frame_id, chunk_index, total_chunks, chunk_data)

Add a chunk and return complete payload if all chunks received.

Returns:

Type	Description
Optional[bytes]	Complete reassembled payload bytes if all chunks received, None otherwise.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

def add_chunk(
    self, frame_id: int, chunk_index: int, total_chunks: int, chunk_data: bytes
) -> Optional[bytes]:
    """Add a chunk and return complete payload if all chunks received.

    Returns:
        Complete reassembled payload bytes if all chunks received, None otherwise.
    """
    # Initialize buffers for new frame
    if frame_id not in self._chunks:
        self._chunks[frame_id] = {}
        self._total[frame_id] = total_chunks

    # Store chunk
    self._chunks[frame_id][chunk_index] = chunk_data

    # Check if all chunks received
    if len(self._chunks[frame_id]) >= total_chunks:
        # Reassemble in order
        complete_payload = b"".join(
            self._chunks[frame_id][i] for i in range(total_chunks)
        )

        # Clean up
        del self._chunks[frame_id]
        del self._total[frame_id]

        return complete_payload

    return None

VideoFileUploadHandler

Handles video file uploads via data channel.

Protocol: [chunk_index:u32][total_chunks:u32][payload] Auto-completes when all chunks received.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

class VideoFileUploadHandler:
    """Handles video file uploads via data channel.

    Protocol: [chunk_index:u32][total_chunks:u32][payload]
    Auto-completes when all chunks received.
    """

    def __init__(self):
        self._chunks: Dict[int, bytes] = {}
        self._total_chunks: Optional[int] = None
        self._temp_file_path: Optional[str] = None
        self._state = VideoFileUploadState.IDLE
        self.upload_complete_event = asyncio.Event()

    @property
    def temp_file_path(self) -> Optional[str]:
        return self._temp_file_path

    def handle_chunk(self, chunk_index: int, total_chunks: int, data: bytes) -> None:
        """Handle a chunk. Auto-completes when all chunks received."""
        if self._total_chunks is None:
            self._total_chunks = total_chunks
            self._state = VideoFileUploadState.UPLOADING
            logger.info(f"Starting video upload: {total_chunks} chunks")

        self._chunks[chunk_index] = data

        if chunk_index % 100 == 0:
            logger.info(
                "Upload progress: %s/%s chunks", len(self._chunks), total_chunks
            )

        # Auto-complete when all chunks received
        # TODO: Handle the file writing without keeping all chunks in memory
        if len(self._chunks) == total_chunks:
            self._write_to_temp_file()
            self._state = VideoFileUploadState.COMPLETE
            self.upload_complete_event.set()

    def _write_to_temp_file(self) -> None:
        """Reassemble chunks and write to temp file."""
        import tempfile

        total_size = 0
        with tempfile.NamedTemporaryFile(mode="wb", suffix=".mp4", delete=False) as f:
            for i in range(self._total_chunks):
                chunk_data = self._chunks[i]
                f.write(chunk_data)
                total_size += len(chunk_data)
            self._temp_file_path = f.name

        logger.info(
            "Video upload complete: {total_size} bytes -> %s", self._temp_file_path
        )
        self._chunks.clear()  # Free memory

    def try_start_processing(self) -> Optional[str]:
        """Atomically check if upload is complete and transition to PROCESSING.

        Returns video path if processing should start, None otherwise.
        This ensures process_video_file() is only triggered once.
        """
        if self._state == VideoFileUploadState.COMPLETE:
            self._state = VideoFileUploadState.PROCESSING
            return self._temp_file_path
        return None

    def cleanup(self) -> None:
        """Clean up temp file."""
        if self._temp_file_path:
            import os

            try:
                os.unlink(self._temp_file_path)
            except Exception:
                pass
            self._temp_file_path = None

cleanup()

Clean up temp file.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

def cleanup(self) -> None:
    """Clean up temp file."""
    if self._temp_file_path:
        import os

        try:
            os.unlink(self._temp_file_path)
        except Exception:
            pass
        self._temp_file_path = None

handle_chunk(chunk_index, total_chunks, data)

Handle a chunk. Auto-completes when all chunks received.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

def handle_chunk(self, chunk_index: int, total_chunks: int, data: bytes) -> None:
    """Handle a chunk. Auto-completes when all chunks received."""
    if self._total_chunks is None:
        self._total_chunks = total_chunks
        self._state = VideoFileUploadState.UPLOADING
        logger.info(f"Starting video upload: {total_chunks} chunks")

    self._chunks[chunk_index] = data

    if chunk_index % 100 == 0:
        logger.info(
            "Upload progress: %s/%s chunks", len(self._chunks), total_chunks
        )

    # Auto-complete when all chunks received
    # TODO: Handle the file writing without keeping all chunks in memory
    if len(self._chunks) == total_chunks:
        self._write_to_temp_file()
        self._state = VideoFileUploadState.COMPLETE
        self.upload_complete_event.set()

try_start_processing()

Atomically check if upload is complete and transition to PROCESSING.

Returns video path if processing should start, None otherwise. This ensures process_video_file() is only triggered once.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

def try_start_processing(self) -> Optional[str]:
    """Atomically check if upload is complete and transition to PROCESSING.

    Returns video path if processing should start, None otherwise.
    This ensures process_video_file() is only triggered once.
    """
    if self._state == VideoFileUploadState.COMPLETE:
        self._state = VideoFileUploadState.PROCESSING
        return self._temp_file_path
    return None

VideoFrameProcessor

Base class for processing video frames through workflow.

Can be used independently for data-only processing (no video track output) or as a base for VideoTransformTrackWithLoop when video output is needed.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

class VideoFrameProcessor:
    """Base class for processing video frames through workflow.

    Can be used independently for data-only processing (no video track output)
    or as a base for VideoTransformTrackWithLoop when video output is needed.
    """

    def __init__(
        self,
        asyncio_loop: asyncio.AbstractEventLoop,
        workflow_configuration: WorkflowConfiguration,
        api_key: str,
        model_manager: Optional[ModelManager] = None,
        data_output: Optional[List[str]] = None,
        stream_output: Optional[str] = None,
        has_video_track: bool = True,
        declared_fps: float = 30,
        termination_date: Optional[datetime.datetime] = None,
        terminate_event: Optional[asyncio.Event] = None,
        use_data_channel_frames: bool = False,
        heartbeat_callback: Optional[Callable[[], None]] = None,
        realtime_processing: bool = True,
    ):
        self._loop = asyncio_loop
        self._termination_date = termination_date
        self._terminate_event = terminate_event
        self.track: Optional[MediaStreamTrack] = None
        self._track_active: bool = False
        self._av_logging_set: bool = False
        self._received_frames = 0
        self._declared_fps = declared_fps
        self._stop_processing = False
        self.heartbeat_callback = heartbeat_callback
        self.use_data_channel_frames = use_data_channel_frames
        self._data_frame_queue: "asyncio.Queue[Optional[VideoFrame]]" = asyncio.Queue()
        self._chunk_reassembler = (
            ChunkReassembler()
        )  # For reassembling inbound frame chunks

        self.has_video_track = has_video_track
        self.stream_output = stream_output
        self.data_channel: Optional[RTCDataChannel] = None

        # Video file upload support
        self.video_upload_handler: Optional[VideoFileUploadHandler] = None
        self._track_ready_event: asyncio.Event = asyncio.Event()
        self.realtime_processing = realtime_processing

        if data_output is None:
            self.data_output = None
            self._data_mode = DataOutputMode.NONE
        elif isinstance(data_output, list):
            self.data_output = [f for f in data_output if f]
            if self.data_output == ["*"]:
                self._data_mode = DataOutputMode.ALL
            elif len(self.data_output) == 0:
                self._data_mode = DataOutputMode.NONE
            else:
                self._data_mode = DataOutputMode.SPECIFIC
        else:
            raise WebRTCConfigurationError(
                f"data_output must be list or None, got {type(data_output).__name__}"
            )

        self._validate_output_fields(workflow_configuration)

        self._inference_pipeline = InferencePipeline.init_with_workflow(
            video_reference=VideoFrameProducer,
            workflow_specification=workflow_configuration.workflow_specification,
            workspace_name=workflow_configuration.workspace_name,
            workflow_id=workflow_configuration.workflow_id,
            api_key=api_key,
            image_input_name=workflow_configuration.image_input_name,
            workflows_parameters=workflow_configuration.workflows_parameters,
            workflows_thread_pool_workers=workflow_configuration.workflows_thread_pool_workers,
            cancel_thread_pool_tasks_on_exit=workflow_configuration.cancel_thread_pool_tasks_on_exit,
            video_metadata_input_name=workflow_configuration.video_metadata_input_name,
            model_manager=model_manager,
        )

    def set_track(self, track: MediaStreamTrack):
        if not self.track:
            self.track = track
            self._track_ready_event.set()

    def close(self):
        self._track_active = False
        self._stop_processing = True
        # Clean up video upload handler if present
        if self.video_upload_handler is not None:
            self.video_upload_handler.cleanup()

    def _check_termination(self):
        """Check if we should terminate based on timeout"""
        if (
            self._termination_date
            and self._termination_date < datetime.datetime.now()
            or self._terminate_event
            and self._terminate_event.is_set()
        ):
            logger.info("Timeout reached, terminating inference pipeline")
            self._terminate_event.set()
            return True
        return False

    async def _send_data_output(
        self,
        workflow_output: Dict[str, Any],
        frame_timestamp: datetime.datetime,
        frame: VideoFrame,
        errors: List[str],
    ):
        if not self.data_channel or self.data_channel.readyState != "open":
            return

        video_metadata = WebRTCVideoMetadata(
            frame_id=self._received_frames,
            received_at=frame_timestamp.isoformat(),
            pts=frame.pts,
            time_base=frame.time_base,
            declared_fps=self._declared_fps,
        )

        webrtc_output = WebRTCOutput(
            serialized_output_data=None,
            video_metadata=video_metadata,
            errors=errors.copy(),
        )

        if self._data_mode == DataOutputMode.NONE:
            # Even empty responses use binary protocol
            json_bytes = await asyncio.to_thread(
                lambda: json.dumps(webrtc_output.model_dump()).encode("utf-8")
            )
            await send_chunked_data(
                self.data_channel,
                self._received_frames,
                json_bytes,
                heartbeat_callback=self.heartbeat_callback,
            )
            return

        if self._data_mode == DataOutputMode.ALL:
            fields_to_send = list(workflow_output.keys())
        else:
            fields_to_send = self.data_output

        serialized_outputs = {}

        for field_name in fields_to_send:
            if field_name not in workflow_output:
                webrtc_output.errors.append(
                    f"Requested output '{field_name}' not found in workflow outputs"
                )
                continue

            output_data = workflow_output[field_name]

            if self._data_mode == DataOutputMode.ALL and isinstance(
                output_data, WorkflowImageData
            ):
                continue

            try:
                serialized_value = serialize_wildcard_kind(output_data)
                serialized_outputs[field_name] = serialized_value
            except Exception as e:
                webrtc_output.errors.append(f"{field_name}: {e}")
                serialized_outputs[field_name] = {"__serialization_error__": str(e)}

        # Set serialized outputs
        if serialized_outputs:
            webrtc_output.serialized_output_data = serialized_outputs

        # Send using binary chunked protocol
        json_bytes = json.dumps(webrtc_output.model_dump(mode="json")).encode("utf-8")
        await send_chunked_data(
            self.data_channel,
            self._received_frames,
            json_bytes,
            heartbeat_callback=self.heartbeat_callback,
        )

    async def _send_processing_complete(self):
        """Send final message indicating processing is complete."""
        if not self.data_channel or self.data_channel.readyState != "open":
            return

        completion_output = WebRTCOutput(
            processing_complete=True,
            video_metadata=WebRTCVideoMetadata(
                frame_id=self._received_frames,
                received_at=datetime.datetime.now().isoformat(),
            ),
        )
        json_bytes = json.dumps(completion_output.model_dump()).encode("utf-8")
        await send_chunked_data(
            self.data_channel, self._received_frames + 1, json_bytes
        )
        logger.info(
            "Sent processing_complete signal after %s frames", self._received_frames
        )

    async def _handle_data_channel_frame(self, message: bytes) -> None:
        """Handle incoming binary frame chunk from upstream_frames data channel.

        Uses standard binary protocol with 12-byte header + JPEG chunk payload.
        """
        try:
            # Parse message
            frame_id, chunk_index, total_chunks, jpeg_chunk = (
                parse_chunked_binary_message(message)
            )

            # Add chunk and check if complete
            jpeg_bytes = self._chunk_reassembler.add_chunk(
                frame_id, chunk_index, total_chunks, jpeg_chunk
            )

            if jpeg_bytes is None:
                # Still waiting for more chunks
                return

            # All chunks received - decode and queue frame
            if frame_id % 100 == 1:
                logger.info(
                    f"Received frame {frame_id}: {total_chunks} chunk(s), {len(jpeg_bytes)} bytes JPEG"
                )

            def _decode_to_frame(jpeg_bytes: bytes) -> VideoFrame:
                nparr = np.frombuffer(jpeg_bytes, np.uint8)
                np_image = cv2.imdecode(nparr, cv2.IMREAD_COLOR)

                if np_image is None:
                    raise ValueError("cv2.imdecode returned None")

                return VideoFrame.from_ndarray(np_image, format="bgr24")

            try:
                video_frame = await asyncio.to_thread(_decode_to_frame, jpeg_bytes)
            except Exception as e:
                logger.error(f"Failed to decode JPEG for frame {frame_id}: {e}")
                return

            self._data_frame_queue.put_nowait((frame_id, video_frame))

            if frame_id % 100 == 1:
                logger.info(f"Queued frame {frame_id}")

        except Exception as e:
            logger.error(f"Error handling frame chunk: {e}", exc_info=True)

    async def process_frames_data_only(self):
        """Process frames for data extraction only, without video track output.

        This is used when stream_output=[] and no video track is needed.
        """
        # Silencing swscaler warnings in multi-threading environment
        if not self._av_logging_set:
            av_logging.set_libav_level(av_logging.ERROR)
            self._av_logging_set = True

        logger.info(
            f"Starting data-only frame processing (use_data_channel_frames={self.use_data_channel_frames})"
        )

        try:
            while not self._stop_processing:
                if self._check_termination():
                    break
                if self.heartbeat_callback:
                    self.heartbeat_callback()

                # Get frame from appropriate source
                if self.use_data_channel_frames:
                    # Wait for frame from data channel queue
                    item = await self._data_frame_queue.get()
                    if item is None:
                        logger.info("Received stop signal from data channel")
                        break
                    frame_id, frame = item
                    self._received_frames = frame_id
                else:
                    # Get frame from media track (existing behavior)
                    if not self.track or self.track.readyState == "ended":
                        break

                    # Drain queue if using PlayerStreamTrack (RTSP)
                    if (
                        isinstance(self.track, PlayerStreamTrack)
                        and self.realtime_processing
                    ):
                        while self.track._queue.qsize() > 30:
                            self.track._queue.get_nowait()

                    frame = await self.track.recv()
                    self._received_frames += 1

                frame_timestamp = datetime.datetime.now()

                workflow_output, _, errors = await self._process_frame_async(
                    frame=frame,
                    frame_id=self._received_frames,
                    render_output=False,
                    include_errors_on_frame=False,
                )

                # Send data via data channel (await for backpressure)
                await self._send_data_output(
                    workflow_output, frame_timestamp, frame, errors
                )

        except asyncio.CancelledError as exc:
            logger.info("Data-only processing cancelled: %s", exc)
        except MediaStreamError as exc:
            logger.info("Stream ended in data-only processing: %s", exc)
        except Exception as exc:
            logger.error("Error in data-only processing: %s", exc)
        finally:
            # Send completion signal to client
            await self._send_processing_complete()

    @staticmethod
    def _ensure_workflow_specification(
        workflow_configuration: WorkflowConfiguration, api_key: str
    ) -> None:
        has_specification = workflow_configuration.workflow_specification is not None
        has_workspace_and_workflow_id = (
            workflow_configuration.workspace_name is not None
            and workflow_configuration.workflow_id is not None
        )

        if not has_specification and not has_workspace_and_workflow_id:
            raise WebRTCConfigurationError(
                "Either 'workflow_specification' or both 'workspace_name' and 'workflow_id' must be provided"
            )

        if not has_specification and has_workspace_and_workflow_id:
            try:
                workflow_configuration.workflow_specification = (
                    get_workflow_specification(
                        api_key=api_key,
                        workspace_id=workflow_configuration.workspace_name,
                        workflow_id=workflow_configuration.workflow_id,
                    )
                )
                workflow_configuration.workspace_name = None
                workflow_configuration.workflow_id = None
            except Exception as e:
                raise WebRTCConfigurationError(
                    f"Failed to fetch workflow specification from API: {str(e)}"
                )

    def _validate_output_fields(
        self, workflow_configuration: WorkflowConfiguration
    ) -> None:
        if workflow_configuration.workflow_specification is None:
            return

        workflow_outputs = workflow_configuration.workflow_specification.get(
            "outputs", []
        )
        available_output_names = [o.get("name") for o in workflow_outputs]

        if self._data_mode == DataOutputMode.SPECIFIC:
            invalid_fields = [
                field
                for field in self.data_output
                if field not in available_output_names
            ]
            if invalid_fields:
                raise WebRTCConfigurationError(
                    f"Invalid data_output fields: {invalid_fields}. "
                    f"Available workflow outputs: {available_output_names}"
                )

        if self.stream_output and self.stream_output not in available_output_names:
            raise WebRTCConfigurationError(
                f"Invalid stream_output field: '{self.stream_output}'. "
                f"Available workflow outputs: {available_output_names}"
            )

    async def _process_frame_async(
        self,
        frame: VideoFrame,
        frame_id: int,
        stream_output: Optional[str] = None,
        render_output: bool = True,
        include_errors_on_frame: bool = True,
    ) -> Tuple[Dict[str, Any], Optional[VideoFrame], List[str]]:
        """Async wrapper for process_frame using executor."""
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(
            None,
            process_frame,
            frame,
            frame_id,
            self._inference_pipeline,
            stream_output,
            render_output,
            include_errors_on_frame,
        )

process_frames_data_only() async

Process frames for data extraction only, without video track output.

This is used when stream_output=[] and no video track is needed.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

async def process_frames_data_only(self):
    """Process frames for data extraction only, without video track output.

    This is used when stream_output=[] and no video track is needed.
    """
    # Silencing swscaler warnings in multi-threading environment
    if not self._av_logging_set:
        av_logging.set_libav_level(av_logging.ERROR)
        self._av_logging_set = True

    logger.info(
        f"Starting data-only frame processing (use_data_channel_frames={self.use_data_channel_frames})"
    )

    try:
        while not self._stop_processing:
            if self._check_termination():
                break
            if self.heartbeat_callback:
                self.heartbeat_callback()

            # Get frame from appropriate source
            if self.use_data_channel_frames:
                # Wait for frame from data channel queue
                item = await self._data_frame_queue.get()
                if item is None:
                    logger.info("Received stop signal from data channel")
                    break
                frame_id, frame = item
                self._received_frames = frame_id
            else:
                # Get frame from media track (existing behavior)
                if not self.track or self.track.readyState == "ended":
                    break

                # Drain queue if using PlayerStreamTrack (RTSP)
                if (
                    isinstance(self.track, PlayerStreamTrack)
                    and self.realtime_processing
                ):
                    while self.track._queue.qsize() > 30:
                        self.track._queue.get_nowait()

                frame = await self.track.recv()
                self._received_frames += 1

            frame_timestamp = datetime.datetime.now()

            workflow_output, _, errors = await self._process_frame_async(
                frame=frame,
                frame_id=self._received_frames,
                render_output=False,
                include_errors_on_frame=False,
            )

            # Send data via data channel (await for backpressure)
            await self._send_data_output(
                workflow_output, frame_timestamp, frame, errors
            )

    except asyncio.CancelledError as exc:
        logger.info("Data-only processing cancelled: %s", exc)
    except MediaStreamError as exc:
        logger.info("Stream ended in data-only processing: %s", exc)
    except Exception as exc:
        logger.error("Error in data-only processing: %s", exc)
    finally:
        # Send completion signal to client
        await self._send_processing_complete()

VideoTransformTrackWithLoop

Bases: VideoStreamTrack, VideoFrameProcessor

Video track that processes frames through workflow and sends video back.

Inherits from both VideoStreamTrack (for WebRTC video track functionality) and VideoFrameProcessor (for workflow processing logic).

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

class VideoTransformTrackWithLoop(VideoStreamTrack, VideoFrameProcessor):
    """Video track that processes frames through workflow and sends video back.

    Inherits from both VideoStreamTrack (for WebRTC video track functionality)
    and VideoFrameProcessor (for workflow processing logic).
    """

    def __init__(
        self,
        asyncio_loop: asyncio.AbstractEventLoop,
        workflow_configuration: WorkflowConfiguration,
        api_key: str,
        model_manager: Optional[ModelManager] = None,
        data_output: Optional[List[str]] = None,
        stream_output: Optional[str] = None,
        has_video_track: bool = True,
        declared_fps: float = 30,
        termination_date: Optional[datetime.datetime] = None,
        terminate_event: Optional[asyncio.Event] = None,
        use_data_channel_frames: bool = False,
        heartbeat_callback: Optional[Callable[[], None]] = None,
        realtime_processing: bool = True,
        *args,
        **kwargs,
    ):
        VideoStreamTrack.__init__(self, *args, **kwargs)
        VideoFrameProcessor.__init__(
            self,
            asyncio_loop=asyncio_loop,
            workflow_configuration=workflow_configuration,
            api_key=api_key,
            data_output=data_output,
            stream_output=stream_output,
            has_video_track=has_video_track,
            declared_fps=declared_fps,
            termination_date=termination_date,
            terminate_event=terminate_event,
            use_data_channel_frames=use_data_channel_frames,
            model_manager=model_manager,
            heartbeat_callback=heartbeat_callback,
            realtime_processing=realtime_processing,
        )

    async def _auto_detect_stream_output(
        self, frame: VideoFrame, frame_id: int
    ) -> None:
        workflow_output_for_detect, _, _ = await self._process_frame_async(
            frame=frame,
            frame_id=frame_id,
            render_output=False,
            include_errors_on_frame=False,
        )
        detected_output = detect_image_output(workflow_output_for_detect)
        if detected_output:
            self.stream_output = detected_output
            logger.info(f"Auto-detected stream_output: {detected_output}")
        else:
            logger.warning("No image output detected, will use fallback")
            self.stream_output = ""

    async def recv(self):
        # Silencing swscaler warnings in multi-threading environment
        if not self._av_logging_set:
            av_logging.set_libav_level(av_logging.ERROR)
            self._av_logging_set = True

        if self.heartbeat_callback:
            self.heartbeat_callback()

        # Check if we should terminate
        if self._check_termination():
            raise MediaStreamError("Processing terminated due to timeout")

        # Wait for track to be ready (video file upload case)
        if self.track is None:
            logger.info("Waiting for track to be ready...")
            await self._track_ready_event.wait()
            if self.track is None:
                raise MediaStreamError("Track not available after wait")

        # Drain queue if using PlayerStreamTrack (RTSP/video file)
        if isinstance(self.track, PlayerStreamTrack) and self.realtime_processing:
            while self.track._queue.qsize() > 30:
                self.track._queue.get_nowait()

        frame: VideoFrame = await self.track.recv()
        self._received_frames += 1
        frame_timestamp = datetime.datetime.now()

        if self.stream_output is None and self._received_frames == 1:
            await self._auto_detect_stream_output(frame, self._received_frames)

        workflow_output, new_frame, errors = await self._process_frame_async(
            frame=frame,
            frame_id=self._received_frames,
            stream_output=self.stream_output,
            render_output=True,
            include_errors_on_frame=True,
        )

        new_frame.pts = frame.pts
        new_frame.time_base = frame.time_base

        await self._send_data_output(workflow_output, frame_timestamp, frame, errors)

        return new_frame

create_chunked_binary_message(frame_id, chunk_index, total_chunks, payload)

Create a binary message with standard 12-byte header.

Format: [frame_id: 4][chunk_index: 4][total_chunks: 4][payload: N] All integers are uint32 little-endian.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

def create_chunked_binary_message(
    frame_id: int, chunk_index: int, total_chunks: int, payload: bytes
) -> bytes:
    """Create a binary message with standard 12-byte header.

    Format: [frame_id: 4][chunk_index: 4][total_chunks: 4][payload: N]
    All integers are uint32 little-endian.
    """
    header = struct.pack("<III", frame_id, chunk_index, total_chunks)
    return header + payload

parse_chunked_binary_message(message)

Parse a binary message with standard 12-byte header.

Returns: (frame_id, chunk_index, total_chunks, payload)

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

def parse_chunked_binary_message(message: bytes) -> Tuple[int, int, int, bytes]:
    """Parse a binary message with standard 12-byte header.

    Returns: (frame_id, chunk_index, total_chunks, payload)
    """
    if len(message) < 12:
        raise ValueError(f"Message too short: {len(message)} bytes (expected >= 12)")

    frame_id, chunk_index, total_chunks = struct.unpack("<III", message[0:12])
    payload = message[12:]
    return frame_id, chunk_index, total_chunks, payload

send_chunked_data(data_channel, frame_id, payload_bytes, chunk_size=CHUNK_SIZE, heartbeat_callback=None) async

Send payload via data channel with rate limiting.

Automatically chunks large payloads and rate limits to prevent SCTP buffer overflow.

Parameters:

Name	Type	Description	Default
data_channel	RTCDataChannel	RTCDataChannel to send on	required
frame_id	int	Frame identifier	required
payload_bytes	bytes	Data to send (JPEG, JSON UTF-8, etc.)	required
chunk_size	int	Maximum chunk size (default 48KB)	CHUNK_SIZE

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

async def send_chunked_data(
    data_channel: RTCDataChannel,
    frame_id: int,
    payload_bytes: bytes,
    chunk_size: int = CHUNK_SIZE,
    heartbeat_callback: Optional[Callable[[], None]] = None,
) -> None:
    """Send payload via data channel with rate limiting.

    Automatically chunks large payloads and rate limits to prevent
    SCTP buffer overflow.

    Args:
        data_channel: RTCDataChannel to send on
        frame_id: Frame identifier
        payload_bytes: Data to send (JPEG, JSON UTF-8, etc.)
        chunk_size: Maximum chunk size (default 48KB)
    """
    if data_channel.readyState != "open":
        logger.warning(f"Cannot send response for frame {frame_id}, channel not open")
        return

    sleep_count = 0
    while data_channel.bufferedAmount > WEBRTC_DATA_CHANNEL_BUFFER_SIZE_LIMIT:
        sleep_count += 1
        if sleep_count % 10 == 0:
            logger.debug(
                "Waiting for data channel buffer to drain. Data channel buffer size: %s",
                data_channel.bufferedAmount,
            )
        if heartbeat_callback:
            heartbeat_callback()
        await asyncio.sleep(WEBRTC_DATA_CHANNEL_BUFFER_DRAINING_DELAY)

    total_chunks = (
        len(payload_bytes) + chunk_size - 1
    ) // chunk_size  # Ceiling division

    if frame_id % 100 == 1:
        logger.info(
            f"Sending response for frame {frame_id}: {total_chunks} chunk(s), {len(payload_bytes)} bytes"
        )

    view = memoryview(payload_bytes)
    for chunk_index in range(total_chunks):
        if data_channel.readyState != "open":
            logger.warning("Channel closed while sending frame %s", frame_id)
            return

        start = chunk_index * chunk_size
        end = min(start + chunk_size, len(payload_bytes))
        chunk_data = view[start:end]

        message = create_chunked_binary_message(
            frame_id, chunk_index, total_chunks, chunk_data
        )
        data_channel.send(message)