Création d’un plugin GStreamer personnalisé pour NVIDIA DeepStream

un pipeline prêt pour la production pour l’analyse vidéo multi-flux : décodage, suivi, affichage à l’écran et courtage de messages accélérés par le matériel, le tout câblé via GStreamer. Pour les modèles de détection standard exportés vers TensorRT, nvinfer gère tout.

Toutefois, le cas courant a des limites. Les modèles en langage de vision, le post-traitement personnalisé, les cadres de délimitation pivotés ou la nécessité de remplacer à chaud les modèles au moment de l’exécution sont des endroits où nvinferLes hypothèses de s’effondrent. Parfois, vous disposez d’une pile d’inférence PyTorch mature que votre équipe a soigneusement réglée et vous souhaitez que DeepStream appelle que plutôt que de le réimplémenter dans un fichier de configuration.

Il convient de noter que pour les modèles de la famille YOLO en particulier, DeepStream-Yolo de Marcos Luciano a déjà réalisé un excellent travail en implémentant un post-traitement personnalisé en C++. Si le C++ est sur la table, commencez par là. Cet article prend un angle différent : obtenir le même résultat entièrement en Python, en utilisant un plugin GStreamer personnalisé avec pyservicemaker sans sacrifier le débit.

L’information clé qui rend cela possible : les éléments en aval tels que nvtracker, nvdsosdet nvmsgconv peu importe quel élément a produit des métadonnées de détection. Écrivez correctement dans la structure des métadonnées de DeepStream et le reste de l’écosystème fonctionne comme si nvinfer n’a jamais été sur la photo.

Métadonnées DeepStream

Chaque tampon circulant dans un pipeline DeepStream contient plus que des données de pixels. A partir du moment où les images passent nvstreammuxchaque GstBuffer a un NvDsBatchMeta structure qui y est attachée. La hiérarchie est simple et peut être trouvée dans le documentation officielle.

NvDsBatchMeta
├── NvDsUserMeta                        (batch-level custom metadata)
└── NvDsFrameMeta                       (one per source stream)
    ├── NvDsUserMeta                    (frame-level custom metadata)
    └── NvDsObjectMeta                  (one per detected object)
        ├── NvDsClassifierMeta
        └── NvDsUserMeta                (object-level custom metadata)

NvDsBatchMeta décrit l’ensemble du lot. Chaque NvDsFrameMeta correspond à un flux source et contient des informations au niveau de la trame telles que l’ID source et le numéro de trame. Chaque NvDsObjectMeta représente une seule détection, ce qui signifie que lorsque notre plugin écrit des détections, nous écrirons un NvDsObjectMeta pour chacun.

La chose essentielle à comprendre est que rien de tout cela n’appartient à nvinfer. Il s’agit d’un contrat de données partagées. Tout élément GStreamer du pipeline peut y lire, y écrire, ou les deux :

• nvtracker lit les cadres de délimitation des objets et écrit les identifiants de suivi.
• nvdsosd lit les boîtes et les étiquettes pour dessiner des superpositions.
• nvmsgconv lit toute la structure pour produire des charges utiles de message.

Notre plugin personnalisé écrira simplement les détections dans cette structure de la même manière nvinfer le ferait et tout en aval les récupère sans modification. Une contrainte importante qui mérite d’être comprise avant d’écrire du code : NvDsObjectMeta cas ne peut pas être construit directement à partir de Python. Tenter d’instancier la classe soulève un No constructor defined! erreur au moment de l’exécution.

La raison est architecturale. DeepStream gère ses objets de métadonnées via des pools de mémoire, des blocs pré-alloués qui sont recyclés entre les trames pour éviter la surcharge liée à l’allocation et à la désallocation répétées du tas dans un pipeline à haut débit. Ces piscines appartiennent à NvDsBatchMeta et habite du côté C de la frontière. Les liaisons Python exposent l’accès à ces pools, mais n’exposent délibérément pas de constructeur côté Python, car la création d’un NvDsObjectMeta en dehors du pool contournerait la gestion du cycle de vie qui maintient l’utilisation de la mémoire de DeepStream prévisible. La bonne façon d’en obtenir un est de le demander au lot : batch_meta.acquire_object_meta()qui vous remet une instance pré-allouée du pool. Une fois la trame terminée, DeepStream la renvoie automatiquement au pool.

Le pont Python : pyservicemaker

Pour interagir avec les métadonnées de DeepStream depuis Python, nous utiliserons pyservicemakerle SDK Python actuel et pris en charge par NVIDIA pour DeepStream. Le documentation officielle couvre les bases des pipelines et des flux, mais ne montre pas comment écrire et attacher des métadonnées à partir d’un élément d’inférence personnalisé. C’est la lacune que cet article comble.

L’abstraction clé est BatchMetadataOperator. Le sous-classer et l’implémenter handle_metadata(batch_meta) vous donne accès à l’intégralité NvDsBatchMeta pour chaque tampon circulant dans le pipeline. À partir de là, itérer des frames est aussi simple que d’utiliser batch_meta.frame_items et fixer un objet de détection.

pyservicemaker fournit également un Buffer envelopper autour Gst.Buffer qui expose batch_meta directement et, surtout, un extract(batch_id) méthode qui renvoie un handle DLPack à la mémoire GPU de chaque image. C’est ce qui rend possible l’inférence sans copie, car nous pouvons transmettre la trame directement à TensorRT sans jamais quitter le GPU.

Plutôt que d’utiliser BatchMetadataOperator autonome via une sonde, nous intégrerons le même modèle directement dans le fichier de notre plugin personnalisé do_transform_ip qui nous permet de contrôler le cycle de vie, les propriétés et la négociation des plafonds de l’élément ainsi que l’accès aux métadonnées. Mais d’abord, nous devons créer ce plugin.

Un plugin Python GStreamer détectable

GStreamer découvre les plugins au moment de l’exécution en analysant les répertoires répertoriés dans GST_PLUGIN_PATH. Pour les plugins Python en particulier, il regarde à l’intérieur d’un python/ sous-répertoire dans chacun de ces chemins. Cela signifie que votre plugin n’est qu’un .py fichier déposé au bon endroit, pas de compilation, pas de CMake, pas de bibliothèque partagée. Le compromis est que le modèle d’enregistrement est strict et qu’une erreur produit des échecs silencieux qui sont véritablement difficiles à déboguer.

$GST_PLUGIN_PATH/
└── python/
    └── gstexampleplugin.py   # your plugin

Ensemble GST_PLUGIN_PATH pointer sur le répertoire parent et GStreamer trouvera python/gstexampleplugin.py automatiquement lors de la prochaine exécution du pipeline.

Le squelette du plugin

Voici le squelette minimal d’un élément d’inférence passthrough : il reçoit des tampons vidéo par lots, exécute l’inférence, attache des métadonnées et transmet le tampon en aval sans modification.

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',                     # name
        'Filter/Effect/Video',                  # classification
        'Custom inference element',             # description
        'Your Name'                             # author
    )

    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 path',
            'Path to the .engine file',
            '',
            GObject.ParamFlags.READWRITE
        ),
        'confidence-threshold': (
            float,
            'Confidence threshold',
            'Minimum confidence to attach a detection',
            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):
        # Load your TensorRT engine here
        self.engine = load_engine(self.model_engine) # This function should be implemented
        return True

    def do_transform_ip(self, gst_buffer: Gst.Buffer) -> Gst.FlowReturn:
        """In-place transform: attach metadata, pass buffer unchanged."""
        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
        batch = torch.stack(frames, dim=0)

        # Run your model inference
        results = self.engine(batch)
        
        # Now we will need to iterate over the results for each frame
        # and attach it to the object_meta in case it is detection/segmentation
        # otherwise we can do it as user_meta
        # The following is pseudocode, which depends on your inference
        for frame_meta in batch_meta.frame_items:
            for det in results:
                obj = batch_meta.acquire_object_meta()
                # Fill the obj with each detection
                ...
                frame_meta.append(obj)
 
        return Gst.FlowReturn.OK


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

Quelques points à noter à propos de ce squelette :

GstBase.BaseTransform est la bonne classe de base pour un filtre sur place, celle qui reçoit un tampon, le modifie (en attachant des métadonnées) et le transmet en aval. Nous remplaçons do_transform_ip plutôt que do_transform car nous n’allouons pas de nouveau tampon de sortie.

__gstmetadata__ et __gsttemplates__ ne sont pas facultatifs. GStreamer n’enregistrera pas l’élément sans eux. La chaîne des majuscules video/x-raw(memory:NVMM) indique à GStreamer que cet élément fonctionne avec la mémoire NVIDIA, ce qui est essentiel pour rester sur GPU dans un pipeline DeepStream.

__gproperties__ expose model-engine et confidence-threshold comme propriétés GStreamer de première classe, ce qui signifie que vous pouvez les définir à partir d’un gst-launch ligne de commande ou à partir du code du pipeline Python sans toucher à la source.

Les deux dernières lignes sont obligatoires pour l’inscription : GObject.type_register indique au système de types GObject la classe et __gstelementfactory__ indique à GStreamer quel nom d’élément exposer et quelle classe instancier.

Vérification du plugin. Une fois le fichier en place et le cache vidé, vérifiez l’enregistrement avec :

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

Vous devriez voir les métadonnées des éléments, les modèles de pad et les deux propriétés répertoriées. Si vous les voyez, GStreamer connaît votre plugin et vous êtes prêt à le déposer dans un pipeline.

Exemple d’inférence de bout en bout avec Ultralytics

Une fois le squelette du plugin en place, il est temps de remplir la logique d’inférence. Le code de travail complet est disponible sous forme de L’essentiel de GitHub. Une fois que vous l’avez détectable, vous pouvez l’inspecter comme nous l’avons fait auparavant ou lancer le pipeline. Voici un exemple simple qui effectue simplement une inférence et affiche les 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

Inspecter le code

Problème de compatibilité

Si vous lisez le code, vous avez peut-être réalisé que nous remplaçons le tuple objet, mais seulement à l’intérieur ultralytics.nn.backends.tensorrt module puisque c’est là que se situe le problème. Il existe un cas limite de compatibilité connu entre les Liaisons TensorRT Python et le Cadre de wrapper GStreamer Python (PyGObject) cela fera planter votre pipeline avec le tristement célèbre message « Défaut de segmentation (core dumped) ». C’est pourquoi il était nécessaire de créer cet extrait de code qui nous aide à préserver le comportement souhaité :

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

Cela remplace le tuple référence à l’intérieur de l’espace de noms du backend Ultralytics lors de l’exécution avec une version qui revient à un accès basé sur l’index pour Dims objets, laissant tout le reste intact. Ce n’est pas élégant, mais c’est chirurgical et cela doit se produire au moment de l’importation, avant qu’un modèle ne soit instancié.

La boucle d’inférence

La boucle d’inférence elle-même est assez simple :

1. Extraire les images des tampons
2. Prétraitement + inférence
3. Attachez les résultats aux métadonnées d’objet de chaque image si les éléments en aval du pipeline sont des plugins Deepstream.

Ci-dessous, vous trouverez l’extrait de code pour la copie zéro à l’aide de DLPack :

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

Prétraitement de l’entrée

Modèles YOLO, lors du passage dans un torch.Tensorattendez-vous à une forme d’entrée fixe (N, 3, 640, 640) selon les documents. Cependant, les cadres se détachent nvstreammux sera quelle que soit la résolution de votre source. L’approche utilisée est le letterboxing : redimensionnez le cadre pour l’adapter aux dimensions cibles tout en préservant les proportions, puis remplissez l’espace restant. L’idée clé ici est que nous pouvons le faire entièrement sur le GPU, sur l’ensemble du lot en même temps, sans jamais toucher à la mémoire du CPU.

Avec l’extraction de trames, le letterboxing, l’inférence et l’inversion de coordonnées, tout se passe sur le GPU en un seul do_transform_ip appel, le plugin se comporte exactement comme nvinfer du point de vue de chaque élément en aval, mais avec toute la flexibilité d’une pile d’inférence Python en dessous.

À partir de là, le reste du pipeline DeepStream prend le relais : nvtracker attribue des identifiants, nvdsosd dessine des superpositions et nvmsgconv sérialise les charges utiles.

Points pratiques à retenir et prochaines étapes

Si vous avez suivi jusqu’ici, vous disposez d’un modèle de travail pour remplacer nvinfer avec votre propre élément d’inférence Python, et plus important encore, vous comprenez pourquoi chaque pièce est telle qu’elle est.

Le schéma se généralise. Tout ce qui est décrit ici : le squelette du plugin, le prétraitement par lots et la pièce jointe des métadonnées est indépendant du modèle. Échanger Ultralytics YOLO contre Le rfdetr de Roboflow est simple et le GStreamer et pyservicemaker l’échafaudage reste identique. Il en va de même pour les architectures plus exotiques : celles de NVIDIA deepstream_reference_apps Le référentiel comprend un exemple fonctionnel d’intégration d’un modèle Vision-Langage via vLLM en utilisant exactement cette approche de plugin, qu’il vaut la peine d’étudier si vous allez au-delà de la détection dans la compréhension vidéo.

Le code complet du plugin est disponible sous forme de L’essentiel de GitHub. Si vous construisez quelque chose par-dessus : un modèle différent, une configuration multi-flux ou une intégration VLM, je serais curieux de savoir comment cela se passe. Bon codage !