使用input-selector时GStreamer出现Bus error问题求助

首先得明确：input-selector要求所有输入Pad的媒体格式（Caps）必须完全匹配，这是你遇到无描述Bus error的最可能原因。你的示例中，一个输入是H264编码流，另一个是PNG解码后的Raw视频流，Caps完全不兼容，当消费端（IINA/VLC）请求流时，GStreamer的Caps协商失败就会触发总线错误。

一、先搞懂input-selector的关键规则 #

• 所有输入Pad必须协商出相同的Caps，否则切换或启动时会触发格式不兼容错误
• 切换Pad时，建议先确保目标Pad已经处于激活/链接状态（比如对应的分支已经成功启动）
• 可以通过capsfilter强制统一所有输入的格式，避免协商失败

二、修正后的gst-launch命令行示例 #

下面的示例将两个输入都统一为H264编码的流，确保Caps匹配：

gst-launch-1.0 --gst-debug-level=3 \
input-selector name=iselect \
filesrc location={INPUT_VIDEO_FILE_PATH} ! qtdemux ! h264parse ! capsfilter caps="video/x-h264,stream-format=avc,alignment=au" ! iselect.sink_0 \
filesrc location={PLACEHOLDER_IMAGE_PATH} ! pngdec ! imagefreeze ! x264enc tune=zerolatency ! capsfilter caps="video/x-h264,stream-format=avc,alignment=au" ! iselect.sink_1 \
iselect. ! gdppay ! shmsink socket-path=/tmp/shm1

这里用capsfilter强制两个输入的H264格式参数一致，同时把静态图片分支编码成H264，和视频分支格式对齐。

三、Python手动构建的正确示例 #

结合你提供的代码，修正Caps匹配问题，并添加错误捕获逻辑：

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

Gst.init(None)

def bus_call(bus, message, loop):
    t = message.type
    if t == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print(f"Error: {err}, Debug: {debug}")
        loop.quit()
    elif t == Gst.MessageType.EOS:
        print("End-of-stream")
        loop.quit()
    return True

def create_selector_pipeline():
    pipeline = Gst.Pipeline.new("input-selector-pipeline")
    
    # 创建input-selector
    iselect = Gst.ElementFactory.make("input-selector", "iselect")
    if not iselect:
        print("Failed to create input-selector")
        return None
    
    # 分支1：视频文件流，统一Caps为H264
    filesrc1 = Gst.ElementFactory.make("filesrc", "filesrc1")
    filesrc1.set_property("location", "input_video.mp4")
    qtdemux = Gst.ElementFactory.make("qtdemux", "qtdemux")
    h264parse1 = Gst.ElementFactory.make("h264parse", "h264parse1")
    capsfilter1 = Gst.ElementFactory.make("capsfilter", "capsfilter1")
    caps1 = Gst.Caps.from_string("video/x-h264,stream-format=avc,alignment=au")
    capsfilter1.set_property("caps", caps1)
    
    # 分支2：静态图片流，编码为H264，统一Caps
    filesrc2 = Gst.ElementFactory.make("filesrc", "filesrc2")
    filesrc2.set_property("location", "placeholder.png")
    pngdec = Gst.ElementFactory.make("pngdec", "pngdec")
    imagefreeze = Gst.ElementFactory.make("imagefreeze", "imagefreeze")
    x264enc = Gst.ElementFactory.make("x264enc", "x264enc")
    x264enc.set_property("tune", "zerolatency")
    h264parse2 = Gst.ElementFactory.make("h264parse", "h264parse2")
    capsfilter2 = Gst.ElementFactory.make("capsfilter", "capsfilter2")
    capsfilter2.set_property("caps", caps1)  # 和分支1用相同的Caps
    
    # 输出分支
    gdppay = Gst.ElementFactory.make("gdppay", "gdppay")
    shmsink = Gst.ElementFactory.make("shmsink", "shmsink")
    shmsink.set_property("socket-path", "/tmp/shm1")
    shmsink.set_property("wait-for-connection", False)
    
    # 添加所有元素到流水线
    elements = [iselect, filesrc1, qtdemux, h264parse1, capsfilter1,
                filesrc2, pngdec, imagefreeze, x264enc, h264parse2, capsfilter2,
                gdppay, shmsink]
    for elem in elements:
        pipeline.add(elem)
    
    # 链接分支1：注意qtdemux需要动态链接pad
    def link_qtdemux_pad(src_pad, data):
        if src_pad.get_name().startswith("video_"):
            src_pad.link(h264parse1.get_static_pad("sink"))
        return True
    qtdemux.connect("pad-added", link_qtdemux_pad)
    Gst.Element.link_many(filesrc1, qtdemux)
    Gst.Element.link_many(h264parse1, capsfilter1, iselect.get_static_pad("sink_0"))
    
    # 链接分支2
    Gst.Element.link_many(filesrc2, pngdec, imagefreeze, x264enc, h264parse2, capsfilter2, iselect.get_static_pad("sink_1"))
    
    # 链接输出
    Gst.Element.link_many(iselect, gdppay, shmsink)
    
    return pipeline

if __name__ == "__main__":
    pipeline = create_selector_pipeline()
    if not pipeline:
        exit(-1)
    
    # 设置Bus监听，捕获详细错误
    loop = GLib.MainLoop()
    bus = pipeline.get_bus()
    bus.add_signal_watch()
    bus.connect("message", bus_call, loop)
    
    # 启动流水线
    pipeline.set_state(Gst.State.PLAYING)
    
    # 示例：切换到视频分支（sink_0）
    iselect = pipeline.get_by_name("iselect")
    target_pad = iselect.get_static_pad("sink_0")
    iselect.set_property("active-pad", target_pad)
    print(f"Switched to active pad: {target_pad.get_name()}")
    
    try:
        loop.run()
    except KeyboardInterrupt:
        pass
    
    # 清理
    pipeline.set_state(Gst.State.NULL)

四、进阶：实现带HTTP API的RTSP输出服务 #

针对你的最终目标，下面是一个结合GstRtspServer和Flask的示例，通过HTTP请求切换输入源：

import gi
gi.require_version('Gst', '1.0')
gi.require_version('GstRtspServer', '1.0')
from gi.repository import Gst, GstRtspServer, GLib
from flask import Flask, request

Gst.init(None)
app = Flask(__name__)
active_pad_name = "sink_0"  # 默认激活视频分支

class RTSPMediaFactory(GstRtspServer.RTSPMediaFactory):
    def __init__(self):
        super().__init__()
        self.iselect = None
    
    def do_create_element(self, url):
        # 构建包含input-selector的流水线
        pipeline_str = """
            input-selector name=iselect
            filesrc location=input_video.mp4 ! qtdemux ! h264parse ! capsfilter caps="video/x-h264,stream-format=avc,alignment=au" ! iselect.sink_0
            filesrc location=placeholder.png ! pngdec ! imagefreeze ! x264enc tune=zerolatency ! h264parse ! capsfilter caps="video/x-h264,stream-format=avc,alignment=au" ! iselect.sink_1
            iselect. ! rtph264pay name=pay0 pt=96
        """
        pipeline = Gst.parse_launch(pipeline_str)
        self.iselect = pipeline.get_by_name("iselect")
        # 设置初始激活pad
        self.switch_pad(active_pad_name)
        return pipeline
    
    def switch_pad(self, pad_name):
        if self.iselect:
            target_pad = self.iselect.get_static_pad(pad_name)
            if target_pad:
                self.iselect.set_property("active-pad", target_pad)
                global active_pad_name
                active_pad_name = pad_name
                print(f"Switched to pad: {pad_name}")

class RTSPServer(GstRtspServer.RTSPServer):
    def __init__(self):
        super().__init__()
        self.factory = RTSPMediaFactory()
        self.factory.set_shared(True)
        mount_points = self.get_mount_points()
        mount_points.add_factory("/stream", self.factory)
        self.attach(None)

@app.route('/switch-source', methods=['POST'])
def switch_source():
    data = request.json
    pad_name = data.get('pad_name', 'sink_0')
    if pad_name in ["sink_0", "sink_1"]:
        server.factory.switch_pad(pad_name)
        return {"status": "success", "active_pad": pad_name}
    else:
        return {"status": "error", "message": "Invalid pad name. Use sink_0 or sink_1"}

if __name__ == '__main__':
    server = RTSPServer()
    loop = GLib.MainLoop()
    
    # 启动Flask线程
    import threading
    flask_thread = threading.Thread(target=app.run, kwargs={'host': '0.0.0.0', 'port': 5000})
    flask_thread.daemon = True
    flask_thread.start()
    
    print("RTSP server running at rtsp://localhost:8554/stream")
    print("HTTP API: POST http://localhost:5000/switch-source with {\"pad_name\": \"sink_0\"} or {\"pad_name\": \"sink_1\"}")
    
    try:
        loop.run()
    except KeyboardInterrupt:
        pass

五、调试技巧 #

• 启动时添加--gst-debug-level=4或更高，查看Caps协商的详细日志，定位格式不匹配的问题
• 用gst-inspect-1.0 input-selector查看官方文档，确认输入输出的Caps要求
• 始终通过Bus的ERROR消息捕获详细错误信息，不要只依赖无描述的Bus error提示

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