Deepstream搭配工业相机进行目标检测实战
·
1. 环境介绍：
工业相机型号：海康威视MV-CT089G-13GC01-PRO
相机输出图像格式：MONO8 640*640
相机数据传输方式：千兆网线
计算设备：Jetson Orin NX (16GB)
deepsteam版本：7.1
2. 环境配置
2.1 DeepStream 版本支持与限制 (Jetson 平台)
在安装前，请务必确认您的 Jetson 硬件、JetPack 版本与 DeepStream 版本的兼容性。以下是近期主流版本的支持情况：
| DeepStream 版本 | 支持的平台 | 主要操作系统版本 | JetPack / L4T | 主要新增/限制 | 备注 |
|---|---|---|---|---|---|
| 8.0 | Jetson Thor | L4T Ubuntu 24.04 | JetPack 7.0 GA / 38.2 | 全新 3D 跟踪、Inference Builder、TAO 模型支持；不支持旧 Jetson（如 Orin 系列） | 最新架构专属 |
| 7.1 | Jetson Orin 系列 | L4T Ubuntu 22.04 | JetPack 6.1 GA / 36.4 | 更成熟的 Python 框架和稳定性改进 | Python 支持更好 |
| 7.0 | Jetson Orin 系列 | L4T Ubuntu 22.04 | JetPack 6.0 GA / 36.3 | Python API、Service Maker 初版 | 功能增强版 |
| 6.4 | Jetson Orin 系列 | L4T Ubuntu 22.04 | JetPack 6.0 DP / 36.2 | 支持最新 Orin 系列 Jetson | 旧版本；一般稳定 |
| 6.3 | Orin, Xavier 系列 | L4T Ubuntu 20.04 | JetPack 5.1.2 / 35.4 | Xavier 系列的最后一个主要支持版本；不支持老款 Nano/TX1/TX2 | 广泛用于 Xavier |
| 6.0.1 | Nano, TX1/TX2, Xavier | L4T Ubuntu 18.04 | JetPack 4.6.x / 32.7 | 老款 Jetson Nano (Maxwell 架构) 的最终支持版本；不支持 Orin | 老设备专属 |
注意：如果您使用的是老款的 Jetson Nano (4GB/2GB)，您最高只能安装 JetPack 4.6.x 和 DeepStream 6.0.1。
如何查询您的系统信息？
在决定安装哪个版本之前，请在终端运行以下命令确认您的环境：
- 查询 JetPack / L4T 版本:
(输出示例：cat /etc/nv_tegra_release# R35 (release), REVISION: 4.1表示 L4T 35.4.1，对应 JetPack 5.1.2) - 查询 Ubuntu 系统版本:
lsb_release -a - 查询 Python 版本:
python3 --version
2.2 安装 DeepStream SDK
在 Jetson 平台上，最便捷的安装方式是直接使用 apt 包管理器从 NVIDIA 的官方软件源进行安装。
-
直接安装:
sudo apt-get update sudo apt-get install deepstream-7.1(注：版本号
7.1请根据您当前 JetPack 支持的最新版本进行替换，例如deepstream-6.3) -
验证安装: 运行自带的示例程序，确保基础环境正常。
deepstream-app -c /opt/nvidia/deepstream/deepstream/samples/configs/deepstream-app/source30_1080p_dec_infer-resnet_tiled_display_int8.txt
2.3 安装 Python 绑定 (pyds)
DeepStream 底层是 C/C++，要用 Python 开发，必须安装 pyds。
-
安装基础依赖:
sudo apt-get install python3-gi python3-dev python3-gst-1.0 python-gi-dev git python-dev \ python3 python3-pip python3.8-dev cmake g++ build-essential libglib2.0-dev \ libglib2.0-dev-bin libgstreamer1.0-dev libtool m4 autoconf automake libgirepository1.0-dev libcairo2-dev -
安装 pyds (通过 pip): NVIDIA 官方现在提供了预编译的
.whl包，可以直接通过pip安装，这是最推荐的方式。- 首先，前往 NVIDIA 官方的 DeepStream Python Apps GitHub 仓库。
- 下载与您的 DeepStream 版本和 Python 版本匹配的
.whl文件（例如pyds-1.1.11-py3-none-linux_aarch64.whl）。 - 执行安装：
pip3 install ./pyds-*.whl
(注：如果您使用的是 Conda 虚拟环境，请确保在激活环境后执行
pip install)
3. YOLO 模型准备
DeepStream 的核心推理引擎是 nvinfer，它底层依赖 TensorRT。因此，我们需要将 YOLO 模型转换为 TensorRT 引擎 (.engine 或 .trt)。
3.1 导出 ONNX
首先，使用 Ultralytics 官方库将 PyTorch 模型 (.pt) 导出为 ONNX 格式。 这里我们使用 DeepStream-Yolo 提供的导出脚本，以确保导出的 ONNX 模型与 DeepStream 兼容。
-
安装依赖:
pip3 install ultralytics onnx onnxslim onnxruntime -
下载模型:
wget https://github.com/ultralytics/assets/releases/download/v8.4.0/yolo11n.pt -P model/ -
导出模型: 使用
DeepStream-Yolo/utils/export_yolo11.py脚本导出模型。python3 DeepStream-Yolo/utils/export_yolo11.py -w model/yolo11n.pt --dynamic这将在
model/目录下生成yolo11n.onnx文件。
3.2 转换为 TensorRT 引擎 (自动转换)
在 DeepStream 中，通常不需要手动使用 trtexec 工具转换模型。 只要在配置文件 (config_infer_yolo.txt) 中正确指定了 onnx-file 的路径，并且配置了 custom-lib-path 指向 DeepStream-Yolo 编译出的 .so 库，DeepStream 在首次运行时，如果发现 model-engine-file 指定的引擎文件不存在，它会自动调用 DeepStream-Yolo 库中的自定义引擎构建函数 (NvDsInferYoloCudaEngineGet)，利用 TensorRT 将 ONNX 模型编译为 .engine 文件。
注意：首次运行自动编译引擎可能需要几分钟到十几分钟不等，请耐心等待。编译完成后，引擎文件会被保存在指定的路径，下次运行将直接加载，速度极快。
4. DeepStream 配置文件 (config_infer_yolo.txt)
nvinfer 插件需要一个配置文件来告诉它如何加载模型、处理输入和解析输出。
关键配置项解析：
[property]
gpu-id=0
# 指定 TensorRT 引擎路径
model-engine-file=model/yolo11n.onnx_b1_gpu0_fp16.engine
# 指定 ONNX 路径 (如果引擎不存在，DeepStream 会尝试自动用 ONNX 生成)
onnx-file=model/yolo11n.onnx
# 批处理大小，必须与 streammux 的 batch-size 匹配
batch-size=1
# 精度: 0=FP32, 1=INT8, 2=FP16
network-mode=2
# 类别数量 (COCO 数据集为 80)
num-detected-classes=80
# 缩放和归一化
# YOLO 通常需要 RGB 输入，且像素值归一化到 0-1
net-scale-factor=0.0039215697906911388 # 1/255
model-color-format=0 # 0=RGB, 1=BGR
# 自定义解析函数 (非常重要！)
# 因为 YOLO 的输出张量格式不是 DeepStream 默认支持的，必须提供 C++ 编写的解析库
parse-bbox-func-name=NvDsInferParseYolo
custom-lib-path=DeepStream-Yolo/nvdsinfer_custom_impl_Yolo/libnvdsinfer_custom_impl_Yolo.so
注意：custom-lib-path 指向的 .so 文件是解析 YOLO 输出的 C++ 动态库，通常可以在 GitHub 上的 DeepStream-Yolo 开源项目中找到并编译。
5. 编译 DeepStream-Yolo 自定义解析库
为了让 DeepStream 能够正确解析 YOLO 模型的输出，我们需要编译 DeepStream-Yolo 提供的自定义 C++ 解析库 (libnvdsinfer_custom_impl_Yolo.so)。
5.1 确定 CUDA 版本
首先，需要确定您的系统上安装的 CUDA 版本。可以通过以下命令查看：
cat /usr/local/cuda/version.json | grep version
或者查看 /usr/local/ 目录下的 cuda 软链接指向的版本。 根据前面的系统信息，JetPack 6.1 (L4T 36.4) 对应的 DeepStream 7.1 默认使用 CUDA 12.6。
5.2 编译 .so 动态库
进入 DeepStream-Yolo/nvdsinfer_custom_impl_Yolo 目录，设置 CUDA_VER 环境变量并执行 make 命令：
cd DeepStream-Yolo/nvdsinfer_custom_impl_Yolo
# 设置 CUDA 版本，请根据实际情况修改 (例如 12.6)
export CUDA_VER=12.6
# 清理旧的编译文件并重新编译
make clean && make
编译成功后，会在当前目录下生成 libnvdsinfer_custom_impl_Yolo.so 文件。 在 config_infer_yolo.txt 配置文件中，custom-lib-path 参数需要指向这个生成的 .so 文件的路径。
6.代码实战
6.1 源码
import sys
import time
import os
import ctypes
import numpy as np
from ctypes import *
# Set environment variables for GStreamer
os.environ["GI_TYPELIB_PATH"] = "/usr/lib/aarch64-linux-gnu/girepository-1.0"
os.environ["LD_LIBRARY_PATH"] = os.environ.get("LD_LIBRARY_PATH", "") + ":/home/dt/deepstream/DeepStream-Yolo/nvdsinfer_custom_impl_Yolo:/opt/MVS/lib/aarch64"
if not os.getenv("MVCAM_COMMON_RUNENV"):
os.environ["MVCAM_COMMON_RUNENV"] = "/opt/MVS/lib"
try:
sys.path.append("/opt/MVS/Samples/aarch64/Python/MvImport")
from MvCameraControl_class import *
except ImportError:
print("[Error] MVS SDK not found! Please assume it is installed at /opt/MVS.")
class MvCamera(): pass
import gi
import pyds
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib, GObject
G_STREAMMUX_WIDTH = 640
G_STREAMMUX_HEIGHT = 640
G_STREAMMUX_BATCH_SIZE = 1
G_STREAMMUX_TIMEOUT = 4000000
class HikvisionDeepStreamApp:
def __init__(self, config_file="config_infer_yolo.txt"):
self.config_file = config_file
self.cam = MvCamera()
self.pipeline = None
self.loop = None
self.elements = {}
self.running = False
self.total_frame_count = 0
self.total_latency_ms = 0.0
self.pipeline_start_time = 0
self.frame_start_times = {}
self.pushed_frames = 0
Gst.init(None)
self.loop = GLib.MainLoop()
def run(self):
try:
self._open_camera()
self._build_pipeline()
self._start_pipeline()
import threading
self.running = True
if "appsrc" in self.elements['source'].get_factory().get_name():
self.grab_thread = threading.Thread(target=self._grab_and_push_frames)
self.grab_thread.start()
self.start_time = time.time()
GLib.timeout_add_seconds(30, self._stop_pipeline)
self.loop.run()
end_time = time.time()
duration = end_time - self.start_time
fps = self.total_frame_count / duration if duration > 0 else 0
avg_latency = self.total_latency_ms / self.total_frame_count if self.total_frame_count > 0 else 0
print(f"\n--- Performance Results ---")
print(f"Total Frames: {self.total_frame_count}")
print(f"Duration: {duration:.2f} seconds")
print(f"Throughput: {fps:.2f} FPS")
print(f"Average Latency: {avg_latency:.2f} ms")
readme_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "README.md")
with open(readme_path, "a") as f:
f.write("\n## Hikvision Camera Performance Test\n")
f.write(f"- **Total Frames**: {self.total_frame_count}\n")
f.write(f"- **Duration**: {duration:.2f} seconds\n")
f.write(f"- **Throughput**: {fps:.2f} FPS\n")
f.write(f"- **Average Latency**: {avg_latency:.2f} ms\n")
except KeyboardInterrupt:
print("\n[App] KeyboardInterrupt detected. Exiting...")
except Exception as e:
print(f"[App] Critical Error: {e}")
finally:
self.running = False
if hasattr(self, 'grab_thread'):
self.grab_thread.join()
self._cleanup()
def _stop_pipeline(self):
print("[App] 30 seconds elapsed. Stopping pipeline...")
self.running = False
if "appsrc" in self.elements['source'].get_factory().get_name():
self.elements['source'].emit("end-of-stream")
else:
self.pipeline.send_event(Gst.Event.new_eos())
return False
def _open_camera(self):
print("[App] Discovering Hikvision Camera (MVS)...")
deviceList = MV_CC_DEVICE_INFO_LIST()
tlayerType = MV_GIGE_DEVICE | MV_USB_DEVICE
if MvCamera.MV_CC_EnumDevices(tlayerType, deviceList) != 0:
raise RuntimeError("Enum devices failed!")
if deviceList.nDeviceNum == 0:
raise RuntimeError("No cameras found!")
print(f"[App] Found {deviceList.nDeviceNum} cameras.")
stDeviceList = cast(deviceList.pDeviceInfo[0], POINTER(MV_CC_DEVICE_INFO)).contents
if MvCamera.MV_CC_CreateHandle(self.cam, stDeviceList) != 0:
raise RuntimeError("Create handle failed!")
if MvCamera.MV_CC_OpenDevice(self.cam) != 0:
raise RuntimeError("Open device failed!")
# --- Full Framerate Configuration ---
# 1. Disable Frame Rate Control (Let camera run at max speed)
ret = self.cam.MV_CC_SetBoolValue("AcquisitionFrameRateEnable", False)
if ret != 0:
print(f"[Camera] Warning: Failed to disable AcquisitionFrameRateEnable (Error: {ret:x})")
# 2. Get the actual Resulting Frame Rate
stFloatParam = MVCC_FLOATVALUE()
memset(byref(stFloatParam), 0, sizeof(MVCC_FLOATVALUE))
ret = self.cam.MV_CC_GetFloatValue("ResultingFrameRate", stFloatParam)
if ret == 0:
self.camera_fps = int(stFloatParam.fCurValue)
print(f"[Camera] Resulting Frame Rate: {stFloatParam.fCurValue:.2f} FPS")
else:
self.camera_fps = 30 # Default fallback
print(f"[Camera] Warning: Failed to get ResultingFrameRate (Error: {ret:x}), using default 30 FPS")
if MvCamera.MV_CC_StartGrabbing(self.cam) != 0:
raise RuntimeError("Start grabbing failed!")
stParam = MVCC_INTVALUE()
memset(byref(stParam), 0, sizeof(MVCC_INTVALUE))
if MvCamera.MV_CC_GetIntValue(self.cam, "PayloadSize", stParam) != 0:
raise RuntimeError("Get PayloadSize failed!")
self.payload_size = stParam.nCurValue
self.data_buf = (c_ubyte * self.payload_size)()
def _grab_and_push_frames(self):
print("[App] Frame grabbing thread started.")
stFrameInfo = MV_FRAME_OUT_INFO_EX()
memset(byref(stFrameInfo), 0, sizeof(MV_FRAME_OUT_INFO_EX))
appsrc = self.elements['source']
self.pushed_frames = 0
while self.running:
ret = self.cam.MV_CC_GetOneFrameTimeout(byref(self.data_buf), self.payload_size, stFrameInfo, 1000)
if ret == 0:
start_time = time.time()
data = string_at(byref(self.data_buf), stFrameInfo.nWidth * stFrameInfo.nHeight)
buf = Gst.Buffer.new_allocate(None, len(data), None)
buf.fill(0, data)
fps_n = getattr(self, 'camera_fps', 30)
duration = Gst.util_uint64_scale(1, Gst.SECOND, fps_n)
timestamp = self.pushed_frames * duration
buf.pts = timestamp
buf.dts = timestamp
buf.duration = duration
# Use PTS as the unique key for latency tracking
self.frame_start_times[timestamp] = start_time
retval = appsrc.emit("push-buffer", buf)
if retval != Gst.FlowReturn.OK:
print(f"[AppSrc] Push buffer failed: {retval}")
break
self.pushed_frames += 1
else:
pass
print("[App] Frame grabbing thread stopped.")
def _build_pipeline(self):
print("[App] Building Pipeline for Hikvision Camera...")
self.pipeline = Gst.Pipeline()
if not self.pipeline:
raise RuntimeError("Unable to create GstPipeline")
self._create_elements()
self._configure_elements()
for name, element in self.elements.items():
self.pipeline.add(element)
self._link_elements()
self._attach_probes()
self._setup_bus()
print("[App] Pipeline built successfully.")
def _create_elements(self):
print("[App] Using 'appsrc' for Hikvision SDK Integration.")
self.elements['source'] = self._make_elm("appsrc", "camera-source")
self.elements['capsfilter'] = self._make_elm("capsfilter", "source-caps")
self.elements['nvvidconv_src'] = self._make_elm("nvvideoconvert", "gpu-uploader")
self.elements['capsfilter_nvmm'] = self._make_elm("capsfilter", "nvmm-caps")
self.elements['streammux'] = self._make_elm("nvstreammux", "stream-muxer")
self.elements['pgie'] = self._make_elm("nvinfer", "primary-inference")
self.elements['nvvidconv_osd'] = self._make_elm("nvvideoconvert", "osd-converter")
self.elements['nvosd'] = self._make_elm("nvdsosd", "onscreendisplay")
self.elements['nvvidconv_enc'] = self._make_elm("nvvideoconvert", "enc-converter")
self.elements['encoder'] = self._make_elm("nvv4l2h264enc", "h264-encoder")
self.elements['parser'] = self._make_elm("h264parse", "h264-parser")
self.elements['mp4mux'] = self._make_elm("qtmux", "muxer")
self.elements['sink'] = self._make_elm("filesink", "file-sink")
def _configure_elements(self):
fps_n = getattr(self, 'camera_fps', 30)
print(f"[App] Configuring pipeline with {fps_n} FPS")
caps_str = f"video/x-raw,format=GRAY8,width={G_STREAMMUX_WIDTH},height={G_STREAMMUX_HEIGHT},framerate={fps_n}/1"
caps = Gst.Caps.from_string(caps_str)
self.elements['capsfilter'].set_property("caps", caps)
nvmm_caps_str = f"video/x-raw(memory:NVMM),format=(string)NV12,width={G_STREAMMUX_WIDTH},height={G_STREAMMUX_HEIGHT},framerate={fps_n}/1"
nvmm_caps = Gst.Caps.from_string(nvmm_caps_str)
self.elements['capsfilter_nvmm'].set_property("caps", nvmm_caps)
if "appsrc" in self.elements['source'].get_factory().get_name():
self.elements['source'].set_property("is-live", True)
self.elements['source'].set_property("format", Gst.Format.TIME)
self.elements['source'].set_property("caps", caps)
# CRITICAL: Block push-buffer when queue is full to prevent infinite buffering and huge latency
self.elements['source'].set_property("block", True)
# Allow buffer to hold up to 3 frames to handle slight pipeline jitter without blocking immediately,
# while still preventing large latency buildup.
self.elements['source'].set_property("max-bytes", G_STREAMMUX_WIDTH * G_STREAMMUX_HEIGHT * 2)
self.elements['streammux'].set_property('width', G_STREAMMUX_WIDTH)
self.elements['streammux'].set_property('height', G_STREAMMUX_HEIGHT)
self.elements['streammux'].set_property('batch-size', G_STREAMMUX_BATCH_SIZE)
self.elements['streammux'].set_property('batched-push-timeout', G_STREAMMUX_TIMEOUT)
self.elements['streammux'].set_property('live-source', 1)
self.elements['pgie'].set_property('config-file-path', self.config_file)
self.elements['encoder'].set_property('bitrate', 4000000)
self.elements['encoder'].set_property('preset-level', 1) # UltraFastPreset for low latency
self.elements['encoder'].set_property('insert-sps-pps', 1)
self.elements['sink'].set_property('location', 'output.mp4')
self.elements['sink'].set_property('sync', False)
self.elements['sink'].set_property('async', False)
def _link_elements(self):
source = self.elements['source']
capsfilter = self.elements['capsfilter']
nvvidconv_src = self.elements['nvvidconv_src']
capsfilter_nvmm = self.elements['capsfilter_nvmm']
muxer = self.elements['streammux']
pgie = self.elements['pgie']
nvvidconv_osd = self.elements['nvvidconv_osd']
osd = self.elements['nvosd']
nvvidconv_enc = self.elements['nvvidconv_enc']
encoder = self.elements['encoder']
parser = self.elements['parser']
mp4mux = self.elements['mp4mux']
sink = self.elements['sink']
self._link_sequential([source, capsfilter, nvvidconv_src, capsfilter_nvmm])
sinkpad = muxer.get_request_pad("sink_0")
srcpad = capsfilter_nvmm.get_static_pad("src")
if srcpad.link(sinkpad) != Gst.PadLinkReturn.OK:
raise RuntimeError("Failed to link nvmm-caps to streammux")
self._link_sequential([muxer, pgie, nvvidconv_osd, osd, nvvidconv_enc, encoder, parser])
parser_srcpad = parser.get_static_pad("src")
muxer_sinkpad = mp4mux.get_request_pad("video_0")
if parser_srcpad.link(muxer_sinkpad) != Gst.PadLinkReturn.OK:
raise RuntimeError("Failed to link parser to mp4mux")
self._link_sequential([mp4mux, sink])
def _link_sequential(self, element_list):
for i in range(len(element_list) - 1):
src = element_list[i]
dest = element_list[i+1]
if not src.link(dest):
raise RuntimeError(f"Could not link {src.get_name()} to {dest.get_name()}")
def _attach_probes(self):
osd_sink = self.elements['nvosd'].get_static_pad("sink")
if osd_sink:
osd_sink.add_probe(Gst.PadProbeType.BUFFER, self._probe_perf_stats, 0)
def _probe_perf_stats(self, pad, info, u_data):
gst_buffer = info.get_buffer()
if not gst_buffer: return Gst.PadProbeReturn.OK
batch_meta = pyds.gst_buffer_get_nvds_batch_meta(hash(gst_buffer))
l_frame = batch_meta.frame_meta_list
while l_frame is not None:
try:
frame_meta = pyds.NvDsFrameMeta.cast(l_frame.data)
except StopIteration:
break
self.total_frame_count += 1
pts = frame_meta.buf_pts
if pts in self.frame_start_times:
latency = time.time() - self.frame_start_times[pts]
self.total_latency_ms += latency * 1000
del self.frame_start_times[pts]
if self.total_frame_count % 30 == 0:
avg_lat = self.total_latency_ms / self.total_frame_count if self.total_frame_count > 0 else 0
print(f"[Stats] Frame: {self.total_frame_count} processed, Avg Latency: {avg_lat:.2f} ms")
try:
l_frame = l_frame.next
except StopIteration:
break
return Gst.PadProbeReturn.OK
def _make_elm(self, factory, name):
elm = Gst.ElementFactory.make(factory, name)
if not elm:
raise RuntimeError(f"Failed to create element {factory} ({name})")
return elm
def _setup_bus(self):
bus = self.pipeline.get_bus()
bus.add_signal_watch()
bus.connect("message", self._bus_callback, None)
def _bus_callback(self, bus, message, loop_data):
t = message.type
if t == Gst.MessageType.EOS:
print("[Bus] End-of-Stream (EOS)")
self.loop.quit()
elif t == Gst.MessageType.ERROR:
err, debug = message.parse_error()
print(f"[Bus] Error: {err}: {debug}")
self.loop.quit()
return True
def _start_pipeline(self):
print("[App] Setting pipeline state to PLAYING...")
self.pipeline.set_state(Gst.State.PLAYING)
def _cleanup(self):
print("[App] Cleaning up resources...")
if self.pipeline:
self.pipeline.set_state(Gst.State.NULL)
if self.loop and self.loop.is_running():
self.loop.quit()
if __name__ == '__main__':
script_dir = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(script_dir, "config_infer_yolo.txt")
if not os.path.exists(CONFIG_PATH):
print(f"[Error] Config file not found at: {CONFIG_PATH}")
sys.exit(1)
print(f"[App] Using config file: {CONFIG_PATH}")
app = HikvisionDeepStreamApp(CONFIG_PATH)
app.run()
6.2 代码解析
这份代码是结合了 海康威视工业相机 (MVS SDK) 和 NVIDIA DeepStream (GStreamer) 的核心推理脚本。它的主要功能是：从工业相机实时抓取图像，将其注入到 DeepStream 硬件加速管线中，使用 YOLO 模型进行目标检测，计算端到端延迟和吞吐量，最后将带有检测框的视频编码保存为 MP4 文件。
为了让你彻底看懂这份代码，我将从整体架构到代码细节为你进行极其详细的拆解。
1. 整体管线架构 (Pipeline Architecture)
在深入代码之前，必须先理解 GStreamer 的数据流向（Pipeline）。数据在代码中经历了以下旅程：
[海康相机 (MVS SDK)]
↓ (内存拷贝)
1. appsrc (数据入口，接收 Python 传来的图像内存)
↓ (GRAY8 格式)
2. capsfilter (限制输入格式为 GRAY8，即单通道灰度图)
↓
3. nvvideoconvert (将数据从 CPU 内存上传到 GPU 内存)
↓
4. capsfilter (将格式转换为 DeepStream 要求的 NV12 格式，且在 NVMM 显存中)
↓
5. nvstreammux (将单路视频流打包成 Batch，准备推理)
↓
6. nvinfer (核心推理插件，加载 YOLO 模型进行目标检测)
↓
7. nvvideoconvert (格式转换，准备画框)
↓
8. nvdsosd (On-Screen Display，将检测框和标签画在图像上)
↓
9. nvvideoconvert (格式转换，准备视频编码)
↓
10. nvv4l2h264enc (硬件 H.264 视频编码器)
↓
11. h264parse (解析 H.264 码流)
↓
12. qtmux (将 H.264 码流封装成 MP4 容器格式)
↓
13. filesink (将最终的 MP4 数据写入本地磁盘 output.mp4)
2. 代码逐段详细解析
2.1 环境变量与依赖导入
import sys, time, os, ctypes, numpy as np
from ctypes import *
# 1. 设置 GStreamer 环境变量，确保能找到 Python 绑定库
os.environ["GI_TYPELIB_PATH"] = "/usr/lib/aarch64-linux-gnu/girepository-1.0"
# 2. 设置动态链接库路径，包含自定义的 YOLO 解析库和海康 MVS 库
os.environ["LD_LIBRARY_PATH"] = os.environ.get("LD_LIBRARY_PATH", "") + ":/home/dt/deepstream/DeepStream-Yolo/nvdsinfer_custom_impl_Yolo:/opt/MVS/lib/aarch64"
# 3. 导入海康威视 MVS SDK
try:
sys.path.append("/opt/MVS/Samples/aarch64/Python/MvImport")
from MvCameraControl_class import *
except ImportError:
print("[Error] MVS SDK not found!...")
# 4. 导入 GStreamer 和 DeepStream Python 绑定 (pyds)
import gi
import pyds
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib, GObject
解释：这部分主要是环境准备。DeepStream 依赖底层的 C 库，因此必须正确设置 LD_LIBRARY_PATH，否则会报找不到 .so 文件的错误。pyds 是 NVIDIA 提供的 Python 接口，用于读取推理结果。
2.2 类的初始化 (__init__)
class HikvisionDeepStreamApp:
def __init__(self, config_file="config_infer_yolo.txt"):
self.config_file = config_file
self.cam = MvCamera() # 实例化海康相机对象
self.pipeline = None # GStreamer 管线对象
self.loop = None # GLib 事件循环
self.elements = {} # 存放所有 GStreamer 插件的字典
self.running = False # 控制抓图线程的标志位
# 性能统计变量
self.total_frame_count = 0 # 总处理帧数
self.total_latency_ms = 0.0 # 总延迟时间
self.frame_start_times = {} # 字典：记录每一帧的【时间戳(PTS) -> 抓取时间】
self.pushed_frames = 0 # 已经推入管线的帧数
Gst.init(None) # 初始化 GStreamer
self.loop = GLib.MainLoop() # 创建主循环，用于监听管线消息
2.3 主控流程 (run)
def run(self):
try:
self._open_camera() # 1. 打开并配置相机
self._build_pipeline() # 2. 构建 GStreamer 管线
self._start_pipeline() # 3. 启动管线 (状态设为 PLAYING)
import threading
self.running = True
# 4. 启动一个独立的子线程，专门负责从相机抓图并塞入管线
if "appsrc" in self.elements['source'].get_factory().get_name():
self.grab_thread = threading.Thread(target=self._grab_and_push_frames)
self.grab_thread.start()
self.start_time = time.time()
# 5. 设置一个定时器，30秒后自动调用 _stop_pipeline 停止程序
GLib.timeout_add_seconds(30, self._stop_pipeline)
# 6. 阻塞在这里，直到 self.loop.quit() 被调用
self.loop.run()
# 7. 运行结束后的统计与报告写入 (写入 README.md)
# ... (省略打印和写入文件的代码)
解释：这是程序的骨架。注意这里使用了多线程：主线程运行 GStreamer 的事件循环 (loop.run())，子线程 (grab_thread) 负责疯狂地从相机拿图并喂给 GStreamer。
2.4 相机初始化与全帧率配置 (_open_camera)
def _open_camera(self):
# ... (省略枚举设备和创建句柄的常规代码)
# --- 核心：全帧率配置 ---
# 1. 关闭相机的帧率控制，让相机以其物理极限速度出图
ret = self.cam.MV_CC_SetBoolValue("AcquisitionFrameRateEnable", False)
# 2. 获取相机当前真实的输出帧率 (ResultingFrameRate)
stFloatParam = MVCC_FLOATVALUE()
ret = self.cam.MV_CC_GetFloatValue("ResultingFrameRate", stFloatParam)
if ret == 0:
self.camera_fps = int(stFloatParam.fCurValue) # 例如获取到 91 FPS
# 3. 开始抓图
MvCamera.MV_CC_StartGrabbing(self.cam)
# 4. 获取一帧图像的字节大小 (PayloadSize)，并预先分配好 C 语言级别的内存 buffer
MvCamera.MV_CC_GetIntValue(self.cam, "PayloadSize", stParam)
self.payload_size = stParam.nCurValue
self.data_buf = (c_ubyte * self.payload_size)()
解释：这里最关键的是关闭了 AcquisitionFrameRateEnable，并动态读取了 ResultingFrameRate。这保证了我们能以相机的最高性能（如 91 FPS）进行测试，而不是被限制在默认的 30 FPS。
2.5 抓图与数据注入 (_grab_and_push_frames)
这是最核心的数据桥梁，连接了 MVS SDK 和 DeepStream。
def _grab_and_push_frames(self):
appsrc = self.elements['source']
while self.running:
# 1. 从相机获取一帧数据，存入 self.data_buf
ret = self.cam.MV_CC_GetOneFrameTimeout(byref(self.data_buf), self.payload_size, stFrameInfo, 1000)
if ret == 0:
start_time = time.time() # 记录这一帧产生的绝对时间 (用于算延迟)
# 2. 将 C 内存转换为 Python 字符串/字节流
data = string_at(byref(self.data_buf), stFrameInfo.nWidth * stFrameInfo.nHeight)
# 3. 申请 GStreamer 内存 (Gst.Buffer)，并将图像数据填入
buf = Gst.Buffer.new_allocate(None, len(data), None)
buf.fill(0, data)
# 4. 计算并打上时间戳 (PTS - Presentation Timestamp)
fps_n = getattr(self, 'camera_fps', 30)
duration = Gst.util_uint64_scale(1, Gst.SECOND, fps_n) # 每一帧持续的时间
timestamp = self.pushed_frames * duration # 当前帧的时间戳
buf.pts = timestamp
buf.dts = timestamp
buf.duration = duration
# 5. 将时间戳作为 Key，记录这一帧的起始时间
self.frame_start_times[timestamp] = start_time
# 6. 将 Buffer 推入 appsrc 插件，正式进入 DeepStream 管线
retval = appsrc.emit("push-buffer", buf)
self.pushed_frames += 1
解释：
- 时间戳 (PTS) 非常重要。视频流必须有时间戳，否则后续的复用器 (muxer) 和编码器会混乱。
- 延迟追踪机制：我们把
timestamp当作这一帧的“身份证号”，存入字典self.frame_start_times。等这帧图像在管线末端处理完时，我们再用这个身份证号把起始时间取出来，相减就得到了端到端延迟。
2.6 管线配置与低延迟秘诀 (_configure_elements)
def _configure_elements(self):
# ... (省略常规的宽高、格式配置)
if "appsrc" in self.elements['source'].get_factory().get_name():
self.elements['source'].set_property("is-live", True)
self.elements['source'].set_property("format", Gst.Format.TIME)
# 【极其关键的低延迟配置】
# block=True: 当 appsrc 内部队列满了时，阻塞 push-buffer 操作。
# 这意味着如果 GPU 推理太慢，抓图线程就会被卡住，从而变相“丢弃”相机新来的帧，
# 保证进入管线的永远是最新的帧，防止数据在内存中无限堆积导致 10秒+ 的延迟。
self.elements['source'].set_property("block", True)
# max-bytes: 限制 appsrc 内部队列的最大容量。
# 这里设置为 2 帧的大小 (WIDTH * HEIGHT * 2)。队列极小，强制背压 (Backpressure)。
self.elements['source'].set_property("max-bytes", G_STREAMMUX_WIDTH * G_STREAMMUX_HEIGHT * 2)
# 编码器低延迟配置
self.elements['encoder'].set_property('preset-level', 1) # 1 = UltraFastPreset，牺牲一点压缩率换取极速编码
解释：这是解决“延迟越来越大”问题的核心代码。block=True 和 max-bytes 共同作用，形成了一种背压机制 (Backpressure)。当后端 YOLO 处理不过来时（比如只能跑 49 FPS，但相机给了 91 FPS），队列瞬间填满，push-buffer 被阻塞，相机 SDK 内部的旧帧被覆盖/丢弃，从而保证了极低的实时延迟（~95ms）。
2.7 性能探针与延迟计算 (_attach_probes & _probe_perf_stats)
def _attach_probes(self):
# 在 nvosd (画框插件) 的输入端挂载一个探针 (Probe)
osd_sink = self.elements['nvosd'].get_static_pad("sink")
osd_sink.add_probe(Gst.PadProbeType.BUFFER, self._probe_perf_stats, 0)
def _probe_perf_stats(self, pad, info, u_data):
gst_buffer = info.get_buffer()
# 1. 使用 pyds 解析 DeepStream 附加在 Buffer 上的元数据 (Metadata)
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:
frame_meta = pyds.NvDsFrameMeta.cast(l_frame.data)
self.total_frame_count += 1
# 2. 获取这一帧的 PTS (身份证号)
pts = frame_meta.buf_pts
# 3. 去字典里查这一帧是什么时候从相机抓出来的
if pts in self.frame_start_times:
# 4. 当前时间 - 抓取时间 = 端到端延迟
latency = time.time() - self.frame_start_times[pts]
self.total_latency_ms += latency * 1000
del self.frame_start_times[pts] # 查完就删掉，防止内存泄漏
# 每 30 帧打印一次平均延迟
if self.total_frame_count % 30 == 0:
avg_lat = self.total_latency_ms / self.total_frame_count
print(f"[Stats] Frame: {self.total_frame_count} processed, Avg Latency: {avg_lat:.2f} ms")
l_frame = l_frame.next
return Gst.PadProbeReturn.OK
解释：GStreamer 的 Probe（探针）就像是在水管上开了一个观察孔。当图像流经 nvosd 时，触发这个回调函数。我们利用 pyds 提取出帧的元数据，拿到 PTS，从而计算出这帧图像在管线中度过的时间。
3. 总结
这份代码的精妙之处在于：
- 零拷贝理念的妥协与桥接：虽然从相机到 CPU 内存发生了一次拷贝，但通过
appsrc和nvvideoconvert，数据被迅速转移到了 GPU 显存 (NVMM) 中，后续的推理、画框、编码全部在 GPU 内完成，不再回传 CPU。 - 精准的延迟控制：通过
appsrc的block属性，完美解决了工业相机高帧率与边缘设备算力瓶颈之间的矛盾，实现了“宁可丢帧，绝不延迟”的实时工业检测需求。
DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
17
0- 0
已为社区贡献1条内容
所有评论(0)