539 lines
17 KiB
Python
Executable File
539 lines
17 KiB
Python
Executable File
|
|
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.ERROR)
|
|
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 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)
|
|
|
|
|
|
|
|
|
|
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):
|
|
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
|
|
|
|
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
|
|
|
|
|
|
|
|
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()
|
|
|