> 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/tui-lun/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
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`](https://supervision.roboflow.com/) ヘルパー - `sv.Detections.from_inference(data)` 物体検出およびインスタンスセグメンテーション用、 `sv.KeyPoints.from_inference(data)` キーポイントモデル用。分類予測には `上位`/`confidence` キーが含まれ、セマンティックセグメンテーション予測にはランレングス符号化マスク（`rle_mask`）が含まれます。これは自分でデコードしてください。

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

VLMは `model_id` モードではサポートされていません（各VLMファミリには専用の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をストリーミングする

マルチステップのパイプラインでは、次を渡します `ワークフロー` の代わりに `model_id`。RoboflowワークスペースにIDで保存されたWorkflowを参照するか、完全な仕様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 %}

注記:

* `ワークフロー` と `model_id` は排他的です - 1つだけ渡してください。
* `workspace` が必要なのは `ワークフロー` が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()                                    # デフォルトカメラ
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")   # 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配列
    except RuntimeError:
        pass                 # セッションはまだ接続中 - フレームはスキップされます
    time.sleep(1 / 30)       # 送信を宣言したFPSに合わせます

session.close()
```

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

## 結果の取得

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

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

```python
# 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()` はイテレーター版です - 同じデータをプル型で取得します:

```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` 出力ルーティング、処理動作、ネットワーク設定を制御します:

```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": "..."}` |
| `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/`](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) - 出力保存付きの動画ファイル処理

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