> 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/ko/self-hosted/enterprise/stream-management-api.md).

# 스트림 관리 API

Stream Management API로 Roboflow Inference 비디오 파이프라인을 원격 관리합니다 - Docker에서 실행하고, HTTP로 통합하고, Stream Manager 프로토콜을 사용합니다.

{% hint style="warning" %}
**독립형 서비스로는 중단되었습니다.** Inference SDK WebRTC 클라이언트를 [Inference Server](/deployment/ko/self-hosted/inference-server.md) 와 함께 사용하세요. 현재 비디오 스트리밍 배포에 적합합니다. 동일한 클라이언트는 자체 호스팅 및 서버리스 런타임에서 모델과 워크플로를 실행합니다. 참조: [비디오 처리](https://docs.roboflow.com/workflows/deploy/video-processing).
{% endhint %}

{% hint style="warning" %}
**엔터프라이즈 기능입니다.** 이 페이지의 독립형 서비스는 프로덕션에서 사용하려면 Roboflow Enterprise 라이선스가 필요합니다. 참조: [Roboflow 라이선스](https://roboflow.com/licensing) 자세한 내용은 여기에서 확인하세요. 이는 위에 설명된 통합 비디오 관리 API에는 적용되지 않습니다.
{% endhint %}

현재 통합 서버에는 별도의 [비디오 구성 안내](/deployment/ko/self-hosted/inference-server/configuration/video-configuration.md)가 있으며, 관리되는 프로세스 한도도 포함됩니다. 보류 중인 media-reference 검증은 이 엔터프라이즈 서비스의 요청 스키마도 포함합니다. 참조: [마이그레이션 범위](/deployment/ko/self-hosted/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 플랫폼에서 객체 탐지 모델을 학습한 후 배포할 준비가 되었습니다. 공장에 카메라가 네 대 있으므로, Joe는 Jetson 장치에서 초당 30회 이상의 추론이 가능한 만큼 충분히 작은 모델을 선택합니다. 장치당 계산 예산을 고려하면, 모든 카메라의 영상을 처리하려면 초당 비디오 소스당 약 15프레임으로 처리할 수 있는 Jetson 장치 두 대가 필요합니다.

배포를 간소화하기 위해 Joe는 로컬 네트워크의 모든 Jetson 장치에 Stream Management 컨테이너를 배포합니다. 이를 통해 HTTP로 각 장치와 통신하여 처리 작업을 조율할 수 있습니다. 그는 장치에 명령을 보내고 각 비디오 스트림의 상태에 대한 메트릭을 가져오는 웹 앱을 만듭니다. 마지막으로 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와 스트림 관리자를 별도로 실행

{% 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` - 스트림 관리자 컨테이너의 호스트 이름입니다. 사용하지 않거나 원격 머신을 대상으로 하는 경우 컨테이너 이름으로 변경하세요. `--network host` 를 사용하지 않거나 원격 머신을 대상으로 하는 경우입니다.
* `STREAM_MANAGER_PORT` - 스트림 관리자와 통신하는 데 사용되는 포트입니다. 스트림 관리자 컨테이너와 일치해야 합니다.

**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 Manager 환경에서 접근 가능해야 합니다**. 예를 들어, Docker 컨테이너 내부에서는 `localhost` 호스트 머신의 localhost가 아니라 컨테이너의 localhost에 바인딩됩니다.
{% endhint %}

## 개발자 참고 사항

구현의 핵심 요소는 단일 스레드 TCP 서버로 동작하는 Stream Manager 구성 요소입니다. 이 구성 요소는 TCP 소켓에서 받은 요청을 처리하고 비디오 워커 프로세스를 감독합니다. 멀티프로세싱 큐는 워커와 Stream Manager 사이에 명령과 결과를 전달합니다.

Stream Manager에 대한 요청은 차단 모드에서 순차적으로 처리되므로, 각 요청은 다음 요청이 시작되기 전에 완료되어야 합니다.

### 통신 프로토콜: 요청

Stream Manager는 다음 이진 프로토콜을 허용합니다. 각 페이로드에는 다음이 포함됩니다:

```
[HEADER: 4B, big-endian, unsigned - int value with message size][MESSAGE: utf-8 serialised json of size dictated by header]
```

메시지는 디코딩 후 유효한 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>`. 참조: [모델 ID](https://docs.roboflow.com/models/model-ids) 에서 이 값을 찾으세요.
{% endhint %}

**`terminate`**

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

**`pause`**

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

**`resume`**

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

**`status`**

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

### 통신 프로토콜: 응답

처리할 수 있는 각 요청(시간 초과 또는 소스 연결 해제 없음)에 대해 Stream Manager는 다음 형식으로 결과를 반환합니다:

```
[HEADER: 4B, big-endian, unsigned - int value with result size][RESULT: utf-8 serialised json of size dictated by header]
```

결과에는 다음이 포함됩니다:

* `request_id` - 디버깅을 쉽게 하기 위해 Stream Manager가 할당한 요청 ID를 나타내는 임의 문자열입니다.
* `pipeline_id` - 해당되는 경우 명령과 연결된 파이프라인입니다.
* `response` - 작업 응답의 페이로드입니다.

각 `response` 에는 `status` 다음 두 값 중 하나를 가진 key가 있습니다: `success` 또는 `failure`. 실패한 각 응답에는 오류 처리를 분기하기 위한 `error_type` key와 선택적 `error_class` 와 `error_message` 필드가 있으며, 이 필드에는 오류의 세부 정보가 들어 있습니다. 성공한 응답의 내용은 작업 유형에 따라 달라집니다.

## 향후 작업

* 안전한 원격 제어를 가능하게 하기 위해 API 연결 계층을 보호합니다.
* Stream Manager의 TCP 소켓을 보호합니다.