使用GStreamer推流至RTSPtoWeb时输出画面呈绿色的问题

问题现象 #

使用Python+OpenCV处理RTSP流（叠加时间戳），通过GStreamer重新推流后：

• VLC播放输出流正常，显示带时间戳的画面
• RTSPtoWeb播放时仅显示绿色屏幕，偶尔可见底部约20%画面，其余部分损坏
• 调整比特率后问题依旧

核心原因分析 #

1. OpenCV帧内存对齐问题：OpenCV的Mat对象默认采用行字节对齐（宽度向上取整到4的倍数），直接调用tobytes()会包含额外填充字节，导致GStreamer解析帧数据时格式错乱。
2. GStreamer Pipeline格式匹配问题：手动指定video/x-raw,format=I420可能存在格式转换不兼容，部分播放器对H.264的SPS/PPS发送频率敏感。
3. 跨线程推送缓冲区风险：直接在OpenCV读取线程推送缓冲区到GStreamer，可能导致线程安全问题。

解决方案 #

1. 修复帧内存对齐问题 #

修改push_frame方法，使用np.ascontiguousarray确保内存连续，去除填充字节：

def push_frame(self, frame: np.ndarray):
    """Push a frame to the RTSP stream"""
    if self.appsrc is None:
        return
    
    try:
        # 确保帧内存连续，去除OpenCV对齐填充
        contiguous_frame = np.ascontiguousarray(frame)
        data = contiguous_frame.tobytes()
        buf = Gst.Buffer.new_allocate(None, len(data), None)
        buf.fill(0, data)
        buf.pts = self.pts
        buf.duration = self.frame_duration
        self.pts += self.frame_duration
        # 通过GLib.idle_add保证在GStreamer主线程推送缓冲区
        GLib.idle_add(lambda: self.appsrc.emit("push-buffer", buf) == Gst.FlowReturn.OK)
    except Exception as e:
        print(f"Push error: {e}")

2. 优化GStreamer Pipeline配置 #

调整Pipeline参数，增强兼容性：

def do_create_element(self, url):
    """Create GStreamer pipeline with appsrc"""
    caps = f"video/x-raw,format=BGR,width={self.width},height={self.height},framerate={self.fps}/1"
    
    pipeline = (
        f"appsrc name=source is-live=true do-timestamp=false format=time caps=\"{caps}\" "
        f"! queue max-size-buffers=8 leaky=downstream "
        f"! videoconvert "
        f"! x264enc tune=zerolatency speed-preset=ultrafast bitrate=2000 key-int-max=60 "
        f"! h264parse config-interval=1 "
        f"! rtph264pay name=pay0 pt=96 config-interval=1 mtu=1400"
    )
    
    return Gst.parse_launch(pipeline)

• 移除手动指定的video/x-raw,format=I420，让videoconvert自动选择最优格式
• 修改queue的leaky为downstream，避免丢帧导致画面错乱
• 增加rtph264pay的mtu=1400，适配网络传输MTU限制

3. 同步帧率与时间戳 #

使用输入源的实际帧率而非固定值，避免时间戳漂移：

# 原代码
# rtsp = RTSPServer(width, height, OUTPUT_FPS, OUTPUT_RTSP_PORT, OUTPUT_RTSP_PATH)
# 修改为
rtsp = RTSPServer(width, height, fps, OUTPUT_RTSP_PORT, OUTPUT_RTSP_PATH)

完整修改后的代码 #

import cv2
import numpy as np
import threading
import sys

try:
    import gi
    gi.require_version('Gst', '1.0')
    gi.require_version('GstRtspServer', '1.0')
    from gi.repository import Gst, GstRtspServer, GLib
    Gst.init(None)
    GST_AVAILABLE = True
except Exception as e:
    print(f"GStreamer not available: {e}")
    GST_AVAILABLE = False
    sys.exit(1)


# ========== CONFIGURATION ==========
INPUT_SOURCE = "rtsp://xxxx/xxx"
OUTPUT_RTSP_PORT = "8554"
OUTPUT_RTSP_PATH = "/live"
OUTPUT_FPS = 25



class GstRTSPFactory(GstRtspServer.RTSPMediaFactory):
    
    def __init__(self, width, height, fps):
        super().__init__()
        self.width = width
        self.height = height
        self.fps = fps
        self.set_shared(True)
        self.appsrc = None
        self.pts = 0
        self.frame_duration = Gst.util_uint64_scale_int(1, Gst.SECOND, int(fps))

    def do_create_element(self, url):
        """Create GStreamer pipeline with appsrc"""
        caps = f"video/x-raw,format=BGR,width={self.width},height={self.height},framerate={self.fps}/1"
        
        pipeline = (
            f"appsrc name=source is-live=true do-timestamp=false format=time caps=\"{caps}\" "
            f"! queue max-size-buffers=8 leaky=downstream "
            f"! videoconvert "
            f"! x264enc tune=zerolatency speed-preset=ultrafast bitrate=2000 key-int-max=60 "
            f"! h264parse config-interval=1 "
            f"! rtph264pay name=pay0 pt=96 config-interval=1 mtu=1400"
        )
        
        return Gst.parse_launch(pipeline)

    def do_configure(self, rtsp_media):
        """Get appsrc element from pipeline"""
        element = rtsp_media.get_element()
        self.appsrc = element.get_child_by_name("source")

    def push_frame(self, frame: np.ndarray):
        """Push a frame to the RTSP stream"""
        if self.appsrc is None:
            return
        
        try:
            # 确保帧内存连续，去除OpenCV对齐填充
            contiguous_frame = np.ascontiguousarray(frame)
            data = contiguous_frame.tobytes()
            buf = Gst.Buffer.new_allocate(None, len(data), None)
            buf.fill(0, data)
            buf.pts = self.pts
            buf.duration = self.frame_duration
            self.pts += self.frame_duration
            # 通过GLib.idle_add保证在GStreamer主线程推送缓冲区
            GLib.idle_add(lambda: self.appsrc.emit("push-buffer", buf) == Gst.FlowReturn.OK)
        except Exception as e:
            print(f"Push error: {e}")


class RTSPServer:
    """Simple RTSP server wrapper"""
    
    def __init__(self, width, height, fps, port, path):
        self.width = width
        self.height = height
        self.fps = fps
        self.port = port
        self.path = "/" + path.lstrip("/")
        self.server = None
        self.factory = None
        self.loop = None
        self.loop_thread = None

    def start(self):
        """Start RTSP server"""
        try:
            self.server = GstRtspServer.RTSPServer()
            self.server.set_service(self.port)
            
            self.factory = GstRTSPFactory(self.width, self.height, self.fps)
            self.factory.connect("media-configure", lambda f, m: f.do_configure(m))
            
            mounts = self.server.get_mount_points()
            mounts.add_factory(self.path, self.factory)
            self.server.attach(None)
            
            # Run GLib mainloop in background
            self.loop = GLib.MainLoop()
            self.loop_thread = threading.Thread(target=self.loop.run, daemon=True)
            self.loop_thread.start()
            
            print(f"RTSP Server started: rtsp://localhost:{self.port}{self.path}")
            return True
        except Exception as e:
            print(f"Failed to start RTSP server: {e}")
            return False

    def write_frame(self, frame):
        """Send frame to RTSP stream"""
        if self.factory:
            self.factory.push_frame(frame)

    def stop(self):
        """Stop RTSP server"""
        if self.loop:
            self.loop.quit()
        if self.loop_thread:
            self.loop_thread.join(timeout=2)


def process_frame(frame):
    import datetime
    
    processed = frame.copy()
    
    # Add timestamp
    timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    cv2.putText(processed, timestamp, (10, 30), 
                cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
    
    
    return processed


def main():
    print("Test Gstream")
    
    # Open video source
    source = INPUT_SOURCE
    if source.isdigit():
        source = int(source)
    
    cap = cv2.VideoCapture(source)
    if not cap.isOpened():
        print(f"Cannot open source: {INPUT_SOURCE}")
        return
    
    # Get video properties
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    fps = cap.get(cv2.CAP_PROP_FPS) or OUTPUT_FPS
    
    print(f"Input: {width}x{height} @ {fps:.1f}fps")
    print(f"Source: {INPUT_SOURCE}")
    
    # Start RTSP server - 使用输入源的实际帧率
    rtsp = RTSPServer(width, height, fps, OUTPUT_RTSP_PORT, OUTPUT_RTSP_PATH)
    if not rtsp.start():
        return
    
    print(f"Streaming...")
    
    frame_count = 0
    try:
        while True:
            ret, frame = cap.read()
            if not ret:
                print("EOServices")
                break
            
            frame_count += 1
            
            # Process frame with OpenCV
            processed = process_frame(frame)
            
            # Send to RTSP stream
            rtsp.write_frame(processed)
            
            
            # Check for exit
            key = cv2.waitKey(1) & 0xFF
            if key == ord('q') or key == 27:
                break
            
            # Print progress every 100 frames
            if frame_count % 100 == 0:
                print(f"Frames processed: {frame_count}")
    
    except KeyboardInterrupt:
        print("\n Interrupted by user")
    
    finally:
        print(f"\n Cleanup... ({frame_count} frames processed)")
        cap.release()
        cv2.destroyAllWindows()
        rtsp.stop()
        print("Done")


if __name__ == "__main__":
    if not GST_AVAILABLE:
        print("Please install GStreamer and python-gi bindings")
        sys.exit(1)
    main()

验证步骤 #

1. 运行修改后的脚本，启动RTSP推流
2. 使用RTSPtoWeb连接输出流，确认画面正常显示时间戳
3. 同时用VLC验证兼容性，确保两种播放器都能正常播放

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