基于OpenCV与GStreamer从RTSP流提取时间戳的技术问询

我正在处理海康威视摄像机的RTSP流，该摄像机已通过NTP同步内部时间。我尝试使用OpenCV和GStreamer从视频帧中提取时间戳，目标是获取流中的NTP时间戳（或任何可靠时间戳）并叠加到帧上，但不确定当前提取的时间戳是否正确。

以下是我的实现代码：

import cv2
import multiprocessing as mp
from datetime import datetime
import gi

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


def process_stream(stream_type, rtsp_url):
    if stream_type == "gstreamer":
        gst_pipeline = (
            f'rtspsrc location="{rtsp_url}" latency=0 '
            'do-ptp-sync=true do-timestamp=true ! '
            'rtph264depay ! h264parse ! avdec_h264 ! '
            'videoconvert ! timeoverlay '
            'time-mode="rtc-time" '
            'halign=left valign=bottom '
            'font-desc="Sans, 10" ! '
            'appsink drop=true sync=false'
        )
        cap = cv2.VideoCapture(gst_pipeline, cv2.CAP_GSTREAMER)
        window_name = "RTSP Stream (Camera Time)"
    else:
        cap = cv2.VideoCapture(rtsp_url)
        window_name = "RTSP Stream (Regular)"

    if not cap.isOpened():
        print(f"Error: Could not open {stream_type} RTSP stream.")
        return

    cv2.namedWindow(window_name, cv2.WINDOW_NORMAL)

    while True:
        ret, frame = cap.read()
        if not ret:
            print(f"Error: Failed to receive frame from {stream_type} stream.")
            break

        if stream_type == "gstreamer":
            ntp_timestamp = cap.get(cv2.CAP_PROP_POS_MSEC)
            if ntp_timestamp > 0:
                cv2.putText(frame, f"NTP Timestamp: {ntp_timestamp:.2f}", (10, 100), 
                            cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2, cv2.LINE_AA)

        frame = cv2.resize(frame, (960, 540))
        cv2.imshow(window_name, frame)
        if cv2.waitKey(1) & 0xFF == ord("q"):
            break

    cap.release()
    cv2.destroyWindow(window_name)


def main():
    rtsp_url = "rtsp://admin:@192.168.1.11:554"  # Hikvision RTSP Stream
    gst_process = mp.Process(target=process_stream, args=("gstreamer", rtsp_url))
    gst_process.start()
    gst_process.join()

if __name__ == "__main__":
    mp.set_start_method('spawn')  # Required for Windows compatibility
    main()

我的问题：

1. 使用cap.get(cv2.CAP_PROP_POS_MSEC)提取NTP时间戳是否正确？
2. 若不正确，如何从RTSP流中可靠提取时间戳？
3. 是否有更好的基于GStreamer的方法来获取并显示时间戳？

1. cap.get(cv2.CAP_PROP_POS_MSEC)是否能提取NTP时间戳？ #

不正确。cv2.CAP_PROP_POS_MSEC返回的是从本地启动流到当前帧的播放时长（毫秒级相对时间），和摄像机端的NTP同步时间没有直接关联，无法作为全局可靠时间戳使用。

2. 如何从RTSP流中可靠提取时间戳？ #

要获取摄像机同步后的NTP时间，可从以下两种方向入手：

方法1：解析GStreamer缓冲区的PTS时间 #

启用rtspsrc的do-ptp-sync=true和do-timestamp=true后，GStreamer会为每个帧缓冲区打上经过PTP/NTP同步的PTS（Presentation Time Stamp），这个时间戳对应摄像机的采集时间（若本地系统也做了NTP同步，会和摄像机时间一致）。

你可以直接通过GStreamer的appsink获取缓冲区的PTS，转换为UTC时间：

import cv2
import gi
from datetime import datetime
gi.require_version('Gst', '1.0')
gi.require_version('GstVideo', '1.0')
from gi.repository import Gst, GstVideo

Gst.init(None)

rtsp_url = "rtsp://admin:@192.168.1.11:554"
pipeline_str = (
    f'rtspsrc location="{rtsp_url}" latency=0 do-ptp-sync=true do-timestamp=true ! '
    'rtph264depay ! h264parse ! avdec_h264 ! videoconvert ! '
    'appsink emit-signals=true drop=true sync=false name=sink'
)

pipeline = Gst.parse_launch(pipeline_str)
sink = pipeline.get_by_name("sink")

def on_new_sample(sink):
    sample = sink.emit("pull-sample")
    buffer = sample.get_buffer()
    pts = buffer.get_pts()
    
    if pts != Gst.CLOCK_TIME_NONE:
        # 转换为毫秒，再转为UTC时间
        pts_ms = pts / 1_000_000
        utc_time = datetime.utcfromtimestamp(pts_ms / 1000)
        time_str = utc_time.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
        
        # 转换为OpenCV帧进行处理
        caps = sample.get_caps()
        frame_info = GstVideo.VideoFrameInfo.new(caps)
        frame = GstVideo.VideoFrame.new(frame_info)
        frame.map(buffer, Gst.MapFlags.READ)
        # 转换为BGR格式（OpenCV默认）
        cv_frame = frame.to_ndarray()[:, :, ::-1]
        frame.unmap()
        
        # 叠加时间戳到帧上
        cv2.putText(cv_frame, f"NTP Time: {time_str}", 
                    (10, 100), cv2.FONT_HERSHEY_SIMPLEX, 1, (0,255,0), 2)
        cv2.imshow("RTSP Stream", cv_frame)
        if cv2.waitKey(1) & 0xFF == ord('q'):
            pipeline.set_state(Gst.State.NULL)
            cv2.destroyAllWindows()
            exit()
    
    return Gst.FlowReturn.OK

sink.connect("new-sample", on_new_sample)
pipeline.set_state(Gst.State.PLAYING)

# 启动GStreamer主循环
loop = Gst.MainLoop()
loop.run()

方法2：读取摄像机元数据或SDK时间 #

• OSD提取：启用摄像机的OSD时间叠加，直接从帧中识别（但依赖图像识别，可靠性略低）
• SDK查询：使用海康威视官方Python SDK（如hikvisionapi）直接调用接口获取摄像机当前NTP时间，再和帧的采集顺序关联

3. 更好的GStreamer时间戳显示方法 #

方法1：用timeoverlay直接显示缓冲区时间 #

修改你现有Pipeline中的timeoverlay配置，将time-mode改为buffer-time，这样会直接显示缓冲区的PTS时间（即摄像机同步后的时间），无需额外代码：

videoconvert ! timeoverlay time-mode="buffer-time" halign=left valign=bottom font-desc="Sans, 10" ! appsink drop=true sync=false

这个方法性能最优，无需Python层处理，完全由GStreamer内核完成。

方法2：自定义textoverlay叠加动态时间 #

如果需要自定义时间格式（如本地时区、额外信息），可以使用textoverlay元素结合摄像机SDK获取的时间，动态更新叠加文本：

import cv2
import gi
import threading
from datetime import datetime
gi.require_version('Gst', '1.0')
from gi.repository import Gst

Gst.init(None)

rtsp_url = "rtsp://admin:@192.168.1.11:554"
pipeline_str = (
    f'rtspsrc location="{rtsp_url}" latency=0 do-ptp-sync=true do-timestamp=true ! '
    'rtph264depay ! h264parse ! avdec_h264 ! videoconvert ! '
    'textoverlay name=overlay halign=left valign=bottom font-desc="Sans, 10" ! '
    'appsink emit-signals=true drop=true sync=false name=sink'
)

pipeline = Gst.parse_launch(pipeline_str)
overlay = pipeline.get_by_name("overlay")
sink = pipeline.get_by_name("sink")

# 模拟通过SDK获取摄像机当前时间（替换为实际SDK调用）
def get_camera_ntp_time():
    return datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]

# 定时更新overlay文本
def update_overlay():
    while True:
        camera_time = get_camera_ntp_time()
        overlay.set_property("text", f"Camera NTP Time: {camera_time}")
        threading.Event().wait(0.1)

threading.Thread(target=update_overlay, daemon=True).start()

# 处理帧显示
def on_new_sample(sink):
    sample = sink.emit("pull-sample")
    buffer = sample.get_buffer()
    caps = sample.get_caps()
    
    # 转换为OpenCV帧
    from gi.repository import GstVideo
    frame_info = GstVideo.VideoFrameInfo.new(caps)
    frame = GstVideo.VideoFrame.new(frame_info)
    frame.map(buffer, Gst.MapFlags.READ)
    cv_frame = frame.to_ndarray()[:, :, ::-1]
    frame.unmap()
    
    cv2.imshow("RTSP Stream", cv_frame)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        pipeline.set_state(Gst.State.NULL)
        cv2.destroyAllWindows()
        exit()
    return Gst.FlowReturn.OK

sink.connect("new-sample", on_new_sample)
pipeline.set_state(Gst.State.PLAYING)

loop = Gst.MainLoop()
loop.run()

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