如何将OpenCV捕获并处理后的ZED相机流推送至RTSP服务器？

我正在用OpenCV捕获ZED相机的流，并用cv2.imshow显示。因为需要在推送RTSP流前从流中提取数据，所以必须通过OpenCV处理，但不知道如何将处理后的OpenCV流推送为RTSP流。

现有ZED相机OpenCV流捕获代码 #

def main() :

    # Create a ZED camera object
    zed = sl.Camera()

    # Set configuration parameters
    input_type = sl.InputType()
    if len(sys.argv) >= 2 :
        input_type.set_from_svo_file(sys.argv[1])
    init = sl.InitParameters(input_t=input_type)
    init.camera_resolution = sl.RESOLUTION.HD1080
    init.depth_mode = sl.DEPTH_MODE.ULTRA
    init.coordinate_units = sl.UNIT.METER

    # Open the camera
    err = zed.open(init)
    if err != sl.ERROR_CODE.SUCCESS :
        print(repr(err))
        zed.close()
        exit(1)


    # Set runtime parameters after opening the camera
    runtime = sl.RuntimeParameters()
    runtime.sensing_mode = sl.SENSING_MODE.FILL
    # Setting the depth confidence parameters
    runtime.confidence_threshold = 100
    runtime.textureness_confidence_threshold = 100

    # Prepare new image size to retrieve half-resolution images
    image_size = zed.get_camera_information().camera_resolution
    image_size.width = image_size.width /2
    image_size.height = image_size.height /2

    # Declare your sl.Mat matrices
    image_zed = sl.Mat(image_size.width, image_size.height, sl.MAT_TYPE.U8_C4)
    point_cloud = sl.Mat()


    key = ' '
    while key != 113 :
        err = zed.grab(runtime)
        def click_event (event, x, y, flags, param):
            if event == cv2.EVENT_LBUTTONDOWN:
                # Retrieve the left image, depth image in the half-resolution
                zed.retrieve_image(image_zed, sl.VIEW.LEFT, sl.MEM.CPU, image_size)
                # Retrieve the RGBA point cloud in half resolution
                zed.retrieve_measure(point_cloud, sl.MEASURE.XYZRGBA, sl.MEM.CPU, image_size)
            
                err, point_cloud_value = point_cloud.get_value(x, y)

                distance = math.sqrt(point_cloud_value[0] * point_cloud_value[0] +
                                     point_cloud_value[1] * point_cloud_value[1] +
                                     point_cloud_value[2] * point_cloud_value[2])
                                                     
                print("Distance to Camera at ({}, {}) (image center): {:1.3} m".format(x, y,  distance))
        
        if err == sl.ERROR_CODE.SUCCESS :
            # Retrieve the left image, depth image in the half-resolution
            zed.retrieve_image(image_zed, sl.VIEW.LEFT, sl.MEM.CPU, image_size)
            # Retrieve the RGBA point cloud in half resolution
            zed.retrieve_measure(point_cloud, sl.MEASURE.XYZRGBA, sl.MEM.CPU, image_size)

            # To recover data from sl.Mat to use it with opencv, use the get_data() method
            # It returns a numpy array that can be used as a matrix with opencv
            image_ocv = image_zed.get_data()
        
            cv2.imshow("Image", image_ocv)     
            cv2.setMouseCallback('Image', click_event)
        
            key = cv2.waitKey(10)


    cv2.destroyAllWindows()
    zed.close()

    print("\nFINISH")

if __name__ == "__main__":
    main()

现有RTSP服务器代码 #

import gi
gi.require_version('Gst','1.0')
gi.require_version('GstVideo','1.0')
gi.require_version('GstRtspServer','1.0')
from gi.repository import GObject, Gst, GstVideo, GstRtspServer

Gst.init(None)

mainloop = GObject.MainLoop()
server = GstRtspServer.RTSPServer()
mounts = server.get_mount_points()

factory = GstRtspServer.RTSPMediaFactory()
factory.set_launch('( zedsrc stream-type=0 ! videoconvert ! videoscale ! video/x-raw,format=YUY2,width=1280,height=720,framerate=30/1 ! nvvidconv ! nvv4l2h264enc insert-sps-pps=1 idrinterval=1 insert-vui=1 ! rtph264pay name=pay0 pt=96 )')
#factory.set_launch('(v4l2src device=/dev/video0 io-mode=2 ! image/jpeg,width=1280,height=720,framerate=30/1 ! nvjpegdec ! video/x-raw ! nvvidconv ! nvv4l2h264enc ! rtph264pay name=pay0 pt=96)')

mounts.add_factory('/test', factory)
server.attach(None)

print('stream ready at rtsp://127.0.0.1:8554/test')
mainloop.run()

解决方案：整合OpenCV处理与RTSP推送 #

核心思路是用GStreamer的appsrc元素作为RTSP流的数据源，在OpenCV捕获并处理每帧后，将帧转换为GStreamer兼容格式，再通过appsrc推送。

完整整合代码 #

import sys
import math
import cv2
import sl
import gi

# 初始化GStreamer相关库
gi.require_version('Gst','1.0')
gi.require_version('GstVideo','1.0')
gi.require_version('GstRtspServer','1.0')
from gi.repository import GObject, Gst, GstVideo, GstRtspServer

# 全局变量用于传递OpenCV帧，线程锁保证安全
shared_frame = None
frame_lock = GObject.RecMutex()

class CustomRTSPMediaFactory(GstRtspServer.RTSPMediaFactory):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.pipeline = None
        self.appsrc = None

    def do_create_element(self, url):
        # 构建包含appsrc的GStreamer管道，匹配OpenCV输出的帧格式
        pipeline_str = (
            'appsrc name=source is-live=true format=GST_FORMAT_TIME '
            'caps=video/x-raw,format=RGBA,width=960,height=540,framerate=30/1 '
            '! videoconvert ! video/x-raw,format=YUY2 '
            '! nvvidconv ! nvv4l2h264enc insert-sps-pps=1 idrinterval=30 insert-vui=1 '
            '! rtph264pay name=pay0 pt=96'
        )
        self.pipeline = Gst.parse_launch(pipeline_str)
        self.appsrc = self.pipeline.get_by_name('source')
        
        # 回调函数：当appsrc需要数据时，从共享变量中获取帧
        def need_data(src, length):
            global shared_frame, frame_lock
            with frame_lock:
                if shared_frame is not None:
                    # 将OpenCV的numpy数组转换为GStreamer缓冲区
                    buf = Gst.Buffer.new_wrapped(shared_frame.tobytes())
                    # 设置缓冲区时间戳，模拟30fps帧率
                    pts = Gst.util_uint64_scale(src.get_current_running_time(), 1, Gst.SECOND)
                    buf.pts = pts
                    buf.dts = pts
                    buf.duration = Gst.SECOND // 30
                    # 推送缓冲区到appsrc
                    src.emit('push-buffer', buf)
        
        self.appsrc.connect('need-data', need_data)
        return self.pipeline

def main():
    global shared_frame, frame_lock

    # 初始化GStreamer和多线程支持
    Gst.init(None)
    GObject.threads_init()
    mainloop = GObject.MainLoop()

    # 启动RTSP服务器
    server = GstRtspServer.RTSPServer()
    mounts = server.get_mount_points()
    factory = CustomRTSPMediaFactory()
    factory.set_shared(True)  # 允许多个客户端同时连接
    mounts.add_factory('/test', factory)
    server.attach(None)
    print('RTSP流已就绪：rtsp://127.0.0.1:8554/test')

    # 启动RTSP主循环线程，避免阻塞相机捕获逻辑
    import threading
    rtsp_thread = threading.Thread(target=mainloop.run)
    rtsp_thread.daemon = True
    rtsp_thread.start()

    # 初始化ZED相机
    zed = sl.Camera()
    input_type = sl.InputType()
    if len(sys.argv) >= 2:
        input_type.set_from_svo_file(sys.argv[1])
    init = sl.InitParameters(input_t=input_type)
    init.camera_resolution = sl.RESOLUTION.HD1080
    init.depth_mode = sl.DEPTH_MODE.ULTRA
    init.coordinate_units = sl.UNIT.METER

    err = zed.open(init)
    if err != sl.ERROR_CODE.SUCCESS:
        print(repr(err))
        zed.close()
        exit(1)

    runtime = sl.RuntimeParameters()
    runtime.sensing_mode = sl.SENSING_MODE.FILL
    runtime.confidence_threshold = 100
    runtime.textureness_confidence_threshold = 100

    # 设置半分辨率输出（HD1080 -> 960x540）
    image_size = zed.get_camera_information().camera_resolution
    image_size.width = image_size.width // 2
    image_size.height = image_size.height // 2

    image_zed = sl.Mat(image_size.width, image_size.height, sl.MAT_TYPE.U8_C4)
    point_cloud = sl.Mat()

    # 鼠标点击事件处理函数
    def click_event(event, x, y, flags, param):
        if event == cv2.EVENT_LBUTTONDOWN:
            err, point_cloud_value = point_cloud.get_value(x, y)
            distance = math.sqrt(point_cloud_value[0]**2 + point_cloud_value[1]**2 + point_cloud_value[2]**2)
            print(f"坐标({x}, {y})到相机的距离: {distance:.3f} m")

    cv2.namedWindow("Image")
    cv2.setMouseCallback("Image", click_event)

    key = -1
    while key != 113:  # 按'q'键退出程序
        err = zed.grab(runtime)
        if err == sl.ERROR_CODE.SUCCESS:
            # 获取左目图像和点云数据
            zed.retrieve_image(image_zed, sl.VIEW.LEFT, sl.MEM.CPU, image_size)
            zed.retrieve_measure(point_cloud, sl.MEASURE.XYZRGBA, sl.MEM.CPU, image_size)
            
            # 转换为OpenCV兼容格式
            image_ocv = image_zed.get_data()

            # 在这里添加你的自定义OpenCV处理逻辑
            # 示例：在图像上叠加文本
            # cv2.putText(image_ocv, "RTSP Streaming", (10,30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0,255,0), 2)

            # 更新共享帧，供RTSP线程推送
            with frame_lock:
                shared_frame = image_ocv.copy()

            # 显示处理后的图像
            cv2.imshow("Image", image_ocv)

        key = cv2.waitKey(10)

    # 清理资源
    cv2.destroyAllWindows()
    zed.close()
    mainloop.quit()
    rtsp_thread.join()
    print("\n程序结束")

if __name__ == "__main__":
    main()

关键说明 #

• 自定义RTSP工厂：继承GstRtspServer.RTSPMediaFactory，构建包含appsrc的管道，通过回调函数获取OpenCV处理后的帧。
• 线程安全：用GObject.RecMutex保证共享帧的读写安全，避免多线程冲突。
• 格式匹配：appsrc的caps参数严格匹配OpenCV输出的RGBA格式、960x540分辨率和30fps帧率，避免格式错误。
• 硬件加速：保留原代码中的NVIDIA硬件编码组件，提升流推送性能。

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