> For the complete documentation index, see [llms.txt](https://docs.roboflow.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.roboflow.com/reference/ja/inference/inference-sdk/webrtc.md).

# WebRTC ストリーミング

Python WebRTC クライアントを使って、セルフホスト型の推論サーバーに動画をストリーミングし、既存のレガシーなホスト型統合も維持します。

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

{% hint style="info" %}
新しい管理対象のライブカメラ環境では、 [Cloud Streaming](https://docs.roboflow.com/deployment/roboflow-cloud/cloud-streaming). この WebRTC リファレンスは、セルフホストの Inference Server および非推奨の [Serverless Video Streaming API](https://docs.roboflow.com/deployment/roboflow-cloud/serverless-api/serverless-video-streaming-api). Cloud Streaming は `api_url` このクライアントの代替ではありません。
{% endhint %}

設定してください `api_url` この WebRTC クライアントで使用するランタイムの

* セルフホスト: `http://localhost:9001`
* 非推奨の Serverless Video Streaming API: `https://serverless.roboflow.com`

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

```bash
pip install "inference-sdk[webrtc]"
```

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

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

```python
import cv2
import supervision as sv
from inference_sdk import InferenceHTTPClient, InferenceConfiguration
from inference_sdk.webrtc import WebcamSource

# ROBOFLOW_API_KEY をあなたの Roboflow API キーに置き換えてください
client = InferenceHTTPClient(
    api_url="http://localhost:9001",
    api_key="ROBOFLOW_API_KEY",
).configure(InferenceConfiguration(api_key_transport="header"))

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 の照会によって自動的に解決され、対応するモデルブロックが選択されます。対応するタスクタイプ:

* `オブジェクト検出`
* `インスタンスセグメンテーション`
* `セマンティックセグメンテーション`
* `分類`
* `マルチラベル分類`
* `キーポイント検出`

`data` は、シリアライズされた predictions dict がそのまま渡されます。その形状はタスクタイプに従います。検出系モデルでは inference-response 形式なので、対応する [`supervision`](https://supervision.roboflow.com/) ヘルパーを使って変換できます - `sv.Detections.from_inference(data)` オブジェクト検出とインスタンスセグメンテーションでは、 `sv.KeyPoints.from_inference(data)` キーポイントモデルでは、分類予測には `トップ`/`confidence` キーが含まれ、セマンティックセグメンテーションの予測にはランレングスエンコードされたマスク（`rle_mask`）が含まれ、それを自分でデコードします。

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

VLM は `model_id` モードではサポートされていません（各 VLM ファミリーには専用の Workflow ブロックがあるため、それらを包む汎用ブロックがありません）- 代わりに完全な [`workflow`](#streaming-a-workflow) を使ってストリーミングしてください。

**タスクタイプの検索をスキップする:** 渡してください `task_type` を明示的に渡してネットワーク呼び出しを避けます - エアギャップ環境やセルフホスト環境で便利です:

```python
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`](#frame-metadata) そのフレームの VideoMetadata です。

## Workflow をストリーミングする

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

{% tabs %}
{% tab title="Workflow ID" %}

```python
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()
```

{% endtab %}

{% tab title="Workflow 仕様" %}

```python
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()
```

{% endtab %}
{% endtabs %}

注意:

* `workflow` および `model_id` は排他的です - どちらか 1 つだけを渡してください。
* `workspace` が必要なのは `workflow` が ID 文字列の場合です。仕様 dict には不要です。
* `image_input` （デフォルト `"image"`）は、動画フレームがバインドされる Workflow の画像入力の名前を指定します。
* workflow モードでは、 `on_frame` ハンドラーは受け取ります `(frame, metadata)` - 予測データは別途次を通じて届きます [`on_data`](#receiving-data-on_data) ハンドラーへ、次によってルーティングされます `data_output` の名前に基づいて `StreamConfig`.

## 動画ソース

の最初の引数は `stream()` 動画の取得元を指定します:

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

### WebcamSource

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

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

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

### RTSPSource

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

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

### LocalStreamSource

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

```python
source = LocalStreamSource("rtsp://192.168.1.10/stream")   # also rtsps://, rtmp://, rtmps://
```

### MJPEGSource

と同様ですが `RTSPSource`、サーバーが取得する MJPEG ストリーム用です:

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

### VideoFileSource

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

```python
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

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

```python
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 array
    except RuntimeError:
        pass                 # セッションはまだ接続中 - フレームはスキップされます
    time.sleep(1 / 30)       # 宣言した FPS に合わせて送信速度を調整

session.close()
```

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

## 結果の消費

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

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

```python
# 1. run() - 終了時に自動で close します（ハンドラー使用時に推奨）
session.run()

# 2. コンテキストマネージャー - 終了時に自動で close します（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()` はイテレーター版に相当します - 同じデータを pull ベースで取得します:

```python
# 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 つ登録できます:

```python
@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` ハンドラーは、空でないエラーリストを持つフレームに対してのみ発火します:

```python
@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` は WebRTC の出力ルーティング、処理動作、ネットワーク設定を制御します。その `requested_plan`, `requested_region`、および `processing_timeout` フィールドは、Cloud Streaming のジョブ設定ではなく、従来のホスト型セッションを表します:

```python
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)
```

| フィールド                | デフォルト  | 説明                                                                         |
| -------------------- | ------ | -------------------------------------------------------------------------- |
| `ストリーム出力`            | `[]`   | ワークフローの出力名を動画としてストリーミングで返します                                               |
| `data_output`        | `[]`   | データチャネル経由で配信されるワークフロー出力名                                                   |
| `リアルタイム処理`           | `True` | リアルタイムを維持するためにフレームを落とします。設定は `False` すべてのフレームをキューに入れて処理するには                |
| `declared_fps`       | `None` | 自動検出のないソース向けのFPS指定（例： `ManualSource`)                                      |
| `TURNサーバー`           | `None` | TURNサーバー設定： `{"urls": "turn:...", "username": "...", "credential": "..."}` |
| `ワークフローパラメータ`        | `{}`   | ワークフロー実行に渡されるパラメータ                                                         |
| `requested_plan`     | `None` | Roboflowのサーバーレスエンドポイント向けのコンピュートプラン（例： `"webrtc-gpu-small"`)                |
| `requested_region`   | `None` | サーバーレスエンドポイントの処理リージョン（例： `"us"`, `"eu"`)                                   |
| `processing_timeout` | `None` | サーバー側のセッション時間制限（秒）（サーバーレスエンドポイント）                                          |

〜では `model_id` mode、空欄 `ストリーム出力` / `data_output` は自動的に入力されます（`["image"]` および `["predictions"]`）；それ以外に指定した設定は保持されます。

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

## 実行可能な例

完全に動作するスクリプトは以下にあります： [`examples/webrtc_sdk/`](https://github.com/roboflow/inference/tree/main/examples/webrtc_sdk) Inferenceリポジトリのディレクトリ：

* [`webcam_basic.py`](https://github.com/roboflow/inference/blob/main/examples/webrtc_sdk/webcam_basic.py) - 基本的なWebカメラストリーミング
* [`rtsp_basic.py`](https://github.com/roboflow/inference/blob/main/examples/webrtc_sdk/rtsp_basic.py) - RTSPストリーム処理
* [`mjpeg_basic.py`](https://github.com/roboflow/inference/blob/main/examples/webrtc_sdk/mjpeg_basic.py) - MJPEGストリーム処理
* [`video_file_basic.py`](https://github.com/roboflow/inference/blob/main/examples/webrtc_sdk/video_file_basic.py) - 出力を保存しながら動画ファイルを処理

次のいずれかのコンテナを起動して、ローカルのInference Serverでこれらの例を実行してください。既存のホスト型統合では次を使用できます [非推奨のServerless Video Streaming API](https://docs.roboflow.com/deployment/roboflow-cloud/serverless-api/serverless-video-streaming-api)；新しい管理対象のライブカメラデプロイでは次を使用してください [Cloud Streaming](https://docs.roboflow.com/deployment/roboflow-cloud/cloud-streaming).

```bash
# 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
```