WebRTC Streaming

WebRTC के माध्यम से video को Inference server पर stream करें और webcam, RTSP cameras, video files, या manually भेजे गए frames से, model ID या Workflow का उपयोग करके live predictions प्राप्त करें।

इस्तेमाल करें इन्फरेंस-sdk मॉडल या वर्कफ़्लो के माध्यम से वीडियो स्ट्रीम करने के लिए WebRTC क्लाइंट। वीडियो फ़्रेम एक कनेक्शन के माध्यम से Inference Server तक जाते हैं, और प्रोसेस किए गए फ़्रेम तथा प्रेडिक्शन डेटा लगातार वापस आते हैं।

यही क्लाइंट self-hosted Inference Server और Serverless Video Streaming API के साथ काम करता है। सेट करें api_url को उस runtime पर जिसे आप चाहते हैं:

• Self-hosted: http://localhost:9001

• Serverless: https://serverless.roboflow.com

WebRTC स्ट्रीमिंग के लिए अतिरिक्त dependencies की आवश्यकता होती है:

pip install "inference-sdk[webrtc]"

मॉडल स्ट्रीम करें

एक model_id पास करें ताकि एक मॉडल के माध्यम से वीडियो स्ट्रीम किया जा सके। SDK आवश्यक single-model Workflow बनाता है, और आपका on_frame handler प्रत्येक वीडियो फ़्रेम को उसके prediction data के साथ प्राप्त करता है:

import cv2
का उपयोग करके visualize करते हैं
from inference_sdk import InferenceHTTPClient
from inference_sdk.webrtc import WebcamSource

# ROBOFLOW_API_KEY को अपने Roboflow API Key से बदलें
client = InferenceHTTPClient(
    api_url="http://localhost:9001",
    api_key="ROBOFLOW_API_KEY",
)

session = client.webrtc.stream(
    source=WebcamSource(),
    model_id="rfdetr-nano",
)

box_annotator = sv.BoxAnnotator()

@session.on_frame
def show(frame, data):
    # data raw predictions dict है, जैसा कि server द्वारा बिल्कुल वैसा ही लौटाया गया है
    # (जब इस frame के लिए predictions उपलब्ध न हों, तब None)
    if data is None:
        return
    detections = sv.Detections.from_inference(data)
    annotated = box_annotator.annotate(frame.copy(), detections)
    cv2.imshow("Preview", annotated)
    if cv2.waitKey(1) & 0xFF == ord("q"):
        session.close()

session.run()  # stream समाप्त होने तक या session.close() बुलाए जाने तक ब्लॉक करता है

model_id generic Workflow model block के साथ किसी भी task type के लिए काम करता है। मॉडल का task type Roboflow API lookup के माध्यम से अपने-आप resolve हो जाता है, और matching model block आपके लिए चुना जाता है। समर्थित task types:

• object-detection

• instance-segmentation

• semantic-segmentation

• classification

• multi-label-classification

• keypoint-detection

data serialized predictions dict को ज्यों का त्यों पास किया जाता है - इसका shape task type के अनुसार होता है। detection-family models के लिए यह inference-response-shaped होता है, इसलिए आप इसे matching supervision helper - sv.Detections.from_inference(data) का उपयोग object detection और instance segmentation के लिए करें, sv.KeyPoints.from_inference(data) का उपयोग keypoint models के लिए करें। Classification predictions में top/confidence keys होते हैं, और semantic-segmentation predictions में run-length-encoded masks (rle_mask) होते हैं जिन्हें आप स्वयं decode करते हैं।

जब किसी frame के लिए predictions उपलब्ध नहीं होते (उदा., live stream frame के लिए paired prediction message कभी नहीं पहुँचा), data है None - उपयोग से पहले अपने handler में इसकी जाँच करें।

VLMs को model_id mode में support नहीं किया जाता (प्रत्येक VLM family का अपना dedicated Workflow block होता है, इसलिए उन्हें wrap करने के लिए कोई generic block नहीं है) - इसके बजाय उन्हें एक full workflow के साथ stream करें।

task-type lookup को छोड़ना: पास करें task_type को network call से बचने के लिए explicitly - air-gapped या self-hosted deployments के लिए उपयोगी:

session = client.webrtc.stream(
    source=WebcamSource(),
    model_id="my-project/3",
    task_type="object-detection",
)

में model_id mode, on_frame handlers इनमें से कोई भी ले सकते हैं (frame, data) या (frame, data, metadata) - तीसरा argument VideoMetadata frame के लिए है।

Workflow स्ट्रीम करें

multi-step pipelines के लिए, workflow पास करें model_idके बजाय एक

Workflow ID

import cv2
from inference_sdk import InferenceHTTPClient
from inference_sdk.webrtc import WebcamSource, StreamConfig

client = InferenceHTTPClient(
    api_url="http://localhost:9001",
    api_key="ROBOFLOW_API_KEY",
)

session = client.webrtc.stream(
    source=WebcamSource(),
    workflow="my-workflow-id",
    workspace="my-workspace-name",
    config=StreamConfig(
        stream_output=["output_image"],   # workflow output वीडियो के रूप में वापस stream होता है
        data_output=["predictions"],      # workflow outputs data channel के माध्यम से भेजे जाते हैं
    ),
)

@session.on_frame
def show(frame, metadata):
    cv2.imshow("Preview", frame)
    if cv2.waitKey(1) & 0xFF == ord("q"):
        session.close()

@session.on_data("predictions")
def handle_predictions(predictions, metadata):
    print(f"Frame {metadata.frame_id}: {predictions}")

session.run()

Workflow विनिर्देश

import cv2
from inference_sdk import InferenceHTTPClient
from inference_sdk.webrtc import WebcamSource, StreamConfig

client = InferenceHTTPClient(
    api_url="http://localhost:9001",
    api_key="ROBOFLOW_API_KEY",
)

workflow_spec = {
    "version": "1.0",
    "inputs": [{"type": "InferenceImage", "name": "image"}],
    "steps": [
        {
            "type": "roboflow_core/roboflow_object_detection_model@v2",
            "name": "model",
            "images": "$inputs.image",
            "model_id": "rfdetr-nano",
        }
    ],
    "outputs": [
        {
            "type": "JsonField",
            "name": "predictions",
            "selector": "$steps.model.predictions",
        },
        {"type": "JsonField", "name": "image", "selector": "$inputs.image"},
    ],
}

session = client.webrtc.stream(
    source=WebcamSource(),
    workflow=workflow_spec,
    config=StreamConfig(
        stream_output=["image"],
        data_output=["predictions"],
    ),
)

@session.on_frame
def show(frame, metadata):
    cv2.imshow("Preview", frame)
    if cv2.waitKey(1) & 0xFF == ord("q"):
        session.close()

session.run()

नोट्स:

• workflow और model_id आपस में परस्पर exclusive हैं - ठीक एक पास करें।

• workspace की आवश्यकता होती है जब workflow एक ID string हो; specification dict के लिए इसकी आवश्यकता नहीं होती।

• image_input (default "image") Workflow image input का नाम बताता है जिससे वीडियो फ़्रेम bind होते हैं।

• workflow mode में, on_frame handlers को प्राप्त होते हैं (frame, metadata) - prediction data अलग से on_data handlers के माध्यम से आता है, जिन्हें data_output नामों द्वारा route किया जाता है StreamConfig.

वीडियो स्रोत

का पहला argument stream() यह चुनता है कि वीडियो कहाँ से आएगा:

from inference_sdk.webrtc import (
    WebcamSource,
    RTSPSource,
    LocalStreamSource,
    MJPEGSource,
    VideoFileSource,
    ManualSource,
)

WebcamSource

स्थानीय कैमरा डिवाइस से फ़्रेम कैप्चर करता है और उन्हें server को भेजता है:

source = WebcamSource()                                    # default camera
source = WebcamSource(device_id=1, resolution=(1920, 1080))

कैमरे का FPS अपने-आप detect होता है और server को रिपोर्ट किया जाता है।

RTSPSource

यह server RTSP camera से जुड़ता है और processed वीडियो आपको वापस stream करता है - जब camera server से पहुँचा जा सकता हो तब इसका उपयोग करें:

source = RTSPSource("rtsp://user:[e-mail]/stream")

LocalStreamSource

एक RTSP/RTMP stream को कैप्चर करता है स्थानीय रूप से (client machine पर) और फ़्रेम server को भेजता है - जब camera केवल आपकी machine से reachable हो, server से नहीं, तब इसका उपयोग करें:

source = LocalStreamSource("rtsp://192.168.1.10/stream")   # साथ ही rtsps://, rtmp://, rtmps://

MJPEGSource

जैसे RTSPSourceलेकिन server द्वारा captured MJPEG streams के लिए:

source = MJPEGSource("http://camera.local/mjpeg")

VideoFileSource

data channel के माध्यम से एक video file को server पर upload करता है; server उसे process करता है और परिणाम वापस stream करता है। prerecorded video के लिए frame-by-frame streaming से अधिक efficient:

source = VideoFileSource("video.mp4")

# upload progress track करें और मूल FPS (live-preview pacing) पर process करें
source = VideoFileSource(
    "video.mp4",
    on_upload_progress=lambda uploaded, total: print(f"{uploaded}/{total} chunks"),
    realtime_processing=True,   # default False = जितनी तेज़ी से संभव हो process करें
)

डिफ़ॉल्ट रूप से फ़्रेम data channel के माध्यम से वापस आते हैं (order और quality की गारंटी के साथ)। पास करें use_datachannel_frames=False ताकि उनके बजाय hardware-accelerated WebRTC video track के माध्यम से उन्हें प्राप्त किया जा सके (कम bandwidth)।

ManualSource

फ़्रेम programmatically भेजें - तब उपयोगी जब फ़्रेम custom pipeline से आते हों:

import threading
import time

import cv2
from inference_sdk.webrtc import ManualSource, StreamConfig

source = ManualSource()
session = client.webrtc.stream(
    source=source,
    model_id="rfdetr-nano",
    config=StreamConfig(declared_fps=30),
)

@session.on_frame
def handle(frame, data):
    print(data)

# run() connection स्थापित करता है और handlers को dispatch करता है; इसे एक
# background thread में शुरू करें ताकि यह thread फ़्रेम feed कर सके।
threading.Thread(target=session.run, daemon=True).start()

cap = cv2.VideoCapture("video.mp4")
while True:
    ret, frame = cap.read()
    if not ret:
        break
    try:
        source.send(frame)   # BGR numpy array
    except RuntimeError:
        pass                 # session अभी connect हो रहा है - frame छोड़ दिया गया
    time.sleep(1 / 30)       # भेजने की गति घोषित FPS के अनुसार रखें

session.close()

send() उत्पन्न करता है RuntimeError जब तक connection स्थापित नहीं हो जाता, और यदि आप stream के consume होने से तेज़ भेजते हैं तो queued frames oldest-first के अनुसार drop हो जाते हैं। ManualSource में FPS auto-detection नहीं है, इसलिए frame rate को StreamConfig(declared_fps=...).

परिभाषित करें

परिणामों का उपभोग

stream() लौटाता है एक WebRTCSession. Connection first use पर lazily शुरू होता है (run(), video(), या wait()) और resources release करने के लिए इसे बंद करना आवश्यक है। तीन समकक्ष patterns:

# 1. run() - बाहर निकलते ही auto-closes होता है (handlers के साथ recommended)
session.run()

# 2. Context manager - बाहर निकलते ही auto-closes होता है (video() iterator के साथ recommended)
with client.webrtc.stream(source=source, model_id="rfdetr-nano") as session:
    for frame, data in session.video():
        ...

# 3. Manual - आपको स्वयं close() कॉल करना होगा
session = client.webrtc.stream(source=source, model_id="rfdetr-nano")
for frame, data in session.video():
    ...
session.close()

session.close() idempotent है और handler के अंदर से कॉल करने के लिए सुरक्षित है - यह run() और video() iterator समाप्त कर देता है। session.wait(timeout=None) आपके द्वारा स्वयं फ़्रेम consume किए बिना stream समाप्त होने तक ब्लॉक करता है।

फ़्रेम प्राप्त करना: on_frame और video()

@session.on_frame जब run(). session.video() का उपयोग किया जाता है, तो हर processed video frame के लिए बुलाए जाने वाला handler पंजीकृत करता है।

# model_id mode: (frame, data) - data raw predictions dict है
# (जब frame के लिए predictions उपलब्ध न हों, तब None)
for frame, data in session.video():
    ...

# workflow mode: (frame, metadata)
for frame, metadata in session.video():
    ...

फ़्रेम BGR numpy arrays होते हैं। यदि आपका handler realtime mode में पीछे रह जाता है, तो सबसे पुराने frames drop कर दिए जाते हैं ताकि stream live बनी रहे।

डेटा प्राप्त करना: on_data

में सूचीबद्ध Workflow outputs StreamConfig.data_output data channel के माध्यम से आते हैं। हर output name के लिए अलग handlers पंजीकृत करें, या पूरे payload के लिए एक global handler:

@session.on_data("predictions")           # एक single output field
def handle_predictions(predictions, metadata):
    print(f"Frame {metadata.frame_id}: {predictions}")

@session.on_data                          # global: full output dict
def handle_all(data, metadata):
    print(data)

Handlers स्वीकार कर सकते हैं (value, metadata) या केवल (value) - signature अपने-आप detect हो जाता है।

त्रुटियों को संभालना: on_error

Server हर data channel message के साथ per-frame errors (workflow execution failures, output serialization failures) रिपोर्ट करता है। on_error handlers केवल non-empty error list वाले frames के लिए चलेंगे:

@session.on_error
def on_err(errors, metadata):
    print(f"Frame {metadata.frame_id} failed: {errors}")
    session.close()   # उदाहरण: पहली error पर बाहर निकलें

ये server-side per-frame failures हैं; connection और setup errors exceptions के रूप में run() से दिखाई देते हैं। Errors metadata.errors से भी हर frame पर जुड़ी होती हैं, इसलिए on_frame / on_data handlers उन्हें सीधे inspect कर सकते हैं।

Frame metadata

VideoMetadata हर frame और data message के साथ आती है:

Attribute	विवरण
frame_id	stream में frame का अद्वितीय पहचानकर्ता
received_at	जब server ने frame प्राप्त किया
pts / time_base	वीडियो stream का presentation timestamp
declared_fps / measured_fps	घोषित बनाम मापा गया stream FPS
errors	server द्वारा रिपोर्ट की गई per-frame errors (जब frame साफ़-साफ़ process हुआ हो तब empty)

StreamConfig

StreamConfig output routing, processing behavior, और network settings को नियंत्रित करता है:

from inference_sdk.webrtc import StreamConfig

config = StreamConfig(
    stream_output=["output_image"],
    data_output=["predictions"],
    realtime_processing=True,
)
session = client.webrtc.stream(source=source, workflow="...", workspace="...", config=config)

फ़ील्ड	डिफ़ॉल्ट	विवरण
stream_output	[]	Workflow output names जो video के रूप में वापस stream होते हैं
data_output	[]	Workflow output names जो data channel के माध्यम से दिए जाते हैं
realtime_processing	True	real time में बने रहने के लिए frames drop करें; सेट करें False ताकि हर frame queue होकर process हो
declared_fps	None	auto-detection के बिना sources के लिए FPS declaration (उदा., ManualSource)
turn_server	None	TURN server config: {"urls": "turn:...", "username": "...", "credential": "..."}
workflow_parameters	{}	Workflow execution को दिए गए parameters
requested_plan	None	Roboflow serverless endpoints के लिए compute plan (उदा., "webrtc-gpu-small")
requested_region	None	serverless endpoints के लिए processing region (उदा., "us", "eu")
processing_timeout	None	server-side session time limit, seconds में (serverless endpoints)

में model_id mode, empty stream_output / data_output अपने-आप भर दिए जाते हैं (["image"] और ["predictions"]); आपके द्वारा दिए गए अन्य सभी settings सुरक्षित रखे जाते हैं।

TURN servers: Roboflow-hosted endpoints से जुड़ते समय TURN configuration अपने-आप fetch हो जाती है। सख्त NATs या firewalls के पीछे self-hosted servers के लिए, turn_server को explicitly प्रदान करें; जब यह set नहीं होता तो direct connection का प्रयास किया जाता है।

चलाने योग्य उदाहरण

पूरे working scripts यहाँ उपलब्ध हैं examples/webrtc_sdk/ Inference repository की directory में:

• webcam_basic.py - बेसिक वेबकैम स्ट्रीमिंग

• rtsp_basic.py - RTSP स्ट्रीम प्रोसेसिंग

• mjpeg_basic.py - MJPEG स्ट्रीम प्रोसेसिंग

• video_file_basic.py - आउटपुट सहेजने के साथ वीडियो फ़ाइल प्रोसेसिंग

आप Roboflow Serverless Hosted API (https://serverless.roboflow.com) बिना किसी सेटअप के, या विकास के लिए स्थानीय सर्वर के विरुद्ध स्ट्रीम कर सकते हैं:

# CPU
docker run -p 9001:9001 roboflow/roboflow-inference-server-cpu:latest

# GPU
docker run --gpus all -p 9001:9001 roboflow/roboflow-inference-server-gpu:latest