如何在GStreamer-Python中获取流媒体源的状态？

我完全懂你碰到的痛点——IP摄像头的RTSP流经常莫名其妙中断，要么RTSP连接直接丢了，要么屏幕全黑但摄像头IP还能通，可现有的Bus消息根本没给出你需要的异常信号，没法及时触发重连对吧？结合你给出的代码，我给你几个实用的解决方案：

1. 监控rtspsrc的专属消息与连接状态属性 #

rtspsrc作为RTSP流的核心源元素，会发出很多**Gst.MessageType.ELEMENT类型的自定义消息**，这些消息里藏着RTSP连接的关键状态（比如会话超时、连接丢失），但你之前的代码没处理这类消息。同时它还有个直接反映连接状态的属性connection-state，可以定期检查。

修改Bus消息处理逻辑 #

在你现有的消息解析循环里，加入对ELEMENT类型消息的处理：

elif message.type == Gst.MessageType.ELEMENT:
    # 解析RTSP相关的元素消息
    struct = message.get_structure()
    if struct is not None and struct.has_name("GstRTSPMessage"):
        # 提取RTSP消息的状态信息，比如连接断开、会话终止
        rtsp_type = struct.get_value("type")
        if rtsp_type == "GST_RTSP_MESSAGE_TYPE_TEARDOWN" or rtsp_type == "GST_RTSP_MESSAGE_TYPE_BYE":
            print("RTSP stream connection was terminated!")
            # 这里触发你的重连逻辑

定期检查connection-state属性 #

在主循环里，定期获取rtspsrc的连接状态：

import time

while True:
    # 检查rtspsrc的连接状态
    conn_state = source.get_property("connection-state")
    print(f"Current RTSP connection state: {conn_state}")
    if conn_state == "disconnected":
        print("RTSP connection lost, initiating reconnect...")
        # 触发重连逻辑

    # 原有的帧显示和消息处理逻辑
    if image_arr is not None:
        cv2.imshow("Receive Image from Pipeline Buffer", image_arr)
        if cv2.waitKey(1) == ord('q'):
            break

    message = bus.timed_pop_filtered(10000, Gst.MessageType.ANY)
    # ... 原有的消息处理代码 ...

2. 监控视频帧的流动（解决黑屏无帧的情况） #

有时候RTSP连接还在，但摄像头已经停止输出有效帧（表现为黑屏），这时候Bus不会发错误消息，但你的new_buffer回调会停止被调用。可以加个计时器来检测这种情况：

新增帧超时检测逻辑 #

1. 先定义全局变量记录最后收到帧的时间：

last_frame_time = time.time()

2. 在new_buffer回调里更新这个时间：

def new_buffer(sink, data):
    global image_arr, last_frame_time
    sample = sink.emit("pull-sample")
    arr = gst_to_opencv(sample)
    image_arr = arr
    last_frame_time = time.time()  # 更新最后收到帧的时间
    return Gst.FlowReturn.OK

3. 在主循环里检查超时：

while True:
    # 检查帧超时（比如超过5秒没收到新帧）
    if time.time() - last_frame_time > 5:
        print("No video frames received for 5 seconds! Stream is likely dead.")
        # 触发重连逻辑

    # 原有的帧显示、消息处理逻辑...

3. 利用Pad Probe监控数据流 #

还可以给管道中的关键Pad（比如解码器的输出Pad）加一个探针（Probe），实时监控有没有数据流过，这比监控回调更底层可靠：

添加Pad Probe #

def pad_probe_callback(pad, info, user_data):
    # 更新最后收到数据的时间
    user_data["last_data_time"] = time.time()
    return Gst.PadProbeReturn.OK

# 在创建解码器后，给它的src pad加探针
decoder_src_pad = decoder.get_static_pad("src")
probe_data = {"last_data_time": time.time()}
decoder_src_pad.add_probe(Gst.PadProbeType.BUFFER, pad_probe_callback, probe_data)

然后在主循环里检查probe_data["last_data_time"]的超时情况即可，逻辑和帧超时检测类似。

4. 重连逻辑的实现 #

不管用哪种检测方式，当判定流异常后，你需要重启管道：

1. 先将现有管道设置为NULL状态，释放资源：

pipeline.set_state(Gst.State.NULL)

2. 重新调用你的管道创建逻辑（可以把创建管道的代码封装成一个独立函数，比如create_rtsp_pipeline()），重新初始化元素、设置属性、链接并启动。

备注：内容来源于stack exchange，提问作者doruk.sonmez