Stream Management API

About

The Stream Management API generated predictions from Roboflow object detection models on online video streams. It added an HTTP management layer for remotely controlling video workers.

This is useful for scenarios including, but not limited to:

• Performing inference across multiple online video streams simultaneously.
• Executing inference on multiple devices that need coordination.
• Establishing a monitoring layer to oversee video processing.

Example use case

Joe wants to monitor objects in the footage captured by a fleet of IP cameras installed in his factory. After training an object detection model on the Roboflow platform, he is ready for deployment. With four cameras in his factory, Joe picks a model compact enough for over 30 inferences per second on his Jetson devices. Given that computational budget per device, he needs two Jetson devices to process footage from all cameras, at roughly 15 frames per second per video source.

To streamline deployment, Joe deploys Stream Management containers to all Jetson devices in his local network. This lets him communicate with each device over HTTP to orchestrate processing tasks. He builds a web app to send commands to the devices and retrieve metrics about the status of each video stream. Finally, he implements a UDP server to receive predictions, using the supervision package to track objects in the footage.

How to run

In Docker with docker compose

The most common use cases are packaged as Docker Compose configurations. Where custom configuration is required inside the containers, such as passing camera devices through, the separate-container options below may suit better.

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

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

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

In Docker, running the API and stream manager separately

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

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

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

Configuration parameters

Stream Management API

• STREAM_MANAGER_HOST - hostname for the stream manager container. Change it to the container name if --network host is not used, or if you are targeting a remote machine.
• STREAM_MANAGER_PORT - port used to communicate with the stream manager. Must match the stream manager container.

Stream Manager

• PORT - the port the server runs on.
• Mount a volume at the container's /tmp/cache to enable permanent storage of models, for faster inference pipeline initialization.
• Camera connectivity must be enabled at the level of this container, so if devices need to be passed into Docker, do it here.

Building the images (optional)

# 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 .

Bare-metal deployment

In some cases you need to deploy the application at host level. This is possible, but you must resolve the environment the way the Stream Manager and Stream Management API dockerfiles do for your platform. Once that is done, run:

python -m inference.enterprise.stream_management.manager.app  # runs the manager
python -m inference.enterprise.stream_management.api.app      # runs the management API

How to integrate

After running the roboflow-inference-stream-management-api container, the HTTP API is available at http://127.0.0.1:8080 with the default configuration.

Call wget http://127.0.0.1:8080/openapi.json to get the OpenAPI specification of the API, which you can render in the Swagger editor.

An example Python client:

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

Developer notes

The pivotal element of the implementation is the Stream Manager component, which operates as a single-threaded TCP server. It processes requests received from a TCP socket and supervises video worker processes. Multiprocessing queues carry commands and results between the workers and the Stream Manager.

Requests to the Stream Manager are handled sequentially in blocking mode, so each request must conclude before the next one starts.

Communication protocol: requests

The Stream Manager accepts the following binary protocol. Each payload contains:

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

The message must be valid JSON after decoding and must represent a valid command.

list_pipelines

{
  "type": "list_pipelines"
}

init

{
  "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
  }
}

terminate

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

pause

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

resume

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

status

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

Communication protocol: responses

For each request that can be processed (without timeout or source disconnection), the Stream Manager returns a result in this format:

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

The result contains:

• request_id - a random string representing the request ID assigned by the Stream Manager, to ease debugging.
• pipeline_id - the pipeline the command is associated with, when applicable.
• response - the payload of the operation response.

Each response has a status key with one of two values: success or failure. Each failed response contains an error_type key to dispatch error handling, plus optional error_class and error_message fields with inner details of the error. The content of successful responses depends on the type of operation.

Future work

• Securing the API connection layer, to enable safe remote control.
• Securing the TCP socket of the Stream Manager.