thebears/deepstream
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

YACWC

Browse Source
This commit is contained in:
Ishan S. Patel
2025-06-17 14:08:34 -04:00
parent 81bb9a702a
commit 1d87063655
5 changed files with 1578 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files

368
deepstream_obj_det_pre_queue.py
View File

@@ -1,32 +1,43 @@
import sys
sys.path.append('/opt/nvidia/deepstream/deepstream/sources/deepstream_python_apps/apps')
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
gi.require_version("Gst", "1.0")
import argparse
import ctypes
from functools import partial
import numpy as np
import pyds
from common.bus_call import bus_call
from common.platform_info import PlatformInfo
from CommonCode.settings import LogColorize, get_logger
from gi.repository import GLib, Gst
pfm = LogColorize.watch_and_fix_permissions
logger = get_logger(__name__,'/var/log/ml_vision_logs/00_watch_and_fix_permissions', stdout=True, systemd=False)
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.ERROR)
os.environ.pop("DISPLAY",':0')
os.environ.pop("DISPLAY", ":0")
target_width_detect = 1280
target_height_detect = 720
os.environ["GST_DEBUG_DUMP_DOT_DIR"] = "/tmp"
os.putenv("GST_DEBUG_DUMP_DIR_DIR", "/tmp")
target_width_embed = 512
target_height_embed = 512
MUXER_BATCH_TIMEOUT_USEC = 1000000
def print_pipeline_structure(pipeline):
"""
Recursively prints elements in the pipeline and their properties.
@@ -37,7 +48,10 @@ def print_pipeline_structure(pipeline):
def _print_element_properties(element, indent=0):
spaces = " " * indent
print(spaces + f"Element: {element.get_name()} (Type: {element.get_factory().get_name()})")
print(
spaces
+ f"Element: {element.get_name()} (Type: {element.get_factory().get_name()})"
)
# Print its properties
for prop in element.list_properties():
@@ -50,8 +64,8 @@ def print_pipeline_structure(pipeline):
def _print_pipeline_structure(element, indent=0):
spaces = " " * indent
children = element.children if hasattr(element, 'children') else []
children = element.children if hasattr(element, "children") else []
if len(children) > 0:
print(spaces + f"[{element.get_name()}]")
for child in children:
@@ -69,7 +83,7 @@ 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()
@@ -77,20 +91,20 @@ def get_detailed_pipeline_string(pipeline):
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'):
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}\"")
props.append(f'{prop.name}="{val}"')
elif isinstance(val, bool):
props.append(f"{prop.name}={str(val).lower()}")
else:
@@ -98,14 +112,14 @@ def get_detailed_pipeline_string(pipeline):
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:
@@ -113,13 +127,11 @@ def get_detailed_pipeline_string(pipeline):
if ret != Gst.IteratorResult.OK:
break
result.append(get_element_string(element))
return " ! ".join(result)
def embed_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:
@@ -138,8 +150,7 @@ def embed_results_probe(pad,info,u_data, list_add, frame_num = 0):
except StopIteration:
break
frame_number=frame_meta.frame_num
frame_number = frame_meta.frame_num
l_user = frame_meta.frame_user_meta_list
while l_user is not None:
@@ -153,8 +164,8 @@ def embed_results_probe(pad,info,u_data, list_add, frame_num = 0):
break
if (
user_meta.base_meta.meta_type
!= pyds.NvDsMetaType.NVDSINFER_TENSOR_OUTPUT_META
user_meta.base_meta.meta_type
!= pyds.NvDsMetaType.NVDSINFER_TENSOR_OUTPUT_META
):
continue
@@ -167,19 +178,17 @@ def embed_results_probe(pad,info,u_data, list_add, frame_num = 0):
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))
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})
list_add.append({"frame_number": frame_number, "vector": v})
try:
l_user = l_user.next
except StopIteration:
@@ -190,17 +199,14 @@ def embed_results_probe(pad,info,u_data, list_add, frame_num = 0):
frame_meta.bInferDone = True
l_frame = l_frame.next
except StopIteration:
break
break
return Gst.PadProbeReturn.OK
def detector_results_probe(pad,info,u_data, list_add, frame_num = 0):
frame_number=0
num_rects=0
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()
@@ -208,28 +214,25 @@ def detector_results_probe(pad,info,u_data, list_add, frame_num = 0):
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
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:
try:
# Casting l_obj.data to pyds.NvDsObjectMeta
obj_meta=pyds.NvDsObjectMeta.cast(l_obj.data)
obj_meta = pyds.NvDsObjectMeta.cast(l_obj.data)
except StopIteration:
break
@@ -239,8 +242,8 @@ def detector_results_probe(pad,info,u_data, list_add, frame_num = 0):
# strc+=str(getattr(obj_meta.rect_params, param))
# strc+=' '
# target_width
# target_height
# target_width
# target_height
score = obj_meta.confidence
label = obj_meta.obj_label
left = obj_meta.rect_params.left
@@ -250,16 +253,26 @@ def detector_results_probe(pad,info,u_data, list_add, frame_num = 0):
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}
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))
str_pr = "FRAME_PROGRESS: " + pfm(
str(frame_number) + "/" + str(frame_num)
)
logger.info(str_pr)
try:
l_obj=l_obj.next
try:
l_obj = l_obj.next
except StopIteration:
break
@@ -267,155 +280,140 @@ def detector_results_probe(pad,info,u_data, list_add, frame_num = 0):
stream_index = "stream{0}".format(frame_meta.pad_index)
try:
l_frame=l_frame.next
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()
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)
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):
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)
print("features=", features)
if features.contains("memory:NVMM"):
# Get the source bin ghost pad
bin_ghost_pad=source_bin.get_static_pad("src")
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")
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):
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 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:
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"
bin_name = "source-bin-any-format"
nbin=Gst.Bin.new(bin_name)
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")
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.set_property("uri", uri)
uri_decode_bin.connect("pad-added",cb_newpad, nbin)
uri_decode_bin.connect("child-added",decodebin_child_added,nbin)
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))
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):
# def run_inference(file_path):
if True:
file_path = '/home/thebears/local/source/short.mp4'
os.environ.pop("DISPLAY",None)
if not file_path.startswith('file://'):
file_path = 'file://'+file_path
file_path = "/home/thebears/local/source/short.mp4"
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)
source_file = create_source_bin(file_path)
tee = Gst.ElementFactory.make("tee", "nvsink-tee")
tee=Gst.ElementFactory.make("tee", "nvsink-tee")
# DETECT
queue_detect = Gst.ElementFactory.make("queue", "nvtee-detect")
# 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)
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")
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)
fakesink_detect = Gst.ElementFactory.make("fakesink", "fakesink")
fakesink_detect.set_property("enable-last-sample", 0)
fakesink_detect.set_property("sync", 0)
# EMBED
# EMBED
queue_embed=Gst.ElementFactory.make("queue", "nvtee-que-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)
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")
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)
fakesink_embed = Gst.ElementFactory.make("fakesink", "fakesink2")
fakesink_embed.set_property("enable-last-sample", 0)
fakesink_embed.set_property("sync", 0)
# LINKING
# 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)
# 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)
@@ -424,66 +422,75 @@ if True:
pipeline.add(nvvidconv)
source_file.link(nvvidconv)
# nvvidconv.link(capsfilter)
# capsfilter.link(tee)
# 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"))
queue_detect.get_static_pad("src").link(
streammux_detect.get_request_pad("sink_0")
)
streammux_detect.link(nugget_detector)
nugget_detector.link(fakesink_detect)
os.environ["GST_DEBUG_DUMP_DOT_DIR"] = "/tmp"
os.putenv("GST_DEBUG_DUMP_DIR_DIR", "/tmp")
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"))
queue_embed.get_static_pad("src").link(
streammux_embed.get_request_pad("sink_0")
)
streammux_embed.link(nugget_embed)
nugget_embed.link(fakesink_embed)
Gst.debug_bin_to_dot_file(pipeline, Gst.DebugGraphDetails.ALL, "pipeline")
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
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)
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",
)
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)
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
@@ -500,13 +507,12 @@ if True:
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)
bus.connect("message", bus_call, loop)
# start play back and listen to events
print("Starting pipeline \n")
pipeline.set_state(Gst.State.PLAYING)
@@ -519,20 +525,18 @@ if True:
# return detector_list, embed_list
if __name__ == '__main__':
cpath = sys.argv[1]
if cpath.endswith('-i'):
cpath = '/home/thebears/local/source/short.mp4'
if not cpath.startswith('file'):
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()
import json
with open("dump.json", "w") as ff:
json.dump([out[0], out[1]], ff)
sys.exit()