Python3+GStreamer结合OpenCV处理视频卡顿、加速问题求助

我有一段可正常运行的Python3+GStreamer代码，能将视频流发送至服务器与appsink。想借助appsink输出的数据，通过OpenCV完成视频显示与保存，但编写的代码出现显示及流卡顿、保存视频播放加速的问题，求解决建议。

可正常运行的代码 #

import gi
from time import sleep
from datetime import datetime

gi.require_version("Gst", "1.0")
# gi.require_version("GstApp", "1.0")

from gi.repository import Gst #, GstApp

current_datetime = datetime.now()
dt = current_datetime.strftime("%Y-%m-%d_%H-%M-%S")
print(dt)

Gst.init()

pipeRGB =(
    "v4l2src io-mode=4 device=/dev/video0 ! video/x-raw, width=1920, height=1080, framerate=30/1 ! tee name=t "
    "t. ! queue ! nvvidconv ! nvv4l2h264enc ! h264parse ! flvmux ! rtmpsink location='www.google.com live=1' "
    "t. ! queue ! videorate ! video/x-raw, width=1920, height=1080, framerate=1/30 ! queue ! videoconvert ! video/x-raw,format=BGR ! appsink name=sinkRGB"
)

pipelineRGB = Gst.parse_launch(pipeRGB)
print("Parsing RGB Pipeline")
pipelineRGB.set_state(Gst.State.PLAYING)
print("RGB Pipeline playing")

try:
    while True:
        sleep(1)
except KeyboardInterrupt:
    print("""\r
    ============================
    Keyboard Interrupt received
    ============================
    Closing GStreamer Pipeline""")
    pipelineRGB.set_state(Gst.State.NULL)
    pass

print("Finished!")

运行异常的代码 #

import gi
import cv2
import numpy as np
import time
from datetime import datetime


gi.require_version("Gst", "1.0")
from gi.repository import Gst, GLib

current_datetime = datetime.now()
dt = current_datetime.strftime("%Y-%m-%d_%H-%M-%S")
print(dt)

Gst.init()

pipeRGB = (
    "v4l2src io-mode=4 device=/dev/video0 ! video/x-raw, width=1920, height=1080, framerate=30/1 ! tee name=t "
    "t. ! queue ! nvvidconv ! nvv4l2h264enc ! h264parse ! flvmux ! rtmpsink location='rtmp://zkms.waldwaechter.de/live/rgb live=1' "
    "t. ! queue ! videoconvert ! video/x-raw,format=BGR ! appsink name=sinkRGB"
)

pipelineRGB = Gst.parse_launch(pipeRGB)
appsinkRGB = pipelineRGB.get_by_name("sinkRGB")

def on_new_sample(sink):
    sample = sink.emit("pull-sample")
    buf = sample.get_buffer()
    caps = sample.get_caps()
    array = np.ndarray(
        (caps.get_structure(0).get_value('height'),
         caps.get_structure(0).get_value('width'),
         3),
        buffer=buf.extract_dup(0, buf.get_size()),
        dtype=np.uint8)
    return array

appsinkRGB.connect("new-sample", on_new_sample)

actual_fps=30
actual_width=1920
actual_height=1080

date_string= time.strftime("%Y-%m-%d-%H:%M")
gst_writer_str = 'appsrc ! video/x-raw,format=BGR,width=1920,height=1080,framerate=30/1 !  queue ! videoconvert ! video/x-raw,format=BGRx ! nvvidconv ! nvv4l2h264enc maxperf-enable=1 preset-level=4 control-rate=1 bitrate=8000000 ! h264parse ! matroskamux ! filesink location=RGB_' + date_string + '.mp4' 
fourcc = 0 #RAW
out = cv2.VideoWriter(gst_writer_str, cv2.CAP_GSTREAMER, fourcc, actual_fps, (actual_width, actual_height), True)
pipelineRGB.set_state(Gst.State.PLAYING)
while True:
    frame = on_new_sample(appsinkRGB)
    if frame is not None:
        out.write(frame)
        cv2.imshow('RGB Video', frame)
        if cv2.waitKey(1) & 0xFF == ord('q'):
            break
pipelineRGB.set_state(Gst.State.NULL)
out.release()
cv2.destroyAllWindows()

print("Finished!")

解决建议 #

1. 修正appsink信号回调逻辑 #

原代码中on_new_sample作为信号回调返回numpy数组不符合GStreamer规范，会导致管道阻塞或丢帧。回调需返回Gst.FlowReturn枚举值表示处理状态，且应在回调内部完成帧的显示与写入操作：

def on_new_sample(sink):
    sample = sink.emit("pull-sample")
    if not sample:
        return Gst.FlowReturn.ERROR
    
    buf = sample.get_buffer()
    caps = sample.get_caps()
    struct = caps.get_structure(0)
    height = struct.get_value('height')
    width = struct.get_value('width')
    
    # 直接映射缓冲区内存，避免数据拷贝
    success, map_info = buf.map(Gst.MapFlags.READ)
    if not success:
        return Gst.FlowReturn.ERROR
    
    array = np.ndarray((height, width, 3), buffer=map_info.data, dtype=np.uint8)
    # 处理帧
    out.write(array)
    cv2.imshow('RGB Video', array)
    cv2.waitKey(1)
    
    buf.unmap(map_info)
    return Gst.FlowReturn.OK

2. 使用GLib主循环处理管道事件 #

原代码主循环主动调用on_new_sample会和信号回调冲突，需依赖GStreamer的GLib主循环处理异步事件：

# 连接信号后启动管道
appsinkRGB.connect("new-sample", on_new_sample)
pipelineRGB.set_state(Gst.State.PLAYING)

# 运行GLib主循环
loop = GLib.MainLoop()
try:
    loop.run()
except KeyboardInterrupt:
    pass

# 清理资源
pipelineRGB.set_state(Gst.State.NULL)
out.release()
cv2.destroyAllWindows()

3. 优化管道队列与同步设置 #

• 给appsink分支添加带缓存控制的队列，并开启同步，保证帧率稳定：

  t. ! queue leaky=2 max-size-buffers=30 ! videoconvert ! video/x-raw,format=BGR ! appsink name=sinkRGB sync=true
  

• 给写入视频的appsrc添加同步和时间格式参数，避免保存视频加速：

  gst_writer_str = 'appsrc sync=true format=TIME ! video/x-raw,format=BGR,width=1920,height=1080,framerate=30/1 ! queue ! videoconvert ! video/x-raw,format=BGRx ! nvvidconv ! nvv4l2h264enc maxperf-enable=1 preset-level=4 control-rate=1 bitrate=8000000 ! h264parse ! matroskamux ! filesink location=RGB_' + date_string + '.mp4'
  

4. 降低OpenCV显示负载 #

如果原分辨率显示卡顿，可缩小帧后再显示，减少CPU消耗：

resized_frame = cv2.resize(array, (960, 540))
cv2.imshow('RGB Video', resized_frame)

内容的提问来源于stack exchange，提问作者Ihssane Oubari