/deepstream-dev

DeepStream Service Maker for Python (pyservicemaker) API Reference

Introduction

The DeepStream Service Maker provides a high-level Python API (pyservicemaker) for building DeepStream applications. It abstracts away the complexity of GStreamer C API and provides a more intuitive, Pythonic interface for constructing video analytics pipelines.

Installation

The pyservicemaker package is installed as part of DeepStream SDK:

pip install /opt/nvidia/deepstream/deepstream/service-maker/python/pyservicemaker*.whl pyyaml

Inside a virtual environment: pyservicemaker is installed system-wide but is NOT accessible from a standard venv. If the application uses a virtual environment, you must install it inside the venv:

python3 -m venv venv
source venv/bin/activate
pip install /opt/nvidia/deepstream/deepstream/service-maker/python/pyservicemaker*.whl pyyaml

Two API Approaches

Service Maker provides two APIs for building pipelines:

1. Pipeline API: Low-level, element-by-element pipeline construction
2. Flow API: High-level, declarative pipeline construction

Pipeline API

The Pipeline API provides fine-grained control over pipeline construction, similar to GStreamer C API but with Python syntax.

Core Classes

Pipeline

Main class for creating and managing DeepStream pipelines.

Constructor:

from pyservicemaker import Pipeline

# Create empty pipeline
pipeline = Pipeline("pipeline-name")

# Create pipeline from YAML config
pipeline = Pipeline("pipeline-name", "/path/to/config.yml")

Methods:

add(element_type, name, properties=None)

Add a GStreamer element to the pipeline.

Parameters:

• element_type (str): GStreamer element factory name (e.g., "nvinfer", "nvstreammux")
• name (str): Unique name for the element
• properties (dict, optional): Element properties as key-value pairs

Returns: Pipeline instance (for method chaining)

Example:

pipeline.add("filesrc", "src", {"location": "/path/to/video.h264"})
pipeline.add("h264parse", "parser")
pipeline.add("nvv4l2decoder", "decoder")
pipeline.add("nvstreammux", "mux", {"batch-size": 1, "width": 1920, "height": 1080})
pipeline.add("nvinfer", "infer", {"config-file-path": "/path/to/config.yml"})

link(*element_names)

Link elements in sequence. Elements are connected in the order specified.

Parameters:

• *element_names: Variable number of element names or tuples for request pads

Returns: Pipeline instance (for method chaining)

Example:

# Simple linear linking
pipeline.link("src", "parser", "decoder", "mux", "infer", "sink")

# Linking with request pads (for nvstreammux)
pipeline.link(("decoder", "mux"), ("", "sink_%u"))
# This connects decoder src pad to mux sink_0 pad

Request Pad Linking: For elements with dynamic pads (like nvstreammux), use tuple syntax:

# Format: (source_element, sink_element), (source_pad, sink_pad_template)
pipeline.link(("decoder1", "mux"), ("", "sink_%u"))  # Connects to sink_0
pipeline.link(("decoder2", "mux"), ("", "sink_%u"))  # Connects to sink_1

CRITICAL: Always use "sink_%u" pad template, NOT "sink_0", "sink_1", or f"sink_{i}"

• "sink_%u" is a GStreamer pad template that automatically assigns sink pads (sink_0, sink_1, sink_2, etc.)
• Using literal pad names like "sink_0" or f"sink_{i}" will FAIL because these pads don't exist until requested
• The %u format specifier tells GStreamer to automatically assign the next available sink pad index

Examples with different source types:

# With nvv4l2decoder (decoded video source)
pipeline.link((f"decoder{i}", "mux"), ("", "sink_%u"))  # CORRECT

# With nvurisrcbin (RTSP/file source with dynamic pads)
pipeline.link((f"src{i}", "mux"), ("", "sink_%u"))  # CORRECT - nvurisrcbin has dynamic src pad

# WRONG - DO NOT USE:
pipeline.link((f"src{i}", "mux"), ("", f"sink_{i}"))  # INCORRECT - will fail!
pipeline.link((f"src{i}", "mux"), ("", "sink_0"))     # INCORRECT - pad doesn't exist!

attach(target, what, name='', tips='', properties=None)

Attach a probe (or other custom object) to a named element in the pipeline.

Parameters:

• target (str): Name of the pipeline element to attach to
• what: Probe instance or name of a built-in probe module (e.g. "measure_fps_probe")
• name (str, optional): Name for the probe. Not needed when what is an explicitly created Probe object.
• tips (str, optional): Extra information for the custom object
• properties (dict, optional): Properties to set on the object. Not applicable for explicitly created Probe objects.

CRITICAL: The parameter is name, NOT probe_name. Using probe_name will raise TypeError.

Returns: Pipeline instance (for method chaining)

Example:

from pyservicemaker import Probe, BatchMetadataOperator

class MyProbe(BatchMetadataOperator):
    def handle_metadata(self, batch_meta):
        # Process metadata
        pass

pipeline.attach("infer", Probe("my-probe", MyProbe()))
# Or attach built-in probe by module name, giving it a name
pipeline.attach("infer", "measure_fps_probe", name="fps-probe")

start()

Start the pipeline (set to PLAYING state).

Returns: Pipeline instance (for method chaining)

Example:

pipeline.start()

wait()

Wait for pipeline to finish (blocking call until EOS or error).

Returns: None

Example:

pipeline.start().wait()

set(properties)

Set properties on an element (when element is accessed via indexing).

Parameters:

• properties (dict): Properties to set

Example:

pipeline["infer"].set({"batch-size": 4})

Element Access via Indexing

Access elements by name to get/set properties:

# Get element
infer_element = pipeline["infer"]

# Set properties
pipeline["infer"].set({"batch-size": 4})

# Get properties
batch_size = pipeline["infer"].get("batch-size")

Complete Pipeline API Example

from pyservicemaker import Pipeline, Probe, BatchMetadataOperator
import platform

PIPELINE_NAME = "my-pipeline"
CONFIG_FILE = "/path/to/inference_config.txt"  # Must be INI-style text format, NOT YAML
VIDEO_FILE = "/path/to/video.h264"

class ObjectCounter(BatchMetadataOperator):
    def handle_metadata(self, batch_meta):
        for frame_meta in batch_meta.frame_items:
            # IMPORTANT: object_items returns an ITERATOR, not a list
            # You cannot use len() directly - iterate and count instead
            obj_count = 0
            for obj in frame_meta.object_items:
                obj_count += 1
            print(f"Frame {frame_meta.frame_number}: {obj_count} objects")

# Create pipeline
pipeline = (Pipeline(PIPELINE_NAME)
    .add("filesrc", "src", {"location": VIDEO_FILE})
    .add("h264parse", "parser")
    .add("nvv4l2decoder", "decoder")
    .add("nvstreammux", "mux", {
        "batch-size": 1,
        "width": 1920,
        "height": 1080
    })
    .add("nvinfer", "infer", {"config-file-path": CONFIG_FILE})
    .add("nvosdbin", "osd")
    .add("nv3dsink" if platform.processor() == "aarch64" else "nveglglessink", "sink")
    .link("src", "parser", "decoder")
    .link(("decoder", "mux"), ("", "sink_%u"))
    .link("mux", "infer", "osd", "sink")
    .attach("infer", Probe("counter", ObjectCounter()))
    .start()
    .wait())

Flow API

The Flow API provides a high-level, declarative interface for common pipeline patterns.

Core Classes

Flow

High-level API for building pipelines using method chaining.

Constructor:

from pyservicemaker import Flow, Pipeline

pipeline = Pipeline("pipeline-name")
flow = Flow(pipeline)

Methods:

batch_capture(sources, record_config=None, **kwargs)

Configure batch capture from multiple sources.

Parameters:

• sources (list): List of source file paths or URIs
• record_config (class RecordConfig): Optional smart recording (see full table in record_config details section below). If None, no smart recording is configured on sources.
• kwargs (dict): Optional overrides merged into mux and/or source properties (see kwargs dict details section below).

record_config details: RecordConfig instance should be constructed as description in record_config Construction examples section. The following RecordConfig fields can be used to configure smart recording.

record_config Construction examples:

from pyservicemaker import RecordConfig

# Local smart recording (minimal)
rec_local = RecordConfig()  # recording_type defaults to "local"

# Local with explicit paths and cache
rec_local = RecordConfig(
    recording_type="local",
    rec_cache=20,
    rec_container=0,
    rec_dir_path="/data/recordings",
    rec_mode=0,
)

# Cloud smart recording (all cloud fields required)
rec_cloud = RecordConfig(
    recording_type="cloud",
    proto_lib="/path/to/broker_library.so",
    conn_str="localhost;9092",
    msgconv_config_file="/path/to/dstest5_msgconv_sample_config.txt",
    proto_config_file="/path/to/cfg_kafka.txt",
    topic_list="sr-test",
    rec_cache=20,
    rec_dir_path=".",
    rec_mode=0,
)

kwargs dict details: Any matching hyphenated name in the merged kwargs dict overrides the default value of the corresponding property, the following keys are supported:

• gpu_id (int): Used as the gpu-id property of nvstreammux and as gpu-id on each nvurisrcbin.
• width (int): Used as the width property of nvstreammux, default value is 1920.
• height (int): Used as the height property of nvstreammux, default value is 1080.
• batch_size (int): Used as the batch-size property of nvstreammux, default value is the number of URIs (if non-empty).
• batched_push_timeout (int): Used as the batched-push-timeout property of nvstreammux, default value is 33000.
• buffer_pool_size (int): Used as the buffer-pool-size property of nvstreammux, default value is 4.
• drop_pipeline_eos (bool): Used as the drop-pipeline-eos property of nvstreammux, default value is False.
• live_source (bool): Used as the live-source property of nvstreammux, default value is False.
• file_loop(bool): Used as the file-loop property of nvstreammux, default value is False.

Returns: Flow instance (for method chaining)

Example:

flow.batch_capture([
    "/path/to/video1.h264",
    "/path/to/video2.h264",
    "rtsp://camera-ip/stream"
])

# Mux resolution and batching setting
flow.batch_capture(uris, width=1280, height=720, batch_size=4)

# GPU and file loop for file sources
flow.batch_capture(uris, gpu_id=0, file_loop=True)

# Combine with YAML: kwargs override missing keys from source-config.properties
flow.batch_capture("/path/to/sources.yaml", width=1920, height=1080, live_source=True)

Important: batch_capture function sets the nvstreammux batch-size according to the input stream number by default, it is not necessary to set 'batch-size' with batch_capture unless you want to support dynamic source adding/removing.

infer(config_file_path, with_triton, **kwargs)

Add inference stage to the pipeline.

Parameters:

• config_file_path (str): Path to inference configuration file
• with_triton (bool): If False (default), adds nvinfer. If True, adds nvinferserver for Triton-based inference.
• kwargs (dict): Optional properties passed to gst-nvinfer or gst-nvinferserver plugin of DeepStream. Underscores in keyword names are converted to hyphens for GStreamer properties (e.g. batch_size → batch-size). Common overrides include batch_size, unique_id, model_engine_file, gpu_id, and other keys supported by nvinfer / nvinferserver for your install.

Returns: Flow instance (for method chaining)

Notes: For multiple streams inferencing case, batch_size property should be set as the same value as the stream number.

Examples:

flow.infer("/path/to/pgie_config.yml")

#set nvinfer/nvinferserver properties with Flow.infer function
flow.infer("/path/to/pgie_config.yml",unique_id=5, batch_size=4)

track(**kwargs)

Add tracker for object tracking. Must be used after primary inference.

Parameters: The following keyword arguments(kwargs) are passed to nvtrack as properties.

Notes: Example tracker configs (paths may vary by installation):

• NvDCF (performance): config_tracker_NvDCF_perf.yml
• NvDCF (accuracy): config_tracker_NvDCF_accuracy.yml
• NvSORT: config_tracker_NvSORT.yml
• IOU: config_tracker_IOU.yml
• NvDeepSORT: config_tracker_NvDeepSORT.yml

Example:

flow = flow.track(ll_config_file=config_tracker_NvDCF_perf.yml, ll_lib_file=libnvds_nvmultiobjecttracker.so)

analyze(config_file_path,**kwargs)

Add analytics for region-of-interest (ROI), line-crossing, overcrowding and direction analytics. The result will be output as AnalyticsFrameMeta in frame meta and AnalyticsObjInfo in object meta.

Parameters:

• config_file_path (str): Path to analytics configuration file
• kwargs (dict): Optional properties passed to gst-nvdsanalytics plugin of DeepStream

Notes: analytics MUST follow tracker to work properly.

Example:

from pyservicemaker import Pipeline, Flow, BatchMetadataOperator, Probe, RenderMode

PGIE_CONFIG = "/path/to/config_infer_primary.yml"
TRACKER_LL_CONFIG = "/path/to/config_tracker_NvDCF_perf.yml"
TRACKER_LL_LIB = "/path/to/libnvds_nvmultiobjecttracker.so"
ANALYTICS_CONFIG = "/path/to/config_analytics.txt"  # nvdsanalytics config
SOURCE = "/path/to/source_list.yaml"

class AnalyticsProbe(BatchMetadataOperator):
    def handle_metadata(self, batch_meta):
        for frame_meta in batch_meta.frame_items:
            # Frame-level analytics (ROI counts, line-cross counts)
            for user_meta in frame_meta.nvdsanalytics_frame_items:
                afm = user_meta.as_nvdsanalytics_frame()
                if afm:
                    print(f"Frame {frame_meta.frame_number}: unique_id={afm.unique_id} "
                          f"obj_in_roi_cnt={afm.obj_in_roi_cnt} obj_lc_curr_cnt={afm.obj_lc_curr_cnt} "
                          f"obj_cnt={afm.obj_cnt} oc_status={afm.oc_status}")

            # Object-level analytics (which ROI/line each object is in)
            for obj_meta in frame_meta.object_items:
                for user_meta in obj_meta.nvdsanalytics_obj_items:
                    aoi = user_meta.as_nvdsanalytics_obj()
                    if aoi:
                        print(f"  object_id={obj_meta.object_id} roi_status={aoi.roi_status} "
                              f"lc_status={aoi.lc_status} dir_status={aoi.dir_status} obj_status={aoi.obj_status}")

pipeline = Pipeline("analytics-demo")
flow = Flow(pipeline).batch_capture(SOURCE, width=1920, height=1080)
flow = flow.infer(PGIE_CONFIG)
flow = flow.track(ll_config_file=TRACKER_LL_CONFIG, ll_lib_file=TRACKER_LL_LIB)
flow = flow.analyze(ANALYTICS_CONFIG)
flow = flow.attach(what=Probe("analytics_probe", AnalyticsProbe()))
flow = flow.render(RenderMode.DISCARD, sync=False)
flow()

attach(what, name='', tips='', properties=None)

Attach a probe to the current flow.

Parameters:

• what: Probe instance or element name
• name (str, optional): Name for the probe. Not applicable when what is an explicitly created Probe object.
• tips (str, optional): Extra information for the custom object
• properties (dict, optional): Properties to set on the object.

Returns: Flow instance (for method chaining)

Example:

from pyservicemaker import Probe
# Attach a custom probe (name is embedded in the Probe object)
flow.attach(Probe("my-probe", MyProbe()))

# Attach built-in probe by module name and name the probe by 'name'
flow = flow.attach(
            what="measure_fps_probe",
            name="fps_probe"
        )

render()

Add rendering stage to the pipeline.

Returns: Flow instance (for method chaining)

Example:

flow.render()

__call__() (Invocation)

Execute the pipeline (start and wait).

Example:

flow()  # Starts and waits for completion

Complete Flow API Example

from pyservicemaker import Pipeline, Flow, Probe, BatchMetadataOperator

class ObjectCounter(BatchMetadataOperator):
    def handle_metadata(self, batch_meta):
        for frame_meta in batch_meta.frame_items:
            # IMPORTANT: object_items is an ITERATOR - cannot use len()
            obj_count = 0
            for obj in frame_meta.object_items:
                obj_count += 1
            print(f"Frame {frame_meta.frame_number}: {obj_count} objects")

def main():
    pipeline = Pipeline("my-pipeline")
    flow = Flow(pipeline)
    
    flow.batch_capture(["/path/to/video.h264"]) \
        .infer("/path/to/inference_config.txt") \  # Must be INI-style text format
        .attach(Probe("counter", ObjectCounter())) \
        .render()()
    
if __name__ == "__main__":
    main()

Metadata API

CRITICAL: Iterator Handling

WARNING: Properties like frame_meta.object_items, frame_meta.tensor_items, and frame_meta.user_items return ITERATORS, not lists!

Common Mistakes to Avoid:

# WRONG - Will crash with "TypeError: object of type 'iterator' has no len()"
count = len(frame_meta.object_items)

# WRONG - Iterator can only be consumed once
for obj in frame_meta.object_items:
    process(obj)
for obj in frame_meta.object_items:  # This loop will be empty!
    do_something(obj)

Correct Patterns:

# CORRECT - Count by iterating
obj_count = 0
for obj in frame_meta.object_items:
    obj_count += 1
    process(obj)

# CORRECT - If you need to iterate multiple times, convert to list first
# (only if you actually need multiple iterations)
object_list = list(frame_meta.object_items)
count = len(object_list)
for obj in object_list:
    process(obj)

BatchMetadataOperator

Base class for implementing custom metadata processing.

Methods:

handle_metadata(batch_meta)

Override this method to process batch metadata.

Parameters:

• batch_meta: BatchMetadata object containing frame and object metadata

Example:

class MyOperator(BatchMetadataOperator):
    def handle_metadata(self, batch_meta):
        for frame_meta in batch_meta.frame_items:
            # Process each frame
            # NOTE: object_items is an ITERATOR, not a list!
            for object_meta in frame_meta.object_items:
                # Process each object
                pass

BatchMetadata Object

Properties:

• frame_items: List of FrameMetadata objects
• Methods for acquiring metadata objects

Methods:

• acquire_object_meta(): Create new object metadata
• acquire_display_meta(): Create new display metadata
• acquire_user_meta(): Create new user metadata
• acquire_event_message_meta(): Create new EventMessageUserMetadata for nvmsgconv (see EventMessageUserMetadata section below)

FrameMetadata Object

Properties:

• frame_number: Frame number (int)
• pad_index: Source pad index (int)
• batch_id: Location of frame in the batch (int)
• source_id: Source ID of the frame, e.g., camera ID (int)
• source_width: Width of the frame at input to streammux (int)
• source_height: Height of the frame at input to streammux (int)
• pipeline_width: Width of the frame at output of streammux (int)
• pipeline_height: Height of the frame at output of streammux (int)
• buffer_pts: Presentation timestamp (PTS) of the frame in nanoseconds (int)
• ntp_timestamp: NTP timestamp of the frame (int)
• object_items: ITERATOR of ObjectMetadata objects (NOT a list - cannot use len())
• tensor_items: ITERATOR of TensorOutputUserMetadata objects (NOT a list - cannot use len())
• segmentation_items: ITERATOR of SegmentationUserMetadata objects (NOT a list - cannot use len())
• nvdsanalytics_frame_items: ITERATOR of AnalyticsFrameMeta objects (NOT a list - cannot use len()) IMPORTANT: The *_items properties return iterators that can only be consumed once. See "CRITICAL: Iterator Handling" section above.

NOTE: There is no timestamp property. Use buffer_pts for PTS timestamp or ntp_timestamp for NTP timestamp.

Methods:

• append(meta): Add metadata to frame

ObjectMetadata Object

Properties:

• class_id: Class ID (int)
• confidence: Confidence score (float)
• object_id: Unique tracking ID assigned by tracker (int). Value is 0xFFFFFFFFFFFFFFFF (UNTRACKED_OBJECT_ID) if object has not been tracked.
• tracker_confidence: Confidence value from tracker (float). Set to -0.1 for KLT and IOU trackers.
• rect_params: Rectangle parameters object
  • left: Left coordinate (float)
  • top: Top coordinate (float)
  • width: Width (float)
  • height: Height (float)
  • border_width: Border width (int)
  • border_color: Border color (Color object)
• label: String describing the object class
• text_params: Text parameters for OSD display (NvOSD_TextParams)
• mask_params: Bbox-local mask parameters for a per-object OSD overlay (NvOSD_MaskParams); it does not represent a full-frame segmentation map.
• classifier_items: ITERATOR of ClassifierMetadata objects. (NOT a list - cannot use len())
• tensor_items: ITERATOR of TensorOutputUserMetadata objects. (NOT a list - cannot use len())
• nvdsanalytics_obj_items: ITERATOR of AnalyticsObjInfo objects. (NOT a list - cannot use len())

Note: The attribute is object_id, NOT tracking_id. This is the unique ID assigned by the tracker to track objects across frames.

RectParams Object

Properties:

• left, top, width, height: Coordinates and dimensions
• border_width: Border width
• border_color: Border color (Color object)

TensorOutputUserMetadata Object

Methods:

• as_tensor_output(): Get tensor output object
  • get_layers(): Get output layers dictionary

Example:

for user_meta in frame_meta.tensor_items:
    tensor_output = user_meta.as_tensor_output()
    layers = tensor_output.get_layers()
    # layers is a dict: {"layer_name": tensor, ...}

SegmentationUserMetadata Object

Properties:

• unique_id: Unique id of the component that generates the segmentation output.
• classes: Number of classes in the segmentation output. |
• width, height: Width and height of the segmentation mask array.
• class_map: Class map array of the segmentation output; shape (height, width), dtype int. Each pixel holds the class index.
• class_probabilities_map: Class probabilities map array; shape (height, width, classes), dtype float. Optional; may be empty if not produced by the model.

Example:

from pyservicemaker import Pipeline, Flow, BatchMetadataOperator

class MyOperator(BatchMetadataOperator):
    def handle_metadata(self, batch_meta):
        for frame_meta in batch_meta.frame_items:
            # frame_meta is FrameMetadata
            for user_meta in frame_meta.segmentation_items:
                # user_meta is UserMetadata (segmentation type)
                seg_meta = user_meta.as_segmentation()
                if seg_meta:  # cast is valid when meta type matches
                    # Use SegmentationUserMetadata attributes
                    print("unique_id:", seg_meta.unique_id)
                    print("classes:", seg_meta.classes)
                    print("width:", seg_meta.width, "height:", seg_meta.height)
                    # class_map: (height, width) int array
                    print("class_map shape:", seg_meta.class_map.shape)
                    # class_probabilities_map: (height, width, classes) float array, if present
                    if seg_meta.class_probabilities_map.size > 0:
                        print("class_probabilities_map shape:", seg_meta.class_probabilities_map.shape)

AnalyticsFrameMeta object

Properties:

• oc_status: Map of overcrowding status per ROI (key = ROI label). Type: dict[str, bool]
• obj_in_roi_cnt: Map of count of valid objects in each ROI (key = ROI label). Type: dict[str, int]
• obj_lc_curr_cnt: Map of line-crossing count in the current frame per line (key = line/ROI label). Type: dict[str, int] | |
• obj_lc_cum_cnt: Map of cumulative line-crossing count per line (key = line/ROI label). Type: dict[str, int]
• unique_id: Unique identifier for the nvdsanalytics instance.
• obj_cnt: Map of object count per class ID (key = class ID). Type: dict[int, int]

Example:

from pyservicemaker import Pipeline, Flow, BatchMetadataOperator

class MyOperator(BatchMetadataOperator):
    def handle_metadata(self, batch_meta):
        for frame_meta in batch_meta.frame_items:
            # frame_meta is FrameMetadata
            for user_meta in frame_meta.nvdsanalytics_frame_items:
                # user_meta is UserMetadata (nvdsanalytics frame type)
                analytics_frame_meta = user_meta.as_nvdsanalytics_frame()
                if analytics_frame_meta:  # cast is valid when meta type matches
                    # Use AnalyticsFrameMeta attributes
                    print("Frame {0} component id: {1}".format(analytics_frame_meta.unique_id))
                    print("Frame {0} overcrowding status: {1}".format(frame_meta.frame_number, analytics_frame_meta.oc_status))
                    print("Frame {0} object in ROI count: {1}".format(frame_meta.frame_number, analytics_frame_meta.obj_in_roi_cnt))
                    print("Frame {0} object line crossing current count: {1}".format(frame_meta.frame_number, analytics_frame_meta.obj_lc_curr_cnt))
                    print("Frame {0} object line crossing cumulative count: {1}".format(frame_meta.frame_number, analytics_frame_meta.obj_lc_cum_cnt))
                    print("Frame {0} object count: {1}".format(frame_meta.frame_number,, analytics_frame_meta.obj_cnt))

AnalyticsObjInfo object

Properties:

• roi_status: Array of ROI labels in which this object is present. Type: list[str].
• oc_status: Array of OverCrowding labels in which this object is present. Type: list[str].
• lc_status: Array of line-crossing labels which this object has crossed. Type: list[str].
• dir_status: Direction string for the tracked object.
• unique_id: Unique identifier for the nvdsanalytics instance.
• obj_status: Status string for the tracked object.

Note: AnalyticsObjInfo is stored as user metadata on the object. ObjectMetadata exposes an iterator nvdsanalytics_obj_items over user metadata of type NVDS_USER_OBJ_META_NVDSANALYTICS; each element is a UserMetadata instance, which you cast to AnalyticsObjInfo using as_nvdsanalytics_obj().

Example:

from pyservicemaker import Pipeline, Flow, BatchMetadataOperator

class MyOperator(BatchMetadataOperator):
    def handle_metadata(self, batch_meta):
        for frame_meta in batch_meta.frame_items:
            for obj_meta in frame_meta.object_items:
                # obj_meta is ObjectMetadata
                for user_meta in obj_meta.nvdsanalytics_obj_items:
                    # user_meta is UserMetadata (nvdsanalytics object type)
                    analytics_obj = user_meta.as_nvdsanalytics_obj()
                    if analytics_obj:  # cast is valid when meta type matches
                        # Use AnalyticsObjInfo attributes
                        print("Object {0} ROI status: {1}".format(object_meta.object_id, analytics_obj.roi_status))
                        print("Object {0} overcrowding status: {1}".format(object_meta.object_id, analytics_obj.oc_status))
                        print("Object {0} line crossing status: {1}".format(obj_meta.object_id, analytics_obj.lc_status))
                        print("Object {0} moving in direction: {1}".format(obj_meta.object_id, analytics_obj.dir_status))
                        print("Object {0} unique ID: {1}".format(object_meta.object_id, analytics_obj.unique_id))
                        print("Object {0} status: {1}".format(object_meta.object_id, analytics_obj.obj_status))

ClassifierMetadata object

Properties:

• n_labels: Number of output labels of the classifier.
• unique_component_id: Unique id of the component that generates the classifier metadata.

Methods:

• get_n_label(n): Returns the nth label of the classifier (0-based index n).

Example:

from pyservicemaker import Pipeline, Flow, BatchMetadataOperator

class MyOperator(BatchMetadataOperator):
    def handle_metadata(self, batch_meta):
        for frame_meta in batch_meta.frame_items:
            for obj_meta in frame_meta.object_items:
                for classifier_meta in obj_meta.classifier_items:
                    # classifier_meta is ClassifierMetadata
                    print("n_labels:", classifier_meta.n_labels)
                    print("unique_component_id:", classifier_meta.unique_component_id)
                    for i in range(classifier_meta.n_labels):
                        label = classifier_meta.get_n_label(i)
                        print(f"  label[{i}]:", label)

OSD (On-Screen Display) API

osd Module

Provides classes for creating OSD elements.

Text

Text display element.

Properties:

• display_text: Text content (bytes)
• x_offset: X position (int)
• y_offset: Y position (int)
• font: Font object
• set_bg_color: Enable background color (bool)
• bg_color: Background color (Color object)

Font

Font specification.

Properties:

• name: Font family (FontFamily enum)
• size: Font size (int)
• color: Font color (Color object)

FontFamily Enum

• Serif
• Sans
• Mono

Color

Color specification (RGBA).

Properties:

• Red, Green, Blue, Alpha values (0.0 to 1.0)

Constructor:

color = osd.Color(1.0, 0.0, 0.0, 1.0)  # Red, fully opaque

DisplayMeta Object

Methods:

• add_text(text): Add text element
• add_rect(rect): Add rectangle element
• add_line(line): Add line element
• add_circle(circle): Add circle element

Example: Adding Text Overlay

from pyservicemaker import osd

display_meta = batch_meta.acquire_display_meta()
text = osd.Text()
text.display_text = b"Object Count: 5"
text.x_offset = 10
text.y_offset = 12
text.font.name = osd.FontFamily.Serif
text.font.size = 12
text.font.color = osd.Color(1.0, 1.0, 1.0, 1.0)
text.set_bg_color = True
text.bg_color = osd.Color(0.0, 0.0, 0.0, 1.0)
display_meta.add_text(text)
frame_meta.append(display_meta)

Postprocessing API

postprocessing Module

Provides classes for custom postprocessing.

ObjectDetectorOutputConverter

Base class for converting tensor outputs to object detections.

Methods:

__call__(output_layers)

Convert tensor outputs to list of bounding boxes.

Parameters:

• output_layers (dict): Dictionary of layer names to tensors

Returns: List of bounding boxes [class_id, confidence, x1, y1, x2, y2]

Example:

from pyservicemaker import postprocessing
import torch

class MyConverter(postprocessing.ObjectDetectorOutputConverter):
    def __call__(self, output_layers):
        outputs = []
        bbox_tensor = output_layers.get('bbox_layer')
        conf_tensor = output_layers.get('conf_layer')
        
        if bbox_tensor and conf_tensor:
            # Convert DLPack tensors to PyTorch
            bbox = torch.utils.dlpack.from_dlpack(bbox_tensor)
            conf = torch.utils.dlpack.from_dlpack(conf_tensor)
            
            # Process and convert to format: [class_id, confidence, x1, y1, x2, y2]
            # ... processing logic ...
            
        return outputs

Usage:

converter = MyConverter()
objects = converter(output_layers)
# objects is list of [class_id, confidence, x1, y1, x2, y2]

Probe API

Probe Class

Wrapper for attaching callback functions to pipeline elements.

Constructor (two overloads):

from pyservicemaker import Probe

# Overload 1: Metadata-level probe (most common)
probe = Probe("probe-name", BatchMetadataOperator())

# Overload 2: Buffer-level probe (for raw buffer access)
probe = Probe("probe-name", BufferOperator())

Parameters:

• name (str): Name of the probe
• operator: BatchMetadataOperator instance or BufferOperator instance

Built-in Probes:

• "measure_fps_probe": Measures FPS
• "measure_latency_probe": Measures latency
• "add_message_meta_probe": Automatically generates EventMessageUserMetadata (NvDsEventMsgMeta) from object metadata for downstream nvmsgconv consumption. Use this when msg2p-newapi=0 and you don't need custom control over sensor mappings.

Example:

# Custom probe
probe = Probe("my-probe", MyOperator())

# Built-in probe
pipeline.attach("infer", "measure_fps_probe", "fps-probe")

# Built-in message meta probe (for Kafka with msg2p-newapi=0)
pipeline.attach("osd", "add_message_meta_probe", "metadata generator")

BufferOperator Class

Low-level probe interface for accessing raw Buffer objects flowing through a pad. Use BufferOperator instead of BatchMetadataOperator when you need to inspect or count raw buffers that do NOT carry batch metadata — e.g., on the src pad of nvdsdynamicsrcbin (before any nvstreammux).

Methods to Override:

handle_buffer(buffer)

Called for every buffer that passes through the probed pad.

Parameters:

• buffer (Buffer): The buffer flowing through the pad

Returns: bool — True to pass the buffer downstream (keep), False to drop it.

Buffer Object Properties/Methods (available inside handle_buffer):

• buffer.timestamp (int): PTS timestamp of the buffer
• buffer.get_chunk_id(batch_id) (int): Chunk/source ID assigned by nvdsdynamicsrcbin. Always 0 for uridecodebin.
• buffer.extract(batch_id) → Tensor: Extract frame data as a tensor

Example:

from pyservicemaker import Pipeline, Probe, BufferOperator

class MyBufferProbe(BufferOperator):
    def __init__(self):
        super().__init__()
        self.count = 0

    def handle_buffer(self, buffer):
        self.count += 1
        print(f"Buffer #{self.count}  ts={buffer.timestamp}")
        return True

probe = MyBufferProbe()
pipeline.attach("dynamicsrcbin", Probe("buf-probe", probe), tips="src")

EventMessageUserMetadata

EventMessageUserMetadata wraps NvDsEventMsgMeta and is required by nvmsgconv when msg2p-newapi is 0 (the default / legacy API). Without it, nvmsgconv silently produces zero messages.

It is acquired from the BatchMetadata pool and must be populated and appended to the corresponding FrameMetadata.

Acquiring and Generating Event Message Metadata

event_msg = batch_meta.acquire_event_message_meta()  # Acquire from pool
event_msg.generate(object_meta, frame_meta, sensor_id, uri, labels)  # Populate
frame_meta.append(event_msg)  # Attach to frame

Parameters for generate():

• object_meta (ObjectMetadata): The detected object to create a message for
• frame_meta (FrameMetadata): The frame containing the object
• sensor_id (str): Camera/sensor identifier string (e.g., "Camera1")
• uri (str): Source URI of the stream (e.g., "file:///path/to/video.mp4")
• labels (list[str]): List of class label strings matching class IDs (e.g., ["person", "bag", "face"])

Two Approaches

Approach 1: Built-in Probe (Simple)

Use the built-in "add_message_meta_probe" -- no custom Python class needed:

# Attach AFTER inference/tracker, BEFORE nvmsgconv
pipeline.attach("osd", "add_message_meta_probe", "metadata generator")

Reference: deepstream_test4_app sample (/opt/nvidia/deepstream/deepstream/service-maker/sources/apps/python/pipeline_api/deepstream_test4_app/deepstream_test4.py)

Approach 2: Custom EventMessageGenerator (Full Control)

For multi-camera pipelines where you need control over sensor mappings:

from pyservicemaker import Pipeline, Probe, BatchMetadataOperator, SensorInfo

class EventMessageGenerator(BatchMetadataOperator):
    """Generate EventMessageUserMetadata for downstream nvmsgconv."""

    def __init__(self, sensor_map, labels):
        super().__init__()
        self._sensor_map = sensor_map  # dict: source_id -> SensorInfo or str
        self._labels = labels          # list of class label strings

    def handle_metadata(self, batch_meta, frame_interval=1):
        for frame_meta in batch_meta.frame_items:
            frame_num = frame_meta.frame_number
            for object_meta in frame_meta.object_items:
                if not (frame_num % frame_interval):
                    event_msg = batch_meta.acquire_event_message_meta()
                    if event_msg:
                        source_id = frame_meta.source_id
                        sensor_info = self._sensor_map.get(source_id)
                        sensor_id = sensor_info.sensor_id if sensor_info else "N/A"
                        uri = sensor_info.uri if sensor_info else "N/A"
                        event_msg.generate(
                            object_meta, frame_meta, sensor_id, uri, self._labels
                        )
                        frame_meta.append(event_msg)

# Attach probe upstream of nvmsgconv
labels = ["car", "bicycle", "person", "roadsign"]
sensor_map = {0: SensorInfo(sensor_id="Camera1", sensor_name="cam1", uri="file:///video1.mp4")}
pipeline.attach("tracker", Probe("event_msg_gen", EventMessageGenerator(sensor_map, labels)))

Reference: deepstream_test5_app sample (/opt/nvidia/deepstream/deepstream/service-maker/sources/apps/python/pipeline_api/deepstream_test5_app/deepstream_test5.py)

SensorInfo Class

Used to map source IDs to sensor metadata for EventMessageGenerator:

from pyservicemaker import SensorInfo

sensor_info = SensorInfo(
    sensor_id="Camera1",       # Unique sensor identifier string
    sensor_name="front_cam",   # Human-readable name
    uri="rtsp://host/stream1"  # Source URI
)

YAML Configuration Support

Pipelines can be created from YAML configuration files (for pipeline structure definition):

pipeline = Pipeline("pipeline-name", "/path/to/pipeline_config.yml")

Note: This YAML config is for pipeline structure (elements, links, probes). The nvinfer config-file-path can point to either a YAML file (.yml) or INI-style text file (.txt) - both formats are supported.

YAML Structure Example (Pipeline Definition)

pipeline:
  name: my-pipeline
  elements:
    - name: src
      type: filesrc
      properties:
        location: /path/to/video.h264
    
    - name: parser
      type: h264parse
    
    - name: decoder
      type: nvv4l2decoder
    
    - name: mux
      type: nvstreammux
      properties:
        batch-size: 1
        width: 1920
        height: 1080
    
    - name: infer
      type: nvinfer
      properties:
        # nvinfer supports both YAML (.yml) and INI-style (.txt) config formats
        config-file-path: /path/to/pgie_config.yml
    
    - name: osd
      type: nvosdbin
    
    - name: sink
      type: nveglglessink
  
  links:
    - [src, parser, decoder]
    - [decoder, mux]
    - [mux, infer, osd, sink]
  
  probes:
    - element: infer
      probe-name: my-probe
      probe-type: custom
      operator: MyOperator

nvinfer Configuration (Both Formats Supported)

The config-file-path for nvinfer supports both YAML and INI-style text formats:

YAML Format (.yml) - Recommended:

# pgie_config.yml - YAML format for nvinfer
property:
  gpu-id: 0
  net-scale-factor: 0.00392156862745098
  onnx-file: /opt/nvidia/deepstream/deepstream/samples/models/Primary_Detector/resnet18_trafficcamnet_pruned.onnx
  labelfile-path: /opt/nvidia/deepstream/deepstream/samples/models/Primary_Detector/labels.txt
  batch-size: 1
  process-mode: 1
  model-color-format: 0
  network-mode: 2
  num-detected-classes: 4
  cluster-mode: 2

class-attrs-all:
  topk: 20
  pre-cluster-threshold: 0.2

INI-style Format (.txt):

# pgie_config.txt - INI-style format for nvinfer
[property]
gpu-id=0
net-scale-factor=0.00392156862745098
onnx-file=/opt/nvidia/deepstream/deepstream/samples/models/Primary_Detector/resnet18_trafficcamnet_pruned.onnx
labelfile-path=/opt/nvidia/deepstream/deepstream/samples/models/Primary_Detector/labels.txt
batch-size=1
process-mode=1
model-color-format=0
network-mode=2
num-detected-classes=4
cluster-mode=2

[class-attrs-all]
topk=20
pre-cluster-threshold=0.2

Common Patterns and Examples

Pattern 1: Single Stream with Detection

from pyservicemaker import Pipeline, Probe, BatchMetadataOperator
import platform

def single_stream_detection(video_path, config_path):
    pipeline = (Pipeline("single-stream")
        .add("filesrc", "src", {"location": video_path})
        .add("h264parse", "parser")
        .add("nvv4l2decoder", "decoder")
        .add("nvstreammux", "mux", {"batch-size": 1, "width": 1920, "height": 1080})
        .add("nvinfer", "infer", {"config-file-path": config_path})
        .add("nvosdbin", "osd")
        .add("nv3dsink" if platform.processor() == "aarch64" else "nveglglessink", "sink")
        .link("src", "parser", "decoder")
        .link(("decoder", "mux"), ("", "sink_%u"))
        .link("mux", "infer", "osd", "sink")
        .start()
        .wait())

Pattern 2: Multi-Stream with Detection

Pattern 2a: Multi-Stream from Files

def multi_stream_detection(video_paths, config_path):
    pipeline = Pipeline("multi-stream")
    
    # Add sources
    for i, path in enumerate(video_paths):
        pipeline.add("filesrc", f"src{i}", {"location": path})
        pipeline.add("h264parse", f"parser{i}")
        pipeline.add("nvv4l2decoder", f"decoder{i}")
    
    # Add muxer
    pipeline.add("nvstreammux", "mux", {
        "batch-size": len(video_paths),
        "width": 1920,
        "height": 1080
    })
    
    # Add processing elements
    pipeline.add("nvinfer", "infer", {"config-file-path": config_path})
    pipeline.add("nvosdbin", "osd")
    pipeline.add("nveglglessink", "sink")
    
    # Link sources to muxer
    for i in range(len(video_paths)):
        pipeline.link(f"src{i}", f"parser{i}", f"decoder{i}")
        pipeline.link((f"decoder{i}", "mux"), ("", "sink_%u"))  # CRITICAL: Use "sink_%u", NOT f"sink_{i}"
    
    # Link processing chain
    pipeline.link("mux", "infer", "osd", "sink")
    pipeline.start().wait()

Pattern 2b: Multi-Stream RTSP with nvurisrcbin

def multi_rtsp_stream_detection(rtsp_urls, config_path):
    """
    Process multiple RTSP streams using nvurisrcbin.
    
    Args:
        rtsp_urls: List of RTSP stream URLs (e.g., ["rtsp://...", "rtsp://..."])
        config_path: Path to inference config file
    """
    pipeline = Pipeline("multi-rtsp-stream")
    
    # Add RTSP sources with nvurisrcbin (auto-detects codec and creates dynamic pads)
    for i, url in enumerate(rtsp_urls):
        pipeline.add("nvurisrcbin", f"src{i}", {"uri": url})
    
    # Add muxer for batching
    pipeline.add("nvstreammux", "mux", {
        "batch-size": len(rtsp_urls),
        "width": 1920,
        "height": 1080,
        "batched-push-timeout": 40000,
        "live-source": 1  # Important for RTSP streams
    })
    
    # Add processing elements
    pipeline.add("nvinfer", "infer", {"config-file-path": config_path, "batch-size": len(rtsp_urls)})
    pipeline.add("nvmultistreamtiler", "tiler", {"rows": 2, "columns": 2})
    pipeline.add("nvosdbin", "osd")
    pipeline.add("nveglglessink", "sink")
    
    # Link sources to muxer - CRITICAL: Use "sink_%u" pad template, NOT f"sink_{i}"
    for i in range(len(rtsp_urls)):
        # nvurisrcbin has dynamic src pad, so link directly to mux sink pad template
        pipeline.link((f"src{i}", "mux"), ("", "sink_%u"))  # CORRECT - pad template auto-assigns sink_0, sink_1, etc.
        # WRONG: pipeline.link((f"src{i}", "mux"), ("", f"sink_{i}"))  # This will FAIL!
    
    # Link processing chain
    pipeline.link("mux", "infer", "tiler", "osd", "sink")
    pipeline.start().wait()

Pattern 3: Custom Metadata Processing

class CustomProcessor(BatchMetadataOperator):
    def handle_metadata(self, batch_meta):
        for frame_meta in batch_meta.frame_items:
            # Count objects by class
            class_counts = {}
            for obj in frame_meta.object_items:
                class_id = obj.class_id
                class_counts[class_id] = class_counts.get(class_id, 0) + 1
            
            # Add text overlay
            display_meta = batch_meta.acquire_display_meta()
            text = osd.Text()
            text.display_text = f"Objects: {sum(class_counts.values())}".encode('ascii')
            text.x_offset = 10
            text.y_offset = 10
            text.font.name = osd.FontFamily.Serif
            text.font.size = 12
            text.font.color = osd.Color(1.0, 1.0, 1.0, 1.0)
            display_meta.add_text(text)
            frame_meta.append(display_meta)

# Attach probe
pipeline.attach("infer", Probe("processor", CustomProcessor()))

Pattern 4: Tensor-Based Custom Postprocessing

class TensorConverter(postprocessing.ObjectDetectorOutputConverter):
    def __call__(self, output_layers):
        outputs = []
        # Extract tensors
        bbox_layer = output_layers.get('bbox')
        conf_layer = output_layers.get('conf')
        
        if bbox_layer and conf_layer:
            import torch
            bbox = torch.utils.dlpack.from_dlpack(bbox_layer)
            conf = torch.utils.dlpack.from_dlpack(conf_layer)
            
            # Process tensors and convert to [class_id, conf, x1, y1, x2, y2]
            # ... processing logic ...
            
        return outputs

class TensorProcessor(BatchMetadataOperator):
    def __init__(self):
        super().__init__()
        self._converter = TensorConverter()
    
    def handle_metadata(self, batch_meta):
        for frame_meta in batch_meta.frame_items:
            for tensor_meta in frame_meta.tensor_items:
                output_layers = tensor_meta.as_tensor_output().get_layers()
                objects = self._converter(output_layers)
                
                # Create object metadata
                for obj in objects:
                    obj_meta = batch_meta.acquire_object_meta()
                    obj_meta.class_id = obj[0]
                    obj_meta.confidence = obj[1]
                    obj_meta.rect_params.left = obj[2]
                    obj_meta.rect_params.top = obj[3]
                    obj_meta.rect_params.width = obj[4] - obj[2]
                    obj_meta.rect_params.height = obj[5] - obj[3]
                    frame_meta.append(obj_meta)

# Enable tensor output in nvinfer
pipeline["infer"].set({"output-tensor-meta": 1})
pipeline.attach("infer", Probe("tensor-processor", TensorProcessor()))

Pattern 5: Cloud Integration (Kafka)

from kafka import KafkaProducer
import json

class KafkaSender(BatchMetadataOperator):
    def __init__(self, kafka_config):
        super().__init__()
        self.producer = KafkaProducer(
            bootstrap_servers=kafka_config['servers'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        self.topic = kafka_config['topic']
    
    def handle_metadata(self, batch_meta):
        for frame_meta in batch_meta.frame_items:
            objects = [
                {
                    "class_id": obj.class_id,
                    "confidence": obj.confidence,
                    "bbox": {
                        "left": obj.rect_params.left,
                        "top": obj.rect_params.top,
                        "width": obj.rect_params.width,
                        "height": obj.rect_params.height
                    },
                    "object_id": obj.object_id  # Tracking ID assigned by tracker
                }
                for obj in frame_meta.object_items
            ]
            
            message = {
                "frame_number": frame_meta.frame_number,
                "source_id": frame_meta.source_id,
                "buffer_pts": frame_meta.buffer_pts,  # PTS timestamp in nanoseconds
                "objects": objects
            }
            
            self.producer.send(topic=self.topic, value=message)
    
    def __del__(self):
        if hasattr(self, 'producer'):
            self.producer.flush()
            self.producer.close()

# Usage
kafka_config = {
    "servers": "localhost:9092",
    "topic": "analytics"
}
pipeline.attach("infer", Probe("kafka-sender", KafkaSender(kafka_config)))

Best Practices

1. Use Pipeline API for fine-grained control, Flow API for rapid prototyping
2. Always use hardware-accelerated decoders (nvv4l2decoder)
3. Configure appropriate batch sizes for your use case
4. Use probes for custom processing instead of modifying plugins
5. Handle KeyboardInterrupt properly (use multiprocessing.Process)
6. Flush and close Kafka producers in cleanup methods
7. Use tensor metadata for custom postprocessing when needed
8. Match tracker dimensions to inference input dimensions
9. Use YAML configs for complex pipelines to improve maintainability
10. Monitor GPU memory when processing multiple streams
11. Use correct Queue types for inter-process/thread communication:
    • queue.Queue → Use with threading.Thread (same process)
    • multiprocessing.Queue → Use with multiprocessing.Process (cross-process)
    • Using queue.Queue with multiprocessing.Process will silently lose data!

Error Handling

from multiprocessing import Process
import sys

def run_pipeline():
    try:
        pipeline.start().wait()
    except Exception as e:
        print(f"Pipeline error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    process = Process(target=run_pipeline)
    try:
        process.start()
        process.join()
    except KeyboardInterrupt:
        print("\nInterrupted. Terminating...")
        process.terminate()
        process.join()

Pipeline State and Message Handling API

Pipeline States

DeepStream pipelines follow GStreamer state transitions:

Pipeline Methods for State Management

prepare(message_handler)

Prepare the pipeline for activation with a message handler.

Parameters:

• message_handler (callable): Function to receive pipeline messages

Returns: Pipeline instance (for method chaining)

Example:

def on_message(message):
    if isinstance(message, StateTransitionMessage):
        print(f"State changed to: {message.new_state}")
    elif isinstance(message, DynamicSourceMessage):
        print(f"Source event: {message.source_id}")

pipeline.prepare(on_message)

activate()

Activate the pipeline (set to PLAYING state).

Returns: Pipeline instance (for method chaining)

deactivate()

Deactivate the pipeline (set to NULL state).

Returns: Pipeline instance (for method chaining)

wait()

Wait for the pipeline to complete (blocking).

Returns: None

Message Types

StateTransitionMessage

Indicates a pipeline state change.

Properties:

• origin (str): Element name that changed state
• old_state (PipelineState): Previous state
• new_state (PipelineState): New state

Example:

from pyservicemaker import StateTransitionMessage, PipelineState

def on_message(message):
    if isinstance(message, StateTransitionMessage):
        if message.new_state == PipelineState.PLAYING:
            print(f"Element {message.origin} is now playing")
        elif message.new_state == PipelineState.NULL:
            print(f"Element {message.origin} stopped")

DynamicSourceMessage

Indicates a dynamic source change (add/remove).

Properties:

• source_id (int): Unique source identifier
• source_added (bool): True if added, False if removed
• sensor_id (str): Sensor identifier
• sensor_name (str): Human-readable sensor name
• uri (str): Source URI (for added sources)

Example:

from pyservicemaker import DynamicSourceMessage

sensor_map = {}

def on_message(message):
    if isinstance(message, DynamicSourceMessage):
        if message.source_added:
            sensor_map[message.source_id] = {
                "sensor_id": message.sensor_id,
                "sensor_name": message.sensor_name,
                "uri": message.uri
            }
            print(f"Added source: {message.sensor_name}")
        else:
            if message.source_id in sensor_map:
                del sensor_map[message.source_id]
            print(f"Removed source: {message.source_id}")

Complete Message Handling Example

from pyservicemaker import (
    Pipeline, PipelineState, StateTransitionMessage,
    DynamicSourceMessage, SensorInfo, utils
)

def run_pipeline_with_messages(config_file):
    """Pipeline with comprehensive message handling"""
    pipeline = Pipeline("message-aware-pipeline", config_file=config_file)
    
    # Track sources
    active_sources = {}
    
    # Performance monitor
    perf_monitor = utils.PerfMonitor(
        batch_size=4,
        interval=5,
        source_type="nvmultiurisrcbin"
    )
    perf_monitor.apply(pipeline["tiler"], "sink")
    
    def handle_message(message):
        """Handle pipeline messages"""
        if isinstance(message, StateTransitionMessage):
            # Handle state transitions
            if message.new_state == PipelineState.PLAYING:
                if message.origin == "sink":
                    print("Pipeline fully started")
            elif message.new_state == PipelineState.NULL:
                print(f"Element {message.origin} stopped")
        
        elif isinstance(message, DynamicSourceMessage):
            # Handle dynamic source changes
            source_id = message.source_id
            
            if message.source_added:
                # Track new source
                active_sources[source_id] = SensorInfo(
                    sensor_id=message.sensor_id,
                    sensor_name=message.sensor_name,
                    uri=message.uri
                )
                
                # Add to performance monitor
                perf_monitor.add_stream(
                    source_id=source_id,
                    uri=message.uri,
                    sensor_id=message.sensor_id,
                    sensor_name=message.sensor_name
                )
                
                print(f"Source added: {message.sensor_name} ({message.uri})")
            else:
                # Remove source
                if source_id in active_sources:
                    del active_sources[source_id]
                perf_monitor.remove_stream(source_id)
                print(f"Source removed: {source_id}")
    
    # Prepare with message handler
    pipeline.prepare(handle_message)
    
    # Activate and wait
    pipeline.activate()
    pipeline.wait()

# Run
run_pipeline_with_messages("pipeline_config.yaml")

Signal Handling API

Signal Module

The signal module provides classes for custom signal handling.

Emitter Class

Base class for signal emitters.

Methods:

• attach(signal_name, element): Attach signal to element
• set(properties): Set properties on the emitter

Handler Class

Base class for signal handlers.

Smart Recording Signals

Smart recording uses signals for start/stop events.

Signal Names:

• "start-sr": Start smart recording
• "stop-sr": Stop smart recording
• "sr-done": Recording complete

Example:

from pyservicemaker import Pipeline, CommonFactory

pipeline = Pipeline("smart-recording")
# ... build pipeline ...

# Create smart recording controller
sr_controller = CommonFactory.create("smart_recording_action", "sr_controller")

if sr_controller:
    sr_controller.set({
        "proto-lib": "/opt/nvidia/deepstream/deepstream/lib/libnvds_kafka_proto.so",
        "conn-str": "localhost;9092",
        "topic-list": "sr-events"
    })
    
    # Attach signals to source element
    sr_controller.attach("start-sr", pipeline["src"])
    sr_controller.attach("stop-sr", pipeline["src"])
    
    # Attach signal handler for completion
    pipeline.attach("src", "smart_recording_signal", "sr", "sr-done")

Dynamic Source Management

nvmultiurisrcbin Properties

For dynamic source management, use nvmultiurisrcbin:

Adding/Removing Sources Dynamically

Sources are added/removed via REST API or programmatically through source management APIs.

from pyservicemaker import Pipeline, SourceConfig, SensorInfo

# Load initial sources from config
source_config = SourceConfig()
source_config.load("sources.yaml")

# Create pipeline
pipeline = Pipeline("dynamic-sources", config_file="pipeline.yaml")

# Initial sensors
for i, sensor in enumerate(source_config.sensor_list):
    print(f"Initial source {i}: {sensor.sensor_name}")

# Handle dynamic changes via message handler
def on_message(message):
    if isinstance(message, DynamicSourceMessage):
        if message.source_added:
            print(f"New source: {message.sensor_name}")
        else:
            print(f"Source removed: {message.source_id}")

pipeline.prepare(on_message)
pipeline.activate()
pipeline.wait()

SourceManager API (nvdsdynamicsrcbin)

SourceManager is a SignalEmitter that dynamically adds and removes sources on nvdsdynamicsrcbin at runtime. Unlike nvmultiurisrcbin (which uses REST API / config-based management), SourceManager gives direct programmatic control over individual file/URI sources through signal actions.

Import

from pyservicemaker._pydeepstream.signal import SourceManager

Class: SourceManager

Inherits from signal.Emitter → Object.

Constructor:

source_mgr = SourceManager("source_manager")

Parameters:

• name (str): Name of the SourceManager instance

Methods

attach(action_name, element)

Attach the SourceManager to a pipeline element for a given action. Must be called for each action before using it.

Supported actions:

• "add-source" — enables add_source()
• "remove-source" — enables remove_source()
• "terminate" — enables terminate()

Parameters:

• action_name (str): One of "add-source", "remove-source", "terminate"
• element: The pipeline element (Node) to attach to — must be an nvdsdynamicsrcbin

Example:

dsb_node = pipeline["dynamicsrcbin"]
source_mgr.attach("add-source", dsb_node)
source_mgr.attach("remove-source", dsb_node)
source_mgr.attach("terminate", dsb_node)

add_source(source_name)

Add a source (file path or URI) to the nvdsdynamicsrcbin.

Parameters:

• source_name (str): File path or URI of the source to add

Returns: int — a unique source ID (>= 0), or -1 if the add failed

Example:

sid = source_mgr.add_source("/path/to/video.h264")
if sid < 0:
    print("Failed to add source")

remove_source(source_id)

Remove a previously added source by its ID.

Parameters:

• source_id (int): The unique ID returned by add_source()

Example:

source_mgr.remove_source(sid)

terminate()

Signal that no more sources will be added. After all currently queued sources finish processing, an EOS (End of Stream) is sent downstream.

Example:

source_mgr.terminate()

This comprehensive API reference should help you build DeepStream applications using the Python Service Maker API effectively.