un canal listo para producción para análisis de video de transmisión múltiple: decodificación acelerada por hardware, seguimiento, visualización en pantalla e intermediación de mensajes, todo conectado a través de GStreamer. Para los modelos de detección estándar exportados a TensorRT, nvinfer se encarga de todo.
Sin embargo, el caso común tiene límites. Los modelos de lenguaje de visión, el posprocesamiento personalizado, los cuadros delimitadores rotados o la necesidad de intercambiar modelos en caliente en tiempo de ejecución son lugares donde las suposiciones de nvinfer fallan. A veces tienes una pila de inferencia de PyTorch madura que tu equipo ha ajustado cuidadosamente y quieres que DeepStream la llame en lugar de volver a implementarla en un archivo de configuración.
Vale la pena señalar que específicamente para los modelos de la familia YOLO, DeepStream-Yolo de Marcos Luciano ya ha realizado un excelente trabajo implementando el posprocesamiento personalizado en C++. Si C++ está sobre la mesa, comience por ahí. Este artículo adopta un ángulo diferente: lograr el mismo resultado completamente en Python, utilizando un complemento GStreamer personalizado con pyservicemaker sin sacrificar el rendimiento.
La idea clave que hace esto posible: a los elementos posteriores como nvtracker, nvdsosd y nvmsgconv no les importa qué elemento produjo metadatos de detección. Escriba correctamente en la estructura de metadatos de DeepStream y el resto del ecosistema funcionará como si nvinfer nunca hubiera estado en escena.
Metadatos de transmisión profunda
Cada búfer que fluye a través de una tubería DeepStream transporta más que datos de píxeles. Desde el momento en que los fotogramas pasan por nvstreammux, cada GstBuffer tiene una estructura NvDsBatchMeta adjunta. La jerarquía es sencilla y se puede encontrar en la documentación oficial.
NvDsBatchMeta ├── NvDsUserMeta (metadatos personalizados a nivel de lote) └── NvDsFrameMeta (uno por flujo de origen) ├── NvDsUserMeta (metadatos personalizados a nivel de marco) └── NvDsObjectMeta (uno por objeto detectado) ├── NvDsClassifierMeta └── NvDsUserMeta (metadatos personalizados a nivel de objeto)
NvDsBatchMeta describe todo el lote. Cada NvDsFrameMeta corresponde a una secuencia de origen y transporta información a nivel de fotograma, como el ID de origen y el número de fotograma. Cada NvDsObjectMeta representa una única detección, lo que significa que cuando nuestro complemento escribe detecciones, escribiremos un NvDsObjectMeta para cada una.
Lo fundamental que hay que entender es que nada de esto es propiedad de nvinfer. Es un contrato de datos compartidos. Cualquier elemento GStreamer en la canalización puede leerlo, escribirle o ambas cosas:
nvtracker lee cuadros delimitadores de objetos y escribe ID de seguimiento. nvdsosd lee cuadros y etiquetas para dibujar superposiciones. nvmsgconv lee toda la estructura para producir cargas útiles de mensajes.
Nuestro complemento personalizado simplemente escribirá las detecciones en esta estructura de la misma manera que lo haría nvinfer y todo lo posterior las recogerá sin modificaciones. Una restricción importante que vale la pena comprender antes de escribir cualquier código: las instancias de NvDsObjectMeta no se pueden construir directamente desde Python. Al intentar crear una instancia de la clase, aparece un mensaje ¡No hay constructor definido! error en tiempo de ejecución.
La razón es arquitectónica. DeepStream administra sus objetos de metadatos a través de grupos de memoria, bloques preasignados que se reciclan entre fotogramas para evitar la sobrecarga de la asignación y desasignación repetidas del montón en una canalización de alto rendimiento. Estos grupos son propiedad de NvDsBatchMeta y se encuentran en el lado C del límite. Los enlaces de Python exponen el acceso a esos grupos, pero deliberadamente no exponen un constructor del lado de Python, porque crear un NvDsObjectMeta fuera del grupo omitiría la administración del ciclo de vida que mantiene predecible el uso de la memoria de DeepStream. La forma correcta de obtener uno es pedírselo al lote: batch_meta.acquire_object_meta(), que le entrega una instancia preasignada del grupo. Cuando el marco finaliza, DeepStream lo devuelve al grupo automáticamente.
El puente Python: pyservicemaker
Para interactuar con los metadatos de DeepStream desde Python, usaremos pyservicemaker, el SDK de Python compatible actual de NVIDIA para DeepStream. La documentación oficial cubre los conceptos básicos de canalizaciones y flujos, pero no llega a mostrar cómo escribir y adjuntar metadatos desde un elemento de inferencia personalizado. Ese es el vacío que llena este artículo.
La abstracción clave es BatchMetadataOperator. Subclasificarlo e implementar handle_metadata(batch_meta) le brinda acceso al NvDsBatchMeta completo para cada búfer que fluye a través de la canalización. A partir de ahí, iterar fotogramas es tan sencillo como utilizar Batch_meta.frame_items y adjuntar un objeto de detección.
pyservicemaker también proporciona un contenedor de búfer alrededor de Gst.Buffer que expone Batch_meta directamente y, lo que es más importante, un método extract(batch_id) que devuelve un identificador de DLPack a la memoria de GPU de cada cuadro. Eso es lo que hace posible la inferencia de copia cero, ya que podemos pasar el cuadro directamente a TensorRT sin tener que salir de la GPU.
En lugar de usar BatchMetadataOperator de forma independiente a través de una sonda, incorporaremos el mismo patrón directamente al método do_transform_ip de nuestro complemento personalizado, que nos brinda control sobre el ciclo de vida, las propiedades y la negociación de límites del elemento junto con el acceso a los metadatos. Pero primero, necesitamos crear ese complemento.
Un complemento Python GStreamer reconocible
GStreamer descubre complementos en tiempo de ejecución escaneando los directorios enumerados en GST_PLUGIN_PATH. Para los complementos de Python específicamente, busca dentro de un subdirectorio python/ dentro de cada una de esas rutas. Eso significa que su complemento es solo un archivo .py colocado en el lugar correcto, sin compilación, sin CMake, sin biblioteca compartida. La desventaja es que el patrón de registro es estricto y hacerlo mal produce fallas silenciosas que son realmente difíciles de depurar.
$GST_PLUGIN_PATH/ └── python/ └── gstexampleplugin.py # tu complemento
Configure GST_PLUGIN_PATH para que apunte al directorio principal y GStreamer encontrará python/gstexampleplugin.py automáticamente en la siguiente ejecución de la canalización.
El esqueleto del complemento
Este es el esqueleto mínimo para un elemento de inferencia de transferencia: recibe buffers de video por lotes, ejecuta la inferencia, adjunta metadatos y pasa el buffer sin modificar.
importar gi gi.require_version('Gst', '1.0') gi.require_version('GstBase', '1.0') desde gi.repository importar Gst, GstBase, GObject importar antorcha desde pyservicemaker importar Buffer GST_PLUGIN_NAME = "gstexampleplugin" Gst.init(None) clase GstExamplePlugin(GstBase.BaseTransform): __gstmetadata__ = ( 'GstExamplePlugin', # nombre 'Filtro/Efecto/Video', # clasificación 'Elemento de inferencia personalizado', # descripción 'Su nombre' # autor ) src_format = Gst.Caps.from_string( "video/x-raw(memoria:NVMM), formato=RGB, " "ancho=(int)[ 1, 2147483647 ], altura=(int)[ 1, 2147483647 ], " "framerate=(fracción)[ 0/1, 2147483647/1 ]" ) fregadero_formato = Gst.Caps.from_string( "video/x-raw(memoria:NVMM), formato=RGB, " "ancho=(int)[ 1, 2147483647 ], altura=(int)[ 1, 2147483647 ], " "framerate=(fracción)[ 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, fregadero_format ) __gsttemplates__ = (src_pad_template, fregadero_pad_template) __gproperties__ = { 'model-engine': ( str, 'Ruta del motor TensorRT', 'Ruta al archivo .engine', '', GObject.ParamFlags.READWRITE ), 'umbral de confianza': ( float, 'Umbral de confianza', 'Confianza mínima para adjuntar una detección', 0.0, 1.0, 0.5, GObject.ParamFlags.READWRITE ), } def __init__(self): super().__init__() self.model_engine = '' self.confidence_threshold = 0.5 self.engine = Ninguno def do_get_property(self, prop): if prop.name == 'model-engine': devuelve self.model_engine elif prop.name == 'confidence-threshold': devuelve self.confidence_threshold def do_set_property(self, prop, value): if prop.name == 'model-engine': self.model_engine = valor elif prop.name == 'confidence-threshold': self.confidence_threshold = value def do_start(self): # Cargue su motor TensorRT aquí self.engine = load_engine(self.model_engine) # Esta función debe implementarse return True def do_transform_ip(self, gst_buffer: Gst.Buffer) -> Gst.FlowReturn: """Transformación in situ: adjunte metadatos, pase el búfer sin cambios.""" buffer = Buffer(gst_buffer) lote_meta = buffer.batch_meta marcos =[]para frame_meta en lote_meta.frame_items: t = torch.utils.dlpack.from_dlpack(buffer.extract(frame_meta.batch_id)) frames.append lote = torch.stack(frames, dim=0) # Ejecute los resultados de inferencia de su modelo = self.engine(batch) # Ahora necesitaremos iterar sobre los resultados para cada cuadro # y adjuntarlo al object_meta en caso de que sea así detección/segmentación # de lo contrario podemos hacerlo como user_meta # El siguiente es un pseudocódigo, que depende de su inferencia para frame_meta en batch_meta.frame_items: para det en los resultados: obj = batch_meta.acquire_object_meta() # Complete el obj con cada detección… frame_meta.append(obj) return Gst.FlowReturn.OK # — Registro — GObject.type_register(GstExamplePlugin) __gstelementfactory__ = (GST_PLUGIN_NAME, Gst.Rank.NONE, GstExamplePlugin)
Algunas cosas que vale la pena señalar sobre este esqueleto:
GstBase.BaseTransform es la clase base adecuada para un filtro local, uno que recibe un búfer, lo modifica (adjuntando metadatos) y lo pasa en sentido descendente. Anulamos do_transform_ip en lugar de do_transform porque no estamos asignando un nuevo búfer de salida.
__gstmetadata__ y __gsttemplates__ no son opcionales. GStreamer no registrará el elemento sin ellos. La cadena de mayúsculas video/x-raw(memory:NVMM) le indica a GStreamer que este elemento funciona con la memoria NVIDIA, que es esencial para permanecer en la GPU en una canalización DeepStream.
__gproperties__ expone el motor de modelo y el umbral de confianza como propiedades GStreamer de primera clase, lo que significa que puede configurarlas desde una línea de comando gst-launch o desde el código de canalización de Python sin tocar el código fuente.
Las dos últimas líneas son necesarias para el registro: GObject.type_register le informa al sistema de tipos GObject sobre la clase y __gstelementfactory__ le dice a GStreamer qué nombre de elemento exponer y qué clase crear una instancia.
Verificando el complemento. Una vez que el archivo esté en su lugar y el caché esté limpio, verifique el registro con:
GST_PLUGIN_PATH=/ruta/a/sus/complementos gst-inspect-1.0 gstexampleplugin
Debería ver los metadatos del elemento, las plantillas de pad y ambas propiedades en la lista. Si los ve, GStreamer conoce su complemento y está listo para colocarlo en una canalización.
Ejemplo de inferencia de un extremo a otro con ultralíticos
Con el esqueleto del complemento en su lugar, es hora de completar la lógica de inferencia. El código de trabajo completo está disponible como GitHub Gist. Una vez que lo tenga reconocible, podrá inspeccionarlo como lo hicimos antes o iniciar la canalización. Aquí hay un ejemplo simple que simplemente realiza inferencia y muestra los fps:
gst-launch-1.0 -v nvstreammux nombre = m ancho = 1280 alto = 720 tamaño de lote = 1 lotes-push-timeout = 33000! nvvideoconvert nvbuf-memory-type=0 ! 'vídeo/x-raw(memoria:NVMM), formato=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
Inspeccionando el código
Problema de compatibilidad
Si está leyendo el código, es posible que se haya dado cuenta de que estamos anulando el objeto tupla, pero solo dentro del módulo ultralytics.nn.backends.tensorrt, ya que es donde está el problema. Existe un caso de compatibilidad conocido entre los enlaces de TensorRT Python y el marco contenedor GStreamer Python (PyGObject) que bloqueará su canalización con el infame mensaje "Error de segmentación (núcleo volcado)". Por eso fue necesario crear este fragmento de código que nos ayuda a preservar el comportamiento previsto:
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 = tupla_segura
Esto reemplaza la referencia de tupla dentro del espacio de nombres del backend de Ultralytics en tiempo de ejecución con una versión que recurre al acceso basado en índices para objetos Dims, dejando todo lo demás intacto. No es elegante, pero es quirúrgico y debe ocurrir en el momento de la importación, antes de que se cree una instancia de cualquier modelo.
El bucle de inferencia
El ciclo de inferencia en sí es bastante sencillo:
Extraiga los fotogramas de los buffers Preproceso + inferencia Adjunte los resultados a los metadatos del objeto de cada fotograma si los elementos posteriores de la canalización son complementos profundos.
A continuación, se encuentra el fragmento de código para copia cero usando DLPack:
marcos =[]para frame_meta en lote_meta.frame_items: t = torch.utils.dlpack.from_dlpack(buffer.extract(frame_meta.batch_id)) frames.append lote = torch.stack(frames, dim=0)
Preprocesamiento de la entrada
Los modelos YOLO, al pasar por un soplete.Tensor, esperan una forma de entrada fija (N, 3, 640, 640) según la documentación. Sin embargo, los fotogramas que salen de nvstreammux tendrán cualquier resolución que tenga su fuente. El enfoque que se utiliza es el formato letterboxing: escale el marco para que se ajuste a las dimensiones objetivo conservando la relación de aspecto, luego rellene el espacio restante. La idea clave aquí es que podemos hacer esto completamente en la GPU, en todo el lote a la vez, sin tocar la memoria de la CPU.
Con la extracción de fotogramas, el formato letterboxing, la inferencia y la inversión de coordenadas que ocurren en la GPU dentro de una sola llamada do_transform_ip, el complemento se comporta exactamente como nvinfer desde la perspectiva de cada elemento posterior, pero con la total flexibilidad de una pila de inferencia de Python debajo.
A partir de aquí, el resto del proceso de DeepStream toma el control: nvtracker asigna ID, nvdsosd dibuja superposiciones y nvmsgconv serializa cargas útiles.
Conclusiones prácticas y próximos pasos
Si ha seguido hasta aquí, tendrá un patrón de trabajo para reemplazar nvinfer con su propio elemento de inferencia de Python y, lo que es más importante, comprenderá por qué cada pieza es como es.
El patrón se generaliza. Todo lo que se describe aquí: el esqueleto del complemento, el preprocesamiento por lotes y el archivo adjunto de metadatos es independiente del modelo. Cambiar Ultralytics YOLO por rfdetr de Roboflow es sencillo y el andamiaje de GStreamer y pyservicemaker sigue siendo idéntico. Lo mismo se aplica a arquitecturas más exóticas: el propio repositorio deepstream_reference_apps de NVIDIA incluye un ejemplo práctico de integración de un modelo de visión-lenguaje a través de vLLM usando exactamente este enfoque de complemento, que vale la pena estudiar si está avanzando más allá de la detección hacia la comprensión del video.
El código completo del complemento está disponible como GitHub Gist. Si construye algo encima: un modelo diferente, una configuración de transmisión múltiple o una integración VLM, tendría curiosidad por saber cómo funciona. ¡Feliz codificación!