> 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 ストリーミング

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

同じクライアントは、セルフホストの Inference Server と Serverless Video Streaming API の両方で動作します。次を設定します `api_url` 使用したい実行環境に対して:

* セルフホスト: `http://localhost:9001`
* サーバーレス: `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 Key に置き換えてください
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 は生の predictions 辞書で、サーバーが返すものと完全に同じです
    # （このフレームで予測が利用できない場合は 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 の検索によって自動的に解決され、対応するモデルブロックが自動で選択されます。対応するタスクタイプ:

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

`データ` は、そのままシリアライズされた predictions 辞書で、その形状はタスクタイプに従います。検出系モデルでは inference-response 形式なので、対応する [`supervision`](https://supervision.roboflow.com/) ヘルパー - `sv.Detections.from_inference(data)` 物体検出およびインスタンスセグメンテーション用、 `キーポイントモデル用。分類予測には` 上位 `信頼度`/`キーが含まれ、セマンティックセグメンテーションの予測にはランレングス符号化マスク（` rle\_mask`rle_mask`）が含まれており、これは自分でデコードします。

フレームの予測が利用できない場合（例: ライブストリームのフレームに対応する予測メッセージが届かなかった場合）、 `データ` は `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) そのフレーム用です。

## 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(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} チャンク"),
    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 配列
    except RuntimeError:
        pass                 # セッションはまだ接続中 - フレームをスキップ
    time.sleep(1 / 30)       # 送信間隔を宣言した FPS に合わせる

session.close()
```

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

## 結果の受信

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

`stream()` 次を返します `WebRTCSession`。接続は最初の使用時に遅延開始されます（`run()`, `video()`、または `wait()`）し、リソースを解放するには閉じる必要があります。等価なパターンは 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()` はイテレータ版です - 同じデータをプル型で取得します:

```python
# model_id モード: (frame, data) - data は生の predictions 辞書
# （フレームの予測が利用できない場合は 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"Frame {metadata.frame_id} failed: {errors}")
    session.close()   # 例: 最初のエラーで中断する
```

これらはサーバー側のフレームごとの失敗です。接続エラーやセットアップエラーは次から例外として表れます `run()` 代わりに。エラーはまた `metadata.errors` 各フレームで、したがって `on_frame` / `on_data` ハンドラはそれらを直接確認できます。

### フレームメタデータ

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

| 属性                              | 説明                                     |
| ------------------------------- | -------------------------------------- |
| `frame_id`                      | ストリーム内のフレームの一意な識別子                     |
| `received_at`                   | サーバーがフレームを受信した時刻                       |
| `pts` / `time_base`             | 動画ストリームのプレゼンテーションタイムスタンプ               |
| `declared_fps` / `measured_fps` | 宣言値と実測のストリーム FPS                       |
| `errors`                        | サーバーが報告するフレームごとのエラー（フレームが正常に処理された場合は空） |

## StreamConfig

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

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

| 項目                    | デフォルト  | 説明                                                                          |
| --------------------- | ------ | --------------------------------------------------------------------------- |
| `stream_output`       | `[]`   | Workflow の出力名が動画としてストリーミングで返されます                                            |
| `data_output`         | `[]`   | Workflow の出力名がデータチャネル経由で配信されます                                              |
| `realtime_processing` | `True` | リアルタイムに追従するためフレームを破棄します。次を設定すると `False` 全フレームをキューに入れて処理します                  |
| `declared_fps`        | `None` | 自動検出のないソース向けの FPS 宣言（例: `ManualSource`)                                     |
| `turn_server`         | `None` | TURN サーバー設定: `{"urls": "turn:...", "username": "...", "credential": "..."}` |
| `ワークフローパラメータ`         | `{}`   | ワークフロー実行に渡されるパラメータ                                                          |
| `要求されたプラン`            | `None` | Roboflow のサーバーレスエンドポイント用の計算プラン（例: `"webrtc-gpu-small"`)                     |
| `要求されたリージョン`          | `None` | サーバーレスエンドポイントの処理リージョン（例: `"us"`, `"eu"`)                                    |
| `処理タイムアウト`            | `None` | サーバー側セッションの制限時間（秒）（サーバーレスエンドポイント）                                           |

次の `model_id` mode、空 `stream_output` / `data_output` は自動的に入力されます（`["image"]` および `["predictions"]`）；その他に指定した設定は保持されます。

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

## 実行可能な例

完全な動作スクリプトは以下にあります: [`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) - 基本的なウェブカメラ配信
* [`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) - 出力保存付きの動画ファイル処理

Roboflow Serverless Hosted API（`https://serverless.roboflow.com`）はセットアップ不要で利用でき、開発用にローカルサーバーに対しても利用できます:

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