WebRTC ストリーミング

WebRTC 経由で動画を Inference server にストリーミングし、ウェブカメラ、RTSP カメラ、動画ファイル、または手動送信フレームから、モデル ID または Workflow を使ってライブ予測を受け取ります。

次を使用します inference-sdk WebRTCクライアントは、モデルまたはWorkflowを通して動画をストリーミングします。動画フレームは1つの接続でInference Serverに送られ、処理済みフレームと予測データが継続的に返されます。

同じクライアントは、セルフホストのInference ServerとServerless Video Streaming APIで動作します。設定 api_url を、使用したいランタイムに:

セルフホスト: http://localhost:9001
サーバーレス: https://serverless.roboflow.com

WebRTCストリーミングには追加の依存関係が必要です:

pip install "inference-sdk[webrtc]"

モデルをストリーミングする

次を渡します model_id 1つのモデルを通して動画をストリーミングします。SDKが必要な単一モデルのWorkflowを構築し、あなたの on_frame ハンドラーは各動画フレームを予測データとともに受け取ります:

import cv2
import supervision as sv
from inference_sdk import InferenceHTTPClient
from inference_sdk.webrtc import WebcamSource

# ROBOFLOW_API_KEY を Roboflow の API キーに置き換えてください
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 は、サーバーから返されたそのままの生の予測dictです
    # （このフレームで予測が利用できない場合はNone）
    if data is None:
        return
    detections = sv.Detections.from_inference(data)
    annotated = box_annotator.annotate(frame.copy(), detections)
    cv2.imshow("プレビュー", annotated)
    if cv2.waitKey(1) & 0xFF == ord("q"):
        session.close()

session.run()  # ストリームが終了するか session.close() が呼ばれるまでブロックします

model_id 汎用のWorkflowモデルブロックを使えば、任意のタスクタイプで動作します。モデルのタスクタイプはRoboflow APIの検索によって自動的に解決され、対応するモデルブロックが選択されます。対応するタスクタイプ:

object-detection
instance-segmentation
semantic-segmentation
classification
multi-label-classification
keypoint-detection

data は、シリアライズされた予測dictがそのまま渡されます - その形状はタスクタイプに従います。検出系モデルではinference-response形式なので、対応する supervision ヘルパー - sv.Detections.from_inference(data) 物体検出およびインスタンスセグメンテーション用、 sv.KeyPoints.from_inference(data) キーポイントモデル用。分類予測には 上位/confidence キーが含まれ、セマンティックセグメンテーション予測にはランレングス符号化マスク（rle_mask）が含まれます。これは自分でデコードしてください。

フレームの予測が利用できない場合（例: ライブストリームのフレームに対応する予測メッセージが届かなかった場合）、 data は None - 使用前にハンドラーで確認してください。

VLMは model_id モードではサポートされていません（各VLMファミリには専用のWorkflowブロックがあるため、包み込むための汎用ブロックがありません） - 完全な ワークフロー の代わりにストリーミングしてください。

タスクタイプ検索をスキップする場合: を渡して task_type を明示的に指定してネットワーク呼び出しを回避します - エアギャップ環境やセルフホスト環境で便利です:

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

では、 model_id モードでは、 on_frame ハンドラーは次のいずれかを受け取れます (frame, data) または (frame, data, metadata) - 3番目の引数は VideoMetadata です。

Workflowをストリーミングする

マルチステップのパイプラインでは、次を渡します ワークフロー の代わりに model_id。RoboflowワークスペースにIDで保存されたWorkflowを参照するか、完全な仕様dictを指定してください:

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の出力が動画として返されます
        data_output=["predictions"],      # workflowの出力がデータチャネル経由で送られます
    ),
)

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

@session.on_data("predictions")
def handle_predictions(predictions, metadata):
    print(f"フレーム {metadata.frame_id}: {predictions}")

session.run()

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("プレビュー", frame)
    if cv2.waitKey(1) & 0xFF == ord("q"):
        session.close()

session.run()

注記:

ワークフロー と model_id は排他的です - 1つだけ渡してください。
workspace が必要なのは ワークフロー がID文字列の場合です。仕様dictでは不要です。
image_input （デフォルト "image"）は、動画フレームをバインドするWorkflowの画像入力名を指定します。
workflowモードでは、 on_frame ハンドラーは (frame, metadata) - 予測データは別途次を通じて届きます on_data ハンドラーへ、次でルーティングされます data_output の名前で StreamConfig.

動画ソース

の最初の引数は stream() 動画の送信元を選択します:

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

WebcamSource

ローカルのカメラデバイスからフレームを取得し、サーバーへ送信します:

source = WebcamSource()                                    # デフォルトカメラ
source = WebcamSource(device_id=1, resolution=(1920, 1080))

カメラのFPSは自動検出され、サーバーに報告されます。

RTSPSource

この サーバー RTSPカメラに接続し、処理済み動画をあなたに返します - カメラがサーバーから到達可能な場合に使用してください:

source = RTSPSource("rtsp://user:pass@camera.local/stream")

LocalStreamSource

RTSP/RTMPストリームを取得します ローカルで （クライアントマシン上で）フレームを取得し、サーバーへ送信します - カメラがあなたのマシンからのみ到達可能で、サーバーからは到達できない場合に使用してください:

source = LocalStreamSource("rtsp://192.168.1.10/stream")   # rtsps://、rtmp://、rtmps:// も可

MJPEGSource

と同様に RTSPSource、ただしサーバーが取得するMJPEGストリーム向けです:

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

VideoFileSource

データチャネル経由で動画ファイルをサーバーへアップロードします。サーバーが処理し、結果をストリームで返します。録画済み動画ではフレームごとのストリーミングより効率的です:

source = VideoFileSource("video.mp4")

# アップロード進捗を追跡し、元のFPSで処理します（ライブプレビューのペース）
source = VideoFileSource(
    "video.mp4",
    on_upload_progress=lambda uploaded, total: print(f"{uploaded}/{total} chunks"),
    realtime_processing=True,   # デフォルト False = 可能な限り高速に処理
)

デフォルトでは、フレームはデータチャネル経由で返されます（順序と品質が保証されます）。次を渡すと use_datachannel_frames=False ハードウェアアクセラレーションされたWebRTC動画トラック経由で受け取れます（帯域幅が少ない）。

ManualSource

フレームをプログラムで送信します - フレームが独自のパイプラインから来る場合に便利です:

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() は接続を確立し、ハンドラーをディスパッチします; これは次の中で開始してください
# バックグラウンドスレッド。そうすればこのスレッドでフレームを供給できます。
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配列
    except RuntimeError:
        pass                 # セッションはまだ接続中 - フレームはスキップされます
    time.sleep(1 / 30)       # 送信を宣言したFPSに合わせます

session.close()

send() は次を送出します RuntimeError 接続が確立されるまで、またストリームの消費より速く送ると、キューに入ったフレームは古いものから順に破棄されます。 ManualSource FPSの自動検出はないため、フレームレートを次で宣言します StreamConfig(declared_fps=...).

結果の取得

セッションのライフサイクル

stream() は WebRTCSessionを返します。接続は初回使用時に遅延開始され（run(), video()、または wait()）リソースを解放するには閉じる必要があります。等価な3つのパターン:

# 1. run() - 終了時に自動で閉じます（ハンドラーと併用する場合に推奨）
session.run()

# 2. コンテキストマネージャー - 退出時に自動で閉じます（video()イテレーターと併用する場合に推奨）
with client.webrtc.stream(source=source, model_id="rfdetr-nano") as session:
    for frame, data in session.video():
        ...

# 3. 手動 - 自分で close() を呼ぶ必要があります
session = client.webrtc.stream(source=source, model_id="rfdetr-nano")
for frame, data in session.video():
    ...
session.close()

session.close() は冪等で、ハンドラー内から呼んでも安全です - これは run() および video() イテレーターを終了します。 session.wait(timeout=None) 自分でフレームを消費せずに、ストリームが終了するまでブロックします。

フレームを受け取る: `on_frame` と `video()`

@session.on_frame を使用する際、処理済みの各動画フレームで呼び出されるハンドラーを登録します run(). session.video() はイテレーター版です - 同じデータをプル型で取得します:

# model_idモード: (frame, data) - data は生の予測dictです
# （このフレームで予測が利用できない場合はNone）
for frame, data in session.video():
    ...

# workflowモード: (frame, metadata)
for frame, metadata in session.video():
    ...

フレームはBGRのnumpy配列です。realtimeモードでハンドラーが追いつかない場合、ストリームがライブのまま維持されるように古いフレームが破棄されます。

データを受け取る: `on_data`

に列挙されたWorkflow出力は StreamConfig.data_output がデータチャネル経由で届きます。出力名ごとにハンドラーを登録するか、ペイロード全体に対するグローバルハンドラーを1つ登録してください:

@session.on_data("predictions")           # 単一の出力フィールド
def handle_predictions(predictions, metadata):
    print(f"フレーム {metadata.frame_id}: {predictions}")

@session.on_data                          # グローバル: 出力dict全体
def handle_all(data, metadata):
    print(data)

ハンドラーは次のいずれかを受け取れます (value, metadata) または単に (value) - シグネチャは自動検出されます。

エラーの処理: `on_error`

サーバーは各データチャネルメッセージとともに、フレームごとのエラー（workflow実行失敗、出力シリアライズ失敗）を報告します。 on_error ハンドラーは、空でないエラー一覧を持つフレームに対してのみ起動します:

@session.on_error
def on_err(errors, metadata):
    print(f"フレーム {metadata.frame_id} が失敗しました: {errors}")
    session.close()   # 例: 最初のエラーで中断

これらはサーバー側のフレームごとの失敗です; 接続およびセットアップのエラーは、次からの例外として表面化します run() その代わりに。エラーは次にも付与されます metadata.errors 各フレームに対して付与されるため、 on_frame / on_data ハンドラーから直接確認できます。

フレームメタデータ

VideoMetadata は各フレームとデータメッセージに付随します:

属性

説明

frame_id

ストリーム内のフレームの一意の識別子

received_at

サーバーがフレームを受信した時刻

pts / time_base

動画ストリームのプレゼンテーションタイムスタンプ

declared_fps / measured_fps

宣言されたFPSと実測FPS

errors

サーバーが報告するフレームごとのエラー（フレームが正常に処理された場合は空）

StreamConfig

StreamConfig 出力ルーティング、処理動作、ネットワーク設定を制御します:

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出力名

data_output

[]

データチャネル経由で送られるWorkflow出力名

realtime_processing

True

リアルタイムで追従するためにフレームを間引きます; 次を設定すると False すべてのフレームをキューに入れて処理します

declared_fps

None

自動検出のないソース向けのFPS宣言（例: ManualSource)

turn_server

None

TURNサーバー設定: {"urls": "turn:...", "username": "...", "credential": "..."}

workflow_parameters

{}

Workflow実行に渡されるパラメータ

requested_plan

None

Roboflowのサーバーレスエンドポイント向けの計算プラン（例: "webrtc-gpu-small")

requested_region

None

サーバーレスエンドポイントの処理リージョン（例: "us", "eu")

processing_timeout

None

サーバー側セッションのタイムリミット（秒）（サーバーレスエンドポイント）

では、 model_id モードでは、空の stream_output / data_output は自動的に埋められます（["image"] と ["predictions"]）; それ以外に指定した設定は保持されます。

TURNサーバー: Roboflowホストのエンドポイントに接続する際、TURN設定は自動的に取得されます。制限のあるNATやファイアウォールの背後にあるセルフホストサーバーでは、次を指定してください turn_server を明示的に。設定されていない場合は直接接続が試行されます。

実行可能な例

完全に動作するスクリプトは次にあります examples/webrtc_sdk/ Inferenceリポジトリのディレクトリ:

webcam_basic.py - 基本的なWebカメラ配信
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

前へモデル管理次へワークフロー

最終更新 6 日前

役に立ちましたか？

モデルをストリーミングする

Workflowをストリーミングする

動画ソース

WebcamSource

RTSPSource

LocalStreamSource

MJPEGSource

VideoFileSource

ManualSource

結果の取得

セッションのライフサイクル

フレームを受け取る: on_frame と video()

データを受け取る: on_data

エラーの処理: on_error

フレームメタデータ

StreamConfig

実行可能な例

フレームを受け取る: `on_frame` と `video()`

データを受け取る: `on_data`

エラーの処理: `on_error`