Gstreamer教程 6

这一节说的是打印过程中的pad属性。

python代码如下：

import gi

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

# 初始化 GStreamer
Gst.init(None)


def print_field(field, value, pfx):
    str_value = Gst.value_serialize(value)
    print(f"{pfx}  {field}: {str_value}")


def print_caps(caps, pfx):
    if caps.is_any():
        print(f"{pfx}ANY")
        return
    if caps.is_empty():
        print(f"{pfx}EMPTY")
        return
    for i in range(caps.get_size()):
        structure = caps.get_structure(i)
        print(f"{pfx}{structure.get_name()}")
        for j in range(structure.n_fields()):
            field_name = structure.nth_field_name(j)
            print_field(field_name, structure.get_value(field_name), pfx)


def print_pad_templates_information(factory):
    print(f"Pad Templates for {factory.get_metadata('long-name')}:")
    if factory.get_num_pad_templates() == 0:
        print("  none")
        return

    pads = factory.get_static_pad_templates()
    for pad in pads:
        # print(dir(pad))，这里可以打印下他有那些属性
        padtemplate = pad

        if padtemplate.direction == Gst.PadDirection.SRC:
            print(f"  SRC template: '{padtemplate.name_template}'")
        elif padtemplate.direction == Gst.PadDirection.SINK:
            print(f"  SINK template: '{padtemplate.name_template}'")
        else:
            print(f"  UNKNOWN!!! template: '{padtemplate.name_template}'")

        if padtemplate.presence == Gst.PadPresence.ALWAYS:
            print("    Availability: Always")
        elif padtemplate.presence == Gst.PadPresence.SOMETIMES:
            print("    Availability: Sometimes")
        elif padtemplate.presence == Gst.PadPresence.REQUEST:
            print("    Availability: On request")
        else:
            print("    Availability: UNKNOWN!!!")

        if padtemplate.static_caps.string:
            caps = Gst.Caps.from_string(padtemplate.static_caps.string)
            print("    Capabilities:")
            print_caps(caps, "      ")


def print_pad_capabilities(element, pad_name):
    pad = element.get_static_pad(pad_name)
    if not pad:
        print(f"Could not retrieve pad '{pad_name}'")
        return

    caps = pad.get_current_caps()
    if not caps:
        caps = pad.query_caps(None)

    print(f"Caps for the {pad_name} pad:")
    print_caps(caps, "      ")


def main():
    # 创建元素工厂
    source_factory = Gst.ElementFactory.find("videotestsrc")
    sink_factory = Gst.ElementFactory.find("autovideosink")

    if not source_factory or not sink_factory:
        print("Not all element factories could be created.")
        exit(-1)

    # 打印这些工厂的 Pad 模板信息
    print_pad_templates_information(source_factory)
    print_pad_templates_information(sink_factory)

    # 要求工厂实例化实际元素
    source = source_factory.create("source")
    sink = sink_factory.create("sink")

    # 创建空管道
    pipeline = Gst.Pipeline.new("test-pipeline")

    if not pipeline or not source or not sink:
        print("Not all elements could be created.")
        exit(-1)

    # 构建管道
    pipeline.add(source)
    pipeline.add(sink)

    if not source.link(sink):
        print("Elements could not be linked.")
        pipeline.unref()
        exit(-1)

    # 打印初始协商能力（在 NULL 状态下）
    print("In NULL state:")
    print_pad_capabilities(sink, "sink")

    # 开始播放
    ret = pipeline.set_state(Gst.State.PLAYING)

    if ret == Gst.StateChangeReturn.FAILURE:
        print("Unable to set the pipeline to the playing state (check the bus for error messages).")

        # 等待错误、EOS 或状态变化
    bus = pipeline.get_bus()
    terminate = False

    while not terminate:
        msg = bus.timed_pop_filtered(Gst.CLOCK_TIME_NONE,
                                     Gst.MessageType.ERROR | Gst.MessageType.EOS | Gst.MessageType.STATE_CHANGED)

        # 解析消息
        if msg is not None:
            if msg.type == Gst.MessageType.ERROR:
                err, debug_info = msg.parse_error()
                print(f"Error received from element {msg.src.get_name()}: {err.message}")
                print(f"Debugging information: {debug_info or 'none'}")
                terminate = True
            elif msg.type == Gst.MessageType.EOS:
                print("End-Of-Stream reached.")
                terminate = True
            elif msg.type == Gst.MessageType.STATE_CHANGED:
                # 我们只对来自管道的状态变更消息感兴趣
                if msg.src == pipeline:
                    old_state, new_state, pending_state = msg.parse_state_changed()
                    print(
                        f"\nPipeline state changed from {Gst.Element.state_get_name(old_state)} to {Gst.Element.state_get_name(new_state)}:")
                    # 打印 sink 元素的当前能力
                    print_pad_capabilities(sink, "sink")

    pipeline.set_state(Gst.State.NULL)
    pipeline.unref()


if __name__ == "__main__":
    main()