使用GStreamer appsrc管道无法保存图像问题求助

需求为将OpenCV VideoCapture读取的帧保存，操作逻辑是把NumPy格式的帧转换为GStreamer buffer后推入appsrc管道，但执行后无图像文件生成。已确认帧被正常读取并转换为buffer，但调用appsrc.emit("push-buffer", buffer)后无输出结果。

import gi
import cv2

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

# Initialize GStreamer
Gst.init(None)

# Set up GStreamer pipeline
# pipeline = Gst.parse_launch("mfvideosrc device-index=0 ! videoconvert ! video/x-raw,format=BGR ! "
#                             "jpegenc ! multifilesink location=frame//frame_%05d.jpg")

pipeline = Gst.parse_launch("appsrc name=source ! videoconvert ! video/x-raw,format=BGR ! "
                            "jpegenc ! multifilesink location=frame//frame_%05d.jpg")

# Get the appsrc element from the pipeline
appsrc = pipeline.get_by_name("source")

# Set the appsrc format and properties
# appsrc.set_property("format", Gst.Format.TIME)
# appsrc.set_property("is-live", True)

# Set the video format (width, height, and framerate) according to your requirements
video_width = 1280
video_height = 720
video_framerate = 30

# Create a VideoCapture object
video_capture = cv2.VideoCapture(0)  # Replace with your video file path or device index if using a camera

# Set the desired video properties
video_capture.set(cv2.CAP_PROP_FRAME_WIDTH, video_width)
video_capture.set(cv2.CAP_PROP_FRAME_HEIGHT, video_height)
video_capture.set(cv2.CAP_PROP_FPS, video_framerate)

# Start the GStreamer pipeline
pipeline.set_state(Gst.State.PLAYING)

# # Run the main loop
# loop = GLib.MainLoop()
# try:
#     loop.run()
# except KeyboardInterrupt:
#     pass


if not video_capture.isOpened():
    print('Failed to open camera')
    exit(-1)

# Main loop to read frames from VideoCapture and save them

try:
    frame_count = 0
    while True:
        # Read frame from VideoCapture
        ret, frame = video_capture.read()

        if not ret:
            break

        # # Convert the frame to RGB if needed
        # frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)

        print(frame.shape)
        # Convert the frame to a GStreamer buffer
        buffer = Gst.Buffer.new_wrapped(frame.tobytes())

        print(buffer.get_size())

        # # Set the buffer timestamp
        # buffer.pts = buffer.dts = frame_count * (1e9 / video_framerate)

        # Push the buffer into the appsrc element
        appsrc.emit("push-buffer", buffer)

except KeyboardInterrupt:
    pass

# Stop the GStreamer pipeline
pipeline.set_state(Gst.State.NULL)

原代码存在几个核心问题，导致管道无法正常处理并保存帧，修复点如下：

• 启动GLib主循环：GStreamer依赖GLib消息循环处理管道事件，原代码注释掉主循环后，管道无法处理推送的buffer数据
• 为appsrc设置caps：必须明确告知管道输入视频的格式、分辨率、帧率，否则下游元素无法识别输入数据
• 配置appsrc核心属性：设置format为Gst.Format.TIME、is-live为True，适配实时流场景
• 设置buffer时间戳：为每个buffer添加pts/dts和duration参数，确保管道时序逻辑正常

修改后的完整代码 #

import gi
import cv2

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

# Initialize GStreamer
Gst.init(None)

# Set up GStreamer pipeline
pipeline = Gst.parse_launch("appsrc name=source ! videoconvert ! video/x-raw,format=BGR ! "
                            "jpegenc ! multifilesink location=frame//frame_%05d.jpg")

# Get the appsrc element from the pipeline
appsrc = pipeline.get_by_name("source")

# Configure appsrc properties
appsrc.set_property("format", Gst.Format.TIME)
appsrc.set_property("is-live", True)

# Set video parameters
video_width = 1280
video_height = 720
video_framerate = 30
frame_duration = int(1e9 / video_framerate)  # Nanoseconds per frame

# Create caps for appsrc
caps = Gst.Caps.from_string(f"video/x-raw,format=BGR,width={video_width},height={video_height},framerate={video_framerate}/1")
appsrc.set_property("caps", caps)

# Create VideoCapture object
video_capture = cv2.VideoCapture(0)
video_capture.set(cv2.CAP_PROP_FRAME_WIDTH, video_width)
video_capture.set(cv2.CAP_PROP_FRAME_HEIGHT, video_height)
video_capture.set(cv2.CAP_PROP_FPS, video_framerate)

if not video_capture.isOpened():
    print('Failed to open camera')
    exit(-1)

# Start pipeline
pipeline.set_state(Gst.State.PLAYING)

# Frame processing function
def push_frame():
    global frame_count
    ret, frame = video_capture.read()
    if not ret:
        video_capture.release()
        pipeline.set_state(Gst.State.NULL)
        loop.quit()
        return False

    # Convert frame to GStreamer buffer
    buffer = Gst.Buffer.new_wrapped(frame.tobytes())
    # Set timestamps
    buffer.pts = buffer.dts = frame_count * frame_duration
    buffer.duration = frame_duration

    # Push buffer to appsrc
    ret_val = appsrc.emit("push-buffer", buffer)
    if ret_val != Gst.FlowReturn.OK:
        print(f"Push buffer failed: {ret_val}")
        return False

    frame_count += 1
    return True

frame_count = 0
# Add timeout to push frames at specified framerate
GLib.timeout_add(int(1000 / video_framerate), push_frame)

# Run main loop
loop = GLib.MainLoop()
try:
    loop.run()
except KeyboardInterrupt:
    video_capture.release()
    pipeline.set_state(Gst.State.NULL)
    loop.quit()

内容的提问来源于stack exchange，提问作者Uvais Karni