Блог

У меня есть конвейер GStreamer с appsink

 filesrc location=test.mp4 ! decodebin ! video/x-raw ! queue max-size-bytes=0 max-size-time=100000000 ! appsink name=appSink sync=false max-buffers=1 drop=false
  

Я извлекаю образец из appsink, затем получаю буфер, сопоставляю его только для чтения и сохраняю информацию о карте (для последующего доступа к необработанной памяти).

 sample: Gst.Sample = self.__sink.pull_sample()
self.__buffer: Gst.Buffer = sample.get_buffer()
self.__buffer_map: Gst.MapInfo = self.__buffer.map(Gst.MapFlags.READ)
  

Затем я хотел бы использовать те же данные (без их копирования) в конвейере вывода, например:

 appsrc name=appSrc block=true ! video/x-raw,format=(string)NV12,width=1920,height=1080,framerate=30/1 ! videoconvert ! ximagesink
  

Для этого это лучшее, что я мог придумать:

 shared_buffer_memory = self.__buffer.get_all_memory()
buf = Gst.Buffer.new()
buf.insert_memory(-1, shared_buffer_memory)
self.__src.push_buffer(buf)
#sleep(0.05) 
  

Если я сделаю это, я правильно увижу первый кадр, но все остальные кадры будут зелеными. Это нормально, потому что базовая память buf выходит за рамки и освобождается тем временем. Если я помещаю sleep(0.05) после push_buffer, он отображает все кадры, но я думаю, что память за буфером освобождается дважды:

 ** (python3.9:6745): CRITICAL **: 09:23:54.645: gst_vaapi_image_unmap: assertion 'image != NULL' failed
  

Я хотел бы получить фреймы из appsink, а затем отправить их в выходные конвейеры без копирования. Каков наилучший подход для этого? (в моем случае один конвейер не подходит).

Ответ не мой, я получил его по IRC-каналу #gstreamer:

В документации говорится следующее:

AppSrc.push_buffer(buffer): добавляет буфер в очередь буферов, которые элемент appsrc отправит в исходную панель. Эта функция становится владельцем буфера.

Это было то, что ввело меня в заблуждение. В Python нужно легко передавать один и тот же буфер нескольким приложениям.

Похоже, что это поток, который вы хотите завершить, прежде чем переходить к коду, например:

 from threading import Thread
from time import sleep

def a():
    print('start')
    sleep(1)
    print('middle')
    sleep(1)
    print('end')
c = Thread(target=a)
print('Hello')
c.start()
print('Bye')
sleep(3)
  

Вывод:

 Hello
startBye

middle
end
  

Скажем c , это поток, вы можете найти способ добавить c.join() в свой код определенную строку, чтобы программа знала, что нужно дождаться завершения процесса, прежде чем продолжить программу:

 from threading import Thread
from time import sleep

def a():
    print('start')
    sleep(1)
    print('middle')
    sleep(1)
    print('end')
c = Thread(target=a)
print('Hello')
c.start()
a.join()
print('Bye')
sleep(3)
  

Вывод:

 Hello
start
middle
end
Bye
  

Обновить

Глядя на структуру вашего кода:

 shared_buffer_memory = self.__buffer.get_all_memory()
buf = Gst.Buffer.new()
buf.insert_memory(-1, shared_buffer_memory)
self.__src.push_buffer(buf)
#sleep(0.05) 
  

Вы говорите, что он работает sleep(0.05) , но без него он не работает должным образом. Причина, по которой это может произойти, связана с потоком. Для того, чтобы это работало, вам нужно отменить действие потока, добавив .join() что-нибудь в свой код.