Stream Management API

Stream Management API के साथ Roboflow Inference video pipelines को remotely manage करें - इसे Docker में चलाएँ, HTTP के माध्यम से integrate करें, और Stream Manager protocol का उपयोग करें।

एक स्वतंत्र सेवा के रूप में बंद कर दी गई है। एक Inference Server का उपयोग करें, वर्तमान वीडियो स्ट्रीमिंग परिनियोजन के लिए। वही क्लाइंट स्व-होस्टेड और सर्वरलेस रनटाइम पर मॉडल और वर्कफ़्लो चलाता है। देखें वीडियो प्रोसेसिंग.

एंटरप्राइज़ सुविधा। इस पृष्ठ पर मौजूद स्वतंत्र सेवा को उत्पादन में उपयोग करने के लिए Roboflow Enterprise लाइसेंस की आवश्यकता है। देखें Roboflow लाइसेंसिंग विवरण के लिए। यह ऊपर वर्णित एकीकृत वीडियो प्रबंधन API पर लागू नहीं होता।

के बारे में

Stream Management API ने ऑनलाइन वीडियो स्ट्रीम पर Roboflow ऑब्जेक्ट डिटेक्शन मॉडलों से पूर्वानुमान उत्पन्न किए। इसने वीडियो वर्करों को दूरस्थ रूप से नियंत्रित करने के लिए एक HTTP प्रबंधन परत जोड़ी।

यह उन परिदृश्यों के लिए उपयोगी है, जिनमें निम्न शामिल हैं, लेकिन केवल इन्हीं तक सीमित नहीं हैं:

एक साथ कई ऑनलाइन वीडियो स्ट्रीम पर इन्फ़रेंस करना।
कई डिवाइसों पर इन्फ़रेंस चलाना जिन्हें समन्वय की आवश्यकता हो।
वीडियो प्रोसेसिंग की निगरानी के लिए एक मॉनिटरिंग परत स्थापित करना।

उदाहरण उपयोग मामला

Joe अपने कारखाने में लगे IP कैमरों के एक बेड़े द्वारा कैप्चर किए गए फुटेज में वस्तुओं की निगरानी करना चाहता है। Roboflow प्लेटफ़ॉर्म पर एक ऑब्जेक्ट डिटेक्शन मॉडल प्रशिक्षित करने के बाद, वह परिनियोजन के लिए तैयार है। अपने कारखाने में चार कैमरों के साथ, Joe ऐसा मॉडल चुनता है जो उसके Jetson डिवाइसों पर प्रति सेकंड 30 से अधिक इन्फ़रेंस के लिए पर्याप्त रूप से कॉम्पैक्ट हो। प्रति डिवाइस उस कंप्यूटेशनल बजट को देखते हुए, उसे सभी कैमरों के फुटेज को प्रोसेस करने के लिए दो Jetson डिवाइसों की आवश्यकता है, लगभग प्रति वीडियो स्रोत 15 फ़्रेम प्रति सेकंड की दर से।

परिनियोजन को सुव्यवस्थित करने के लिए, Joe अपने लोकल नेटवर्क में सभी Jetson डिवाइसों पर Stream Management कंटेनर परिनियोजित करता है। इससे वह HTTP के माध्यम से प्रत्येक डिवाइस से संवाद करके प्रोसेसिंग कार्यों का समन्वय कर सकता है। वह डिवाइसों को कमांड भेजने और प्रत्येक वीडियो स्ट्रीम की स्थिति के बारे में मेट्रिक्स प्राप्त करने के लिए एक वेब ऐप बनाता है। अंत में, वह supervision पैकेज का उपयोग करके फुटेज में वस्तुओं को ट्रैक करने के लिए एक UDP सर्वर लागू करता है।

इसे कैसे चलाएँ

Docker में `docker compose`

सबसे सामान्य उपयोग मामले Docker Compose कॉन्फ़िगरेशन के रूप में पैकेज किए गए हैं। जहाँ कंटेनरों के अंदर कस्टम कॉन्फ़िगरेशन की आवश्यकता होती है, जैसे कैमरा डिवाइस पास करना, वहाँ नीचे दिए गए अलग-कंटेनर विकल्प अधिक उपयुक्त हो सकते हैं।

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

Jetson डिवाइसों पर, कुछ ऑपरेशन (जैसे कंटेनर बूटस्ट्रैप या मॉडल इनिशियलाइज़ेशन) अन्य प्लेटफ़ॉर्म की तुलना में अधिक समय लेते हैं। Docker Compose परिभाषा वर्तमान में Stream Manager TCP सॉकेट पोर्ट के खुलने की प्रतीक्षा नहीं करती, इसलिए HTTP API के लिए शुरुआती अनुरोधों का उत्तर HTTP 503 के साथ मिल सकता है।

Docker में, API और stream manager को अलग-अलग चलाना

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

कॉन्फ़िगरेशन पैरामीटर

स्ट्रीम प्रबंधन API

STREAM_MANAGER_HOST - stream manager कंटेनर का होस्टनाम। यदि --network host का उपयोग नहीं किया जाता, या यदि आप किसी दूरस्थ मशीन को लक्षित कर रहे हैं, तो इसे कंटेनर नाम में बदलें।
STREAM_MANAGER_PORT - Stream Manager के साथ संचार करने के लिए उपयोग किया जाने वाला पोर्ट। Stream manager कंटेनर से मेल खाना चाहिए।

Stream Manager

PORT - वह पोर्ट जिस पर सर्वर चलता है।
कंटेनर के /tmp/cache पर एक वॉल्यूम माउंट करें ताकि मॉडलों का स्थायी संग्रहण सक्षम हो, और इन्फ़रेंस पाइपलाइन इनिशियलाइज़ेशन तेज़ हो सके।
कैमरा कनेक्टिविटी को इस कंटेनर के स्तर पर सक्षम होना चाहिए, इसलिए यदि डिवाइसों को Docker में पास करना हो, तो इसे यहीं करें।

इमेज बनाना (वैकल्पिक)

# 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 करते हैं। यह हो जाने के बाद, चलाएँ:

python -m inference.enterprise.stream_management.manager.app  # manager चलाता है
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 एडिटर.

एक उदाहरण 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()

initialise_pipeline() को दिया जाना चाहिए एक video_reference और sink_configuration जहाँ प्रत्येक संसाधन (वीडियो फ़ाइल या कैमरा डिवाइस) और URI (स्ट्रीम संदर्भ, सिंक संदर्भ) Stream Manager वातावरण से पहुँचा जा सकने वाला हो. उदाहरण के लिए, Docker कंटेनरों के अंदर localhost कंटेनर के localhost से बाइंड होता है, होस्ट मशीन के localhost से नहीं।

डेवलपर नोट्स

कार्यान्वयन का महत्वपूर्ण तत्व Stream Manager घटक है, जो एक सिंगल-थ्रेडेड TCP सर्वर के रूप में काम करता है। यह 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

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

मॉडल ID इस स्ट्रिंग से बना होता है <project_id>/<version_id>. देखें model IDs इन मानों को खोजने के लिए।

terminate

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

pause

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

resume

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

स्थिति

{
  "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 - अनुरोध ID का प्रतिनिधित्व करने वाली एक यादृच्छिक स्ट्रिंग, जो Stream Manager द्वारा असाइन की गई होती है, ताकि डिबगिंग आसान हो।
pipeline_id - वह पाइपलाइन जिससे कमांड संबंधित है, जब लागू हो।
response - ऑपरेशन प्रतिक्रिया का पेलोड।

प्रत्येक response में एक स्थिति key होती है जिसके दो में से एक मान हो सकता है: success या failure. प्रत्येक विफल प्रतिक्रिया में एक error_type key होती है, जो त्रुटि प्रबंधन को रूट करने के लिए उपयोग होती है, साथ ही वैकल्पिक error_class और error_message फ़ील्ड्स होती हैं जिनमें त्रुटि का आंतरिक विवरण होता है। सफल प्रतिक्रियाओं की सामग्री ऑपरेशन के प्रकार पर निर्भर करती है।

भविष्य का कार्य

सुरक्षित रिमोट नियंत्रण सक्षम करने के लिए API कनेक्शन लेयर को सुरक्षित करना।
Stream Manager के TCP सॉकेट को सुरक्षित करना।

पिछलाSecure Gateway अगलाDeployment Manager

अंतिम अपडेट 5 दिन पहले

क्या यह उपयोगी था?