Building a Custom GStreamer Plugin for NVIDIA DeepStream

为 NVIDIA DeepStream 构建自定义 GStreamer 插件 | Towards Data Science

深度学习

为 NVIDIA DeepStream 构建自定义 GStreamer 插件

为什么在 DeepStream 中需要自定义推理？

David Redó Nieto

2026 年 6 月 19 日

10 分钟阅读

分享

照片由 Julian Hochgesang 在 Unsplash 上提供

NVIDIA DeepStream 为您提供了一个用于多流视频分析的生产就绪管道：硬件加速解码、跟踪、屏幕显示和消息代理，所有这些都通过 GStreamer 连接。对于导出到 TensorRT 的标准检测模型，nvinfer 处理所有内容。

然而，常见情况有其局限性。视觉语言模型、自定义后处理、旋转边界框或在运行时热交换模型的需求，这些是 nvinfer 的假设失效的地方。有时候，您的团队已经精心调整了一个成熟的 PyTorch 推理堆栈，您希望 DeepStream 调用这个堆栈，而不是在配置文件中重新实现它。

值得注意的是，对于 YOLO 家族模型，特别是 DeepStream-Yolo 由 Marcos Luciano 已经出色地完成了在 C++ 中实现自定义后处理的工作。如果考虑使用 C++，请从那里开始。本文则采取不同的角度：完全使用 Python，通过 pyservicemaker 创建自定义 GStreamer 插件，同时不牺牲吞吐量，实现相同的结果。

使这一切成为可能的关键见解是：下游元素如 nvtracker、nvdsosd 和 nvmsgconv 并不关心哪个元素生成了检测元数据。只要正确地写入 DeepStream 的元数据结构，其余生态系统的工作方式就好像 nvinfer 从未出现过一样。

DeepStream 元数据

每个流经 DeepStream 管道的缓冲区都包含的不仅仅是像素数据。从帧通过 nvstreammux 的那一刻起，每个 GstBuffer 都附带了一个 NvDsBatchMeta 结构。层次结构是直接的，可以在官方文档中找到：

NvDsBatchMeta
├── NvDsUserMeta                        (批次级别的自定义元数据)
└── NvDsFrameMeta                       (每个源流一个)
    ├── NvDsUserMeta                    (帧级别的自定义元数据)
    └── NvDsObjectMeta                  (每个检测对象一个)
        ├── NvDsClassifierMeta
        └── NvDsUserMeta                (对象级别的自定义元数据)

NvDsBatchMeta 描述了整个批次。每个 NvDsFrameMeta 对应一个源流，并携带帧级别的信息，如源 ID 和帧号。每个 NvDsObjectMeta 代表一个检测，这意味着当我们的插件写入检测时，我们将为每个检测写入一个 NvDsObjectMeta。

需要理解的关键点是，这些内容并不由 nvinfer 所拥有。这是一个共享的数据契约。管道中的任何 GStreamer 元素都可以从它读取、写入它，或者两者都做：

• nvtracker 读取对象的边界框并写入跟踪 ID。

• nvdsosd 读取边界框和标签以绘制叠加层。

• nvmsgconv 读取整个结构以生成消息负载。

我们的自定义插件将像 nvinfer 一样将检测写入此结构，下游的所有内容都会自动获取这些检测，无需任何修改。在我们编写任何代码之前，有一个重要的约束需要理解：不能直接从 Python 构造 NvDsObjectMeta 实例。尝试实例化该类会在运行时引发“未定义的构造函数！”错误。

原因与架构有关。DeepStream 通过内存池来管理其元数据对象，这些内存池是预先分配的块，它们在帧之间被重复使用，以避免在高吞吐量流水线中重复进行堆内存分配和释放的开销。这些内存池由 NvDsBatchMeta 所拥有，并存在于边界的一侧（C 侧）。Python 绑定提供了对这些内存池的访问，但故意没有暴露 Python 侧的构造函数，因为如果在内存池之外创建 NvDsObjectMeta 对象，将绕过生命周期管理，从而使得 DeepStream 的内存使用变得不可预测。正确的方式是向批次请求一个对象：batch_meta.acquire_object_meta()，它会从内存池中提供一个预先分配的实例。当帧处理完成时，DeepStream 会自动将其返回到内存池中。

Python 桥接：pyservicemaker

为了从 Python 与 DeepStream 的元数据进行交互，我们将使用 pyservicemaker，这是 NVIDIA 当前支持的 DeepStream Python SDK。官方文档涵盖了流水线和流程的基础知识，但并未展示如何编写和附加自定义推理元素的元数据。这就是本文填补的空白。

关键的抽象是 BatchMetadataOperator。通过继承它并实现 handle_metadata(batch_meta) 方法，您可以访问流水线中每个缓冲区的完整 NvDsBatchMeta。从那里开始，迭代帧就像使用 batch_meta.frame_items 一样简单，并可以附加一个检测对象。

pyservicemaker 还提供了 Gst.Buffer 的包装器，直接暴露 batch_meta，并且重要的是提供了一个 extract(batch_id) 方法，该方法返回每个帧的 GPU 内存的 DLPack 句柄。这使得零拷贝推理成为可能，因为我们可以直接将帧传递给 TensorRT，而无需离开 GPU。

我们不会通过探针单独使用 BatchMetadataOperator，而是将相同的模式直接整合到我们自定义插件的 do_transform_ip 方法中，这样我们就可以在元数据访问的同时，控制元素的生命周期、属性和 caps 协商。但首先，我们需要构建该插件。

可发现的 Python GStreamer 插件

GStreamer 在运行时通过扫描 GST_PLUGIN_PATH 中列出的目录来发现插件。对于 Python 插件，它会在每个路径中的 python/ 子目录中查找。这意味着你的插件只是一个 .py 文件放在正确的位置，无需编译、无需 CMake、无需共享库。权衡是注册模式是严格的，如果出错，将导致难以调试的静默失败。

$GST_PLUGIN_PATH/
└── python/
    └── gstexampleplugin.py   # 你的插件

将 GST_PLUGIN_PATH 设置为指向父目录，GStreamer 将在下一次流水线运行时自动找到 python/gstexampleplugin.py。

插件骨架

这是一个通过推理元素的最小骨架：它接收批量视频缓冲区，运行推理，附加元数据，并将缓冲区未经修改地传递到下游。

import gi
gi.require_version('Gst', '1.0')
gi.require_version('GstBase', '1.0')
from gi.repository import Gst, GstBase, GObject

import torch
from pyservicemaker import Buffer

GST_PLUGIN_NAME = "gstexampleplugin"

Gst.init(None)

class GstExamplePlugin(GstBase.BaseTransform):

__gstmetadata__ = ( 'GstExamplePlugin', # 名称 'Filter/Effect/Video', # 分类 '自定义推理元素', # 描述 '你的名字' # 作者 )

src_format = Gst.Caps.from_string( "video/x-raw(memory:NVMM), format=RGB, " "width=(int)[ 1, 2147483647 ], height=(int)[ 1, 2147483647 ], " "framerate=(fraction)[ 0/1, 2147483647/1 ]" ) sink_format = Gst.Caps.from_string( "video/x-raw(memory:NVMM), format=RGB, " "width=(int)[ 1, 2147483647 ], height=(int)[ 1, 2147483647 ], " "framerate=(fraction)[ 0/1, 2147483647/1 ]" )

src_pad_template = Gst.PadTemplate.new( "src", Gst.PadDirection.SRC, Gst.PadPresence.ALWAYS, src_format ) sink_pad_template = Gst.PadTemplate.new( "sink", Gst.PadDirection.SINK, Gst.PadPresence.ALWAYS, sink_format ) __gsttemplates__ = (src_pad_template, sink_pad_template)

__gproperties__ = { 'model-engine': ( str, 'TensorRT 引擎路径', '.engine 文件的路径', '', GObject.ParamFlags.READWRITE ), 'confidence-threshold': ( float, '置信度阈值', '附加检测所需的最小置信度', 0.0, 1.0, 0.5, GObject.ParamFlags.READWRITE ), }

def __init__(self): super().__init__() self.model_engine = '' self.confidence_threshold = 0.5 self.engine = None

def do_get_property(self, prop): if prop.name == 'model-engine': return self.model_engine elif prop.name == 'confidence-threshold': return self.confidence_threshold

def do_set_property(self, prop, value): if prop.name == 'model-engine': self.model_engine = value elif prop.name == 'confidence-threshold': self.confidence_threshold = value

def do_start(self):

在此处加载 TensorRT 引擎

self.engine = load_engine(self.model_engine) # 此函数应被实现 return True

def do_transform_ip(self, gst_buffer: Gst.Buffer) -> Gst.FlowReturn: """原地转换：附加元数据，保持缓冲区不变。""" buffer = Buffer(gst_buffer) batch_meta = buffer.batch_meta

frames = [] for frame_meta in batch_meta.frame_items: t = torch.utils.dlpack.from_dlpack(buffer.extract(frame_meta.batch_id)) frames.append(t) batch = torch.stack(frames, dim=0)

运行模型推理

results = self.engine(batch)

现在我们需要遍历每个帧的结果

并将其附加到 object_meta 中，如果是检测/分割

否则可以将其附加到 user_meta

以下为伪代码，具体取决于你的推理

for frame_meta in batch_meta.frame_items: for det in results: obj = batch_meta.acquire_object_meta()

用每个检测结果填充 obj

... frame_meta.append(obj)

return Gst.FlowReturn.OK

--- 注册 ---

GObject.type_register(GstExamplePlugin) __gstelementfactory__ = (GST_PLUGIN_NAME, Gst.Rank.NONE, GstExamplePlugin)

关于这个框架的一些注意事项：

GstBase.BaseTransform 是用于原地过滤器的正确基类，这种过滤器接收一个缓冲区，对其进行修改（通过附加元数据），然后将其传递给下游。我们覆盖的是 do_transform_ip 而不是 do_transform，因为我们没有分配新的输出缓冲区。

__gstmetadata__ 和 __gsttemplates__ 是必不可少的。GStreamer 没有它们将不会注册该元素。caps 字符串 video/x-raw(memory:NVMM) 告诉 GStreamer 该元素可以与 NVIDIA 内存一起工作，这对于在 DeepStream 管道中保持在 GPU 上至关重要。

__gproperties__ 将 model-engine 和 confidence-threshold 作为一流的 GStreamer 属性公开，这意味着你可以从 gst-launch 命令行或从 Python 管道代码中设置它们，而无需修改源代码。

最后两行是注册所必需的：GObject.type_register 告诉 GObject 类型系统有关该类的信息，而 __gstelementfactory__ 告诉 GStreamer 要公开的元素名称以及要实例化的类。

验证插件。一旦文件到位且缓存已清除，使用以下命令验证注册：

GST_PLUGIN_PATH=/path/to/your/plugins gst-inspect-1.0 gstexampleplugin

你应该看到元素元数据、垫模板以及两个属性的列表。如果你看到了这些，GStreamer 就知道你的插件了，你现在可以将其放入管道中使用了。

## 使用 Ultralytics 进行端到端推理的示例

有了插件的框架，现在是时候填充推理逻辑了。完整的可运行代码作为 GitHub Gist 提供。一旦你能够发现它，你可以像之前一样检查它或启动管道。以下是一个简单的示例，它仅执行推理并显示 fps：

gst-launch-1.0 -v \ nvstreammux name=m width=1280 height=720 batch-size=1 \ batched-push-timeout=33000 ! \ nvvideoconvert nvbuf-memory-type=0 ! \ 'video/x-raw(memory:NVMM), format=RGB' ! \ gstyoloplugin model-path=/path/to/yolo26s.engine ! \ fpsdisplaysink text-overlay=false silent=false sync=false \ video-sink=fakesink \ uridecodebin uri=file:///path/to/video.mp4 ! m.sink_0

### 检查代码

#### 兼容性问题

如果你正在阅读代码，你可能已经意识到我们正在覆盖 tuple 对象，但只在 ultralytics.nn.backends.tensorrt 模块中，因为问题就出在这里。TensorRT Python 绑定和 GStreamer Python 封装框架（PyGObject）之间存在一个已知的兼容性边缘情况，这会导致管道崩溃并显示臭名昭著的消息“Segmentation fault (core dumped)”。这就是为什么需要创建这段代码片段来帮助我们保持预期行为的原因：

import ultralytics.nn.backends.tensorrt as trt_backend

_original_tuple = tuple

def safe_tuple(obj): if "tensorrt" in type(obj).__module__ and type(obj).__name__ == "Dims": return _original_tuple(obj[i] for i in range(len(obj))) return _original_tuple(obj)

trt_backend.tuple = safe_tuple

这在运行时将 Ultralytics 后端命名空间中的 tuple 引用替换为一个版本，该版本对 Dims 对象使用基于索引的访问方式，而其他内容保持不变。这并不优雅，但它是精确的，并且需要在导入时发生，任何模型实例化之前。

#### 推理循环

推理循环本身相当直接：

- 从缓冲区中提取帧

- 预处理 + 推理

- 如果流水线下游元素是 deepstream 插件，则将结果附加到每帧对象的元数据中。

下面是使用 DLPack 实现零拷贝的代码片段：

frames = [] for frame_meta in batch_meta.frame_items: t = torch.utils.dlpack.from_dlpack(buffer.extract(frame_meta.batch_id)) frames.append(t) batch = torch.stack(frames, dim=0)

#### 预处理输入

根据文档，YOLO 模型在通过 torch.Tensor 时，期望一个固定的输入形状（N, 3, 640, 640）。然而，从 nvstreammux 输出的帧的分辨率将取决于源的分辨率。所采用的方法是信封填充（letterboxing）：将帧缩放以适应目标尺寸，同时保持宽高比，然后填充剩余空间。关键的洞察是，我们可以在 GPU 上完全实现这一过程，一次性处理整个批次，而无需接触 CPU 内存。

通过在单个 do_transform_ip 调用中完成帧提取、信封填充、推理和坐标反转，该插件在所有下游元素看来的行为与 nvinfer 完全相同，但其下方却具有完整的 Python 推理堆栈的灵活性。

从这里开始，DeepStream 流水线的其余部分将接管：nvtracker 分配 ID，nvdsosd 绘制叠加层，nvmsgconv 序列化负载。

## 实践要点和下一步

如果你一路读到这里，你已经掌握了一种可行的模式，可以将 nvinfer 替换为自己的 Python 推理元素，更重要的是，你理解了为什么每个部分都设计成这样。

这种模式是通用的。这里描述的一切：插件框架、批量预处理和元数据附加，都与模型无关。将 Ultralytics YOLO 替换为 Roboflow 的 rfdetr 是直接的，GStreamer 和 pyservicemaker 的框架保持不变。对于更复杂的架构也是如此：NVIDIA 自己的 deepstream_reference_apps 仓库中包含了一个使用这种插件方法集成 Vision-Language Model（VLM）的完整示例，如果你正在从检测推进到视频理解，值得研究这个示例。

完整的插件代码可在 GitHub Gist 上找到。如果你在其基础上构建了某些内容：使用不同的模型、多流设置或 VLM 集成，我很想听听你的进展。编码愉快！

撰写人

查看 David Redó Nieto 的所有文章

人工智能

,

Deepstream

推理

机器学习

Python

分享这篇文章

- 在 Facebook 上分享

- 在 LinkedIn 上分享

- 在 X 上分享

Towards Data Science 是一个社区出版物。提交你的见解，以触达全球受众并通过 TDS 作者支付计划获得报酬。

将 href 更新为你的实际提交 URL

为 TDS 写作

✦ 结束 CTA ✦