Inference Pipeline

InferencePipeline के साथ video streams पर models चलाएँ: video sources, custom inference logic, Workflows, और sinks।

InferencePipeline का प्रत्यक्ष, इन-प्रोसेस वीडियो इंटरफ़ेस है इन्फरेंस Python पैकेज में। इसका उपयोग तब करें जब आपका एप्लिकेशन Inference Library को एम्बेड करता हो और वीडियो फ़्रेम, कस्टम inference logic, या sinks तक सीधे Python एक्सेस की आवश्यकता हो।

उन एप्लिकेशनों के लिए जो Inference Server चलाते हैं या Serverless का उपयोग करते हैं, मॉडलों और Workflows को इसके साथ स्ट्रीम करें Inference SDK WebRTC क्लाइंट के साथ stream करें।

त्वरित प्रारंभ

Inference के साथ fine-tuned मॉडलों का उपयोग करने के लिए, आपको Roboflow API key की आवश्यकता होगी। यदि आपके पास पहले से Roboflow खाता नहीं है, तो एक निःशुल्क Roboflow खाता बनाएं. फिर, Roboflow dashboard से अपनी API key प्राप्त करें और इसे अपने coding environment में सेट करें:

export ROBOFLOW_API_KEY=<your api key>

Roboflow API keys के बारे में अधिक जानें.

फिर, Inference इंस्टॉल करें:

pip install inference

यदि आपके पास NVIDIA GPU है, तो आप अपने inference को इस तरह तेज़ कर सकते हैं:

pip install --extra-index-url https://download.pytorch.org/whl/cu124 inference-gpu
# कृपया --extra-index-url को अपने OS में स्थापित CUDA version के अनुसार समायोजित करें

अगला, एक Inference Pipeline बनाएं:

# InferencePipeline interface import करें
from inference import InferencePipeline
# render_boxes नामक एक built-in sink import करें (sinks वह logic हैं जो inference के बाद होती है)
from inference.core.interfaces.stream.sinks import render_boxes

api_key = "YOUR_ROBOFLOW_API_KEY"

# एक inference pipeline object बनाएं
pipeline = InferencePipeline.init(
    # model id को एक rfdetr model पर सेट करें (COCO पर pre-trained)
    model_id="rfdetr-large",
    # वीडियो reference (वीडियो का source) सेट करें, यह किसी video file के link/path, RTSP stream url,
    # या device id को दर्शाने वाला integer हो सकता है (आमतौर पर built-in webcams के लिए 0)
    video_reference="https://storage.googleapis.com/com-roboflow-marketing/inference/people-walking.mp4",
    # pipeline को बताएं कि inference results के साथ क्या करना है। render_boxes एक built-in sink है जो वीडियो के ऊपर boxes को render करता है
    on_prediction=render_boxes,
    # roboflow api से models लोड करने के लिए अपनी roboflow api key प्रदान करें
    api_key=api_key,
)

# पाइपलाइन शुरू करें और वीडियो stream को process करने वाले thread से जुड़ें।
pipeline.start()
pipeline.join()

video reference क्या है?

Inference Pipelines कई अलग-अलग प्रकार के video streams को consume कर सकती हैं:

• Device ID (integer): एक integer देने पर pipeline को किसी local device, जैसे webcam, से video stream करने का निर्देश मिलता है। आमतौर पर, built-in webcams device के रूप में दिखाई देती हैं 0.

• Video file (string): video file का path देने पर pipeline फ़ाइल से हर frame पढ़ती है, निर्दिष्ट model के साथ inference चलाती है, फिर on_prediction method को परिणामस्वरूप predictions के प्रत्येक सेट के साथ चलाती है।

• Video URL (string): video URL देना video file path देने के बराबर है और पहले वीडियो डाउनलोड करने की आवश्यकता को समाप्त करता है।

• RTSP URL (string): RTSP URL देने पर pipeline RTSP stream से जितनी तेज़ी से संभव हो frames stream करती है, फिर on_prediction callback को उपलब्ध सबसे नए frame पर चलाती है।

• एक सूची जिसमें ऊपर वर्णित किसी भी मान के elements हो सकते हैं।

कैसे InferencePipeline काम करता है

inference pipeline आरेख

InferencePipeline प्रत्येक प्रदान किए गए video reference के लिए एक video source consumer thread शुरू करता है। videos से frames को एक video multiplexer द्वारा लिया जाता है जो प्रतीक्षा करता है batch_collection_timeout (यदि कोई source frame प्रदान नहीं करता है, तो एक छोटा batch on_video_frame(...)को भेजा जाता है, लेकिन missing frames और predictions को भरा जाता है None before passing to on_prediction(...)). on_prediction(...) में काम कर सकता है SEQUENTIAL mode (एक बार में केवल एक element), या BATCH mode (सभी batch elements एक साथ); यह sink_mode parameter द्वारा नियंत्रित होता है।

स्थिर video files के लिए, InferencePipeline डिफ़ॉल्ट रूप से सभी frames को process करता है। streams के लिए, buffers से frames को हटाना संभव है ताकि हमेशा सबसे हालिया data process हो (जब model inference धीमा हो, तो buffer में अधिक frames जमा हो सकते हैं; stream processing पुराने frames को हटाती है और केवल सबसे हालिया को process करती है)।

स्थिरता बढ़ाने के लिए, streams को process करते समय processing के दौरान connectivity खो जाने पर video sources अपने आप फिर से connect हो जाते हैं। इसका उद्देश्य production environment में failures को रोकना है, जहाँ pipeline लंबे घंटों तक चल सकती है और source downtime को सहजता से संभालने की आवश्यकता होती है।

कस्टम inference logic

InferencePipeline कस्टम inference logic चलाने का समर्थन करता है। model ID पास करने के बजाय, आप एक कस्टम callable पास कर सकते हैं। इस callable को एक VideoFrame स्वीकार करना चाहिए और processing के परिणामों वाली एक dictionary लौटानी चाहिए ( on_video_frame handler के रूप में)। यह model predictions या किसी अन्य processing के परिणाम हो सकते हैं जिन्हें आप चलाना चाहते हैं।

यह ध्यान देने योग्य महत्वपूर्ण बात है कि उपयोग किया जा रहा sink ( on_prediction handler) को on_video_frame(...) response के विशिष्ट format के अनुसार समायोजित करना होगा। इस तरह, आप वीडियो processing को अपनी इच्छानुसार आकार दे सकते हैं।

# यह उदाहरण, संदर्भ implementation है - आपको अपने उद्देश्य के अनुसार code को समायोजित करना होगा
import os
import json
from inference.core.interfaces.camera.entities import VideoFrame
from inference import InferencePipeline
from typing import Any, List

TARGET_DIR = "./my_predictions"

class MyModel:

  def __init__(self, weights_path: str):
    self._model = your_model_loader(weights_path)

  # v0.9.18 से पहले
  def infer(self, video_frame: VideoFrame) -> Any:
    return self._model(video_frame.image)

  # v0.9.18 के बाद
  def infer(self, video_frames: List[VideoFrame]) -> List[Any]:
    # परिणामों को एकल frame के model prediction का प्रतिनिधित्व करने वाले elements की सूची के रूप में लौटाना चाहिए
    # और order अपरिवर्तित रहना चाहिए।
    return self._model([v.image for v in video_frames])

def save_prediction(prediction: dict, video_frame: VideoFrame) -> None:
  with open(os.path.join(TARGET_DIR, f"{video_frame.frame_id}.json")) as f:
    json.dump(prediction, f)

my_model = MyModel("./my_model.pt")

pipeline = InferencePipeline.init_with_custom_logic(
  video_reference="./my_video.mp4",
  on_video_frame=my_model.infer,
  on_prediction=save_prediction,
)

# पाइपलाइन शुरू करें
pipeline.start()
# पाइपलाइन के समाप्त होने की प्रतीक्षा करें
pipeline.join()

InferencePipeline Workflows के साथ

InferencePipeline भी चला सकते हैं Roboflow Workflowsजैसा कि नीचे दिखाया गया है:

from inference import InferencePipeline
from inference.core.interfaces.camera.entities import VideoFrame
from inference.core.interfaces.stream.sinks import render_boxes

def workflows_sink(
    predictions: dict,
    video_frame: VideoFrame,
) -> None:
    render_boxes(
        predictions["predictions"][0],
        video_frame,
        display_statistics=True,
    )


# यहाँ आपको workflow की बहुत बुनियादी परिभाषा मिल सकती है - एक single object detection model के साथ।
workflow_specification = {
    "specification": {
        "version": "1.0",
        "inputs": [
            {"type": "InferenceImage", "name": "image"},
        ],
        "steps": [
            {
                "type": "ObjectDetectionModel",
                "name": "step_1",
                "image": "$inputs.image",
                "model_id": "rfdetr-small",
                "confidence": 0.5,
            }
        ],
        "outputs": [
            {"type": "JsonField", "name": "predictions", "selector": "$steps.step_1.*"},
        ],
    }
}
pipeline = InferencePipeline.init_with_workflow(
    video_reference="./my_video.mp4",
    workflow_specification=workflow_specification,
    on_prediction=workflows_sink,
    image_input_name="image",  # आपके द्वारा परिभाषित WorkflowImage input के नाम के अनुसार समायोजित करें
    video_metadata_input_name="video_metadata" # v0.17.0 से उपलब्ध! आपके द्वारा परिभाषित WorkflowVideoMetadata input के नाम के अनुसार समायोजित करें
)

# पाइपलाइन शुरू करें
pipeline.start()
# पाइपलाइन के समाप्त होने की प्रतीक्षा करें
pipeline.join()

आप InferencePipeline को Roboflow app में registered किसी Workflow के साथ शुरू कर सकते हैं, इसके लिए अपना workspace_name और workflow_id:

pipeline = InferencePipeline.init_with_workflow(
    video_reference="./my_video.mp4",
    workspace_name="<your_workspace>",
    workflow_id="<your_workflow_id_to_be_found_in_workflow_url>",
    on_prediction=workflows_sink,
)

Workflows profiling। आप अंदर अपने Workflow execution का profiling कर सकते हैं InferencePipeline environment variable को export करके ENABLE_WORKFLOWS_PROFILING=True. इसके अतिरिक्त, आप environment variable के माध्यम से profiler buffer में रखे जाने वाले frames की संख्या समायोजित कर सकते हैं WORKFLOWS_PROFILER_BUFFER_SIZE. init_with_workflow(...) एक profiling_directory parameter भी लेता है जो trace को कहाँ सहेजना है, यह निर्धारित करता है।

Sinks

Sinks तय करते हैं कि एक Inference Pipeline को प्रत्येक prediction के साथ क्या करना चाहिए। sink निम्न signature वाली एक function है:

from typing import Union, List, Optional
from inference.core.interfaces.camera.entities import VideoFrame

def on_prediction(
    predictions: Union[dict, List[Optional[dict]]],
    video_frame: Union[VideoFrame, List[Optional[VideoFrame]]],
) -> None:
    for prediction, frame in zip(predictions, video_frame):
        if prediction is None:
            # खाली frame
            continue
        # कुछ processing

Arguments हैं:

• पूर्वानुमान: एक dictionary (या multiple video sources उपयोग करते समय dicts की सूची) जो model के infer(...) method को call करने से प्राप्त response object है।

• video_frame: एक VideoFrame object (या VideoFrames की सूची) जिसमें video frame से metadata और pixel data शामिल है।

उपयोग

आप on_prediction ऐसे अन्य parameters भी बना सकते हैं जो इसके व्यवहार को configure करें, लेकिन उन्हें InferencePipeline init methods में injection से पहले function closure में latch करना होगा।

from functools import partial
from inference.core.interfaces.camera.entities import VideoFrame
from inference import InferencePipeline


def on_prediction(
    predictions: dict,
    video_frame: VideoFrame,
    my_parameter: int,
) -> None:
    # आपको अपनी logic यहाँ implement करनी होगी, जिसमें `my_parameter` का उपयोग हो
    पास करें

pipeline = InferencePipeline.init(
  video_reference="./my_video.mp4",
  model_id="rfdetr-small",
  on_prediction=partial(on_prediction, my_parameter=42),
)

कस्टम sink tutorial

आइए चरण-दर-चरण एक custom sink बनाने की प्रक्रिया देखें। सबसे पहले, एक सरल sink जो frame ID प्रिंट करता है:

from inference import InferencePipeline
# type hinting के लिए VideoFrame import करें
from inference.core.interfaces.camera.entities import VideoFrame

# sink function परिभाषित करें
def my_custom_sink(predictions: dict, video_frame: VideoFrame):
    # video_frame object का frame ID प्रिंट करें
    print(f"Frame ID: {video_frame.frame_id}")

pipeline = InferencePipeline.init(
    model_id="rfdetr-large",
    video_reference="https://storage.googleapis.com/com-roboflow-marketing/inference/people-walking.mp4",
    on_prediction=my_custom_sink,
)

pipeline.start()
pipeline.join()

आउटपुट कुछ ऐसा दिखना चाहिए:

Frame ID: 1
Frame ID: 2
Frame ID: 3

अब आइए कुछ अधिक उपयोगी करते हैं और predictions को विज़ुअलाइज़ करने के लिए अपने custom sink का उपयोग करें Supervision:

from inference import InferencePipeline
from inference.core.interfaces.camera.entities import VideoFrame

# अपनी annotated images दिखाने के लिए opencv import करें
import cv2
# हमारी predictions को visualize करने में मदद के लिए supervision import करें
का उपयोग करके visualize करते हैं

# अपने custom sink में उपयोग करने के लिए एक bounding box annotator और label annotator बनाएं
label_annotator = sv.LabelAnnotator()
box_annotator = sv.BoxAnnotator()

def my_custom_sink(predictions: dict, video_frame: VideoFrame):
    # प्रत्येक prediction के लिए text labels प्राप्त करें
    labels = [p["class"] for p in predictions["predictions"]]
    # अपनी predictions को Supervision Detections api में लोड करें
    detections = sv.Detections.from_inference(predictions)
    # अपने supervision annotator, video_frame, predictions (as supervision Detections), और prediction labels का उपयोग करके frame annotate करें
    image = label_annotator.annotate(
        scene=video_frame.image.copy(), detections=detections, labels=labels
    )
    image = box_annotator.annotate(image, detections=detections)
    # annotated image दिखाएं
    cv2.imshow("Predictions", image)
    cv2.waitKey(1)

pipeline = InferencePipeline.init(
    model_id="rfdetr-large",
    video_reference="https://storage.googleapis.com/com-roboflow-marketing/inference/people-walking.mp4",
    on_prediction=my_custom_sink,
)

pipeline.start()
pipeline.join()

आपको अपनी स्क्रीन पर कुछ ऐसा दिखाई देना चाहिए:

Custom sinks (advanced)

एक custom sink बनाने के लिए, उचित signature के साथ एक नया function परिभाषित करें।

from typing import Union, List, Optional, Any
from inference.core.interfaces.camera.entities import VideoFrame

def on_prediction(
    predictions: Union[Any, List[Optional[dict]]],
    video_frame: Union[VideoFrame, List[Optional[VideoFrame]]],
) -> None:
    if not issubclass(type(predictions), list):
      # यह single code के साथ sequential और batch processing दोनों का समर्थन करने के लिए आवश्यक है
      # यदि आप केवल एक mode का उपयोग करते हैं - तो आप ऐसा function बना सकते हैं जो केवल एक प्रकार के
      # input को संभाले
      predictions = [predictions]
      video_frame = [video_frame]
    for prediction, frame in zip(predictions, video_frame):
        if prediction is None:
            # खाली frame
            continue
        # कुछ processing

InferencePipeline एक sink_mode parameter प्रदान करता है ताकि predictions को आपके sink तक कैसे भेजा जाए, इसे नियंत्रित किया जा सके। SinkMode.SEQUENTIALके साथ, प्रत्येक frame और prediction sink को एक अलग call trigger करता है। SinkMode.BATCHके साथ, frames और predictions की एक सूची sink को प्रदान की जाती है, जो हमेशा video sources के order के अनुसार aligned रहती है, None उन video frames या predictions की जगह पर values के साथ जो batch_collection_timeout. SinkMode.ADAPTIVE डिफ़ॉल्ट mode है: single video input के लिए, pipeline ऐसा व्यवहार करती है जैसे कि वह SinkMode.SEQUENTIAL. multiple videos संभालने के लिए, sink को predictions: List[Optional[dict]] और video_frame: List[Optional[VideoFrame]]. सरल sinks का उपयोग करके multiple videos process करना भी संभव है, लेकिन तब SinkMode.SEQUENTIAL का उपयोग किया जाना चाहिए, जिससे sink प्रत्येक prediction element पर अलग-अलग call हो।

क्यों है Optional में List[Optional[dict]] और List[Optional[VideoFrame]]?

ऐसा हो सकता है कि सभी video sources से video frames एकत्र करना संभव न हो (उदाहरण के लिए जब sources में से कोई एक disconnect हो जाए और पुनः कनेक्शन का प्रयास किया जा रहा हो)। पूर्वानुमान और video_frame को order दिया जाता है ताकि यह video_reference सूची के order से मेल खाए InferencePipelineऔर None elements missing frames की स्थिति में दिखाई देते हैं। हम यह जानकारी sink को प्रदान करते हैं, क्योंकि कुछ sinks को batch से सभी predictions और video frames (भले ही missing हों) की आवश्यकता हो सकती है। उदाहरण के लिए, render_boxes(...) sink को tile mosaic में frames की position बनाए रखने के लिए उस जानकारी की आवश्यकता होती है।

Prediction format

Predictions sink को एक dictionary के रूप में प्रदान की जाती हैं जिसमें key पूर्वानुमानहोती है, जो single frame या frames के batch के लिए predictions रखती है। सामग्री इस पर निर्भर करती है कि पीछे कौन सा model चल रहा है InferencePipeline; Roboflow models के लिए यह dict या dicts की list के रूप में आती है।

model output के आधार पर, predictions अलग दिखती हैं। आपको sink को prediction format के अनुसार समायोजित करना होगा। उदाहरण के लिए, एक Roboflow object detection prediction में निम्न keys होती हैं:

• xx

• : अनुमानित bounding box का केंद्र x coordinate, pixels मेंy

• : अनुमानित bounding box की width, pixels मेंwidth

• height: अनुमानित bounding box की height, pixels में

• confidence: prediction का confidence value (0 और 1 के बीच)

• class: अनुमानित class name

• class_id: अनुमानित class ID

Built-in sinks

Inference में कई built-in sinks हैं जो उपयोग के लिए तैयार हैं (देखें inference/core/interfaces/stream/sinks.py).

render_boxes(...)

render boxes sink predictions को visualize करता है और उन्हें stream पर overlay करता है। यह predictions render करने और annotated frame प्रदर्शित करने के लिए Supervision annotators का उपयोग करता है। यह केवल उन Roboflow models के लिए काम करता है जो detection-based output देते हैं (object-detection, instance-segmentation, keypoint-detection), और predictions के सभी विवरण डिफ़ॉल्ट रूप से प्रदर्शित नहीं हो सकते (जैसे detected keypoints)।

UDPSink(...)

UDP sink predictions को UDP port के माध्यम से broadcast करता है। इस port को आगे processing के लिए client code द्वारा listen किया जा सकता है। यह Python की default JSON serialization का उपयोग करता है, इसलिए predictions serializable होनी चाहिए, अन्यथा एक error उत्पन्न होगा।

multi_sink(...)

multi-sink कई sinks को जोड़ता है ताकि एक ही inference result पर कई actions हो सकें।

VideoFileSink(...)

video file sink predictions को visualize करता है, ठीक render_boxes(...) sink की तरह; हालांकि, annotated frames को प्रदर्शित करने के बजाय, यह उन्हें एक video file में सहेजता है। render_boxes(...) लागू होते हैं।

Model weights download

जब आप inference पहली बार चलाते हैं तो model weights स्वतः डाउनलोड हो जाते हैं। आप इंटरनेट से जुड़े रहते हुए pipeline को एक बार initialize करके weights को पहले से डाउनलोड कर सकते हैं:

from inference import InferencePipeline

pipeline = InferencePipeline.init(
    model_id="rfdetr-base",
    video_reference=0,
    on_prediction=lambda predictions, video_frame: None,
    api_key="YOUR_ROBOFLOW_API_KEY",
)

pipeline.start()
pipeline.terminate()

print("Model weights सफलतापूर्वक डाउनलोड हो गए!")

वैकल्पिक रूप से, पहले से weights डाउनलोड करने के लिए get_model() का उपयोग करें:

from inference import get_model

get_model("rfdetr-base")

आप cache directory की जाँच करके cached models सत्यापित कर सकते हैं:

ls -lh /tmp/cache

आपको प्रत्येक cached model के लिए directories दिखनी चाहिए, जिन्हें आमतौर पर model ID से नामित किया जाता है।

के बारे में और पढ़ें weights caching, persistent storage, और Docker configuration.

अन्य pipeline configuration

Inference Pipelines अत्यधिक configurable हैं। configuration विकल्पों में शामिल हैं:

• max_fps: frame processing की अधिकतम दर निर्धारित करने के लिए उपयोग किया जाता है।

• confidence: inference के लिए उपयोग किया जाने वाला confidence threshold।

• iou_threshold: inference के लिए उपयोग किया जाने वाला IoU threshold।

• video_source_properties: video source को configure करने के लिए properties का वैकल्पिक dictionary, जो cv2 VideoCapture properties के अनुरूप है cv2.CAP_PROP_*. सभी संभावित properties की सूची के लिए OpenCV documentation देखें।

from inference import InferencePipeline
pipeline = InferencePipeline.init(
    ...,
    max_fps=10,
    confidence=0.75,
    iou_threshold=0.4,
    video_source_properties={
        "frame_width": 1920.0,
        "frame_height": 1080.0,
        "fps": 30.0,
    },
)

Inference Pipeline पैरामीटरों की पूरी सूची के लिए, स्रोत देखें inference/core/interfaces/stream/inference_pipeline.py.