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

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,
    ):
        self._loop = asyncio_loop
        self._termination_date = termination_date
        self._terminate_event = terminate_event
        self.track: Optional[RemoteStreamTrack] = 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.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

        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._ensure_workflow_specification(workflow_configuration, api_key)
        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: RemoteStreamTrack):
        if not self.track:
            self.track = track

    def close(self):
        self._track_active = False
        self._stop_processing = True

    def _check_termination(self):
        """Check if we should terminate based on timeout"""
        if (
            self._termination_date
            and self._termination_date < datetime.datetime.now()
            and self._terminate_event
            and not 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 = json.dumps(webrtc_output.model_dump()).encode("utf-8")
            send_chunked_data(self.data_channel, self._received_frames, json_bytes)
            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")
        send_chunked_data(self.data_channel, self._received_frames, json_bytes)

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

            nparr = np.frombuffer(jpeg_bytes, np.uint8)
            np_image = cv2.imdecode(nparr, cv2.IMREAD_COLOR)

            if np_image is None:
                logger.error(f"Failed to decode JPEG for frame {frame_id}")
                return

            video_frame = VideoFrame.from_ndarray(np_image, format="bgr24")
            await self._data_frame_queue.put((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

                # 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):
                        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 self._send_data_output(
                    workflow_output, frame_timestamp, frame, errors
                )

        except asyncio.CancelledError:
            logger.info("Data-only processing cancelled")
        except MediaStreamError:
            logger.info("Stream ended in data-only processing")
        except Exception as exc:
            logger.error("Error in data-only processing: %s", exc)

    @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

            # 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):
                    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 self._send_data_output(
                workflow_output, frame_timestamp, frame, errors
            )

    except asyncio.CancelledError:
        logger.info("Data-only processing cancelled")
    except MediaStreamError:
        logger.info("Stream ended in data-only processing")
    except Exception as exc:
        logger.error("Error in data-only processing: %s", exc)

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,
        *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,
        )

    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

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

        # Drain queue if using PlayerStreamTrack (RTSP)
        if isinstance(self.track, PlayerStreamTrack):
            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)

Send payload via data channel, automatically chunking if needed.

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

def send_chunked_data(
    data_channel: RTCDataChannel,
    frame_id: int,
    payload_bytes: bytes,
    chunk_size: int = CHUNK_SIZE,
) -> None:
    """Send payload via data channel, automatically chunking if needed.

    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

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

    for chunk_index in range(total_chunks):
        start = chunk_index * chunk_size
        end = min(start + chunk_size, len(payload_bytes))
        chunk_data = payload_bytes[start:end]

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