WebRTC 스트리밍

웹캠, RTSP 카메라, 비디오 파일 또는 수동 전송 프레임에서 모델 ID나 Workflow를 사용해 WebRTC를 통해 비디오를 Inference 서버로 스트리밍하고 실시간 예측을 받으세요.

다음을 사용하세요: inference-sdk 모델 또는 워크플로를 통해 비디오를 스트리밍하는 WebRTC 클라이언트입니다. 비디오 프레임은 하나의 연결을 통해 Inference Server로 전달되고, 처리된 프레임과 예측 데이터는 지속적으로 다시 흐릅니다.

동일한 클라이언트는 자체 호스팅 Inference Server와 Serverless Video Streaming API 모두에서 작동합니다. 설정 api_url 을(를) 원하는 런타임으로 지정하세요:

• 자체 호스팅: http://localhost:9001

• 서버리스: https://serverless.roboflow.com

WebRTC 스트리밍에는 추가 종속성이 필요합니다:

pip install "inference-sdk[webrtc]"

모델 스트리밍

다음과 같은 model_id 를 전달하여 하나의 모델을 통해 비디오를 스트리밍합니다. SDK가 필요한 단일 모델 워크플로를 생성하며, 여러분의 on_frame 핸들러는 각 비디오 프레임과 해당 예측 데이터를 받습니다:

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("Preview", 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 헬퍼를 사용해 변환할 수 있습니다 - sv.Detections.from_inference(data) 객체 감지 및 인스턴스 분할용, sv.KeyPoints.from_inference(data) 키포인트 모델용. 분류 예측은 top/confidence 키를 포함하며, 시맨틱 분할 예측은 런렝스 인코딩된 마스크(rle_mask)를 포함하며, 이는 직접 디코딩해야 합니다.

프레임에 대해 예측을 사용할 수 없는 경우(예: 라이브 스트림 프레임에 대한 짝지어진 예측 메시지가 도착하지 않은 경우), data 는 None 입니다 - 사용 전에 핸들러에서 확인하세요.

VLM은 model_id 모드에서 지원되지 않습니다(각 VLM 계열에는 전용 Workflow 블록이 있으므로 이를 감쌀 일반 블록이 없습니다) - 대신 전체 워크플로로 스트리밍하세요.

작업 유형 조회를 건너뛰기: 전달하여 task_type 를 명시적으로 지정해 네트워크 호출을 피하세요 - 에어갭 또는 자체 호스팅 배포에 유용합니다:

session = client.webrtc.stream(
    source=WebcamSource(),
    model_id="my-project/3",
    task_type="object-detection",
)

In model_id 모드에서, on_frame 핸들러는 다음 중 하나를 받을 수 있습니다 (frame, data) 또는 (frame, data, metadata) - 세 번째 인자는 VideoMetadata 입니다.

Workflow 스트리밍

다단계 파이프라인의 경우, 워크플로로 대신 model_id를 전달하세요. Roboflow 작업공간에 저장된 Workflow를 ID로 참조하거나, 전체 사양 dict를 제공할 수 있습니다:

워크플로 ID

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"],   # 워크플로 출력이 비디오로 다시 스트리밍됨
        data_output=["predictions"],      # 워크플로 출력이 데이터 채널을 통해 전달됨
    ),
)

@session.on_frame
def show(frame, metadata):
    cv2.imshow("Preview", frame)
    if cv2.waitKey(1) & 0xFF == ord("q"):
        session.close()

@session.on_data("predictions")
def handle_predictions(predictions, metadata):
    print(f"Frame {metadata.frame_id}: {predictions}")

session.run()

워크플로 사양

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("Preview", frame)
    if cv2.waitKey(1) & 0xFF == ord("q"):
        session.close()

session.run()

참고:

• 워크플로로 및 model_id 는 서로 배타적입니다 - 정확히 하나만 전달하세요.

• workspace 가 필요합니다 when 워크플로로 가 ID 문자열인 경우; 사양 dict에는 필요하지 않습니다.

• image_input (기본값 "image")은 비디오 프레임이 바인딩되는 Workflow 이미지 입력의 이름을 지정합니다.

• 워크플로 모드에서는 on_frame 핸들러가 받는 것은 (frame, metadata) 입니다 - 예측 데이터는 on_data 핸들러를 통해 별도로 도착하며, data_output 이름을 기준으로 라우팅됩니다. StreamConfig.

비디오 소스

의 첫 번째 인자는 stream() 비디오가 어디에서 오는지 선택합니다:

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

WebcamSource

로컬 카메라 장치에서 프레임을 캡처하여 서버로 보냅니다:

source = WebcamSource()                                    # 기본 카메라
source = WebcamSource(device_id=1, resolution=(1920, 1080))

카메라 FPS는 자동 감지되어 서버에 보고됩니다.

RTSPSource

해당 서버 RTSP 카메라에 연결하여 처리된 비디오를 다시 스트리밍합니다 - 카메라가 서버에서 접근 가능할 때 사용하세요:

source = RTSPSource("rtsp://user:[e-mail]/stream")

LocalStreamSource

RTSP/RTMP 스트림을 캡처하여 로컬에서 (클라이언트 머신에서) 서버로 프레임을 보냅니다 - 카메라가 서버가 아니라 여러분의 머신에서만 접근 가능할 때 사용하세요:

source = LocalStreamSource("rtsp://192.168.1.10/stream")   # rtsps://, rtmp://, rtmps://도 가능

MJPEGSource

와 같지만 RTSPSource서버가 캡처한 MJPEG 스트림용입니다:

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

VideoFileSource

데이터 채널을 통해 비디오 파일을 서버에 업로드합니다. 서버가 이를 처리한 뒤 결과를 다시 스트리밍합니다. 녹화된 비디오의 경우 프레임 단위 스트리밍보다 더 효율적입니다:

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

프레임을 프로그래밍 방식으로 전송합니다 - 프레임이 사용자 지정 파이프라인에서 오는 경우 유용합니다:

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()) 자원을 해제하려면 닫아야 합니다. 세 가지 동등한 패턴:

# 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() 는 이터레이터에 해당하는 것으로, 동일한 데이터를 pull 기반으로 제공합니다:

# 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 를 통해 데이터 채널로 도착합니다. 출력 이름별로 핸들러를 등록하거나, 전체 페이로드에 대한 전역 핸들러 하나를 등록할 수 있습니다:

@session.on_data("predictions")           # 단일 출력 필드
def handle_predictions(predictions, metadata):
    print(f"Frame {metadata.frame_id}: {predictions}")

@session.on_data                          # 전역: 전체 출력 dict
def handle_all(data, metadata):
    print(data)

핸들러는 (value, metadata) 또는 (value) 를 받을 수 있습니다 - 서명은 자동 감지됩니다.

오류 처리: on_error

서버는 각 데이터 채널 메시지와 함께 프레임별 오류(워크플로 실행 실패, 출력 직렬화 실패)를 보고합니다. on_error 핸들러는 비어 있지 않은 오류 목록이 있는 프레임에 대해서만 실행됩니다:

@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와 측정된 스트림 FPS
errors	서버가 보고한 프레임별 오류(프레임이 정상 처리되면 비어 있음)

StreamConfig

StreamConfig 출력 라우팅, 처리 동작, 네트워크 설정을 제어합니다:

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	서버 측 세션 시간 제한(초, 서버리스 엔드포인트)

In model_id 모드에서, 비어 있음 stream_output / data_output 는 자동으로 채워집니다(["image"] 및 ["predictions"]); 제공한 다른 설정은 보존됩니다.

TURN 서버: Roboflow 호스팅 엔드포인트에 연결할 때 TURN 설정은 자동으로 가져와집니다. 제한적인 NAT 또는 방화벽 뒤의 자체 호스팅 서버의 경우 turn_server 를 명시적으로 제공하세요; 설정하지 않으면 직접 연결을 시도합니다.

실행 가능한 예제

완전한 동작 스크립트는 Inference 저장소의 examples/webrtc_sdk/ 디렉터리에 있습니다:

• webcam_basic.py - 기본 웹캠 스트리밍

• rtsp_basic.py - RTSP 스트림 처리

• mjpeg_basic.py - MJPEG 스트림 처리

• video_file_basic.py - 출력 저장을 포함한 비디오 파일 처리

Roboflow Serverless 호스티드 API(https://serverless.roboflow.com) 설정 없이 사용할 수 있으며, 개발용 로컬 서버를 대상으로도 사용할 수 있습니다:

# 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