YACWC

deepstream_obj_det.py

@@ -16,14 +16,19 @@ import argparse
 pfm = LogColorize.watch_and_fix_permissions
 
 logger = get_logger(__name__,'/var/log/ml_vision_logs/00_watch_and_fix_permissions', stdout=True, systemd=False)
+Gst.debug_set_default_threshold(Gst.DebugLevel.INFO)
+os.environ.pop("DISPLAY",':0')
+target_width_detect = 1280
+target_height_detect = 720
 
 
-target_width = 1280
+target_width_embed = 512
-target_height = 720
+target_height_embed = 512
-os.environ.pop("DISPLAY",None)
+
 MUXER_BATCH_TIMEOUT_USEC = 1000000
-def embedder_results_probe(pad,info,u_data, list_add, frame_num = 0):
+def embed_results_probe(pad,info,u_data, list_add, frame_num = 0):
     gst_buffer = info.get_buffer()
+    print("HEY I AM PROBING EMBEDDINGS")
     if not gst_buffer:
         print("Unable to get GstBuffer ")
         return
@@ -104,7 +109,7 @@ def detector_results_probe(pad,info,u_data, list_add, frame_num = 0):
     frame_number=0
     num_rects=0
     got_fps = False
-
+    print("HEY I AM PROBING DETECTIONS")
     gst_buffer = info.get_buffer()
     if not gst_buffer:
         print("Unable to get GstBuffer ")
@@ -178,7 +183,6 @@ def detector_results_probe(pad,info,u_data, list_add, frame_num = 0):
 
 
 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()
@@ -215,34 +219,22 @@ def decodebin_child_added(child_proxy,Object,name,user_data):
 
 
 def create_source_bin(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-any-format"
-    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.
     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("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:
@@ -261,86 +253,116 @@ def run_inference(file_path):
 
     pipeline = Gst.Pipeline()
 
-    streammux = Gst.ElementFactory.make("nvstreammux", "Stream-muxer")
-
-    nugget_detector = Gst.ElementFactory.make("nvinfer", "primary-inference")
-    nugget_embedder = Gst.ElementFactory.make("nvinfer", "secondary-inference")
-        
-    streammux.set_property('width', target_width)
-    streammux.set_property('height', target_height)
-    streammux.set_property('batched-push-timeout', MUXER_BATCH_TIMEOUT_USEC)
-    streammux.set_property('enable-padding',1)
-    streammux.set_property('batch-size', 4)
-
-
-#    nugget_detector.set_property('config-file-path', "/home/thebears/DeepStream-Yolo/detector.txt")
-    nugget_detector.set_property('config-file-path', "/home/thebears/DeepStream-Yolo/config_infer_primary_yoloV7.txt")
-    nugget_embedder.set_property('config-file-path', "/home/thebears/DeepStream-Yolo/embedder.txt")
-
-    fakesink1 = Gst.ElementFactory.make("fakesink","fakesink")
-    fakesink1.set_property('enable-last-sample', 0)
-    fakesink1.set_property('sync', 0)
-    pipeline.add(fakesink1)
-
-    fakesink2 = Gst.ElementFactory.make("fakesink","fakesink2")
-    fakesink2.set_property('enable-last-sample', 0)
-    fakesink2.set_property('sync', 0)
-    pipeline.add(fakesink2)
-    
-    pipeline.add(streammux)
-    pipeline.add(nugget_detector)
-    pipeline.add(nugget_embedder)
-
-
-
-#    uri_name =  'file:///home/thebears/railing.mp4'
-#    uri_name = 'file:///home/thebears/railing_00_20250213094806.mp4'
     source_file=create_source_bin(file_path)
 
-    pipeline.add(source_file)
     
-    stream_pad = streammux.request_pad_simple("sink_0")
+
-    source_pad = source_file.get_static_pad("src")
-    source_pad.link(stream_pad)
     
     tee=Gst.ElementFactory.make("tee", "nvsink-tee")
+
+
+
+    
+# DETECT
+    queue_detect=Gst.ElementFactory.make("queue", "nvtee-detect")
+
+    
+    streammux_detect = Gst.ElementFactory.make("nvstreammux", "Stream-muxer-detector")
+    streammux_detect.set_property('width', target_width_detect)
+    streammux_detect.set_property('height', target_height_detect)
+    streammux_detect.set_property('batched-push-timeout', MUXER_BATCH_TIMEOUT_USEC)
+    streammux_detect.set_property('enable-padding',1)
+    streammux_detect.set_property('batch-size', 4)
+
+    nugget_detector = Gst.ElementFactory.make("nvinfer", "primary-inference")
+    nugget_detector.set_property('config-file-path', "/home/thebears/DeepStream-Yolo/config_infer_primary_yoloV7.txt")
+
+    fakesink_detect = Gst.ElementFactory.make("fakesink","fakesink")
+    fakesink_detect.set_property('enable-last-sample', 0)
+    fakesink_detect.set_property('sync', 0)
+
+    
+
+# EMBED
+
+    queue_embed=Gst.ElementFactory.make("queue", "nvtee-que-embed")
+
+    streammux_embed = Gst.ElementFactory.make("nvstreammux", "Stream-muxer-embed")
+    streammux_embed.set_property('width', target_width_embed)
+    streammux_embed.set_property('height', target_height_embed)
+    streammux_embed.set_property('batched-push-timeout', MUXER_BATCH_TIMEOUT_USEC)
+    streammux_embed.set_property('enable-padding',1)
+    streammux_embed.set_property('batch-size', 4)
+
+
+    nugget_embed = Gst.ElementFactory.make("nvinfer", "primary-inference")
+    nugget_embed.set_property('config-file-path', "/home/thebears/DeepStream-Yolo/embedder.txt")
+#    nugget_embed.set_property('config-file-path', "/home/thebears/DeepStream-Yolo/config_infer_primary_yoloV7.txt")
+
+    fakesink_embed = Gst.ElementFactory.make("fakesink","fakesink2")
+    fakesink_embed.set_property('enable-last-sample', 0)
+    fakesink_embed.set_property('sync', 0)
+
+
+
+    
+
+
+
+
+
+
+
+# LINKING
+
+    # Ensure NVMM caps with a capsfilter
+#    capsfilter = Gst.ElementFactory.make("capsfilter", "capsfilter")
+#    capsfilter.set_property("caps", Gst.Caps.from_string("video/x-raw(memory:NVMM), format=NV12"))
+#    pipeline.add(capsfilter)
+
+
+
+    pipeline.add(source_file)
     pipeline.add(tee)
 
-    queue1=Gst.ElementFactory.make("queue", "nvtee-que1")
+    nvvidconv = Gst.ElementFactory.make("nvvidconv", "nvvidconv")
-    queue2=Gst.ElementFactory.make("queue", "nvtee-que2")
+    pipeline.add(nvvidconv)
-    pipeline.add(queue1)
-    pipeline.add(queue2)
 
-    streammux.link(tee)
+    source_file.link(nvvidconv)
+  #  nvvidconv.link(capsfilter)
+ #   capsfilter.link(tee)
+    nvvidconv.link(tee)
 
 
-    tee.link(queue1)
+
-    tee.link(queue2)
+    if True:
+        pipeline.add(queue_detect)
+        pipeline.add(streammux_detect)
+        pipeline.add(nugget_detector)
+        pipeline.add(fakesink_detect)
 
 
-#    preprocess_detector = Gst.ElementFactory.make("nvdspreprocess","preprocess_detector")
+
-#    preprocess_detector.set_property('config-file', "/home/thebears/DeepStream-Yolo/detector_preprocess.txt")
+        tee.get_request_pad("src_%u").link(queue_detect.get_static_pad("sink"))
-#    preprocess_detector.set_property('config-file',pre_file)
+        queue_detect.get_static_pad("src").link(streammux_detect.get_request_pad("sink_0"))
+        streammux_detect.link(nugget_detector)
+        nugget_detector.link(fakesink_detect)
 
 
-#    preprocess_embedder = Gst.ElementFactory.make("nvdspreprocess","preprocess_embedder")
-#    preprocess_embedder.set_property('config-file', "/home/thebears/DeepStream-Yolo/embedder_preprocess.txt")
-#    preprocess_embedder.set_property('config-file',pre_file)
-                                                  
-#    pipeline.add(preprocess_detector)
-#    pipeline.add(preprocess_embedder)
-
-#    queue1.link(preprocess_detector)
-#    preprocess_detector.link(nugget_detector)
-    
-#    queue2.link(preprocess_embedder)
-#    preprocess_embedder.link(nugget_embedder)
 
 
-    queue1.link(nugget_detector)
+    if False:
-    queue2.link(nugget_embedder)
+        pipeline.add(queue_embed)
+        pipeline.add(streammux_embed)
+        pipeline.add(nugget_embed)
+        pipeline.add(fakesink_embed)
 
+        tee.get_request_pad("src_%u").link(queue_embed.get_static_pad("sink"))
+        queue_embed.get_static_pad("src").link(streammux_embed.get_request_pad("sink_0"))        
+        streammux_embed.link(nugget_embed)
+        nugget_embed.link(fakesink_embed)
 
+    print_pipeline_structure(pipeline)
     cmd = f'/usr/bin/ffprobe -v error -select_streams v:0 -count_packets -show_entries stream=nb_read_packets -of csv=p=0 {file_path}'#/srv/ftp/railing/2025/02/28/railing_00_20250228115800.mp4
     try:
         frames = int(os.popen(cmd).read().strip())
@@ -349,12 +371,15 @@ def run_inference(file_path):
 
     logger.info(f"TOTAL_FRAMES: {frames}")
     
-    embedder_list = list()
+    embed_list = list()
+    
+    Gst.debug_bin_to_dot_file(pipeline, Gst.DebugGraphDetails.ALL, "pipeline_structure")
+    embed_results = partial(embed_results_probe, list_add=embed_list, frame_num = frames)
+    nugget_embed.get_static_pad("src").add_probe(Gst.PadProbeType.BUFFER, embed_results, 0)
 
 
-    embedder_results = partial(embedder_results_probe, list_add=embedder_list, frame_num = frames)
-    nugget_embedder.get_static_pad("src").add_probe(Gst.PadProbeType.BUFFER, embedder_results, 0)
     
+    Gst.debug_bin_to_dot_file(pipeline, Gst.DebugGraphDetails.ALL, "/home/thebears/local/source/pipeline_structure")
     
     detector_list = list()
     
@@ -363,7 +388,7 @@ def run_inference(file_path):
     nugget_detector.get_static_pad("src").add_probe(Gst.PadProbeType.BUFFER, detector_results, 0)
     
 
-
+    print("AFTER SETTING STATIC PADS")
     def get_pipeline_string(pipeline):
         if not isinstance(pipeline, Gst.Pipeline):
             return None
@@ -378,8 +403,7 @@ def run_inference(file_path):
 
         return " ! ".join(elements)
 
-    nugget_detector.link(fakesink1)
+    Gst.debug_bin_to_dot_file(pipeline, Gst.DebugGraphDetails.ALL, "pipeline_structure")
-    nugget_embedder.link(fakesink2)
 
 
     # create an event loop and feed gstreamer bus mesages to it
@@ -397,7 +421,45 @@ def run_inference(file_path):
         pass
     # cleanup
     pipeline.set_state(Gst.State.NULL)
-    return  detector_list, embedder_list
+    return  detector_list, embed_list
+
+def print_pipeline_structure(pipeline):
+    """
+    Recursively prints elements in the pipeline and their properties.
+    """
+    if not isinstance(pipeline, Gst.Pipeline):
+        print("Not a valid GStreamer pipeline.")
+        return
+
+    def _print_element_properties(element, indent=0):
+        spaces = " " * indent
+        print(spaces + f"Element: {element.get_name()} (Type: {element.get_factory().get_name()})")
+
+        # Print its properties
+        for prop in element.list_properties():
+            try:
+                val = element.get_property(prop.name)
+                if val != prop.default_value:  # Display only non-default properties
+                    print(spaces + f"  - {prop.name}: {val}")
+            except:
+                pass
+
+    def _print_pipeline_structure(element, indent=0):
+        spaces = " " * indent
+        children = element.children if hasattr(element, 'children') else []
+        
+        if len(children) > 0:
+            print(spaces + f"[{element.get_name()}]")
+            for child in children:
+                _print_pipeline_structure(child, indent + 2)
+        else:
+            _print_element_properties(element, indent)
+
+    print("\nPipeline Structure:")
+    print("===================")
+    _print_pipeline_structure(pipeline)
+    print("===================\n")
+
 
 def get_detailed_pipeline_string(pipeline):
     """Generate a more detailed pipeline string with properties"""

deepstream_obj_det_pre_queue.py

@@ -0,0 +1,533 @@
+
+import sys
+sys.path.append('/opt/nvidia/deepstream/deepstream/sources/deepstream_python_apps/apps')
+import os
+import gi
+gi.require_version('Gst', '1.0')
+from gi.repository import GLib, Gst
+from common.platform_info import PlatformInfo
+from common.bus_call import bus_call
+import numpy as np
+import ctypes
+import pyds
+from functools import partial
+from CommonCode.settings import get_logger, LogColorize
+import argparse
+pfm = LogColorize.watch_and_fix_permissions
+
+logger = get_logger(__name__,'/var/log/ml_vision_logs/00_watch_and_fix_permissions', stdout=True, systemd=False)
+Gst.debug_set_default_threshold(Gst.DebugLevel.INFO)
+os.environ.pop("DISPLAY",':0')
+target_width_detect = 1280
+target_height_detect = 720
+
+
+target_width_embed = 512
+target_height_embed = 512
+
+MUXER_BATCH_TIMEOUT_USEC = 1000000
+def embed_results_probe(pad,info,u_data, list_add, frame_num = 0):
+    gst_buffer = info.get_buffer()
+    print("HEY I AM PROBING EMBEDDINGS")
+    if not gst_buffer:
+        print("Unable to get GstBuffer ")
+        return
+    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:
+        try:
+            # Note that l_frame.data needs a cast to pyds.NvDsFrameMeta
+            # 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_user = frame_meta.frame_user_meta_list
+        while l_user is not None:
+            try:
+                # Note that l_user.data needs a cast to pyds.NvDsUserMeta
+                # The casting also keeps ownership of the underlying memory
+                # in the C code, so the Python garbage collector will leave
+                # it alone.
+                user_meta = pyds.NvDsUserMeta.cast(l_user.data)
+            except StopIteration:
+                break
+
+            if (
+                    user_meta.base_meta.meta_type
+                    != pyds.NvDsMetaType.NVDSINFER_TENSOR_OUTPUT_META
+            ):
+                continue
+
+            tensor_meta = pyds.NvDsInferTensorMeta.cast(user_meta.user_meta_data)
+
+            # Boxes in the tensor meta should be in network resolution which is
+            # found in tensor_meta.network_info. Use this info to scale boxes to
+            # the input frame resolution.
+            layers_info = []
+            if True:
+                for i in range(tensor_meta.num_output_layers):
+                    layer = pyds.get_nvds_LayerInfo(tensor_meta, i)
+                    if layer.layerName=='embedding':
+                        
+                        ptr = ctypes.cast(pyds.get_ptr(layer.buffer), ctypes.POINTER(ctypes.c_float))
+                        num_elements = layer.inferDims.numElements
+                        v = list(np.ctypeslib.as_array(ptr, shape=(num_elements,)))
+                        v = [float(x) for x in v]
+                        
+                        list_add.append({'frame_number':frame_number, 'vector':v})
+
+
+                            
+
+        
+                try:
+                    l_user = l_user.next
+                except StopIteration:
+                    break
+
+        try:
+            # indicate inference is performed on the frame
+            frame_meta.bInferDone = True
+            l_frame = l_frame.next
+        except StopIteration:
+            break            
+    
+
+    return Gst.PadProbeReturn.OK
+
+
+
+
+def detector_results_probe(pad,info,u_data, list_add, frame_num = 0):
+    frame_number=0
+    num_rects=0
+    got_fps = False
+    print("HEY I AM PROBING DETECTIONS")
+    gst_buffer = info.get_buffer()
+    if not gst_buffer:
+        print("Unable to get GstBuffer ")
+        return
+
+
+    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:
+        try:
+            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
+
+        l_user = frame_meta.frame_user_meta_list
+
+
+        
+        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
+
+            # param_extract = ['left','top','width','height']
+            # strc = ''
+            # for param in param_extract:
+            #     strc+=str(getattr(obj_meta.rect_params, param))
+            #     strc+=' '
+
+ #           target_width
+#            target_height
+            score = obj_meta.confidence
+            label = obj_meta.obj_label
+            left = obj_meta.rect_params.left
+            top = obj_meta.rect_params.top
+            width = obj_meta.rect_params.width
+            height = obj_meta.rect_params.height
+            frame_number = frame_number
+            class_id = obj_meta.class_id
+
+            d_add = {'score':score, 'label':label, 'left':left, 'top':top, 'width':width, 'height':height, 'frame_number':frame_number, 'class_id': class_id}
+            list_add.append(d_add)
+            print(frame_number, label, score)
+            if frame_number % 100 == 0:
+                str_pr = 'FRAME_PROGRESS: '+pfm(str(frame_number) + '/' + str(frame_num))
+                logger.info(str_pr)
+        
+
+            try: 
+                l_obj=l_obj.next
+            except StopIteration:
+                break
+
+        # Update frame rate through this probe
+        stream_index = "stream{0}".format(frame_meta.pad_index)
+
+        try:
+            l_frame=l_frame.next
+        except StopIteration:
+            break
+
+    return Gst.PadProbeReturn.OK
+
+
+
+def cb_newpad(decodebin, decoder_src_pad,data):
+    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(uri):
+
+    bin_name="source-bin-any-format"
+
+    nbin=Gst.Bin.new(bin_name)
+    if not nbin:
+        sys.stderr.write(" Unable to create source bin \n")
+
+    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")
+
+    uri_decode_bin.set_property("uri",uri)
+
+    uri_decode_bin.connect("pad-added",cb_newpad, nbin)
+    uri_decode_bin.connect("child-added",decodebin_child_added,nbin)
+
+    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 run_inference(file_path):
+    os.environ.pop("DISPLAY",None)
+    if not file_path.startswith('file://'):
+        file_path = 'file://'+file_path
+
+    platform_info = PlatformInfo()
+    Gst.init(None)
+
+    pipeline = Gst.Pipeline()
+
+    source_file=create_source_bin(file_path)
+
+    
+
+    
+    tee=Gst.ElementFactory.make("tee", "nvsink-tee")
+
+
+
+    
+# DETECT
+    queue_detect=Gst.ElementFactory.make("queue", "nvtee-detect")
+
+    
+    streammux_detect = Gst.ElementFactory.make("nvstreammux", "Stream-muxer-detector")
+    streammux_detect.set_property('width', target_width_detect)
+    streammux_detect.set_property('height', target_height_detect)
+    streammux_detect.set_property('batched-push-timeout', MUXER_BATCH_TIMEOUT_USEC)
+    streammux_detect.set_property('enable-padding',1)
+    streammux_detect.set_property('batch-size', 4)
+
+    nugget_detector = Gst.ElementFactory.make("nvinfer", "primary-inference")
+    nugget_detector.set_property('config-file-path', "/home/thebears/DeepStream-Yolo/config_infer_primary_yoloV7.txt")
+
+    fakesink_detect = Gst.ElementFactory.make("fakesink","fakesink")
+    fakesink_detect.set_property('enable-last-sample', 0)
+    fakesink_detect.set_property('sync', 0)
+
+    
+
+# EMBED
+
+    queue_embed=Gst.ElementFactory.make("queue", "nvtee-que-embed")
+
+    streammux_embed = Gst.ElementFactory.make("nvstreammux", "Stream-muxer-embed")
+    streammux_embed.set_property('width', target_width_embed)
+    streammux_embed.set_property('height', target_height_embed)
+    streammux_embed.set_property('batched-push-timeout', MUXER_BATCH_TIMEOUT_USEC)
+    streammux_embed.set_property('enable-padding',1)
+    streammux_embed.set_property('batch-size', 4)
+
+
+    nugget_embed = Gst.ElementFactory.make("nvinfer", "primary-inference")
+    nugget_embed.set_property('config-file-path', "/home/thebears/DeepStream-Yolo/embedder.txt")
+#    nugget_embed.set_property('config-file-path', "/home/thebears/DeepStream-Yolo/config_infer_primary_yoloV7.txt")
+
+    fakesink_embed = Gst.ElementFactory.make("fakesink","fakesink2")
+    fakesink_embed.set_property('enable-last-sample', 0)
+    fakesink_embed.set_property('sync', 0)
+
+
+
+    
+
+
+
+
+
+
+
+# LINKING
+
+    # Ensure NVMM caps with a capsfilter
+#    capsfilter = Gst.ElementFactory.make("capsfilter", "capsfilter")
+#    capsfilter.set_property("caps", Gst.Caps.from_string("video/x-raw(memory:NVMM), format=NV12"))
+#    pipeline.add(capsfilter)
+
+
+
+    pipeline.add(source_file)
+    pipeline.add(tee)
+
+    nvvidconv = Gst.ElementFactory.make("nvvidconv", "nvvidconv")
+    pipeline.add(nvvidconv)
+
+    source_file.link(nvvidconv)
+  #  nvvidconv.link(capsfilter)
+ #   capsfilter.link(tee)
+    nvvidconv.link(tee)
+
+
+
+    if True:
+        pipeline.add(queue_detect)
+        pipeline.add(streammux_detect)
+        pipeline.add(nugget_detector)
+        pipeline.add(fakesink_detect)
+
+
+
+        tee.get_request_pad("src_%u").link(queue_detect.get_static_pad("sink"))
+        queue_detect.get_static_pad("src").link(streammux_detect.get_request_pad("sink_0"))
+        streammux_detect.link(nugget_detector)
+        nugget_detector.link(fakesink_detect)
+
+
+
+
+    if False:
+        pipeline.add(queue_embed)
+        pipeline.add(streammux_embed)
+        pipeline.add(nugget_embed)
+        pipeline.add(fakesink_embed)
+
+        tee.get_request_pad("src_%u").link(queue_embed.get_static_pad("sink"))
+        queue_embed.get_static_pad("src").link(streammux_embed.get_request_pad("sink_0"))        
+        streammux_embed.link(nugget_embed)
+        nugget_embed.link(fakesink_embed)
+
+    print_pipeline_structure(pipeline)
+    cmd = f'/usr/bin/ffprobe -v error -select_streams v:0 -count_packets -show_entries stream=nb_read_packets -of csv=p=0 {file_path}'#/srv/ftp/railing/2025/02/28/railing_00_20250228115800.mp4
+    try:
+        frames = int(os.popen(cmd).read().strip())
+    except:
+        frames = 0
+
+    logger.info(f"TOTAL_FRAMES: {frames}")
+    
+    embed_list = list()
+    
+    Gst.debug_bin_to_dot_file(pipeline, Gst.DebugGraphDetails.ALL, "pipeline_structure")
+    embed_results = partial(embed_results_probe, list_add=embed_list, frame_num = frames)
+    nugget_embed.get_static_pad("src").add_probe(Gst.PadProbeType.BUFFER, embed_results, 0)
+
+
+    
+    Gst.debug_bin_to_dot_file(pipeline, Gst.DebugGraphDetails.ALL, "/home/thebears/local/source/pipeline_structure")
+    
+    detector_list = list()
+    
+
+    detector_results = partial(detector_results_probe, list_add = detector_list, frame_num = frames)
+    nugget_detector.get_static_pad("src").add_probe(Gst.PadProbeType.BUFFER, detector_results, 0)
+    
+
+    print("AFTER SETTING STATIC PADS")
+    def get_pipeline_string(pipeline):
+        if not isinstance(pipeline, Gst.Pipeline):
+            return None
+
+        elements = []
+        iterator = pipeline.iterate_elements()
+        while True:
+            result, element = iterator.next()
+            if result != Gst.IteratorResult.OK:
+                break
+            elements.append(element.get_name())
+
+        return " ! ".join(elements)
+
+    Gst.debug_bin_to_dot_file(pipeline, Gst.DebugGraphDetails.ALL, "pipeline_structure")
+
+
+    # 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)
+    
+    # start play back and listen to events
+    print("Starting pipeline \n")
+    pipeline.set_state(Gst.State.PLAYING)
+    try:
+        loop.run()
+    except:
+        pass
+    # cleanup
+    pipeline.set_state(Gst.State.NULL)
+    return  detector_list, embed_list
+
+def print_pipeline_structure(pipeline):
+    """
+    Recursively prints elements in the pipeline and their properties.
+    """
+    if not isinstance(pipeline, Gst.Pipeline):
+        print("Not a valid GStreamer pipeline.")
+        return
+
+    def _print_element_properties(element, indent=0):
+        spaces = " " * indent
+        print(spaces + f"Element: {element.get_name()} (Type: {element.get_factory().get_name()})")
+
+        # Print its properties
+        for prop in element.list_properties():
+            try:
+                val = element.get_property(prop.name)
+                if val != prop.default_value:  # Display only non-default properties
+                    print(spaces + f"  - {prop.name}: {val}")
+            except:
+                pass
+
+    def _print_pipeline_structure(element, indent=0):
+        spaces = " " * indent
+        children = element.children if hasattr(element, 'children') else []
+        
+        if len(children) > 0:
+            print(spaces + f"[{element.get_name()}]")
+            for child in children:
+                _print_pipeline_structure(child, indent + 2)
+        else:
+            _print_element_properties(element, indent)
+
+    print("\nPipeline Structure:")
+    print("===================")
+    _print_pipeline_structure(pipeline)
+    print("===================\n")
+
+
+def get_detailed_pipeline_string(pipeline):
+    """Generate a more detailed pipeline string with properties"""
+    if not isinstance(pipeline, Gst.Pipeline):
+        return None
+    
+    def get_element_string(element):
+        # Get element factory name
+        factory = element.get_factory()
+        if factory:
+            element_str = factory.get_name()
+        else:
+            element_str = element.get_name()
+        
+        # Add properties
+        props = []
+        for prop in element.list_properties():
+            # Skip some properties that are typically not set in command line
+            if prop.name in ('name', 'parent'):
+                continue
+            
+            try:
+                val = element.get_property(prop.name)
+                if val is not None and val != prop.default_value:
+                    # Format value appropriately based on type
+                    if isinstance(val, str):
+                        props.append(f"{prop.name}=\"{val}\"")
+                    elif isinstance(val, bool):
+                        props.append(f"{prop.name}={str(val).lower()}")
+                    else:
+                        props.append(f"{prop.name}={val}")
+            except:
+                # Skip properties that can't be read
+                pass
+                
+        if props:
+            element_str += " " + " ".join(props)
+            
+        return element_str
+
+    result = []
+    
+    # Simple approach - just gets top-level elements
+    iterator = pipeline.iterate_elements()
+    while True:
+        ret, element = iterator.next()
+        if ret != Gst.IteratorResult.OK:
+            break
+        result.append(get_element_string(element))
+    
+    return " ! ".join(result)
+
+
+
+    
+if __name__ == '__main__':
+    cpath  = sys.argv[1]
+    if cpath.endswith('-i'):
+        cpath = '/home/thebears/local/source/short.mp4'
+        
+    if not cpath.startswith('file'):
+        cpath = os.path.abspath(cpath)
+
+
+    out = run_inference(cpath)
+    
+    import json
+    with open('dump.json','w') as ff:
+        json.dump([out[0],out[1]],ff)
+    sys.exit()
+

new_temp.py

@@ -0,0 +1,79 @@
+    
+    
+import gi
+gi.require_version('Gst', '1.0')
+#gi.require_version('GstNv', '1.0')
+from gi.repository import Gst, GObject
+
+# Initialize GStreamer
+Gst.init(None)
+
+def main():
+    # Create a pipeline
+    pipeline = Gst.Pipeline()
+
+    # Source element: File or Camera
+    source_bin = Gst.ElementFactory.make("filesrc", "file-source")
+
+    source_bin.set_property("location", '/opt/nvidia/deepstream/deepstream/samples/streams/sample_720p.mp4')    
+    
+    decodebin = Gst.ElementFactory.make("decodebin", "decode-bin")
+    source_bin.link(decodebin)
+
+    # Queue 1 for decodebin output
+    queue1 = Gst.ElementFactory.make("queue", "queue1")
+    pipeline.add(queue1)
+
+    # Splitter: Tee Element
+    tee = Gst.ElementFactory.make("tee", "splitter")
+    pipeline.add(tee)
+
+    # Resizer branch
+    queue_resizer = Gst.ElementFactory.make("queue", "resizer-queue")
+
+
+    resizer = Gst.ElementFactory.make("nvvideoconvert", "video-converter")
+    capsfilter = Gst.ElementFactory.make("capsfilter", "caps-filter")
+    capsfilter.set_property("caps", Gst.Caps.from_string("video/x-raw(memory:NVMM), width=300, height=300"))
+
+    # Model inference element
+    nvinfer = Gst.ElementFactory.make("nvinfer", "primary-inference")
+    nvinfer.set_property("config-file-path", "path_to_config_file.txt")
+
+    # Create a sink for visualizing output
+    fakesink_detect = Gst.ElementFactory.make("fakesink","fakesink")
+    fakesink_detect.set_property('enable-last-sample', 0)
+    fakesink_detect.set_property('sync', 0)
+    sink = fakesink_detect
+
+    # Add and link all elements (Resizer Branch)
+    for element in [queue_resizer, resizer, capsfilter, nvinfer, sink]:
+        pipeline.add(element)
+
+    decodebin.connect("pad-added", lambda decode, src_pad: src_pad.link(queue1.get_static_pad("sink")))
+    queue1.link(tee)
+
+    # Link the tee to the resizer pipeline
+    tee_src_pad1 = tee.get_request_pad("src_%u")
+    queue_resizer_sink_pad = queue_resizer.get_static_pad("sink")
+    tee_src_pad1.link(queue_resizer_sink_pad)
+
+    queue_resizer.link(resizer)
+    resizer.link(capsfilter)
+    capsfilter.link(nvinfer)
+    nvinfer.link(sink)
+
+    # Start the pipeline
+    pipeline.set_state(Gst.State.PLAYING)
+
+    # Run a main loop to listen for EOS or error
+    loop = GObject.MainLoop()
+    try:
+        loop.run()
+    except KeyboardInterrupt:
+        pass
+
+    pipeline.set_state(Gst.State.NULL)
+
+if __name__ == "__main__":
+    main()

new_test.py

@@ -0,0 +1,228 @@
+import sys
+import gi
+gi.require_version('Gst', '1.0')
+from gi.repository import GObject, Gst, GLib
+
+# Initialize GStreamer
+Gst.init(None)
+
+def create_source_bin(uri, source_id):
+    # Create a source bin element
+    bin_name = "source-bin-%02d" % source_id
+    source_bin = Gst.Bin.new(bin_name)
+    if not source_bin:
+        sys.stderr.write("Unable to create source bin\n")
+        return None
+
+    # Create elements
+    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")
+        return None
+
+    # Set properties
+    uri_decode_bin.set_property("uri", uri)
+
+    # Add the URI decode bin to the source bin
+    source_bin.add(uri_decode_bin)
+
+    # Add pad probe for data flow
+    pad = Gst.Ghost.Pad.new_no_target("src", Gst.PadDirection.SRC)
+    if not pad:
+        sys.stderr.write("Failed to add ghost pad in source bin\n")
+        return None
+    source_bin.add_pad(pad)
+
+    # Connect to the "pad-added" signal of the decoder
+    uri_decode_bin.connect("pad-added", cb_newpad, source_bin)
+    return source_bin
+
+def cb_newpad(decoder, pad, source_bin):
+    # Check the media type of the pad
+    pad_caps = pad.get_current_caps()
+    gst_struct = pad_caps.get_structure(0)
+    pad_type = gst_struct.get_name()
+    
+    if pad_type.startswith("video"):
+        # Link the decoder to the ghost pad
+        sink_pad = source_bin.get_static_pad("src")
+        if not sink_pad.is_linked():
+            pad.link(sink_pad)
+
+def main():
+    # Create the main pipeline
+    pipeline = Gst.Pipeline()
+    
+    # Create the source bin
+    source_bin = create_source_bin("file:///path/to/video.mp4", 0)
+    
+    # Create a tee element to split the stream
+    tee = Gst.ElementFactory.make("tee", "tee")
+    
+    # Create two streammux elements
+    streammux1 = Gst.ElementFactory.make("nvstreammux", "stream-muxer1")
+    streammux2 = Gst.ElementFactory.make("nvstreammux", "stream-muxer2")
+    
+    # Set streammux properties
+    streammux1.set_property("width", 1920)
+    streammux1.set_property("height", 1080)
+    streammux1.set_property("batch-size", 1)
+    streammux1.set_property("batched-push-timeout", 4000000)
+    
+    streammux2.set_property("width", 1280)
+    streammux2.set_property("height", 720)
+    streammux2.set_property("batch-size", 1)
+    streammux2.set_property("batched-push-timeout", 4000000)
+    
+    # Create queue elements for each branch
+    queue1 = Gst.ElementFactory.make("queue", "queue1")
+    queue2 = Gst.ElementFactory.make("queue", "queue2")
+    
+    # Add all elements to the pipeline
+    pipeline.add(source_bin)
+    pipeline.add(tee)
+    pipeline.add(queue1)
+    pipeline.add(queue2)
+    pipeline.add(streammux1)
+    pipeline.add(streammux2)
+    
+    # Link the source bin to the tee
+    source_bin.link(tee)
+    
+    # Link tee to the first queue and then to streammux1
+    tee_pad1 = tee.get_request_pad("src_%u")
+    queue1_pad = queue1.get_static_pad("sink")
+    tee_pad1.link(queue1_pad)
+    
+    # Link the queue1 to streammux1
+    sinkpad1 = streammux1.get_request_pad("sink_0")
+    srcpad1 = queue1.get_static_pad("src")
+    srcpad1.link(sinkpad1)
+    
+    # Link tee to the second queue and then to streammux2
+    tee_pad2 = tee.get_request_pad("src_%u")
+    queue2_pad = queue2.get_static_pad("sink")
+    tee_pad2.link(queue2_pad)
+    
+    # Link the queue2 to streammux2
+    sinkpad2 = streammux2.get_request_pad("sink_0")
+    srcpad2 = queue2.get_static_pad("src")
+    srcpad2.link(sinkpad2)
+    
+    # Continue building your pipeline with the two streammux outputs
+    # For example, you can add nvinfer, nvtracker, nvvideoconvert, etc.
+    
+    # Start the pipeline
+    pipeline.set_state(Gst.State.PLAYING)
+    
+    # Run the main loop
+    loop = GLib.MainLoop()
+    loop.run()
+    
+    # Clean up
+    pipeline.set_state(Gst.State.NULL)
+
+if __name__ == "__main__":
+    main()
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+import gi
+gi.require_version('Gst', '1.0')
+from gi.repository import Gst, GObject
+
+def main():
+    # Initialize GStreamer
+    Gst.init(None)
+
+    # Create the pipeline
+    pipeline = Gst.Pipeline()
+
+    # Create a source bin (e.g., decodebin from a file source or cameras)
+    source_bin = create_source_bin("file://path/to/your/video")
+    pipeline.add(source_bin)
+
+    # Create the tee element
+    tee = Gst.ElementFactory.make("tee", "tee")
+    pipeline.add(tee)
+
+    # Create two nvstreammux elements
+    streammux1 = Gst.ElementFactory.make("nvstreammux", "streammux1")
+    streammux1.set_property("batch-size", 1)
+    pipeline.add(streammux1)
+
+    streammux2 = Gst.ElementFactory.make("nvstreammux", "streammux2")
+    streammux2.set_property("batch-size", 1)
+    pipeline.add(streammux2)
+
+    # Request pads from the tee element
+    tee_src_pad_1 = tee.get_request_pad("src_%u")
+    tee_src_pad_2 = tee.get_request_pad("src_%u")
+
+    # Create two queues for each branch
+    queue1 = Gst.ElementFactory.make("queue", "queue1")
+    queue2 = Gst.ElementFactory.make("queue", "queue2")
+    pipeline.add(queue1)
+    pipeline.add(queue2)
+
+    # Link tee to the queues
+    tee_src_pad_1.link(queue1.get_static_pad("sink"))
+    tee_src_pad_2.link(queue2.get_static_pad("sink"))
+
+    # Link queue1 -> streammux1 and queue2 -> streammux2
+    queue1.link(streammux1)
+    queue2.link(streammux2)
+
+    # Add other elements (e.g., processing, display sinks) after each streammux as required.
+    # For simplicity, you can just add a fakesink at the end of each branch for now. 
+
+    fake_sink_1 = Gst.ElementFactory.make("fakesink", "fakesink1")
+    fake_sink_2 = Gst.ElementFactory.make("fakesink2")
+    pipeline.add(fake_sink_1)
+    pipeline.add(fake_sink_2)
+    streammux1.link(fake_sink_1)
+    streammux2.link(fake_sink_2)
+
+    # Connect the source bin to the tee
+    source_bin_src_pad = source_bin.get_static_pad("src")
+    tee_sink_pad = tee.get_static_pad("sink")
+    source_bin_src_pad.link(tee_sink_pad)
+
+    # Start the pipeline
+    pipeline.set_state(Gst.State.PLAYING)
+
+    # Run the pipeline
+    loop = GObject.MainLoop()
+    try:
+        loop.run()
+    except KeyboardInterrupt:
+        print("Exiting...")
+        pipeline.set_state(Gst.State.NULL)
+
+
+def create_source_bin(uri):
+    # Create the source bin (file or live source)
+    uridecodebin = Gst.ElementFactory.make("uridecodebin", "source-bin")
+    uridecodebin.set_property("uri", uri)
+
+    bin_pad = uridecodebin.get_static_pad("src")
+
+    return uridecodebin
+
+
+if __name__ == "__main__":
+    main()