Приложения DeepStream на Python в настраиваемом приложении

Question

По этой ссылке мы можем получить доступ к буферу конвейера gstreamer и преобразовать буферы кадров в массив numpy, я хочу знать, как я могу получить доступ к буферам кадров в памяти графического процессора, а затем передать их в свой пользовательский процессор без преобразования кадров в массив numpy.

У нас есть два решения для использования декодера глубокого потока (более эффективный способ, чем opencv+gstreamer): один способ - нам нужно написать собственный элемент обработки и зарегистрировать в gstreamer, а затем поместить пользовательский элемент в конвейер, а затем выполнить обработку в буфере кадров. этот способ хорош, но нужно писать и знать программирование gstreamer. этот путь - тот же путь глубокого потока. второй способ - мы используем только декодированные кадры из этой ссылки, а затем передаем кадры в специальные процессоры. по этой части у меня два вопроса:

1- Цикл gstreamer такой же, как цикл программирования asyncio?

2- Как вы знаете, если мы добавим дополнительную операцию в функцию проверки площадки, это приведет к снижению производительности, но я хочу знать, можно ли поместить кадры в функцию проверки площадки и выполнить loop.create_task(process(frame)) как асинхронный? по этой причине мы здесь не дожидаемся выполнения обработки. нравится:

def tiler_sink_pad_buffer_probe(pad,info,u_data):
    ....
    ### capture the frames in GPU buffer without converting into numpy
    loop.create_task(process(frame))
    ....
    return Gst.PadProbeReturn.OK

Answer 1

1. Цикл Yeap gstreamer в Python — это asyncio

2. Ну, вы можете сделать это, как я (плохой способ, создавая глобальные переменные)

            ws = None
    loopIO = None
   
    def tiler_sink_pad_buffer_probe(pad,info,u_data):
        global ws
        global loopIO
        ....
        ### capture the frames in GPU buffer converting into numpy
   
        if ws and loopIO:
            _, jpeg_frame = cv2.imencode('.jpg', frame_image)
            str_pic = jpeg_frame.tobytes()
            asyncio.run_coroutine_threadsafe(ws.send(str_pic), loopIO)
        ....
        return Gst.PadProbeReturn.OK
   
    if __name__ == '__main__':
        start_server = websockets.serve(consumer_handler, 'localhost', 8765)
        loopIO = asyncio.get_event_loop()
        loopIO.run_until_complete(start_server)
        wst = threading.Thread(target=asyncio.get_event_loop().run_forever)
        wst.daemon = True
        wst.start()
        sys.exit(main(sys.argv))