Создание кастомного GStreamer плагина для NVIDIA DeepStream на Python: как обойти nvinfer и взять контроль над метаданными

Почему nvinfer — это не серебряная пуля, а ты — не статист

Каждый, кто хоть раз собирал пайплайн на NVIDIA DeepStream, знает nvinfer. Это чёрный ящик, который принимает кадр и выдаёт метаданные с десятками полей — классы, bounding boxes, уверенности, трекинг. И всё бы ничего, пока требования не выходят за рамки стандартного детектора.

Но в реальном продакшене хочется большего: дорисовывать свои оверлеи, фильтровать объекты по нестандартной логике, подмешивать данные из внешнего API, или вообще заменить nvinfer на свой инференс — например, запущенный через CUDA stream interleaving для минимизации latency. DeepStream SDK 7.1 (и новее на 2026 год) предоставляет для этого мощный, но неочевидный инструмент — возможность писать кастомные GStreamer плагины на Python. Никаких C++ имплементаций, make и перекомпиляций. Только Python и GObject Introspection.

В этой статье мы научимся перехватывать управление после nvinfer, создадим свой плагин, который на лету изменяет метаданные DeepStream, и разберём грабли, на которые наступают даже опытные инженеры.

Проблема: nvinfer как диктатор

Стандартный конвейер выглядит так:
filesrc ! decodebin ! nvstreammux ! nvinfer ! nvdsosd ! nveglglessink

nvinfer делает инференс и записывает результат в структуры NvDsObjectMeta, которые прикрепляются к буферу Gst. После этого ты можешь читать эти метаданные через pad probe (событие pad-probe), но изменить их по-человечески — уже проблема. DeepStream сильно завязан на C API, и попытки вставить свою логику через probe заканчиваются либо костылями с пересозданием буфера, либо утечками памяти.

Кроме того, nvinfer не умеет выполнять кастомную постобработку: например, накладывать фильтры по геопозиции, агрегировать статистику по зонам, отправлять данные в Kafka. В теории можно добавлять параллельные элементы, но это увеличивает эвристику и усложняет отладку.

Решение: Python плагин как proxy между nvinfer и sink

Идея проста: мы пишем свой GStreamer элемент, который подключается в пайплайн сразу после nvinfer и до любого потребителя метаданных. Внутри элемента мы получаем буфер, извлекаем объекты DeepStream, делаем с ними что хотим (фильтруем, модифицируем, логируем) и отправляем дальше. Никаких костылей с probe — элемент ведёт себя как полноправный участник пайплайна.

Пошаговый план на 21.06.2026

1Установка окружения

sudo apt-get install python3-gi python3-gi-cairo gir1.2-glib-2.0 gir1.2-gstreamer-1.0
# Убеждаемся, что DeepStream SDK установлен (рекомендуется 7.1)
source /opt/nvidia/deepstream/deepstream-7.1/setup.sh
pip install pyds  # обёртка для NvDs метаданных

Обратите внимание: pyds — официальная библиотека от NVIDIA для доступа к метаданным DeepStream из Python. У неё есть обёртки для NvDsBatchMeta, NvDsFrameMeta, NvDsObjectMeta. Код пишем с её помощью.

2Скелет плагина

GStreamer Python плагин — это класс, унаследованный от Gst.Element. Мы сделаем минимальный элемент с одним входным и одним выходным pad. Важно: для DeepStream наши pads должны быть совместимы с типом видео (NV12) и поддерживать дополнительные кванты (Gst.Query).

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

class MyFilter(Gst.Element):
    __gstmetadata__ = ('Custom Python Filter for DeepStream',
                       'Filter/Video',
                       'Modify DeepStream metadata',
                       'Your Name')

    _sinkpadtemplate = Gst.PadTemplate.new('sink',
        Gst.PadDirection.SINK,
        Gst.PadPresence.ALWAYS,
        Gst.Caps.from_string('video/x-raw(memory:NVMM), format=NV12'))

    _srcpadtemplate = Gst.PadTemplate.new('src',
        Gst.PadDirection.SRC,
        Gst.PadPresence.ALWAYS,
        Gst.Caps.from_string('video/x-raw(memory:NVMM), format=NV12'))

    def __init__(self):
        super().__init__()
        self.sinkpad = Gst.Pad.new_from_template(self._sinkpadtemplate, 'sink')
        self.srcpad = Gst.Pad.new_from_template(self._srcpadtemplate, 'src')
        self.sinkpad.set_chain_function_full(self.chain_func)
        self.add_pad(self.sinkpad)
        self.add_pad(self.srcpad)

    def chain_func(self, pad, buffer):
        # Здесь будем работать с метаданными
        return Gst.FlowReturn.OK

3Извлечение метаданных DeepStream

Внутри chain_func получаем Gst.Buffer. Чтобы добраться до NvDsBatchMeta, используем pyds.gst_buffer_get_nvds_batch_meta(hash(buffer)). Далее проходим по фреймам и объектам.

def chain_func(self, pad, buffer):
    batch_meta = pyds.gst_buffer_get_nvds_batch_meta(hash(buffer))
    if not batch_meta:
        self.srcpad.push(buffer)
        return Gst.FlowReturn.OK

    l_frame = batch_meta.frame_meta_list
    while l_frame is not None:
        frame_meta = pyds.NvDsFrameMeta.cast(l_frame.data)
        # Перебираем объекты в фрейме
        l_obj = frame_meta.obj_meta_list
        while l_obj is not None:
            obj_meta = pyds.NvDsObjectMeta.cast(l_obj.data)
            # Пример: удаляем объекты с низкой уверенностью
            if obj_meta.confidence < 0.5:
                # Удаляем метаданные объекта из lists
                pyds.nvds_remove_obj_meta_from_frame(frame_meta, obj_meta)
            # Или меняем класс
            # obj_meta.class_id = 999
            l_obj = l_obj.next
        l_frame = l_frame.next

    # Передаём дальше
    self.srcpad.push(buffer)
    return Gst.FlowReturn.OK

4Регистрация плагина в GStreamer

Чтобы элемент был доступен в пайплайне, его нужно зарегистрировать. Есть два пути: динамическая загрузка через Gst.Plugin или статическая регистрация в рантайме.

Gst.init(None)

# Регистрируем элемент
def plugin_init(plugin):
    from gi.repository import Gst
    # наш класс MyFilter
    Gst.Element.register(plugin, 'myfilter', Gst.Rank.NONE, MyFilter)
    return True

# Загружаем плагин (в реальности это делает Gst при обнаружении .so)
Gst.Plugin.register_static(Gst.Version(1, 0), 'myfilter', 'Custom Python DeepStream Filter', plugin_init, '1.0', 'LGPL', 'myfilter.py', 'myfilter', 'myfilter')

# Теперь создаём пайплайн
pipeline = Gst.parse_launch('filesrc location=test.mp4 ! decodebin ! nvstreammux ! nvinfer ! myfilter ! nvdsosd ! nveglglessink')
pipeline.set_state(Gst.State.PLAYING)

Вместо статической регистрации проще всего написать небольшую обёртку и запускать как модуль. Но для production используйте Gst.Plugin.load_by_name или подключите через GST_PLUGIN_PATH с .so, скомпилированным из Python (через Cython).

Главный нюанс: совместимость с DeepStream 7.x и 8.x

На 2026 год NVIDIA активно развивает DeepStream. В версии 7.1 добавили поддержку NvDsInferContext для мультимодальных моделей, а в SDK 8 (вышел в середине 2025) полностью переработали систему метаданных: NvDsBatchMeta теперь может содержать вложенные графы. Наш код через pyds должен работать без изменений, потому что pyds абстрагирует структуры.

Но есть подводный камень: начиная с DeepStream 7.0, NVIDIA рекомендует использовать pad-проберы вместо цепных функций из-за многопоточности. Если ваш плагин вызывает блокирующие операции (например, обращение к базе данных), делайте это в отдельном потоке через Gst.Poll или используйте Gst.Bus. Иначе рискуете повесить весь пайплайн.

Грабли №1: Утечки памяти при удалении объектов

Когда мы вызываем pyds.nvds_remove_obj_meta_from_frame, мы не освобождаем память, занятую объектом. В C++ это делает деструктор, но в Python нужно дополнительно вызвать pyds.nvds_destroy_obj_meta(obj_meta). Иначе каждый пропущенный фрейм будет съедать память.

# Правильное удаление
if obj_meta.confidence < 0.5:
    pyds.nvds_remove_obj_meta_from_frame(frame_meta, obj_meta)
    pyds.nvds_destroy_obj_meta(obj_meta)

И не забудьте, что после уничтожения obj_meta перестаёт быть доступным для чтения.

Грабли №2: Изменение buffer'а через GPU

Иногда нужно не только менять метаданные, но и дорисовывать на кадре (например, сложный оверлей). Простейший путь — создать Gst.Buffer с новыми данными, но DeepStream использует NVMM память, которая находится на GPU. Для доступа к ней нужен CUDA или nvbufsurface. Из Python это делается через pyds.get_buffer_surface(buffer_ptr) и pyds.map, но лучше не лезть в сырые пиксели, а добавить отдельный элемент на основе nvdsosd или написать Cython-плагин для рисования.

Когда Python плагин не нужен: альтернативы

Если ваша задача — просто обойти nvinfer и заменить его на другую модель, необязательно писать плагин. Можно использовать nvinferserver или nvinfer_custom с динамической загрузкой библиотеки. Но если вам нужно смешивать несколько потоков данных, обращаться к Redis, или реализовывать сложную бизнес-логику — Python плагин будет единственным адекватным вариантом.

Кстати, недавно вышла статья о запуске LLM на vLLM — в ней рассматривается работа с метаданными из Python, что перекликается с нашей темой.

Финальный код плагина (минимум)

#!/usr/bin/env python3
import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GObject
import pyds

class DeepStreamPythonFilter(Gst.Element):
    __gstmetadata__ = ('DSFilter', 'Filter/Video', 'Filter objects with low confidence', 'DevOps')

    sink_tmpl = Gst.PadTemplate.new('sink', Gst.PadDirection.SINK, Gst.PadPresence.ALWAYS,
                                     Gst.Caps.from_string('video/x-raw(memory:NVMM), format=NV12'))
    src_tmpl = Gst.PadTemplate.new('src', Gst.PadDirection.SRC, Gst.PadPresence.ALWAYS,
                                    Gst.Caps.from_string('video/x-raw(memory:NVMM), format=NV12'))

    def __init__(self):
        super().__init__()
        self.sinkpad = Gst.Pad.new_from_template(self.sink_tmpl, 'sink')
        self.sinkpad.set_chain_function_full(self._chain)
        self.add_pad(self.sinkpad)
        self.srcpad = Gst.Pad.new_from_template(self.src_tmpl, 'src')
        self.add_pad(self.srcpad)

    def _chain(self, pad, buffer):
        batch_meta = pyds.gst_buffer_get_nvds_batch_meta(hash(buffer))
        if not batch_meta:
            return self._push(buffer)

        f_iter = batch_meta.frame_meta_list
        while f_iter is not None:
            f_meta = pyds.NvDsFrameMeta.cast(f_iter.data)
            o_iter = f_meta.obj_meta_list
            prev = None
            while o_iter is not None:
                obj_meta = pyds.NvDsObjectMeta.cast(o_iter.data)
                if obj_meta.confidence < 0.5:
                    next_meta = o_iter.next
                    pyds.nvds_remove_obj_meta_from_frame(f_meta, obj_meta)
                    pyds.nvds_destroy_obj_meta(obj_meta)
                    o_iter = next_meta
                else:
                    prev = o_iter
                    o_iter = o_iter.next
            f_iter = f_iter.next

        return self._push(buffer)

    def _push(self, buffer):
        return self.srcpad.push(buffer)

def register_plugin():
    Gst.init(None)
    Gst.Plugin.register_static(Gst.Version(1,0), 'dspythonfilter', 'Custom Python DeepStream Filter',
                                lambda p: Gst.Element.register(p, 'dspythonfilter', Gst.Rank.NONE, DeepStreamPythonFilter),
                                '1.0', 'LGPL', 'dspythonfilter.py', 'dspythonfilter', 'dspythonfilter')

if __name__ == '__main__':
    register_plugin()
    pipeline_str = 'filesrc location=/opt/nvidia/deepstream/deepstream/samples/streams/sample_720p.mp4 ! qtdemux ! queue ! nvstreammux ! nvinfer config-file-path=/opt/nvidia/deepstream/deepstream/samples/configs/deepstream-app/config_infer_primary.txt ! dspythonfilter ! nvdsosd ! nveglglessink'
    pipeline = Gst.parse_launch(pipeline_str)
    pipeline.set_state(Gst.State.PLAYING)
    loop = GLib.MainLoop()
    try:
        loop.run()
    except KeyboardInterrupt:
        pipeline.set_state(Gst.State.NULL)

Резюме: три важных совета от автора

1. Не копируйте код из старых гайдов (2022-2023). API pyds сильно изменился: устаревшие методы nvds_add_obj_meta_to_frame теперь требуют дополнительно передавать batch_meta. Всегда сверяйтесь с документацией SDK 7.1+.
2. Тестируйте на синтетическом потоке. Запустите videotestsrc вместо файла, чтобы отладка была быстрее. Но помните: videotestsrc не создаёт метаданные DeepStream, поэтому придётся подсунуть искусственный NvDsBatchMeta через pyds.nvds_acquire_batch_meta.
3. Смотрите в сторону многопроцессорных мультиплексоров, если ваш Python плагин блокирует пайплайн. DeepStream сам по себе многопоточен, но GIL в Python может стать узким горлышком. Выход: выносите тяжелые вычисления в C++ через Cython или используйте multiprocessing с передачей буферов через shared memory.

Создание кастомного плагина — это не просто хак, а архитектурное решение, которое даёт полный контроль над конвейером AI-видео. Используйте его с умом, и ваши пайплайны станут гибче, быстрее и понятнее.

Подписаться на канал