Skip to main content

Python Bindings

DeepStream Python bindings provide a Pythonic interface to build video analytics applications using the DeepStream SDK.

Overview​

The Python bindings (pyds) wrap the DeepStream C API, allowing you to:

  • Build GStreamer pipelines using Python
  • Access and manipulate metadata
  • Create custom probe functions
  • Integrate with Python ML/DL frameworks
  • Rapid application development

GitHub Repository: deepstream_python_apps

Installation​

See the Installation guide for detailed instructions.

Quick verification:

import sys
sys.path.append('/opt/nvidia/deepstream/deepstream/lib')
import pyds

print(f"PyDS version: {pyds.__version__}")

Core Concepts​

GStreamer Integration​

DeepStream Python apps use GStreamer's Python bindings (gi):

import gi
gi.require_version('Gst', '1.0')
from gi.repository import GLib, Gst
import pyds

# Initialize GStreamer
Gst.init(None)

# Create pipeline
pipeline = Gst.Pipeline.new("my-pipeline")

# Create elements
source = Gst.ElementFactory.make("filesrc", "source")
decoder = Gst.ElementFactory.make("nvv4l2decoder", "decoder")

# Add to pipeline
pipeline.add(source)
pipeline.add(decoder)

# Link elements
source.link(decoder)

# Start pipeline
pipeline.set_state(Gst.State.PLAYING)

Metadata Access​

Metadata is accessed through probe functions attached to GStreamer pads:

def pad_probe_callback(pad, info, user_data):
    """Probe function to access buffer metadata"""
    # Get buffer from probe info
    gst_buffer = info.get_buffer()
    if not gst_buffer:
        return Gst.PadProbeReturn.OK
    
    # Get batch metadata
    batch_meta = pyds.gst_buffer_get_nvds_batch_meta(hash(gst_buffer))
    
    # Process metadata
    # ...
    
    return Gst.PadProbeReturn.OK

# Attach probe to element pad
pad = element.get_static_pad("sink")
pad.add_probe(Gst.PadProbeType.BUFFER, pad_probe_callback, user_data)

Metadata Classes​

NvDsBatchMeta​

Root metadata object for a batch of frames:

batch_meta = pyds.gst_buffer_get_nvds_batch_meta(hash(gst_buffer))

# Properties
num_frames = batch_meta.num_frames_in_batch
frame_list = batch_meta.frame_meta_list

NvDsFrameMeta​

Metadata for a single frame:

frame_meta = pyds.NvDsFrameMeta.cast(frame_list.data)

# Properties
frame_meta.frame_num          # Frame number
frame_meta.source_id          # Source stream ID
frame_meta.num_obj_meta       # Number of detected objects
frame_meta.buf_pts            # Presentation timestamp
frame_meta.ntp_timestamp      # NTP timestamp
frame_meta.source_frame_width # Original frame width
frame_meta.source_frame_height # Original frame height

NvDsObjectMeta​

Metadata for a detected object:

obj_meta = pyds.NvDsObjectMeta.cast(obj_list.data)

# Identification
obj_meta.object_id        # Unique object ID
obj_meta.class_id         # Class ID
obj_meta.obj_label        # Class label string

# Detection info
obj_meta.confidence       # Detection confidence
obj_meta.tracker_confidence # Tracking confidence

# Bounding box (NvDsRect)
rect = obj_meta.rect_params
rect.left                 # X coordinate
rect.top                  # Y coordinate
rect.width                # Width
rect.height               # Height

# Tracking info
obj_meta.object_id        # Track ID (when tracker is used)

# Display properties
obj_meta.rect_params.border_width
obj_meta.rect_params.border_color  # RGBA tuple

# Text display
obj_meta.text_params.display_text
obj_meta.text_params.x_offset
obj_meta.text_params.y_offset
obj_meta.text_params.font_params.font_name
obj_meta.text_params.font_params.font_size
obj_meta.text_params.font_params.font_color  # RGBA tuple

NvDsClassifierMeta​

Secondary classifier metadata:

# Iterate through classifier metadata
l_classifier = obj_meta.classifier_meta_list
while l_classifier:
    classifier_meta = pyds.NvDsClassifierMeta.cast(l_classifier.data)
    
    # Properties
    classifier_meta.unique_component_id
    classifier_meta.num_labels
    
    # Iterate through labels
    l_label = classifier_meta.label_info_list
    while l_label:
        label_info = pyds.NvDsLabelInfo.cast(l_label.data)
        
        label_info.result_label    # Label string
        label_info.result_class_id # Class ID
        label_info.result_prob     # Probability/confidence
        
        l_label = l_label.next
    
    l_classifier = l_classifier.next

Complete Example: Object Counter​

A practical example counting objects by class:

object_counter.py
#!/usr/bin/env python3

import sys
sys.path.append('/opt/nvidia/deepstream/deepstream/lib')

import gi
gi.require_version('Gst', '1.0')
from gi.repository import GLib, Gst
import pyds
from collections import defaultdict

class ObjectCounter:
    def __init__(self):
        self.counts = defaultdict(int)
        self.total_frames = 0
    
    def update(self, frame_meta):
        """Update counts from frame metadata"""
        self.total_frames += 1
        
        # Reset counts for this frame
        frame_counts = defaultdict(int)
        
        # Count objects
        l_obj = frame_meta.obj_meta_list
        while l_obj:
            try:
                obj_meta = pyds.NvDsObjectMeta.cast(l_obj.data)
                class_name = obj_meta.obj_label
                frame_counts[class_name] += 1
                l_obj = l_obj.next
            except StopIteration:
                break
        
        # Update totals
        for class_name, count in frame_counts.items():
            self.counts[class_name] += count
    
    def get_report(self):
        """Generate report"""
        report = f"\n=== Object Detection Report ===\n"
        report += f"Total Frames: {self.total_frames}\n\n"
        
        if self.counts:
            report += "Object Counts:\n"
            for class_name, count in sorted(self.counts.items()):
                avg = count / self.total_frames
                report += f"  {class_name}: {count} (avg: {avg:.2f} per frame)\n"
        else:
            report += "No objects detected\n"
        
        return report

# Global counter
counter = ObjectCounter()

def osd_sink_pad_buffer_probe(pad, info, u_data):
    """Probe function to count objects"""
    gst_buffer = info.get_buffer()
    if not gst_buffer:
        return Gst.PadProbeReturn.OK
    
    batch_meta = pyds.gst_buffer_get_nvds_batch_meta(hash(gst_buffer))
    l_frame = batch_meta.frame_meta_list
    
    while l_frame:
        try:
            frame_meta = pyds.NvDsFrameMeta.cast(l_frame.data)
            counter.update(frame_meta)
            
            # Optional: Display counts on frame
            display_meta = pyds.nvds_acquire_display_meta_from_pool(batch_meta)
            display_meta.num_labels = 1
            
            py_nvosd_text_params = display_meta.text_params[0]
            py_nvosd_text_params.display_text = f"Frame {frame_meta.frame_num}: {frame_meta.num_obj_meta} objects"
            py_nvosd_text_params.x_offset = 10
            py_nvosd_text_params.y_offset = 10
            py_nvosd_text_params.font_params.font_name = "Serif"
            py_nvosd_text_params.font_params.font_size = 12
            py_nvosd_text_params.font_params.font_color.set(1.0, 1.0, 1.0, 1.0)
            py_nvosd_text_params.set_bg_clr = 1
            py_nvosd_text_params.text_bg_clr.set(0.0, 0.0, 0.0, 0.5)
            
            pyds.nvds_add_display_meta_to_frame(frame_meta, display_meta)
            
            l_frame = l_frame.next
        except StopIteration:
            break
    
    return Gst.PadProbeReturn.OK

def bus_call(bus, message, loop):
    """Handle bus messages"""
    t = message.type
    if t == Gst.MessageType.EOS:
        print("\nEnd of stream")
        print(counter.get_report())
        loop.quit()
    elif t == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print(f"Error: {err}: {debug}")
        loop.quit()
    return True

def main(args):
    if len(args) < 2:
        print("Usage: python3 object_counter.py <video_file>")
        return 1
    
    # Initialize
    Gst.init(None)
    
    # Create pipeline
    pipeline = Gst.Pipeline()
    
    # Create elements
    source = Gst.ElementFactory.make("filesrc", "source")
    h264parser = Gst.ElementFactory.make("h264parse", "h264-parser")
    decoder = Gst.ElementFactory.make("nvv4l2decoder", "decoder")
    streammux = Gst.ElementFactory.make("nvstreammux", "muxer")
    pgie = Gst.ElementFactory.make("nvinfer", "pgie")
    nvvidconv = Gst.ElementFactory.make("nvvideoconvert", "convertor")
    nvosd = Gst.ElementFactory.make("nvdsosd", "osd")
    sink = Gst.ElementFactory.make("nveglglessink", "sink")
    
    if not all([source, h264parser, decoder, streammux, pgie, nvvidconv, nvosd, sink]):
        print("Unable to create elements")
        return 1
    
    # Set properties
    source.set_property('location', args[1])
    streammux.set_property('width', 1920)
    streammux.set_property('height', 1080)
    streammux.set_property('batch-size', 1)
    streammux.set_property('batched-push-timeout', 4000000)
    pgie.set_property('config-file-path', 
                      '/opt/nvidia/deepstream/deepstream/samples/configs/deepstream-app/config_infer_primary.txt')
    sink.set_property('sync', 0)
    
    # Add to pipeline
    pipeline.add(source)
    pipeline.add(h264parser)
    pipeline.add(decoder)
    pipeline.add(streammux)
    pipeline.add(pgie)
    pipeline.add(nvvidconv)
    pipeline.add(nvosd)
    pipeline.add(sink)
    
    # Link elements
    source.link(h264parser)
    h264parser.link(decoder)
    
    sinkpad = streammux.get_request_pad("sink_0")
    srcpad = decoder.get_static_pad("src")
    srcpad.link(sinkpad)
    
    streammux.link(pgie)
    pgie.link(nvvidconv)
    nvvidconv.link(nvosd)
    nvosd.link(sink)
    
    # Add probe
    osdsinkpad = nvosd.get_static_pad("sink")
    osdsinkpad.add_probe(Gst.PadProbeType.BUFFER, osd_sink_pad_buffer_probe, 0)
    
    # Event loop
    loop = GLib.MainLoop()
    bus = pipeline.get_bus()
    bus.add_signal_watch()
    bus.connect("message", bus_call, loop)
    
    # Start
    print("Starting pipeline...")
    pipeline.set_state(Gst.State.PLAYING)
    
    try:
        loop.run()
    except KeyboardInterrupt:
        print("\nInterrupted")
        print(counter.get_report())
    
    # Cleanup
    pipeline.set_state(Gst.State.NULL)
    return 0

if __name__ == '__main__':
    sys.exit(main(sys.argv))

Run the counter:

python3 object_counter.py /opt/nvidia/deepstream/deepstream/samples/streams/sample_720p.h264

Advanced Metadata Operations​

Modifying Object Properties​

def modify_object_display(obj_meta, color=(1.0, 0.0, 0.0, 1.0)):
    """Change object bounding box color"""
    rect_params = obj_meta.rect_params
    rect_params.border_width = 3
    rect_params.border_color.set(*color)
    
    # Modify text
    text_params = obj_meta.text_params
    text_params.display_text = f"{obj_meta.obj_label} ({obj_meta.confidence:.2f})"
    text_params.font_params.font_size = 12
    text_params.font_params.font_color.set(1.0, 1.0, 1.0, 1.0)

Adding Display Metadata​

def add_text_overlay(frame_meta, text, x=10, y=10):
    """Add text overlay to frame"""
    batch_meta = frame_meta.base_meta.batch_meta
    display_meta = pyds.nvds_acquire_display_meta_from_pool(batch_meta)
    
    display_meta.num_labels = 1
    py_nvosd_text_params = display_meta.text_params[0]
    
    # Set text
    py_nvosd_text_params.display_text = text
    
    # Position
    py_nvosd_text_params.x_offset = x
    py_nvosd_text_params.y_offset = y
    
    # Font
    py_nvosd_text_params.font_params.font_name = "Serif"
    py_nvosd_text_params.font_params.font_size = 14
    py_nvosd_text_params.font_params.font_color.set(1.0, 1.0, 1.0, 1.0)
    
    # Background
    py_nvosd_text_params.set_bg_clr = 1
    py_nvosd_text_params.text_bg_clr.set(0.0, 0.0, 0.0, 0.7)
    
    pyds.nvds_add_display_meta_to_frame(frame_meta, display_meta)

Drawing Lines and Rectangles​

def add_roi_rectangle(frame_meta, x, y, width, height):
    """Draw ROI rectangle on frame"""
    batch_meta = frame_meta.base_meta.batch_meta
    display_meta = pyds.nvds_acquire_display_meta_from_pool(batch_meta)
    
    display_meta.num_rects = 1
    rect_params = display_meta.rect_params[0]
    
    # Rectangle properties
    rect_params.left = x
    rect_params.top = y
    rect_params.width = width
    rect_params.height = height
    rect_params.border_width = 2
    rect_params.border_color.set(0.0, 1.0, 0.0, 1.0)  # Green
    rect_params.has_bg_color = 0
    
    pyds.nvds_add_display_meta_to_frame(frame_meta, display_meta)

def add_line(frame_meta, x1, y1, x2, y2):
    """Draw line on frame"""
    batch_meta = frame_meta.base_meta.batch_meta
    display_meta = pyds.nvds_acquire_display_meta_from_pool(batch_meta)
    
    display_meta.num_lines = 1
    line_params = display_meta.line_params[0]
    
    line_params.x1 = x1
    line_params.y1 = y1
    line_params.x2 = x2
    line_params.y2 = y2
    line_params.line_width = 2
    line_params.line_color.set(1.0, 0.0, 0.0, 1.0)  # Red
    
    pyds.nvds_add_display_meta_to_frame(frame_meta, display_meta)

Integration with NumPy/OpenCV​

Extract frame data for processing with NumPy/OpenCV:

import numpy as np
import cv2

def get_numpy_frame(batch_meta, frame_meta):
    """Extract frame as NumPy array"""
    # Get surface from NvBufSurface
    n_frame = pyds.get_nvds_buf_surface(hash(gst_buffer), frame_meta.batch_id)
    
    # Convert to numpy array
    frame_copy = np.array(n_frame, copy=True, order='C')
    
    # Convert from RGBA to BGR (OpenCV format)
    frame_bgr = cv2.cvtColor(frame_copy, cv2.COLOR_RGBA2BGR)
    
    return frame_bgr

def process_with_opencv(frame_bgr, obj_meta):
    """Process frame with OpenCV"""
    # Extract ROI
    x = int(obj_meta.rect_params.left)
    y = int(obj_meta.rect_params.top)
    w = int(obj_meta.rect_params.width)
    h = int(obj_meta.rect_params.height)
    
    roi = frame_bgr[y:y+h, x:x+w]
    
    # Apply OpenCV operations
    gray = cv2.cvtColor(roi, cv2.COLOR_BGR2GRAY)
    edges = cv2.Canny(gray, 50, 150)
    
    return edges

Best Practices​

Memory Management

# Always acquire metadata from pool
display_meta = pyds.nvds_acquire_display_meta_from_pool(batch_meta)

# Don't create metadata objects directly
# BAD: display_meta = pyds.NvDsDisplayMeta()

Iterator Pattern

# Use try-except for safe iteration
l_frame = batch_meta.frame_meta_list
while l_frame:
    try:
        frame_meta = pyds.NvDsFrameMeta.cast(l_frame.data)
        # Process frame_meta
        l_frame = l_frame.next
    except StopIteration:
        break

Probe Return Values

# Always return appropriate probe status
return Gst.PadProbeReturn.OK      # Continue normally
return Gst.PadProbeReturn.DROP    # Drop buffer
return Gst.PadProbeReturn.REMOVE  # Remove probe

Next Steps​

Additional Resources​