> 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/deployment/ja/serufuhosuto/enterprise/stream-management-api.md).

# ストリーム管理 API

Stream Management API を使って Roboflow Inference の動画パイプラインをリモート管理します - Docker で実行し、HTTP 経由で統合し、Stream Manager プロトコルを使用します。

{% hint style="warning" %}
**単独サービスとしては廃止されました。** Inference SDK の WebRTC クライアントを [Inference Server](/deployment/ja/serufuhosuto/inference-server.md) 現在の動画ストリーミングのデプロイで使用します。同じクライアントは、セルフホストおよびサーバーレスのランタイム上でモデルとワークフローを実行します。参照: [動画処理](https://docs.roboflow.com/workflows/deploy/video-processing).
{% endhint %}

{% hint style="warning" %}
**エンタープライズ機能。** このページのスタンドアロン サービスを本番環境で使用するには、Roboflow Enterprise ライセンスが必要です。詳細は [Roboflow Licensing](https://roboflow.com/licensing) をご覧ください。これは上記の統合動画管理 API には適用されません。
{% endhint %}

現在の統合サーバーには個別の [動画構成ガイダンス](/deployment/ja/serufuhosuto/inference-server/configuration/video-configuration.md)があり、管理対象プロセス上限も含まれます。保留中のメディア参照検証は、このエンタープライズ サービスのリクエストスキーマも対象です。参照先: [移行範囲](/deployment/ja/serufuhosuto/inference-server/configuration/security-migration.md#video-source-validation).

## 概要

Stream Management API は、オンライン動画ストリーム上の Roboflow 物体検出モデルから予測を生成していました。また、動画ワーカーをリモートで制御するための HTTP 管理レイヤーを追加していました。

これは、以下を含むがこれらに限定されないシナリオで有用です:

* 複数のオンライン動画ストリームに対して同時に推論を実行する。
* 連携が必要な複数のデバイスで推論を実行する。
* 動画処理を監視するための監視レイヤーを構築する。

![Stream Management の設計](https://storage.googleapis.com/com-roboflow-marketing/inference/stream_management_api_design.jpg)

### 使用例

Joe は、工場に設置された IP カメラ群で撮影された映像内の物体を監視したいと考えています。Roboflow プラットフォームで物体検出モデルを学習した後、デプロイの準備が整いました。工場には 4 台のカメラがあるため、Joe は Jetson デバイス上で 1 秒あたり 30 回を超える推論が可能なほどコンパクトなモデルを選びます。各デバイスの計算予算を考えると、すべてのカメラの映像を処理するには 2 台の Jetson デバイスが必要で、各動画ソースあたりおよそ毎秒 15 フレームで処理します。

デプロイを簡素化するため、Joe はローカルネットワーク内のすべての Jetson デバイスに Stream Management コンテナをデプロイします。これにより、HTTP 経由で各デバイスと通信して処理タスクを調整できます。彼は、デバイスにコマンドを送信し、各動画ストリームの状態に関するメトリクスを取得する Web アプリを構築します。最後に、予測を受信する UDP サーバーを実装し、 `supervision` パッケージを使って映像内の物体を追跡します。

## 実行方法

### Docker で `docker compose`

最も一般的なユースケースは、Docker Compose の設定としてパッケージ化されています。カメラデバイスのパススルーなど、コンテナ内でカスタム設定が必要な場合は、以下の個別コンテナのオプションのほうが適していることがあります。

{% tabs %}
{% tab title="CPU" %}

```bash
docker compose -f ./docker/dockerfiles/stream-management-api.compose-cpu.yaml up
```

{% endtab %}

{% tab title="GPU" %}

```bash
docker compose -f ./docker/dockerfiles/stream-management-api.compose-gpu.yaml up
```

{% endtab %}

{% tab title="Jetson（JetPack 5.1.1）" %}

```bash
docker compose -f ./docker/dockerfiles/stream-management-api.compose-jetson.5.1.1.yaml up
```

Jetson デバイスでは、コンテナの起動やモデルの初期化などの一部の操作に、他のプラットフォームよりも時間がかかります。Docker Compose 定義は現在、Stream Manager の TCP ソケットポートが開くのを待たないため、HTTP API への最初のリクエストには HTTP 503 が返される場合があります。
{% endtab %}
{% endtabs %}

### Docker で、API と Stream Manager を別々に実行する

{% tabs %}
{% tab title="CPU" %}

```bash
docker run -d --name stream_manager --network host roboflow/roboflow-inference-stream-manager-cpu:latest
docker run -d --name stream_management_api --network host roboflow/roboflow-inference-stream-management-api:latest
```

{% endtab %}

{% tab title="GPU" %}

```bash
docker run -d --name stream_manager --network host --runtime nvidia roboflow/roboflow-inference-stream-manager-gpu:latest
docker run -d --name stream_management_api --network host roboflow/roboflow-inference-stream-management-api:latest
```

{% endtab %}

{% tab title="Jetson（JetPack 5.1.1）" %}

```bash
docker run -d --name stream_manager --network host --runtime nvidia roboflow/roboflow-inference-stream-manager-jetson-5.1.1:latest
docker run -d --name stream_management_api --network host roboflow/roboflow-inference-stream-management-api:latest
```

{% endtab %}
{% endtabs %}

#### 設定パラメータ

**Stream Management API**

* `STREAM_MANAGER_HOST` - stream manager コンテナのホスト名。 `--network host` 「docker compose」を使用しない場合、またはリモートマシンを対象にする場合は、コンテナ名に変更してください。
* `STREAM_MANAGER_PORT` - stream manager との通信に使用するポート。stream manager コンテナと一致している必要があります。

**Stream Manager**

* `PORT` - サーバーが実行されるポート。
* コンテナの `/tmp/cache` にボリュームをマウントすると、モデルを永続保存でき、推論パイプラインの初期化を高速化できます。
* カメラ接続はこのコンテナのレベルで有効にする必要があるため、デバイスを Docker に渡す必要がある場合はここで行ってください。

#### イメージのビルド（任意）

```bash
# Stream Management API
docker build -t roboflow/roboflow-inference-stream-management-api:dev -f docker/dockerfiles/Dockerfile.stream_management_api .

# Stream Manager
docker build -t roboflow/roboflow-inference-stream-manager-{device}:dev -f docker/dockerfiles/Dockerfile.onnx.{device}.stream_manager .
```

### ベアメタルデプロイ

場合によっては、アプリケーションをホストレベルでデプロイする必要があります。これは可能ですが、Stream Manager と Stream Management API の Dockerfile がプラットフォーム向けに行うのと同じ方法で環境を整える必要があります。それが完了したら、次を実行します:

```bash
python -m inference.enterprise.stream_management.manager.app  # マネージャーを実行します
python -m inference.enterprise.stream_management.api.app      # 管理 API を実行します
```

## 統合方法

を実行した後、 `roboflow-inference-stream-management-api` コンテナを起動すると、HTTP API は `http://127.0.0.1:8080` でデフォルト設定として利用できます。

次を呼び出して `wget http://127.0.0.1:8080/openapi.json` API の OpenAPI 仕様を取得してください。これは [Swagger エディタ](https://editor.swagger.io/).

Python クライアントの例:

```python
import requests
from typing import Optional

URL = "http://127.0.0.1:8080"

def list_pipelines() -> dict:
    response = requests.get(f"{URL}/list_pipelines")
    return response.json()


def get_pipeline_status(pipeline_id: str) -> dict:
    response = requests.get(f"{URL}/status/{pipeline_id}")
    return response.json()


def pause_pipeline(pipeline_id: str) -> dict:
    response = requests.post(f"{URL}/pause/{pipeline_id}")
    return response.json()


def resume_pipeline(pipeline_id: str) -> dict:
    response = requests.post(f"{URL}/resume/{pipeline_id}")
    return response.json()


def terminate_pipeline(pipeline_id: str) -> dict:
    response = requests.post(f"{URL}/terminate/{pipeline_id}")
    return response.json()


def initialise_pipeline(
    video_reference: str,
    model_id: str,
    api_key: str,
    sink_host: str,
    sink_port: int,
    max_fps: Optional[int] = None,
) -> dict:
    response = requests.post(
        f"{URL}/initialise",
        json={
            "type": "init",
            "sink_configuration": {
                "type": "udp_sink",
                "host": sink_host,
                "port": sink_port,
            },
            "video_reference": video_reference,
            "model_id": model_id,
            "api_key": api_key,
            "max_fps": max_fps,
        },
    )
    return response.json()
```

{% hint style="info" %}
`initialise_pipeline()` には `video_reference` と `sink_configuration` すべてのリソース（動画ファイルまたはカメラデバイス）と URI（stream reference、sink reference）が **Stream Manager 環境から到達可能**である必要があります。たとえば、Docker コンテナ内では `localhost` はホストマシンの localhost ではなく、コンテナの localhost にバインドされます。
{% endhint %}

## 開発者向けメモ

実装の重要な要素は Stream Manager コンポーネントで、単一スレッドの TCP サーバーとして動作します。TCP ソケットから受信したリクエストを処理し、動画ワーカープロセスを監督します。マルチプロセッシングキューが、ワーカーと Stream Manager の間でコマンドと結果を運びます。

Stream Manager へのリクエストはブロッキングモードで順番に処理されるため、次のリクエストは前のリクエストが完了してから開始されます。

### 通信プロトコル: リクエスト

Stream Manager は次のバイナリプロトコルを受け入れます。各ペイロードには以下が含まれます:

```
[HEADER: 4B, big-endian, unsigned - int value with message size][MESSAGE: ヘッダーで指定されたサイズの utf-8 シリアライズ済み JSON]
```

デコード後、メッセージは有効な JSON でなければならず、有効なコマンドを表している必要があります。

**`list_pipelines`**

```json
{
  "type": "list_pipelines"
}
```

**`init`**

```json
{
  "type": "init",
  "model_id": "some/1",
  "video_reference": "rtsp://192.168.0.1:554",
  "sink_configuration": {
    "type": "udp_sink",
    "host": "192.168.0.3",
    "port": 9999
  },
  "api_key": "YOUR_API_KEY",
  "max_fps": 16,
  "model_configuration": {
    "type": "object-detection",
    "class_agnostic_nms": true,
    "confidence": 0.5,
    "iou_threshold": 0.4,
    "max_candidates": 300,
    "max_detections": 3000
  },
  "video_source_properties": {
    "frame_width": 1920,
    "frame_height": 1080,
    "fps": 30
  }
}
```

{% hint style="info" %}
モデル ID は文字列 `<project_id>/<version_id>`です。参照: [で構成されています](https://docs.roboflow.com/models/model-ids) 。これらの値を確認するには
{% endhint %}

**`終了`**

```json
{
  "type": "terminate",
  "pipeline_id": "my_pipeline"
}
```

**`一時停止`**

```json
{
  "type": "mute",
  "pipeline_id": "my_pipeline"
}
```

**`再開`**

```json
{
  "type": "resume",
  "pipeline_id": "my_pipeline"
}
```

**`状態`**

```json
{
  "type": "status",
  "pipeline_id": "my_pipeline"
}
```

### 通信プロトコル: レスポンス

処理可能な各リクエスト（タイムアウトやソース切断がない場合）に対して、Stream Manager は次の形式で結果を返します:

```
[HEADER: 4B, big-endian, unsigned - int value with result size][RESULT: ヘッダーで指定されたサイズの utf-8 シリアライズ済み JSON]
```

結果には以下が含まれます:

* `request_id` - Stream Manager によって割り当てられたリクエスト ID を表すランダム文字列で、デバッグを容易にします。
* `pipeline_id` - 該当する場合、コマンドが関連付けられているパイプライン。
* `response` - 操作レスポンスのペイロード。

各 `response` には `状態` 次のいずれかの値を持つ `success` または `failure`キーがあります。各失敗レスポンスには `error_type` キーがあり、エラー処理を振り分けます。さらに任意の `error_class` と `error_message` フィールドに、エラーの詳細が含まれます。成功レスポンスの内容は操作の種類によって異なります。

## 今後の作業

* 安全なリモート制御を可能にするため、API 接続レイヤーを保護する。
* Stream Manager の TCP ソケットを保護する。