NvdsMetaMux not merging metadata

Please provide complete information as applicable to your setup.

• Hardware Platform (Jetson / GPU) RTX 3070
• DeepStream Version 7.0.0
• Issue Type( bugs)

I have 4 different rtsp sources and this is my pipeline

I am using nvdsmetamux to merge meta data from pgie’s. There are 2 sink pads at metamux sink_0 and sink_1. sink_0 is for detection pgie and sink_1 for pose estimation pgie. if i set active_pad=sink_0 the meta data from src1 and src2 will go to fakesink ( i checked that with probe function) and meta data of src3 and src4 will be lost. if I set active_pad=sink_1 then data from src3 and src4 will go to fakesink and meta data of src1 and src2 will be lost.

Config file for nvdsmetamux

[property]
enable=1
# sink pad name which data will be pass to src pad.
active-pad=sink_0
# default pts-tolerance is 60 ms.
pts-tolerance=60000

[user-configs]

[group-0]
# src-ids-model-<model unique ID>=<source ids>
# mux all source if don't set it.
# src-ids-model-1=0;1
# src-ids-model-2=1;0
# src-ids-model-3=1;2

I think this is normal. The source id of each nvstreammux is independent, which means that the source ids of the two nvstreammux in the above pipeline are both 0/1, which cannot be distinguished when merging.

So if you want to merge metadata from multiple branches, you can only refer to the pipeline of the parallel sample app. First, use nvstreammux to form all streams into a batch, then nvstreamdemux, nvstreammux, and finally merge them using metamux.

Hi Junshengy
I have created the similar pipeline in my python application.

Gstreamer pipeline is similar to deepstream parallel inference app but its not working. pipeline does not send data down the pipleline however

when I remove 1 branch in middle from tee src_2 its started to send data down the pipeline.

If you remove the final nvstreamdemux and use fakesink directly, will it work properly?

Queues are recommended after nvstreamdemux source pads.

nvstreamdemux does not create additional threads inside. This is why a sizable queue between nvstreammux and nvstreamdemux is essential to de-couple them.

Can you share your goal? Your pipeline doesn’t seem to be so complicated. Or can you share a sample code that reproduces the problem?

Hi Junshengy,

So I updated the sample app test_app_3 and attached the code below so that you can reproduce the issue at your end.

So Issues I am facing with current pipline are these

if inveglglessink is in pipline data does not flow down the pipeline so when you replace it with fakesink data start to go down the pipeline but meta data for one of the source is empty as you will be able to see through probe function attached at src pad of metamux plugin.

meta-mux-test.zip (10.4 KB)

Pipeline

Hi Junshengy,
any update?

Sorry for the long delay because a holiday. I’m checking the code and will reply if I find anything.

Modify config_metamux.txt as follows

[property]
enable=1
# sink pad name which data will be pass to src pad.
active-pad=sink_0

# default pts-tolerance is 60 ms.
# pts-tolerance=60000

[user-configs]

[group-0]
# src-ids-model-<model unique ID>=<source ids>
# mux all source if don't set it.
src-ids-model-1=0
src-ids-model-2=1
# src-ids-model-1=0;1
# src-ids-model-2=1;0
# src-ids-model-3=1;2

Modify the pipeline as follows

#!/usr/bin/env python3

################################################################################
# SPDX-FileCopyrightText: Copyright (c) 2019-2023 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################

#Linkedin: https://www.linkedin.com/in/ahsaanraazaa

import sys

from graphviz import pipe_lines

sys.path.append('../')
from pathlib import Path
from os import environ
import gi
import configparser
import argparse
gi.require_version('Gst', '1.0')
from gi.repository import GLib, Gst
from ctypes import *
import time
import sys
import math
from common.bus_call import bus_call
from common.FPS import PERF_DATA

import pyds



no_display = False
silent = False
file_loop = False
perf_data = None
measure_latency = False

MAX_DISPLAY_LEN=64
PGIE_CLASS_ID_VEHICLE = 0
PGIE_CLASS_ID_BICYCLE = 1
PGIE_CLASS_ID_PERSON = 2
PGIE_CLASS_ID_ROADSIGN = 3
MUXER_OUTPUT_WIDTH=1920
MUXER_OUTPUT_HEIGHT=1080
MUXER_BATCH_TIMEOUT_USEC = 33000
TILED_OUTPUT_WIDTH=1280
TILED_OUTPUT_HEIGHT=720
GST_CAPS_FEATURES_NVMM="memory:NVMM"
OSD_PROCESS_MODE= 0
OSD_DISPLAY_TEXT= 1
pgie_classes_str= ["Vehicle", "TwoWheeler", "Person","RoadSign"]

# pgie_src_pad_buffer_probe  will extract metadata received on tiler sink pad
# and update params for drawing rectangle, object information etc.
def pgie_src_pad_buffer_probe(pad,info,u_data):
    frame_number=0
    num_rects=0
    got_fps = False
    gst_buffer = info.get_buffer()
    if not gst_buffer:
        print("Unable to get GstBuffer ")
        return
    # Retrieve batch metadata from the gst_buffer
    # Note that pyds.gst_buffer_get_nvds_batch_meta() expects the
    # C address of gst_buffer as input, which is obtained with hash(gst_buffer)

    # Enable latency measurement via probe if environment variable NVDS_ENABLE_LATENCY_MEASUREMENT=1 is set.
    # To enable component level latency measurement, please set environment variable
    # NVDS_ENABLE_COMPONENT_LATENCY_MEASUREMENT=1 in addition to the above.
    global measure_latency
    if measure_latency:
        num_sources_in_batch = pyds.nvds_measure_buffer_latency(hash(gst_buffer))
        if num_sources_in_batch == 0:
            print("Unable to get number of sources in GstBuffer for latency measurement")

    print("pgie_src_pad_buffer_probe")
    batch_meta = pyds.gst_buffer_get_nvds_batch_meta(hash(gst_buffer))
    l_frame = batch_meta.frame_meta_list
    while l_frame is not None:
        print("l_frame")
        try:
            # Note that l_frame.data needs a cast to pyds.NvDsFrameMeta
            # The casting is done by pyds.NvDsFrameMeta.cast()
            # The casting also keeps ownership of the underlying memory
            # in the C code, so the Python garbage collector will leave
            # it alone.
            frame_meta = pyds.NvDsFrameMeta.cast(l_frame.data)
        except StopIteration:
            break

        frame_number=frame_meta.frame_num
        l_obj=frame_meta.obj_meta_list
        num_rects = frame_meta.num_obj_meta
        obj_counter = {
        PGIE_CLASS_ID_VEHICLE:0,
        PGIE_CLASS_ID_PERSON:0,
        PGIE_CLASS_ID_BICYCLE:0,
        PGIE_CLASS_ID_ROADSIGN:0
        }
        while l_obj is not None:
            try: 
                # Casting l_obj.data to pyds.NvDsObjectMeta
                obj_meta=pyds.NvDsObjectMeta.cast(l_obj.data)
            except StopIteration:
                break
            obj_counter[obj_meta.class_id] += 1
            try: 
                l_obj=l_obj.next
            except StopIteration:
                break
        if not silent:
            print(frame_meta.source_id)
            print("Frame Number=", frame_number, "Number of Objects=",num_rects,"Vehicle_count=",obj_counter[PGIE_CLASS_ID_VEHICLE],"Person_count=",obj_counter[PGIE_CLASS_ID_PERSON])

        # Update frame rate through this probe
        stream_index = "stream{0}".format(frame_meta.pad_index)
        global perf_data
        perf_data.update_fps(stream_index)

        try:
            l_frame=l_frame.next
        except StopIteration:
            break

    return Gst.PadProbeReturn.OK



def cb_newpad(decodebin, decoder_src_pad,data):
    print("In cb_newpad\n")
    caps=decoder_src_pad.get_current_caps()
    if not caps:
        caps = decoder_src_pad.query_caps()
    gststruct=caps.get_structure(0)
    gstname=gststruct.get_name()
    source_bin=data
    features=caps.get_features(0)

    # Need to check if the pad created by the decodebin is for video and not
    # audio.
    print("gstname=",gstname)
    if(gstname.find("video")!=-1):
        # Link the decodebin pad only if decodebin has picked nvidia
        # decoder plugin nvdec_*. We do this by checking if the pad caps contain
        # NVMM memory features.
        print("features=",features)
        if features.contains("memory:NVMM"):
            # Get the source bin ghost pad
            bin_ghost_pad=source_bin.get_static_pad("src")
            if not bin_ghost_pad.set_target(decoder_src_pad):
                sys.stderr.write("Failed to link decoder src pad to source bin ghost pad\n")
        else:
            sys.stderr.write(" Error: Decodebin did not pick nvidia decoder plugin.\n")

def decodebin_child_added(child_proxy,Object,name,user_data):
    print("Decodebin child added:", name, "\n")
    if(name.find("decodebin") != -1):
        Object.connect("child-added",decodebin_child_added,user_data)

    if "source" in name:
        source_element = child_proxy.get_by_name("source")
        if source_element.find_property('drop-on-latency') != None:
            Object.set_property("drop-on-latency", True)



def create_source_bin(index,uri):
    print("Creating source bin")

    # Create a source GstBin to abstract this bin's content from the rest of the
    # pipeline
    bin_name="source-bin-%02d" %index
    print(bin_name)
    nbin=Gst.Bin.new(bin_name)
    if not nbin:
        sys.stderr.write(" Unable to create source bin \n")

    # Source element for reading from the uri.
    # We will use decodebin and let it figure out the container format of the
    # stream and the codec and plug the appropriate demux and decode plugins.
    if file_loop:
        # use nvurisrcbin to enable file-loop
        uri_decode_bin=Gst.ElementFactory.make("nvurisrcbin", "uri-decode-bin")
        uri_decode_bin.set_property("file-loop", 1)
        uri_decode_bin.set_property("cudadec-memtype", 0)
    else:
        uri_decode_bin=Gst.ElementFactory.make("uridecodebin", "uri-decode-bin")
    if not uri_decode_bin:
        sys.stderr.write(" Unable to create uri decode bin \n")
    # We set the input uri to the source element
    uri_decode_bin.set_property("uri",uri)
    # Connect to the "pad-added" signal of the decodebin which generates a
    # callback once a new pad for raw data has beed created by the decodebin
    uri_decode_bin.connect("pad-added",cb_newpad,nbin)
    uri_decode_bin.connect("child-added",decodebin_child_added,nbin)

    # We need to create a ghost pad for the source bin which will act as a proxy
    # for the video decoder src pad. The ghost pad will not have a target right
    # now. Once the decode bin creates the video decoder and generates the
    # cb_newpad callback, we will set the ghost pad target to the video decoder
    # src pad.
    Gst.Bin.add(nbin,uri_decode_bin)
    bin_pad=nbin.add_pad(Gst.GhostPad.new_no_target("src",Gst.PadDirection.SRC))
    if not bin_pad:
        sys.stderr.write(" Failed to add ghost pad in source bin \n")
        return None
    return nbin

def main(args, requested_pgie=None, config=None, disable_probe=False):
    global perf_data
    perf_data = PERF_DATA(len(args))

    number_sources=len(args)

    # Standard GStreamer initialization
    Gst.init(None)

    # Create gstreamer elements */
    # Create Pipeline element that will form a connection of other elements
    print("Creating Pipeline \n ")
    pipeline = Gst.Pipeline()
    is_live = False

    if not pipeline:
        sys.stderr.write(" Unable to create Pipeline \n")

    tee = Gst.ElementFactory.make("tee", "tee-0")
    demux = Gst.ElementFactory.make("nvstreamdemux", "demux")
    tee_b0 = Gst.ElementFactory.make("tee", "tee-b0")
    tee_b1 = Gst.ElementFactory.make("tee", "tee-b1")
    branch0_mux = Gst.ElementFactory.make("nvstreammux", "branch0_mux")
    branch1_mux = Gst.ElementFactory.make("nvstreammux", "branch1_mux")

    queue1=Gst.ElementFactory.make("queue","queue1")
    queue2=Gst.ElementFactory.make("queue","queue2")
    queue3=Gst.ElementFactory.make("queue","queue3")
    queue4=Gst.ElementFactory.make("queue","queue4")
    queue5=Gst.ElementFactory.make("queue","queue5")
    meta_mux = Gst.ElementFactory.make("nvdsmetamux", "meta-mux")
    meta_mux.set_property('config-file', 'config_metamux.txt')

    pipeline.add(meta_mux)
    pipeline.add(queue1)
    pipeline.add(queue2)
    pipeline.add(queue3)
    pipeline.add(queue4)
    pipeline.add(queue5)


    pipeline.add(tee)
    pipeline.add(demux)
    pipeline.add(tee_b0)
    pipeline.add(tee_b1)
    pipeline.add(branch0_mux)
    pipeline.add(branch1_mux)

    tee_pad_src0 = tee.request_pad_simple("src_0")
    meta_mux_sink_pad_0 = meta_mux.request_pad_simple("sink_0")
    tee_pad_src0.link(meta_mux_sink_pad_0)

    tee_pad_src1 = tee.request_pad_simple("src_1")
    demux_sink_pad = demux.get_static_pad("sink")
    tee_pad_src1.link(demux_sink_pad)

    demux_src0 = demux.request_pad_simple("src_0")
    tee_b0_sink_pad = tee_b0.get_static_pad("sink")
    demux_src0.link(tee_b0_sink_pad)
    b0_mux_sink0_pad = branch0_mux.request_pad_simple("sink_0")
    tee_b0_src0_pad = tee_b0.request_pad_simple("src_0")
    tee_b0_src0_pad.link(b0_mux_sink0_pad)

    demux_src1 = demux.request_pad_simple("src_1")
    tee_b1_sink_pad = tee_b1.get_static_pad("sink")
    demux_src1.link(tee_b1_sink_pad)
    b1_mux_sink1_pad = branch1_mux.request_pad_simple("sink_1")
    tee_b1_src0_pad = tee_b1.request_pad_simple("src_0")
    tee_b1_src0_pad.link(b1_mux_sink1_pad)
    print("Creating streamux \n ")

    # Create nvstreammux instance to form batches from one or more sources.
    streammux = Gst.ElementFactory.make("nvstreammux", "Stream-muxer")
    if not streammux:
        sys.stderr.write(" Unable to create NvStreamMux \n")

    pipeline.add(streammux)
    for i in range(number_sources):
        print("Creating source_bin ",i," \n ")
        uri_name=args[i]
        if uri_name.find("rtsp://") == 0 :
            is_live = True
        source_bin=create_source_bin(i, uri_name)
        if not source_bin:
            sys.stderr.write("Unable to create source bin \n")
        pipeline.add(source_bin)
        padname="sink_%u" %i
        sinkpad= streammux.request_pad_simple(padname) 
        if not sinkpad:
            sys.stderr.write("Unable to create sink pad bin \n")
        srcpad=source_bin.get_static_pad("src")
        if not srcpad:
            sys.stderr.write("Unable to create src pad bin \n")
        srcpad.link(sinkpad)

    nvdslogger = None

    print("Creating Pgie \n ")
    if requested_pgie != None and (requested_pgie == 'nvinferserver' or requested_pgie == 'nvinferserver-grpc') :
        pgie = Gst.ElementFactory.make("nvinferserver", "primary-inference")
    elif requested_pgie != None and requested_pgie == 'nvinfer':
        pgie = Gst.ElementFactory.make("nvinfer", "primary-inference")
    else:
        print("create ........")
        pgie = Gst.ElementFactory.make("nvinfer", "primary-inference")
        second_pgie = Gst.ElementFactory.make("nvinfer", "primary-inference-second")

    if not pgie:
        sys.stderr.write(" Unable to create pgie :  %s\n" % requested_pgie)

    pgie_src_pad = pgie.get_static_pad("src")
    second_pgie_src_pad = second_pgie.get_static_pad("src")
    if not pgie_src_pad:
        raise RuntimeError("Failed to get source pad from second_pgie")

    if disable_probe:
        # Use nvdslogger for perf measurement instead of probe function
        print ("Creating nvdslogger \n")
        nvdslogger = Gst.ElementFactory.make("nvdslogger", "nvdslogger")

    print("Creating tiler \n ")
    tiler=Gst.ElementFactory.make("nvmultistreamtiler", "nvtiler")
    if not tiler:
        sys.stderr.write(" Unable to create tiler \n")
    print("Creating nvvidconv \n ")
    nvvidconv = Gst.ElementFactory.make("nvvideoconvert", "convertor")
    if not nvvidconv:
        sys.stderr.write(" Unable to create nvvidconv \n")
    print("Creating nvosd \n ")
    nvosd = Gst.ElementFactory.make("nvdsosd", "onscreendisplay")
    if not nvosd:
        sys.stderr.write(" Unable to create nvosd \n")
    nvosd.set_property('process-mode',OSD_PROCESS_MODE)
    nvosd.set_property('display-text',OSD_DISPLAY_TEXT)


    if no_display:
        print("Creating Fakesink \n")
        sink = Gst.ElementFactory.make("fakesink", "fakesink")
        sink.set_property('enable-last-sample', 0)
        sink.set_property('sync', 0)
    else:
        print("Creating EGLSink \n")
        #sink = Gst.ElementFactory.make("nveglglessink", "nvvideo-renderer")
        sink = Gst.ElementFactory.make("nv3dsink", "nvvideo-renderer")
        if not sink:
            sys.stderr.write(" Unable to create egl sink \n")

    if is_live:
        print("At least one of the sources is live")
        streammux.set_property('live-source', 1)

    streammux.set_property('width', 1920)
    streammux.set_property('height', 1080)
    streammux.set_property('batch-size', number_sources)
    streammux.set_property('batched-push-timeout', MUXER_BATCH_TIMEOUT_USEC)

    branch0_mux.set_property('width', 1920)
    branch0_mux.set_property('height', 1080)
    branch0_mux.set_property('batch-size', 1)
    branch0_mux.set_property('batched-push-timeout', MUXER_BATCH_TIMEOUT_USEC)

    branch1_mux.set_property('width', 1920)
    branch1_mux.set_property('height', 1080)
    branch1_mux.set_property('batch-size', 1)
    branch1_mux.set_property('batched-push-timeout', MUXER_BATCH_TIMEOUT_USEC)

    if requested_pgie == "nvinferserver" and config != None:
        pgie.set_property('config-file-path', config)
    elif requested_pgie == "nvinferserver-grpc" and config != None:
        pgie.set_property('config-file-path', config)
    elif requested_pgie == "nvinfer" and config != None:
        pgie.set_property('config-file-path', config)
    else:
        pgie.set_property('config-file-path', "dstest3_pgie_config.txt")
        second_pgie.set_property('config-file-path', "dstest3_pgie_2_config.txt")

    pgie_batch_size=pgie.get_property("batch-size")
    if(pgie_batch_size != number_sources):
        print("WARNING: Overriding infer-config batch-size",pgie_batch_size," with number of sources ", number_sources," \n")
        pgie.set_property("batch-size",number_sources)
        second_pgie.set_property("batch-size",number_sources)

    tiler_rows=int(math.sqrt(number_sources))
    tiler_columns=int(math.ceil((1.0*number_sources)/tiler_rows))
    tiler.set_property("rows",tiler_rows)
    tiler.set_property("columns",tiler_columns)
    tiler.set_property("width", TILED_OUTPUT_WIDTH)
    tiler.set_property("height", TILED_OUTPUT_HEIGHT)
    sink.set_property("qos",0)

    print("Adding elements to Pipeline \n")
    pipeline.add(pgie)
    pipeline.add(second_pgie)
    if nvdslogger:
        pipeline.add(nvdslogger)
    pipeline.add(tiler)
    pipeline.add(nvvidconv)
    pipeline.add(nvosd)
    pipeline.add(sink)

    print("Linking elements in the Pipeline \n")

    streammux.link(tee)
    branch0_mux.link(pgie)
    branch1_mux.link(second_pgie)


    # pgie.link(queue2)
    # if nvdslogger:
    #     queue2.link(nvdslogger)
    #     nvdslogger.link(tiler)
    # else:
    #     queue2.link(tiler)
    meta_mux.link(tiler)
    tiler.link(queue3)
    queue3.link(nvvidconv)
    nvvidconv.link(queue4)
    queue4.link(nvosd)
    nvosd.link(queue5)
    queue5.link(sink)   

    # create an event loop and feed gstreamer bus mesages to it
    loop = GLib.MainLoop()
    bus = pipeline.get_bus()
    bus.add_signal_watch()
    bus.connect ("message", bus_call, loop)
    pgie_src_pad=pgie.get_static_pad("src")
    if not pgie_src_pad:
        sys.stderr.write(" Unable to get src pad \n")
    else:
        if not disable_probe:
            meta_mux_src_pad = meta_mux.get_static_pad("src")
            meta_mux_src_pad.add_probe(Gst.PadProbeType.BUFFER, pgie_src_pad_buffer_probe, 0)
            # perf callback function to print fps every 5 sec
            GLib.timeout_add(5000, perf_data.perf_print_callback)

    meta_mux_sink_pad_1 = meta_mux.request_pad_simple("sink_1")
    meta_mux_sink_pad_2 = meta_mux.request_pad_simple("sink_2")
    pgie_src_pad.link(meta_mux_sink_pad_1)
    second_pgie_src_pad.link(meta_mux_sink_pad_2)

    # Enable latency measurement via probe if environment variable NVDS_ENABLE_LATENCY_MEASUREMENT=1 is set.
    # To enable component level latency measurement, please set environment variable
    # NVDS_ENABLE_COMPONENT_LATENCY_MEASUREMENT=1 in addition to the above.
    if environ.get('NVDS_ENABLE_LATENCY_MEASUREMENT') == '1':
        print ("Pipeline Latency Measurement enabled!\nPlease set env var NVDS_ENABLE_COMPONENT_LATENCY_MEASUREMENT=1 for Component Latency Measurement")
        global measure_latency
        measure_latency = True

    # List the sources
    print("Now playing...")
    for i, source in enumerate(args):
        print(i, ": ", source)

    print("Starting pipeline \n")
    # start play back and listed to events		
    pipeline.set_state(Gst.State.PLAYING)

    try:
        loop.run()
    except:
        pass
    # cleanup
    print("Exiting app\n")
    pipeline.set_state(Gst.State.NULL)

def parse_args():

    parser = argparse.ArgumentParser(prog="deepstream_test_3",
                    description="deepstream-test3 multi stream, multi model inference reference app")
    parser.add_argument(
        "-i",
        "--input",
        help="Path to input streams",
        nargs="+",
        metavar="URIs",
        default=["a"],
        required=True,
    )
    parser.add_argument(
        "-c",
        "--configfile",
        metavar="config_location.txt",
        default=None,
        help="Choose the config-file to be used with specified pgie",
    )
    parser.add_argument(
        "-g",
        "--pgie",
        default=None,
        help="Choose Primary GPU Inference Engine",
        choices=["nvinfer", "nvinferserver", "nvinferserver-grpc"],
    )
    parser.add_argument(
        "--no-display",
        action="store_true",
        default=False,
        dest='no_display',
        help="Disable display of video output",
    )
    parser.add_argument(
        "--file-loop",
        action="store_true",
        default=False,
        dest='file_loop',
        help="Loop the input file sources after EOS",
    )
    parser.add_argument(
        "--disable-probe",
        action="store_true",
        default=False,
        dest='disable_probe',
        help="Disable the probe function and use nvdslogger for FPS",
    )
    parser.add_argument(
        "-s",
        "--silent",
        action="store_true",
        default=False,
        dest='silent',
        help="Disable verbose output",
    )
    # Check input arguments
    if len(sys.argv) == 1:
        parser.print_help(sys.stderr)
        sys.exit(1)
    args = parser.parse_args()

    stream_paths = args.input
    pgie = args.pgie
    config = args.configfile
    disable_probe = args.disable_probe
    global no_display
    global silent
    global file_loop
    no_display = args.no_display
    silent = args.silent
    file_loop = args.file_loop

    if config and not pgie or pgie and not config:
        sys.stderr.write ("\nEither pgie or configfile is missing. Please specify both! Exiting...\n\n\n\n")
        parser.print_help()
        sys.exit(1)
    if config:
        config_path = Path(config)
        if not config_path.is_file():
            sys.stderr.write ("Specified config-file: %s doesn't exist. Exiting...\n\n" % config)
            sys.exit(1)

    print(vars(args))
    return stream_paths, pgie, config, disable_probe

if __name__ == '__main__':
    stream_paths, pgie, config, disable_probe = parse_args()
    sys.exit(main(stream_paths, pgie, config, disable_probe))

Run the application

python3 deepstream_test_3.py -i file:///opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4 file:///opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4

In fact, the above program is equivalent to the following command line

gst-launch-1.0 uridecodebin uri=file:///opt/nvidia/deepstream/deepstream/samples/streams/sample_walk.mov ! mux.sink_0 \
               uridecodebin uri=file:///opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4 ! mux.sink_1 \
               nvstreammux name=mux gpu-id=0 batch-size=2 width=1920 height=1080 ! queue ! tee name=t \
               t.src_0 ! queue ! meta.sink_0 \
               t.src_1 ! queue ! nvstreamdemux name=demux per-stream-eos=true  \
               demux.src_0 ! queue ! tee name=b0_t \
               demux.src_1 ! queue ! tee name=b1_t \
               b0_t.src_0 ! queue ! b0_m.sink_0 nvstreammux name=b0_m batch-size=1 width=1920 height=1080 ! queue ! nvinfer batch-size=1 config-file-path=/opt/nvidia/deepstream/deepstream/sources/deepstream_python_apps/apps/meta-mux-test/dstest3_pgie_config.txt ! queue ! meta.sink_1 \
               b1_t.src_0 ! queue ! b1_m.sink_1 nvstreammux name=b1_m batch-size=1 width=1920 height=1080 ! queue ! nvinfer batch-size=1 config-file-path=/opt/nvidia/deepstream/deepstream/sources/deepstream_python_apps/apps/meta-mux-test/dstest3_pgie_2_config.txt ! queue ! meta.sink_2 \
               nvdsmetamux name=meta config-file=/opt/nvidia/deepstream/deepstream/sources/deepstream_python_apps/apps/meta-mux-test/config_metamux.txt ! queue ! nvmultistreamtiler width=1920 height=1080 ! queue ! nvvideoconvert ! nvdsosd ! nv3dsink

For DS-7.1, please refer to this repository, This is different from the previous

Hi Junshengy,

Above code is working for two streams but when i updated the code and added the 3rd stream only first stream (stream 0) works other two streams hangs.

issue

Modified config_metamux.txt:

[property]
enable=1
# sink pad name which data will be pass to src pad.
active-pad=sink_0

# default pts-tolerance is 60 ms.
# pts-tolerance=60000

[user-configs]

[group-0]
# src-ids-model-<model unique ID>=<source ids>
# mux all source if don't set it.
src-ids-model-1=0
src-ids-model-2=1;2
# src-ids-model-1=0;1
# src-ids-model-2=1;0
# src-ids-model-3=1;2

Modiefied Pipeline:

#!/usr/bin/env python3

################################################################################
# SPDX-FileCopyrightText: Copyright (c) 2019-2023 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################

#Linkedin: https://www.linkedin.com/in/ahsaanraazaa

import sys

from graphviz import pipe_lines

sys.path.append('../')
from pathlib import Path
from os import environ
import gi
import configparser
import argparse
gi.require_version('Gst', '1.0')
from gi.repository import GLib, Gst
from ctypes import *
import time
import sys
import math
import os
from common.bus_call import bus_call
from common.FPS import PERF_DATA

import pyds

os.environ["GST_DEBUG_DUMP_DOT_DIR"] = "/tmp"
os.putenv('GST_DEBUG_DUMP_DIR', '/tmp')

no_display = False
silent = False
file_loop = False
perf_data = None
measure_latency = False

MAX_DISPLAY_LEN=64
PGIE_CLASS_ID_VEHICLE = 0
PGIE_CLASS_ID_BICYCLE = 1
PGIE_CLASS_ID_PERSON = 2
PGIE_CLASS_ID_ROADSIGN = 3
MUXER_OUTPUT_WIDTH=1920
MUXER_OUTPUT_HEIGHT=1080
MUXER_BATCH_TIMEOUT_USEC = 33000
TILED_OUTPUT_WIDTH=1280
TILED_OUTPUT_HEIGHT=720
GST_CAPS_FEATURES_NVMM="memory:NVMM"
OSD_PROCESS_MODE= 0
OSD_DISPLAY_TEXT= 1
pgie_classes_str= ["Vehicle", "TwoWheeler", "Person","RoadSign"]



def bus_call(bus, message, loop):
    msg_type = message.type
    if msg_type == Gst.MessageType.EOS:
        print("End of stream")
        loop.quit()
    elif msg_type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print(f"Error: {err}, {debug}")
        loop.quit()
    return True

# pgie_src_pad_buffer_probe  will extract metadata received on tiler sink pad
# and update params for drawing rectangle, object information etc.
def pgie_src_pad_buffer_probe(pad,info,u_data):
    frame_number=0
    num_rects=0
    got_fps = False
    gst_buffer = info.get_buffer()
    if not gst_buffer:
        print("Unable to get GstBuffer ")
        return
    # Retrieve batch metadata from the gst_buffer
    # Note that pyds.gst_buffer_get_nvds_batch_meta() expects the
    # C address of gst_buffer as input, which is obtained with hash(gst_buffer)

    # Enable latency measurement via probe if environment variable NVDS_ENABLE_LATENCY_MEASUREMENT=1 is set.
    # To enable component level latency measurement, please set environment variable
    # NVDS_ENABLE_COMPONENT_LATENCY_MEASUREMENT=1 in addition to the above.
    global measure_latency
    if measure_latency:
        num_sources_in_batch = pyds.nvds_measure_buffer_latency(hash(gst_buffer))
        if num_sources_in_batch == 0:
            print("Unable to get number of sources in GstBuffer for latency measurement")

    print("pgie_src_pad_buffer_probe")
    batch_meta = pyds.gst_buffer_get_nvds_batch_meta(hash(gst_buffer))
    l_frame = batch_meta.frame_meta_list
    while l_frame is not None:
        print("l_frame")
        try:
            # Note that l_frame.data needs a cast to pyds.NvDsFrameMeta
            # The casting is done by pyds.NvDsFrameMeta.cast()
            # The casting also keeps ownership of the underlying memory
            # in the C code, so the Python garbage collector will leave
            # it alone.
            frame_meta = pyds.NvDsFrameMeta.cast(l_frame.data)
        except StopIteration:
            break

        frame_number=frame_meta.frame_num
        l_obj=frame_meta.obj_meta_list
        num_rects = frame_meta.num_obj_meta
        obj_counter = {
        PGIE_CLASS_ID_VEHICLE:0,
        PGIE_CLASS_ID_PERSON:0,
        PGIE_CLASS_ID_BICYCLE:0,
        PGIE_CLASS_ID_ROADSIGN:0
        }
        while l_obj is not None:
            try:
                # Casting l_obj.data to pyds.NvDsObjectMeta
                obj_meta=pyds.NvDsObjectMeta.cast(l_obj.data)
            except StopIteration:
                break
            obj_counter[obj_meta.class_id] += 1
            try:
                l_obj=l_obj.next
            except StopIteration:
                break
        if not silent:
            print(frame_meta.source_id)
            print("Frame Number=", frame_number, "Number of Objects=",num_rects,"Vehicle_count=",obj_counter[PGIE_CLASS_ID_VEHICLE],"Person_count=",obj_counter[PGIE_CLASS_ID_PERSON])

        # Update frame rate through this probe
        stream_index = "stream{0}".format(frame_meta.pad_index)
        global perf_data
        perf_data.update_fps(stream_index)

        try:
            l_frame=l_frame.next
        except StopIteration:
            break

    return Gst.PadProbeReturn.OK



def cb_newpad(decodebin, decoder_src_pad,data):
    print("In cb_newpad\n")
    caps=decoder_src_pad.get_current_caps()
    if not caps:
        caps = decoder_src_pad.query_caps()
    gststruct=caps.get_structure(0)
    gstname=gststruct.get_name()
    source_bin=data
    features=caps.get_features(0)

    # Need to check if the pad created by the decodebin is for video and not
    # audio.
    print("gstname=",gstname)
    if(gstname.find("video")!=-1):
        # Link the decodebin pad only if decodebin has picked nvidia
        # decoder plugin nvdec_*. We do this by checking if the pad caps contain
        # NVMM memory features.
        print("features=",features)
        if features.contains("memory:NVMM"):
            # Get the source bin ghost pad
            bin_ghost_pad=source_bin.get_static_pad("src")
            if not bin_ghost_pad.set_target(decoder_src_pad):
                sys.stderr.write("Failed to link decoder src pad to source bin ghost pad\n")
        else:
            sys.stderr.write(" Error: Decodebin did not pick nvidia decoder plugin.\n")

def decodebin_child_added(child_proxy,Object,name,user_data):
    print("Decodebin child added:", name, "\n")
    if(name.find("decodebin") != -1):
        Object.connect("child-added",decodebin_child_added,user_data)

    if "source" in name:
        source_element = child_proxy.get_by_name("source")
        if source_element.find_property('drop-on-latency') != None:
            Object.set_property("drop-on-latency", True)



def create_source_bin(index,uri):
    print("Creating source bin")

    # Create a source GstBin to abstract this bin's content from the rest of the
    # pipeline
    bin_name="source-bin-%02d" %index
    print(bin_name)
    nbin=Gst.Bin.new(bin_name)
    if not nbin:
        sys.stderr.write(" Unable to create source bin \n")

    # Source element for reading from the uri.
    # We will use decodebin and let it figure out the container format of the
    # stream and the codec and plug the appropriate demux and decode plugins.
    if file_loop:
        # use nvurisrcbin to enable file-loop
        uri_decode_bin=Gst.ElementFactory.make("nvurisrcbin", "uri-decode-bin")
        uri_decode_bin.set_property("file-loop", 1)
        uri_decode_bin.set_property("cudadec-memtype", 0)
    else:
        uri_decode_bin=Gst.ElementFactory.make("uridecodebin", "uri-decode-bin")
    if not uri_decode_bin:
        sys.stderr.write(" Unable to create uri decode bin \n")
    # We set the input uri to the source element
    uri_decode_bin.set_property("uri",uri)
    # Connect to the "pad-added" signal of the decodebin which generates a
    # callback once a new pad for raw data has beed created by the decodebin
    uri_decode_bin.connect("pad-added",cb_newpad,nbin)
    uri_decode_bin.connect("child-added",decodebin_child_added,nbin)

    # We need to create a ghost pad for the source bin which will act as a proxy
    # for the video decoder src pad. The ghost pad will not have a target right
    # now. Once the decode bin creates the video decoder and generates the
    # cb_newpad callback, we will set the ghost pad target to the video decoder
    # src pad.
    Gst.Bin.add(nbin,uri_decode_bin)
    bin_pad=nbin.add_pad(Gst.GhostPad.new_no_target("src",Gst.PadDirection.SRC))
    if not bin_pad:
        sys.stderr.write(" Failed to add ghost pad in source bin \n")
        return None
    return nbin

def main(args, requested_pgie=None, config=None, disable_probe=False):
    global perf_data
    perf_data = PERF_DATA(len(args))

    number_sources=len(args)

    # Standard GStreamer initialization
    Gst.init(None)

    # Create gstreamer elements */
    # Create Pipeline element that will form a connection of other elements
    print("Creating Pipeline \n ")
    pipeline = Gst.Pipeline()
    is_live = False

    if not pipeline:
        sys.stderr.write(" Unable to create Pipeline \n")

    tee = Gst.ElementFactory.make("tee", "tee-0")
    demux = Gst.ElementFactory.make("nvstreamdemux", "demux")
    tee_b0 = Gst.ElementFactory.make("tee", "tee-b0")
    tee_b1 = Gst.ElementFactory.make("tee", "tee-b1")
    tee_b2 = Gst.ElementFactory.make("tee", "tee-b2")
    branch0_mux = Gst.ElementFactory.make("nvstreammux", "branch0_mux")
    branch1_mux = Gst.ElementFactory.make("nvstreammux", "branch1_mux")

    queue1=Gst.ElementFactory.make("queue","queue1")
    queue2=Gst.ElementFactory.make("queue","queue2")
    queue3=Gst.ElementFactory.make("queue","queue3")
    queue4=Gst.ElementFactory.make("queue","queue4")
    queue5=Gst.ElementFactory.make("queue","queue5")
    queue6=Gst.ElementFactory.make("queue","queue6")
    queue7=Gst.ElementFactory.make("queue","queue7")
    queue8=Gst.ElementFactory.make("queue","queue8")
    queue9=Gst.ElementFactory.make("queue","queue9")
    queue10=Gst.ElementFactory.make("queue","queue10")

    meta_mux = Gst.ElementFactory.make("nvdsmetamux", "meta-mux")
    meta_mux.set_property('config-file', 'config_metamux.txt')

    pipeline.add(meta_mux)
    pipeline.add(queue1)
    pipeline.add(queue2)
    pipeline.add(queue3)
    pipeline.add(queue4)
    pipeline.add(queue5)
    pipeline.add(queue6)
    pipeline.add(queue7)
    pipeline.add(queue8)
    pipeline.add(queue9)
    pipeline.add(queue10)


    pipeline.add(tee)
    pipeline.add(demux)
    pipeline.add(tee_b0)
    pipeline.add(tee_b1)
    pipeline.add(tee_b2)
    pipeline.add(branch0_mux)
    pipeline.add(branch1_mux)

    tee_pad_src0 = tee.request_pad_simple("src_0")
    meta_mux_sink_pad_0 = meta_mux.request_pad_simple("sink_0")
    queue6_sink_pad = queue6.get_static_pad("sink")
    queue6_src_pad = queue6.get_static_pad("src")
    tee_pad_src0.link(queue6_sink_pad)
    queue6_src_pad.link(meta_mux_sink_pad_0)

    tee_pad_src1 = tee.request_pad_simple("src_1")
    demux_sink_pad = demux.get_static_pad("sink")
    queue7_sink_pad = queue7.get_static_pad("sink")
    queue7_src_pad = queue7.get_static_pad("src")
    tee_pad_src1.link(queue7_sink_pad)
    queue7_src_pad.link(demux_sink_pad)

    demux_src0 = demux.request_pad_simple("src_0")
    tee_b0_sink_pad = tee_b0.get_static_pad("sink")
    demux_src0.link(tee_b0_sink_pad)
    b0_mux_sink0_pad = branch0_mux.request_pad_simple("sink_0")
    tee_b0_src0_pad = tee_b0.request_pad_simple("src_0")
    tee_b0_src0_pad.link(b0_mux_sink0_pad)

    demux_src1 = demux.request_pad_simple("src_1")
    tee_b1_sink_pad = tee_b1.get_static_pad("sink")
    demux_src1.link(tee_b1_sink_pad)
    b1_mux_sink1_pad = branch1_mux.request_pad_simple("sink_1")
    tee_b1_src0_pad = tee_b1.request_pad_simple("src_0")
    tee_b1_src0_pad.link(b1_mux_sink1_pad)

    demux_src2 = demux.request_pad_simple("src_2")
    tee_b2_sink_pad = tee_b2.get_static_pad("sink")
    demux_src2.link(tee_b2_sink_pad)
    b1_mux_sink2_pad = branch1_mux.request_pad_simple("sink_2")
    tee_b2_src0_pad = tee_b2.request_pad_simple("src_0")
    tee_b2_src0_pad.link(b1_mux_sink2_pad)

    print("Creating streamux \n ")

    # Create nvstreammux instance to form batches from one or more sources.
    streammux = Gst.ElementFactory.make("nvstreammux", "Stream-muxer")
    if not streammux:
        sys.stderr.write(" Unable to create NvStreamMux \n")

    pipeline.add(streammux)
    for i in range(number_sources):
        print("Creating source_bin ",i," \n ")
        uri_name=args[i]
        if uri_name.find("rtsp://") == 0 :
            is_live = True
        source_bin=create_source_bin(i, uri_name)
        if not source_bin:
            sys.stderr.write("Unable to create source bin \n")
        pipeline.add(source_bin)
        padname="sink_%u" %i
        sinkpad= streammux.request_pad_simple(padname)
        if not sinkpad:
            sys.stderr.write("Unable to create sink pad bin \n")
        srcpad=source_bin.get_static_pad("src")
        if not srcpad:
            sys.stderr.write("Unable to create src pad bin \n")
        srcpad.link(sinkpad)

    nvdslogger = None

    print("Creating Pgie \n ")
    if requested_pgie != None and (requested_pgie == 'nvinferserver' or requested_pgie == 'nvinferserver-grpc') :
        pgie = Gst.ElementFactory.make("nvinferserver", "primary-inference")
    elif requested_pgie != None and requested_pgie == 'nvinfer':
        pgie = Gst.ElementFactory.make("nvinfer", "primary-inference")
    else:
        print("create ........")
        pgie = Gst.ElementFactory.make("nvinfer", "primary-inference")
        second_pgie = Gst.ElementFactory.make("nvinfer", "primary-inference-second")

    if not pgie:
        sys.stderr.write(" Unable to create pgie :  %s\n" % requested_pgie)

    pgie_src_pad = pgie.get_static_pad("src")
    second_pgie_src_pad = second_pgie.get_static_pad("src")
    if not pgie_src_pad:
        raise RuntimeError("Failed to get source pad from second_pgie")

    if disable_probe:
        # Use nvdslogger for perf measurement instead of probe function
        print ("Creating nvdslogger \n")
        nvdslogger = Gst.ElementFactory.make("nvdslogger", "nvdslogger")

    print("Creating tiler \n ")
    tiler=Gst.ElementFactory.make("nvmultistreamtiler", "nvtiler")
    if not tiler:
        sys.stderr.write(" Unable to create tiler \n")
    print("Creating nvvidconv \n ")
    nvvidconv = Gst.ElementFactory.make("nvvideoconvert", "convertor")
    if not nvvidconv:
        sys.stderr.write(" Unable to create nvvidconv \n")
    print("Creating nvosd \n ")
    nvosd = Gst.ElementFactory.make("nvdsosd", "onscreendisplay")
    if not nvosd:
        sys.stderr.write(" Unable to create nvosd \n")
    nvosd.set_property('process-mode',OSD_PROCESS_MODE)
    nvosd.set_property('display-text',OSD_DISPLAY_TEXT)


    if no_display:
        print("Creating Fakesink \n")
        sink = Gst.ElementFactory.make("fakesink", "fakesink")
        sink.set_property('enable-last-sample', 0)
        sink.set_property('sync', 0)
    else:
        print("Creating EGLSink \n")
        sink = Gst.ElementFactory.make("nveglglessink", "nvvideo-renderer")
        # sink = Gst.ElementFactory.make("nv3dsink", "nvvideo-renderer")
        if not sink:
            sys.stderr.write(" Unable to create egl sink \n")

    if is_live:
        print("At least one of the sources is live")
        streammux.set_property('live-source', 1)

    streammux.set_property('width', 1920)
    streammux.set_property('height', 1080)
    streammux.set_property('batch-size', number_sources)
    streammux.set_property('batched-push-timeout', MUXER_BATCH_TIMEOUT_USEC)

    branch0_mux.set_property('width', 1920)
    branch0_mux.set_property('height', 1080)
    branch0_mux.set_property('batch-size', 1)
    branch0_mux.set_property('batched-push-timeout', MUXER_BATCH_TIMEOUT_USEC)

    branch1_mux.set_property('width', 1920)
    branch1_mux.set_property('height', 1080)
    branch1_mux.set_property('batch-size', 1)
    branch1_mux.set_property('batched-push-timeout', MUXER_BATCH_TIMEOUT_USEC)

    if requested_pgie == "nvinferserver" and config != None:
        pgie.set_property('config-file-path', config)
    elif requested_pgie == "nvinferserver-grpc" and config != None:
        pgie.set_property('config-file-path', config)
    elif requested_pgie == "nvinfer" and config != None:
        pgie.set_property('config-file-path', config)
    else:
        pgie.set_property('config-file-path', "dstest3_pgie_config.txt")
        second_pgie.set_property('config-file-path', "dstest3_pgie_2_config.txt")

    pgie_batch_size=pgie.get_property("batch-size")
    if(pgie_batch_size != number_sources):
        print("WARNING: Overriding infer-config batch-size",pgie_batch_size," with number of sources ", number_sources," \n")
        # pgie.set_property("batch-size",number_sources)
        # second_pgie.set_property("batch-size",number_sources)
        pgie.set_property("batch-size",2)
        second_pgie.set_property("batch-size",1)

    tiler_rows=int(math.sqrt(number_sources))
    tiler_columns=int(math.ceil((1.0*number_sources)/tiler_rows))
    tiler.set_property("rows",tiler_rows)
    tiler.set_property("columns",tiler_columns)
    tiler.set_property("width", TILED_OUTPUT_WIDTH)
    tiler.set_property("height", TILED_OUTPUT_HEIGHT)
    sink.set_property("qos",0)

    print("Adding elements to Pipeline \n")
    pipeline.add(pgie)
    pipeline.add(second_pgie)
    if nvdslogger:
        pipeline.add(nvdslogger)
    pipeline.add(tiler)
    pipeline.add(nvvidconv)
    pipeline.add(nvosd)
    pipeline.add(sink)

    print("Linking elements in the Pipeline \n")

    streammux.link(tee)
    branch0_mux.link(queue1)
    queue1.link(pgie)
    branch1_mux.link(queue2)
    queue2.link(second_pgie)

    # pgie.link(queue2)
    # if nvdslogger:
    #     queue2.link(nvdslogger)
    #     nvdslogger.link(tiler)
    # else:
    #     queue2.link(tiler)
    meta_mux.link(tiler)
    tiler.link(queue3)
    queue3.link(nvvidconv)
    nvvidconv.link(queue4)
    queue4.link(nvosd)
    nvosd.link(queue5)
    queue5.link(sink)

    # create an event loop and feed gstreamer bus mesages to it
    loop = GLib.MainLoop()
    bus = pipeline.get_bus()
    bus.add_signal_watch()
    bus.connect ("message", bus_call, loop)
    pgie_src_pad=pgie.get_static_pad("src")
    if not pgie_src_pad:
        sys.stderr.write(" Unable to get src pad \n")
    else:
        if not disable_probe:
            meta_mux_src_pad = meta_mux.get_static_pad("src")
            meta_mux_src_pad.add_probe(Gst.PadProbeType.BUFFER, pgie_src_pad_buffer_probe, 0)
            # perf callback function to print fps every 5 sec
            GLib.timeout_add(5000, perf_data.perf_print_callback)

    meta_mux_sink_pad_1 = meta_mux.request_pad_simple("sink_1")
    meta_mux_sink_pad_2 = meta_mux.request_pad_simple("sink_2")
    pgie_src_pad.link(meta_mux_sink_pad_1)
    second_pgie_src_pad.link(meta_mux_sink_pad_2)

    # Enable latency measurement via probe if environment variable NVDS_ENABLE_LATENCY_MEASUREMENT=1 is set.
    # To enable component level latency measurement, please set environment variable
    # NVDS_ENABLE_COMPONENT_LATENCY_MEASUREMENT=1 in addition to the above.
    if environ.get('NVDS_ENABLE_LATENCY_MEASUREMENT') == '1':
        print ("Pipeline Latency Measurement enabled!\nPlease set env var NVDS_ENABLE_COMPONENT_LATENCY_MEASUREMENT=1 for Component Latency Measurement")
        global measure_latency
        measure_latency = True

    # pipeline.set_state(Gst.State.PAUSED)

    # List the sources
    print("Now playing...")

    for i, source in enumerate(args):
        print(i, ": ", source)

    print("Starting pipeline \n")
    # start play back and listed to events
    pipeline.set_state(Gst.State.PLAYING)
    Gst.debug_bin_to_dot_file(pipeline, Gst.DebugGraphDetails.ALL, "pipeline")
    try:
        loop.run()
    except:
        pass
    # cleanup
    print("Exiting app\n")
    pipeline.set_state(Gst.State.NULL)

def parse_args():

    parser = argparse.ArgumentParser(prog="deepstream_test_3",
                    description="deepstream-test3 multi stream, multi model inference reference app")
    parser.add_argument(
        "-i",
        "--input",
        help="Path to input streams",
        nargs="+",
        metavar="URIs",
        default=["a"],
        required=True,
    )
    parser.add_argument(
        "-c",
        "--configfile",
        metavar="config_location.txt",
        default=None,
        help="Choose the config-file to be used with specified pgie",
    )
    parser.add_argument(
        "-g",
        "--pgie",
        default=None,
        help="Choose Primary GPU Inference Engine",
        choices=["nvinfer", "nvinferserver", "nvinferserver-grpc"],
    )
    parser.add_argument(
        "--no-display",
        action="store_true",
        default=False,
        dest='no_display',
        help="Disable display of video output",
    )
    parser.add_argument(
        "--file-loop",
        action="store_true",
        default=False,
        dest='file_loop',
        help="Loop the input file sources after EOS",
    )
    parser.add_argument(
        "--disable-probe",
        action="store_true",
        default=False,
        dest='disable_probe',
        help="Disable the probe function and use nvdslogger for FPS",
    )
    parser.add_argument(
        "-s",
        "--silent",
        action="store_true",
        default=False,
        dest='silent',
        help="Disable verbose output",
    )
    # Check input arguments
    if len(sys.argv) == 1:
        parser.print_help(sys.stderr)
        sys.exit(1)
    args = parser.parse_args()

    stream_paths = args.input
    pgie = args.pgie
    config = args.configfile
    disable_probe = args.disable_probe
    global no_display
    global silent
    global file_loop
    no_display = args.no_display
    silent = args.silent
    file_loop = args.file_loop

    if config and not pgie or pgie and not config:
        sys.stderr.write ("\nEither pgie or configfile is missing. Please specify both! Exiting...\n\n\n\n")
        parser.print_help()
        sys.exit(1)
    if config:
        config_path = Path(config)
        if not config_path.is_file():
            sys.stderr.write ("Specified config-file: %s doesn't exist. Exiting...\n\n" % config)
            sys.exit(1)

    print(vars(args))
    return stream_paths, pgie, config, disable_probe

if __name__ == '__main__':
    stream_paths, pgie, config, disable_probe = parse_args()
    sys.exit(main(stream_paths, pgie, config, disable_probe))

Pipeline:

Please refer to the command line below to build the pipeline, Ensure that the pipeline of the Python code is consistent with it.

gst-launch-1.0 uridecodebin uri=file:///opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4 ! mux.sink_0 \
               uridecodebin uri=file:///opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4 ! mux.sink_1 \
               uridecodebin uri=file:///opt/nvidia/deepstream/deepstream/samples/streams/sample_1080p_h264.mp4 ! mux.sink_2 \
               nvstreammux name=mux gpu-id=0 batch-size=3 width=1920 height=1080 ! queue ! tee name=t \
               t.src_0 ! queue ! meta.sink_0 \
               t.src_1 ! queue ! nvstreamdemux name=demux per-stream-eos=true  \
               demux.src_0 ! queue ! tee name=b0_t \
               demux.src_1 ! queue ! tee name=b1_t \
               demux.src_2 ! queue ! tee name=b2_t \
               b0_t.src_0 ! queue ! b0_m.sink_0 nvstreammux name=b0_m batch-size=1 width=1920 height=1080 ! queue ! nvinfer batch-size=1 config-file-path=/opt/nvidia/deepstream/deepstream/sources/deepstream_python_apps/apps/meta-mux-test/dstest3_pgie_config.txt ! queue ! meta.sink_1 \
               b1_t.src_0 ! queue ! b1_m.sink_1 b2_t.src_0 ! queue ! b1_m.sink_2 nvstreammux name=b1_m batch-size=2 width=1920 height=1080 ! queue ! nvinfer batch-size=1 config-file-path=/opt/nvidia/deepstream/deepstream/sources/deepstream_python_apps/apps/meta-mux-test/dstest3_pgie_2_config.txt ! queue ! meta.sink_2 \
               nvdsmetamux name=meta config-file=/opt/nvidia/deepstream/deepstream/sources/deepstream_python_apps/apps/meta-mux-test/config_metamux.txt ! queue ! nvmultistreamtiler width=1920 height=1080 ! queue ! nvvideoconvert ! nvdsosd ! nveglglessink

Try to modify the batch-size property of streammux in branch1