4、Python基础Pipeline：使用Gst.parse_launch()、创建自定义Pipeline、事件循环与Bus监听

好，咱们进入实战环节了。

前面几章聊了不少理论，我知道你们可能有点手痒。这一章我们就用Python来写真正的GStreamer代码。我个人觉得，Python是学习GStreamer最好的语言——语法干净，调试方便，出错了也不容易把系统搞崩。

这一章我们会搞定三件事：用一行字符串搭Pipeline、手写自定义Pipeline、以及监听Bus消息。这三板斧学会了，你就能应付80%的日常开发场景。

4.1 快速上手：Gst.parse_launch()

先来个最简单的。

GStreamer提供了一种"懒人写法"——Gst.parse_launch()。你只需要把Pipeline描述成一段字符串，它就能自动帮你把元件创建好、连接好。我刚开始接触GStreamer时，就是靠这个函数快速跑通的第一个Demo。

看个例子：

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

# 初始化
Gst.init(None)

# 用字符串搭建Pipeline
pipeline_str = "videotestsrc ! videoconvert ! autovideosink"
pipeline = Gst.parse_launch(pipeline_str)

# 启动
pipeline.set_state(Gst.State.PLAYING)

# 跑10秒后退出
loop = GLib.MainLoop()
GLib.timeout_add_seconds(10, loop.quit)
loop.run()

# 清理
pipeline.set_state(Gst.State.NULL)

这段代码干了什么？

• videotestsrc：生成测试视频信号（就是那个彩色条纹图案）
• videoconvert：做色彩空间转换，保证前后格式匹配
• autovideosink：自动选择一个能显示视频的窗口

说白了，就是"生成画面→转格式→显示出来"这么个流程。

我曾经在一个项目里偷懒，用parse_launch()搭了一个20多个元件的Pipeline。结果运行时某个元件参数传错了，报错信息根本看不懂是哪个环节出了问题。后来我老老实实拆成手写，逐个元件排查，十分钟就找到了问题。

4.2 手写自定义Pipeline

字符串写法虽然方便，但不够灵活。比如你想动态插入分支、条件判断、或者给某个元件设置回调，那就得手写。

手写Pipeline的步骤其实很固定：

1. 创建元件
2. 设置属性
3. 添加到Pipeline
4. 连接元件
5. 启动

来，看代码：

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

Gst.init(None)

# 1. 创建Pipeline
pipeline = Gst.Pipeline.new("my-pipeline")

# 2. 创建元件
source = Gst.ElementFactory.make("videotestsrc", "source")
sink = Gst.ElementFactory.make("autovideosink", "sink")
convert = Gst.ElementFactory.make("videoconvert", "converter")

if not source or not sink or not convert:
    print("创建元件失败，请检查GStreamer安装")
    exit(1)

# 3. 设置属性（可选）
source.set_property("pattern", 1)  # 1 = 彩色条纹

# 4. 添加到Pipeline
pipeline.add(source)
pipeline.add(convert)
pipeline.add(sink)

# 5. 连接元件
source.link(convert)
convert.link(sink)

# 6. 启动
pipeline.set_state(Gst.State.PLAYING)

# 事件循环
loop = GLib.MainLoop()
GLib.timeout_add_seconds(10, loop.quit)
loop.run()

pipeline.set_state(Gst.State.NULL)

你可能会问：为什么中间要加个videoconvert？

嗯，这个问题问得好。GStreamer的元件之间传输数据，要求格式必须匹配。videotestsrc输出的是原始视频帧，而autovideosink可能期望某种特定格式。不加转换元件，连接时会报错。我刚开始学的时候就被这个坑过——明明代码看起来没问题，就是跑不起来。

4.3 事件循环与Bus监听

Pipeline跑起来了，然后呢？

你需要知道它运行得怎么样——有没有报错？有没有走到结尾？这时候就需要Bus监听了。

GStreamer的Bus机制，说白了就是一个消息队列。Pipeline里的所有元件都会往Bus上发消息，你只需要注册一个回调函数，就能收到这些消息。

常见的消息类型：

消息类型	含义	常见场景
GST_MESSAGE_ERROR	发生错误	文件找不到、解码失败
GST_MESSAGE_EOS	播放结束	文件播放完毕、录制完成
GST_MESSAGE_STATE_CHANGED	状态变化	从PLAYING切换到PAUSED
GST_MESSAGE_WARNING	警告信息	丢帧、延迟过高

来看一个带Bus监听的完整例子：

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

Gst.init(None)

# 创建Pipeline
pipeline = Gst.Pipeline.new("bus-demo")
source = Gst.ElementFactory.make("videotestsrc", "source")
sink = Gst.ElementFactory.make("autovideosink", "sink")
convert = Gst.ElementFactory.make("videoconvert", "converter")

pipeline.add(source)
pipeline.add(convert)
pipeline.add(sink)
source.link(convert)
convert.link(sink)

# 获取Bus
bus = pipeline.get_bus()
bus.add_signal_watch()

# 注册消息回调
def on_message(bus, message):
    msg_type = message.type
    if msg_type == Gst.MessageType.EOS:
        print("播放结束")
        loop.quit()
    elif msg_type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print(f"错误: {err}, 详情: {debug}")
        loop.quit()
    elif msg_type == Gst.MessageType.WARNING:
        warn, debug = message.parse_warning()
        print(f"警告: {warn}")
    elif msg_type == Gst.MessageType.STATE_CHANGED:
        old, new, pending = message.parse_state_changed()
        print(f"状态变化: {old} -> {new}")

bus.connect("message", on_message)

# 启动
pipeline.set_state(Gst.State.PLAYING)

# 事件循环
loop = GLib.MainLoop()
try:
    loop.run()
except KeyboardInterrupt:
    print("用户中断")
finally:
    pipeline.set_state(Gst.State.NULL)

这段代码里，我做了几件事：

• 用bus.add_signal_watch()开启消息监听
• 用bus.connect("message", on_message)注册回调
• 在回调里根据消息类型做不同处理

我个人习惯把Bus监听放在一个单独的线程里，这样主线程可以干别的事。不过对于初学者，用GLib的主循环就够了，简单直接。

我曾经在一个录制项目中，因为没监听EOS消息，导致录制结束后程序一直卡着不退出。排查了半天才发现——文件录完了，但Pipeline还在等下一个数据。加上EOS监听后，一切正常。

4.4 避坑指南

最后分享几个我踩过的坑：

• 别忘了初始化：Gst.init(None)必须写在最前面，否则所有API调用都会崩溃
• 元件创建失败要检查：make()返回None时，说明这个元件不存在（比如拼写错误或缺少插件）
• link()失败要排查：两个元件连接不上，通常是格式不匹配。可以用Gst.debug_bin_to_dot_file()导出Pipeline拓扑图来分析
• 事件循环别忘启动：loop.run()是阻塞的，记得放在最后

好了，这一章的内容就到这里。下一章我们会深入Pipeline的状态管理——为什么有NULL、READY、PAUSED、PLAYING四种状态？它们之间怎么切换？到时候见。